kis_bot/kis_trader/engine/tail_engine.py

#!/usr/bin/env python3
"""
kis_trader/engine/tail_engine.py — 꼬리잡기 백테스트·실매매 공통 엔진
====================================================
백테스트(backtest_web), 파라미터 탐색(tail_param_search), 실매매(kis_trader TailCatchStrategy)가
모두 동일한 진입/청산 계산식과 '고급 방어 로직'을 쓰도록 통합된 단일 소스 엔진.

■ SCAN vs TRIGGER (돌파 전략과 동일 원칙)
  [SCAN — HTS 조건검색, 널넬하게]
    **실매 권장: 키움 ``tail`` (``SHORT_UNIVERSE_SOURCE=kiwoom_condition``)**
      A) [일] 시가→종가 -10% ~ -0.5% (당일 약세)
      B) 체결강도 85% ~ 400%
      C) 3봉전 대비 거래량 180% ~ 2000%
      F) [일] 저가 대비 종가 +1% ~ +8% (꼬리 회복 구간)
      → ``KiwoomConditionSearchManager`` + ``CONDITION_SHORT_NAME=tail``
    **레거시 KIS ``condition`` (REST 폴링)**
      A) 1봉 등락률 -10% ~ -0.5%  (직전봉 종가 대비)
      B/C 동일 축 — ``ConditionSearchManager`` / ``tail``
    → ``target_candidates_history`` (strategy_id=SHORT) 에 스냅샷 저장.

  [TRIGGER — 본 엔진, 엄격하게 — HTS SCAN 통과 후보에 타점·리스크 재검사]
    반전 패턴 OR(망치·핀바·장악·관통·하라미·도지·샛별) + 당일 회복률·3분봉 회복 위치,
    (선택) 신호봉 거래량 폭증(``TAIL_VOL_MULT``×N봉평균, 0=OFF) + RSI·MA20,
    고점추격·피뢰침, 시간대/쿨다운. 패턴별 ``TAIL_PATTERN_*`` env 로 ON/OFF.
    ``TAIL_SKIP_HTS_SCAN_DUPES=false`` (운영 기본) 이면 TRIGGER 에서
    **3분봉 직전대비 등락**(``TAIL_BAR_CHG_MIN/MAX_PCT``) 을 추가 검사.
    (HTS 일봉 A·1분 G 와 **축이 다름** — 라벨을 HTS A 로 부르지 말 것.
     true 이면 이 3분 등락 재검사만 생략. 시가→저점 낙폭은 ``TAIL_USE_INTRADAY_DROP``.)

■ 엔진 공통 로직 (백테·실매 동일)
  매도 우선순위 (V4):
    1) 트레일(어깨컷) — max_price 갱신 후 되돌림(저가로 터치 판정, 체결=매도선), 발동 수익% 충족 시 **손절·금액손실보다 우선**
    2) ATR 캡 적용 익절 / 3) ATR 캡 적용 손절 / 4) 금액손실컷(트레일 미발동 구간만) / 5) 장마감
  ATR 목표·손절: 배수 × ATR 후 ``TAIL_ATR_*_MIN/MAX_PCT`` % 상·하한 캡 (잡주 과대 목표가 방지).
  백테 청산: ws_ticks 시간순 (``resolve_backtest_sell``, session_low).
  **실매와 동일** — 고점(max_price)·청산 판정은 틱만. 봉 OHLC high/low 로 max 선반영·폴백 금지
  (``TAIL_BACKTEST_TICK_FALLBACK_OHLC`` 기본 False, EOD 도 OHLC 폴백 강제 금지 → 틱 미청산 시 장마감 플랫만).
  진입·지표는 N분봉 그대로.

■ 진입 모드 (``TAIL_ENTRY_MODE``)
  - ``align``: 신호봉 확정 → **직후 봉**(signal+tf, 1M→3M 합성 포함) 시가·첫 틱 시장가
    (구멍으로 먼 다음 3분 시가에 밀리지 않음)
  - ``limit_atr``(기본): 신호봉 확정 → anchor−ATR×mult 지정가 → ``TAIL_LIMIT_VALID_BARS`` 봉 내
    low 터치 시 체결, 미체결 시 다음 봉부터 취소(실매) / 백테 스킵
"""

from datetime import datetime
from typing import List, Dict, Any, Optional, Set, Tuple

from kis_trader.engine.limit_entry_common import (
    compute_atr_limit_price,
    is_limit_atr_entry,
    limit_valid_until_bar_key,
    resolve_limit_anchor_price,
    short_entry_mode,
    tail_limit_params,
    try_limit_fill_on_bar,
)
from kis_trader.engine.indicator_cache import (
    attach_indicator_caches_to_params,
    get_indicator_cache_from_params,
)
from kis_trader.engine.whipsaw_filter import inject_whipsaw_ticks_into_params, whipsaw_reject_for_signal
from kis_trader.backtest.trigger_snapshot_loader import inject_trigger_snapshots_into_params
from kis_trader.engine.orderbook_filter import orderbook_reject_for_entry
from kis_trader.engine.program_filter import program_reject_for_entry
from kis_trader.engine.tick_exit_common import (
    backtest_sell_slip_pct,
    backtest_tick_poll_ms,
    collect_minute_ticks,
    resolve_backtest_sell,
)
from kis_trader.engine.tail_tick_replay import (
    align_entry_execute_time,
    align_entry_price_from_ticks,
    collect_bar_ticks,
    resolve_align_entry_bar,
    tail_backtest_tick_fallback_ohlc,
    tail_backtest_use_tick_exit,
    tail_backtest_wants_tick_replay,
    tail_timeframe_min,
    try_limit_fill_on_bar_with_ticks,
)
from kis_trader.engine.atr_series import compute_atr_series
from kis_trader.engine.strategy_eod import (
    is_strategy_eod_bar,
    parse_eod_hm,
    resolve_strategy_eod_params,
)
from kis_trader.engine.tail_env_keys import (
    tail_env_bool,
    tail_env_float,
    tail_env_int,
)


def _to_bool(v: Any, default: bool = True) -> bool:
    if v is None:
        return default
    if isinstance(v, bool):
        return v
    s = str(v).strip().lower()
    if s in ("1", "true", "t", "y", "yes", "on"):
        return True
    if s in ("0", "false", "f", "n", "no", "off", ""):
        return False
    return default


def _confirmed_candles_only(candles: List[Dict]) -> List[Dict]:
    """확정봉만 사용 (미확정 봉 제외). 없으면 원본 유지."""
    confirmed = [c for c in candles if _to_bool(c.get("is_confirmed", 1), True)]
    return confirmed if confirmed else list(candles)


def resolve_tail_skip_hts_scan_dupes(r: Optional[Dict[str, Any]] = None) -> bool:
    """
    TRIGGER 의 **3분봉 직전대비 등락**(BAR_CHG) 재검사 생략 여부.

    이름에 HTS 가 들어가지만, 끄는 대상은 HTS 일봉 A/1분 G 가 아니라
    ``TAIL_BAR_CHG_*`` (신호 3분봉 vs 직전 3분봉) 이다.

    - ``TAIL_SKIP_HTS_SCAN_DUPES`` 명시(true/false) → 그대로
    - 미설정 → ``SHORT_UNIVERSE_SOURCE`` 가 condition/kiwoom_condition 이면 True
    """
    if r is None:
        try:
            from kis_trader.utils.env import get_strategy_env_dict
            r = get_strategy_env_dict("TAIL") or {}
        except Exception:
            r = {}
    raw = r.get("TAIL_SKIP_HTS_SCAN_DUPES")
    if raw is not None and str(raw).strip() != "":
        return tail_env_bool(r, "TAIL_SKIP_HTS_SCAN_DUPES", True)
    # 엔진 defaults 에 이미 해석된 bool 이 있으면 universe fallback 금지
    if "skip_hts_scan_dupes" in r and r.get("skip_hts_scan_dupes") is not None:
        return _to_bool(r.get("skip_hts_scan_dupes"), False)
    src = str(r.get("SHORT_UNIVERSE_SOURCE") or "kiwoom_condition").strip().lower()
    return src in ("kiwoom_condition", "condition")


def _tail_symbol_reentry_gate(
    daily_cnt: int,
    daily_pnl_krw: float,
    slot_money: float,
    params: Dict[str, Any],
) -> Tuple[bool, str]:
    """종목당 당일 재진입 게이트 — 무지성 횟수 제한 대신 손실예산·최소엣지로 판정.

    ``TAIL_MAX_DAILY`` (기본 3, 안전판)는 그대로 두되, 실질적인 과매매 통제는
    이 게이트가 담당한다:
      - 오늘 그 종목 **첫 진입**(daily_cnt==0)은 항상 통과.
      - **재진입**(daily_cnt>=1)부터는:
        1) 종목 당일 실현손익이 손실한도(KRW·PCT 중 먼저 닿는 쪽=더 작은 금액) 이하면 차단.
           → "3번까지만" 대신 "-3만원까지만" 처럼 손실 크기로 제어.
        2) ``TAIL_REENTRY_REQUIRE_NONNEG``/``TAIL_REENTRY_MIN_EDGE_KRW`` 미충족이면 차단.
           → 수수료·세금(왕복 약 0.21%)도 못 건진 재진입을 걸러 회전매매 방지.
    반환: (allowed, reject_detail) — allowed=False 면 detail 에 사유 메시지.
    """
    if daily_cnt <= 0:
        return True, ""
    loss_limit_krw = abs(float(params.get("symbol_daily_loss_limit_krw", 0) or 0))
    loss_limit_pct = abs(float(params.get("symbol_daily_loss_limit_pct", 0) or 0))
    candidate_limits: List[float] = []
    if loss_limit_krw > 0:
        candidate_limits.append(loss_limit_krw)
    if loss_limit_pct > 0 and slot_money > 0:
        candidate_limits.append(slot_money * loss_limit_pct / 100.0)
    if candidate_limits:
        active_limit = min(candidate_limits)
        if daily_pnl_krw <= -active_limit:
            return False, f"당일 실현 {daily_pnl_krw:+,.0f}원 ≤ -{active_limit:,.0f}원"
    require_nonneg = _to_bool(params.get("reentry_require_nonneg"), False)
    if require_nonneg and daily_pnl_krw < 0:
        return False, f"당일 실현 {daily_pnl_krw:+,.0f}원 < 0 (재진입 최소수익 미달)"
    min_edge = float(params.get("reentry_min_edge_krw", 0) or 0)
    if min_edge > 0 and daily_pnl_krw < min_edge:
        return False, f"당일 실현 {daily_pnl_krw:+,.0f}원 < {min_edge:,.0f}원 (재진입 최소엣지 미달)"
    return True, ""


def resolve_tail_cand_limit(params: Optional[Dict[str, Any]] = None) -> int:
    """
    ``SHORT_CAND_LIMIT`` — 실매 ``BaseStrategy._post_filter_candidates`` 의 후보 하드캡과
    동일한 값. 실매는 유니버스 편입 순서 상위 N개만 매수 체크 대상으로 본다(0=무제한).
    백테도 그 시각 유니버스에서 동일하게 상위 N개만 남겨야 실매와 정합된다.

    ``params["cand_limit"]`` 명시 시 그대로(파라서치 그리드용) → 없으면 DB
    ``SHORT_CAND_LIMIT`` 값을 그대로 읽는다.
    """
    if params is not None and params.get("cand_limit") is not None:
        try:
            return max(0, int(float(params.get("cand_limit"))))
        except (TypeError, ValueError):
            pass
    from kis_trader.utils.env import get_env_int
    return max(0, get_env_int("SHORT_CAND_LIMIT", 0))


def _slot_key(candle_time: str, scan_interval_min: int = 1) -> str:
    """봉 시각을 N분 단위 슬롯 키로 변환.

    신봇 기준: caller 가 ``TradeDBExt.get_universe_by_candle_time()`` 으로
    1분 캔들 시각 키 dict 를 만들어 주입하므로 기본값은 1분(= passthrough).
    ``--fallback-universe`` 시뮬레이션은 caller 가 ``scan_interval_min=5`` 를
    명시해 5분 버킷팅으로 사용.
    """
    date = candle_time[:8]
    hm = int(candle_time[8:12])
    total_min = (hm // 100) * 60 + (hm % 100)
    slot_min = (total_min // scan_interval_min) * scan_interval_min
    slot_hm = (slot_min // 60) * 100 + (slot_min % 60)
    return date + str(slot_hm).zfill(4)


def get_tail_defaults_from_db(db=None, *, env_row: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    env_config 최신 행에서 꼬리잡기 관련 값과 고급 방어 로직 값을 전부 로드.
    백테스트·파라미터서치·실매매가 동일 DB 값을 쓰도록 단일 소스.
    env_row 가 있으면 DB 조회 생략 (웹 백테 env 타임라인용).
    """
    own_db = None
    r: Dict[str, Any] = {}
    try:
        if env_row is not None:
            r = dict(env_row)
        else:
            # SCALP 와 동일 — RAM TTL(get_strategy_env_dict). 매 호출 full snapshot 금지.
            try:
                from kis_trader.utils.env import get_strategy_env_dict
                r = get_strategy_env_dict("SHORT") or {}
            except Exception:
                r = {}
            if not r:
                if db is None:
                    from database import TradeDB
                    own_db = TradeDB()
                    db = own_db
                if hasattr(db, "get_merged_env_snapshot"):
                    r = db.get_merged_env_snapshot()
                elif hasattr(db, "get_latest_env"):
                    latest = db.get_latest_env()
                    r = dict((latest or {}).get("snapshot") or {})
                else:
                    row = db.conn.execute(
                        "SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
                    ).fetchone()
                    r = dict(row) if row else {}
        if r:
            # TAIL_* 키만 사용 (레거시 MIN_DROP_RATE 등 폴백 없음)
            min_drop = tail_env_float(r, "TAIL_MIN_DROP_RATE", 0.03)
            min_rec  = tail_env_float(r, "TAIL_MIN_RECOVERY_RATIO", 0.5)
            tail_ratio = tail_env_float(r, "TAIL_RATIO_MIN", 1.5)
            tail_pct   = tail_env_float(r, "TAIL_PCT_MIN", 0.003)
            sl_pct     = abs(tail_env_float(r, "TAIL_STOP_LOSS_PCT", -0.03))
            tp_pct     = tail_env_float(r, "TAIL_TAKE_PROFIT_PCT", 0.05)
            shoulder_high = tail_env_float(r, "TAIL_SHOULDER_MIN_HIGH_PCT", 0.005)
            shoulder_cut  = tail_env_float(r, "TAIL_SHOULDER_CUT_PCT", 0.003)
            cooldown_sec  = tail_env_int(r, "TAIL_COOLDOWN_SEC", 900)
            rsi_period    = tail_env_int(r, "TAIL_RSI_PERIOD", 14)
            rsi_threshold = tail_env_float(r, "TAIL_RSI_THRESHOLD", 78.0)
            max_rec_3m    = tail_env_float(r, "TAIL_MAX_RECOVERY_3M", 0.8)
            high_chase    = tail_env_float(r, "TAIL_HIGH_CHASE_THR", 0.96)
            time_start    = tail_env_int(r, "TAIL_TIME_START", 930)
            time_end      = tail_env_int(r, "TAIL_TIME_END", 1500)
            max_daily     = tail_env_int(r, "TAIL_MAX_DAILY", 20)
            # 종목당 일일 재진입 게이트 — 무지성 횟수 대신 손실예산·최소엣지 (양수=절댓값)
            symbol_daily_loss_limit_krw = tail_env_float(r, "TAIL_SYMBOL_DAILY_LOSS_LIMIT_KRW", 30000.0)
            symbol_daily_loss_limit_pct = tail_env_float(r, "TAIL_SYMBOL_DAILY_LOSS_LIMIT_PCT", 1.5)
            reentry_require_nonneg = tail_env_bool(r, "TAIL_REENTRY_REQUIRE_NONNEG", False)
            reentry_min_edge_krw = tail_env_float(r, "TAIL_REENTRY_MIN_EDGE_KRW", 0.0)

            min_price          = tail_env_float(r, "TAIL_MIN_PRICE", 1000.0)
            max_daily_change   = tail_env_float(r, "TAIL_MAX_DAILY_CHG", 20.0)
            ma20_max_above     = tail_env_float(r, "TAIL_MA20_MAX_ABOVE_PCT", 3.0)
            stop_atr_mult      = tail_env_float(r, "TAIL_STOP_ATR_MULT", 2.0)
            target_atr_mult    = tail_env_float(r, "TAIL_TARGET_ATR_MULT", 2.5)
            atr_sl_min_pct     = _read_tail_pct_from_row(r, "TAIL_ATR_SL_MIN_PCT", "TAIL_ATR_SL_MIN_PCT", 0.8) * 100.0
            atr_sl_max_pct     = _read_tail_pct_from_row(r, "TAIL_ATR_SL_MAX_PCT", "TAIL_ATR_SL_MAX_PCT", 6.0) * 100.0
            atr_tp_min_pct     = _read_tail_pct_from_row(r, "TAIL_ATR_TP_MIN_PCT", "TAIL_ATR_TP_MIN_PCT", 0.5) * 100.0
            atr_tp_max_pct     = _read_tail_pct_from_row(r, "TAIL_ATR_TP_MAX_PCT", "TAIL_ATR_TP_MAX_PCT", 3.0) * 100.0
            max_loss_krw       = tail_env_int(r, "TAIL_MAX_LOSS_KRW", 200000)
            _min_drop_loss     = r.get("TAIL_MIN_DROP_FOR_LOSS_CUT")
            min_drop_pct_for_loss_cut = 0.015
            if _min_drop_loss not in (None, ""):
                v = float(_min_drop_loss)
                min_drop_pct_for_loss_cut = v / 100.0 if v >= 1 else v
            risk_pct           = float(r.get("RISK_PCT_PER_TRADE") or 0.01)
            kelly_mult         = float(r.get("KELLY_MULTIPLIER") or 0.25)
            min_hold_sec       = float(r.get("MIN_HOLD_AFTER_BUY_SEC") or 30.0)
            capital            = float(r.get("BACKTEST_CAPITAL") or 100000000.0)
            skip_hts_scan_dupes = tail_env_bool(r, "TAIL_SKIP_HTS_SCAN_DUPES", True)
            use_intraday_drop   = tail_env_bool(r, "TAIL_USE_INTRADAY_DROP", False)
            use_ma20_filter     = tail_env_bool(r, "TAIL_USE_MA20_FILTER", False)
            use_rsi_filter      = tail_env_bool(r, "TAIL_USE_RSI_FILTER", True)
            use_daily_range_f   = tail_env_bool(r, "TAIL_USE_DAILY_RANGE_FILTER", True)
            use_high_chase_f    = tail_env_bool(r, "TAIL_USE_HIGH_CHASE_FILTER", True)
            bar_chg_min_pct     = tail_env_float(r, "TAIL_BAR_CHG_MIN_PCT", -10.0)
            bar_chg_max_pct     = tail_env_float(r, "TAIL_BAR_CHG_MAX_PCT", -1.5)
            tail_vol_mult       = tail_env_float(r, "TAIL_VOL_MULT", 0.0)
            tail_vol_win        = tail_env_int(r, "TAIL_VOL_WIN", 5)
            max_stocks = tail_env_int(r, "TAIL_MAX_STOCKS", 3)
            total_budget_krw = tail_env_int(r, "TAIL_TOTAL_BUDGET_KRW", 0)
            slot_money = tail_env_int(r, "TAIL_SLOT_MONEY", 3_000_000)
            # 실매 BaseStrategy._post_filter_candidates 후보 하드캡과 동일 (0=무제한)
            cand_limit = tail_env_int(r, "SHORT_CAND_LIMIT", 0)
            short_max_buy = tail_env_int(r, "TAIL_MAX_BUY_AMOUNT", 0)
            from kis_trader.utils.env import get_env_from_db
            ratchet_tiers = str(r.get("TAIL_RATCHET_TIERS") or get_env_from_db("TAIL_RATCHET_TIERS", "") or "").strip()
            max_hold_bars = tail_env_int(r, "TAIL_MAX_HOLD_BARS", 0)
            eod_enabled = tail_env_bool(r, "TAIL_EOD_ENABLED", True)
            eod_hm = str(r.get("TAIL_EOD_HM") or "15:20").strip() or "15:20"
            # ws_ticks 진입가 재생 기본 ON — tail_tick_replay·모멘텀·돌파와 동일 (실매 체결 정합, env=0 일 때만 OFF)
            backtest_use_tick_db = tail_env_bool(r, "TAIL_BACKTEST_USE_TICK_DB", True)
            backtest_use_tick_exit = tail_env_bool(r, "TAIL_BACKTEST_USE_TICK_EXIT", True)
            backtest_tick_fallback_ohlc = tail_env_bool(r, "TAIL_BACKTEST_TICK_FALLBACK_OHLC", False)
            trail_pct = abs(tail_env_float(r, "TAIL_TRAIL_PCT", 0.0))
            trail_arm_pct = abs(tail_env_float(r, "TAIL_TRAIL_ARM_PCT", 0.0))
            _pat = _load_tail_pattern_params_from_row(r)
            
            whipsaw_filter = tail_env_bool(r, "TAIL_WHIPSAW_FILTER_ENABLED", False)
            whipsaw_subbar_sec = tail_env_int(r, "TAIL_WHIPSAW_SUBBAR_SEC", 10)
            whipsaw_lookback_sec = tail_env_int(r, "TAIL_WHIPSAW_LOOKBACK_SEC", 30)
            whipsaw_dip_pct = tail_env_float(r, "TAIL_WHIPSAW_DIP_PCT", 0.5)
            whipsaw_tol = tail_env_float(r, "TAIL_WHIPSAW_RECOVERY_TOL_PCT", 0.1)
            min_bid_ask_ratio = tail_env_float(r, "TAIL_ORDERBOOK_MIN_BID_ASK_RATIO", 0.0)
            ob_ask_max_mult = tail_env_float(r, "TAIL_ORDERBOOK_ENTRY_ASK_MAX_MULT", 0.0)
        else:
            min_drop, min_rec = 0.03, 0.5
            tail_ratio, tail_pct = 1.5, 0.003
            sl_pct, tp_pct = 0.03, 0.05
            shoulder_high, shoulder_cut = 0.005, 0.003
            cooldown_sec, rsi_period, rsi_threshold = 900, 14, 78.0
            max_rec_3m, high_chase = 0.8, 0.96
            time_start, time_end, max_daily = 930, 1500, 20
            symbol_daily_loss_limit_krw, symbol_daily_loss_limit_pct = 30000.0, 1.5
            reentry_require_nonneg, reentry_min_edge_krw = False, 0.0
            min_price, max_daily_change, ma20_max_above = 1000.0, 20.0, 3.0
            stop_atr_mult, target_atr_mult = 2.0, 2.5
            atr_sl_min_pct, atr_sl_max_pct = 0.8, 6.0
            atr_tp_min_pct, atr_tp_max_pct = 0.5, 3.0
            max_loss_krw, risk_pct, kelly_mult = 200000, 0.01, 0.25
            min_drop_pct_for_loss_cut = 0.015
            min_hold_sec, capital = 30.0, 100000000.0
            skip_hts_scan_dupes, use_intraday_drop = True, False
            use_ma20_filter, use_rsi_filter = False, True
            use_daily_range_f, use_high_chase_f = True, True
            bar_chg_min_pct, bar_chg_max_pct = -10.0, -1.5
            tail_vol_mult, tail_vol_win = 0.0, 5
            max_stocks, total_budget_krw, slot_money, short_max_buy = 3, 0, 3_000_000, 0
            cand_limit = 0
            ratchet_tiers, max_hold_bars = "", 0
            eod_enabled, eod_hm = True, "15:20"
            backtest_use_tick_db, backtest_use_tick_exit = True, True
            backtest_tick_fallback_ohlc = False
            trail_pct, trail_arm_pct = 0.0, 0.0
            _pat = _load_tail_pattern_params_from_row({})
            
            whipsaw_filter = False
            whipsaw_subbar_sec, whipsaw_lookback_sec = 10, 30
            whipsaw_dip_pct, whipsaw_tol = 0.5, 0.1
            min_bid_ask_ratio, ob_ask_max_mult = 0.0, 0.0
    except Exception:
        min_drop, min_rec = 0.03, 0.5
        tail_ratio, tail_pct = 1.5, 0.003
        sl_pct, tp_pct = 0.03, 0.05
        shoulder_high, shoulder_cut = 0.015, 0.03
        cooldown_sec, rsi_period, rsi_threshold = 900, 14, 78.0
        max_rec_3m, high_chase = 0.8, 0.96
        time_start, time_end, max_daily = 930, 1500, 20
        symbol_daily_loss_limit_krw, symbol_daily_loss_limit_pct = 30000.0, 1.5
        reentry_require_nonneg, reentry_min_edge_krw = False, 0.0
        min_price, max_daily_change, ma20_max_above = 1000.0, 20.0, 3.0
        stop_atr_mult, target_atr_mult = 2.5, 5.0
        atr_sl_min_pct, atr_sl_max_pct = 0.8, 6.0
        atr_tp_min_pct, atr_tp_max_pct = 0.5, 5.0
        max_loss_krw, risk_pct, kelly_mult = 200000, 0.01, 0.25
        min_drop_pct_for_loss_cut = 0.015
        min_hold_sec, capital = 30.0, 100000000.0
        skip_hts_scan_dupes, use_intraday_drop = True, False
        use_ma20_filter, use_rsi_filter = False, True
        use_daily_range_f, use_high_chase_f = True, True
        bar_chg_min_pct, bar_chg_max_pct = -10.0, -1.5
        tail_vol_mult, tail_vol_win = 0.0, 5
        max_stocks, total_budget_krw, slot_money, short_max_buy = 3, 0, 3_000_000, 0
        cand_limit = 0
        ratchet_tiers, max_hold_bars = "", 0
        eod_enabled, eod_hm = True, "15:20"
        backtest_use_tick_db, backtest_use_tick_exit = True, True
        backtest_tick_fallback_ohlc = False
        trail_pct, trail_arm_pct = 0.0, 0.0
        _pat = _load_tail_pattern_params_from_row({})
        
        whipsaw_filter = False
        whipsaw_subbar_sec, whipsaw_lookback_sec = 10, 30
        whipsaw_dip_pct, whipsaw_tol = 0.5, 0.1
        min_bid_ask_ratio, ob_ask_max_mult = 0.0, 0.0
    finally:
        if own_db is not None:
            try:
                own_db.close()
            except Exception:
                pass
    return {
        "min_drop_rate":      min_drop,
        "min_recovery_ratio": min_rec,
        "max_rec_3m":         max_rec_3m,
        "tail_ratio_min":     tail_ratio,
        "tail_pct_min":       tail_pct,
        "sl_pct":             sl_pct,
        "tp_pct":             tp_pct,
        "shoulder_min_high":  shoulder_high,
        "shoulder_cut_pct":   shoulder_cut,
        "ratchet_tiers":      ratchet_tiers,
        "max_hold_bars":      max_hold_bars,
        "rsi_period":         rsi_period,
        "rsi_threshold":      rsi_threshold,
        "high_chase_thr":     high_chase,
        "cooldown_min":       cooldown_sec // 60,
        "time_start_hm":      time_start,
        "time_end_hm":        time_end,
        "max_daily":          max_daily,
        # 종목당 일일 재진입 게이트 (기본 0/false=OFF, 동작 불변)
        "symbol_daily_loss_limit_krw": symbol_daily_loss_limit_krw,
        "symbol_daily_loss_limit_pct": symbol_daily_loss_limit_pct,
        "reentry_require_nonneg":      reentry_require_nonneg,
        "reentry_min_edge_krw":        reentry_min_edge_krw,
        # 고급 방어 파라미터 반환
        "min_price":          min_price,
        "max_daily_change":   max_daily_change,
        "ma20_max_above":     ma20_max_above,
        "stop_atr_mult":      stop_atr_mult,
        "target_atr_mult":    target_atr_mult,
        "atr_sl_min_pct":     atr_sl_min_pct,
        "atr_sl_max_pct":     atr_sl_max_pct,
        "atr_tp_min_pct":     atr_tp_min_pct,
        "atr_tp_max_pct":     atr_tp_max_pct,
        "max_loss_krw":       max_loss_krw,
        "min_drop_pct_for_loss_cut": min_drop_pct_for_loss_cut,
        "risk_pct":           risk_pct,
        "kelly_mult":         kelly_mult,
        "min_hold_sec":       min_hold_sec,
        "capital":            capital,
        # SCAN/TRIGGER 분리 플래그
        "skip_hts_scan_dupes": skip_hts_scan_dupes,
        "use_intraday_drop":   use_intraday_drop,
        "use_ma20_filter":     use_ma20_filter,
        "use_rsi_filter":      use_rsi_filter,
        "use_daily_range_filter": use_daily_range_f,
        "use_high_chase_filter":  use_high_chase_f,
        "bar_chg_min_pct":     bar_chg_min_pct,
        "bar_chg_max_pct":     bar_chg_max_pct,
        "tail_vol_mult":       tail_vol_mult,
        "tail_vol_win":        tail_vol_win,
        "max_stocks":          max_stocks,
        "total_budget_krw":    total_budget_krw,
        "slot_money":          slot_money,
        "short_max_buy_amount": short_max_buy,
        "cand_limit":          cand_limit,
        "portfolio_mode":      True,
        "entry_mode":          short_entry_mode(
            {"entry_mode": r.get("TAIL_ENTRY_MODE")} if r else None
        ),
        "limit_atr_mult":      tail_limit_params(
            {
                "limit_atr_mult": tail_env_float(r, "TAIL_LIMIT_ATR_MULT", 1.5) if r else None,
                "limit_anchor": (r.get("TAIL_LIMIT_ANCHOR") if r else None),
                "limit_valid_bars": tail_env_int(r, "TAIL_LIMIT_VALID_BARS", 1) if r else None,
                "limit_fill_slip_pct": tail_env_float(r, "TAIL_LIMIT_FILL_SLIP_PCT", 0.0) if r else None,
            }
        ).get("mult", 1.5),
        "limit_anchor":        tail_limit_params(
            {
                "limit_atr_mult": tail_env_float(r, "TAIL_LIMIT_ATR_MULT", 1.5) if r else None,
                "limit_anchor": (r.get("TAIL_LIMIT_ANCHOR") if r else None),
            }
        ).get("anchor", "signal_low"),
        "limit_valid_bars":    int(tail_limit_params(
            {"limit_valid_bars": tail_env_int(r, "TAIL_LIMIT_VALID_BARS", 1) if r else None}
        ).get("valid_bars", 1)),
        "limit_fill_slip_pct": float(tail_limit_params(
            {"limit_fill_slip_pct": tail_env_float(r, "TAIL_LIMIT_FILL_SLIP_PCT", 0.0) if r else None}
        ).get("fill_slip_pct", 0.0)),
        "backtest_use_tick_db": backtest_use_tick_db,
        "backtest_use_tick_exit": backtest_use_tick_exit,
        "backtest_tick_fallback_ohlc": backtest_tick_fallback_ohlc,
        # 체결량 상한(진입봉 거래량×N%) — 0=OFF. 실매 IOC 미체결 근사 (DB·웹·CLI 공통).
        "backtest_vol_fill_cap_pct": tail_env_float(r, "TAIL_BACKTEST_VOL_FILL_CAP_PCT", 0.0) if r else 0.0,
        "trail_pct":          trail_pct,
        "trail_arm_pct":      trail_arm_pct,
        "eod_enabled":        eod_enabled,
        "eod_hm":             eod_hm,
        "force_eod_exit":     eod_enabled,
        "whipsaw_filter":     whipsaw_filter,
        "whipsaw_subbar_sec": whipsaw_subbar_sec,
        "whipsaw_lookback_sec": whipsaw_lookback_sec,
        "whipsaw_dip_pct":    whipsaw_dip_pct,
        "whipsaw_tol":        whipsaw_tol,
        "min_bid_ask_ratio":  min_bid_ask_ratio,
        "ob_ask_max_mult":    ob_ask_max_mult,
        **_pat,
    }


def compute_rsi_series(closes: List[float], period: int = 14) -> List[Optional[float]]:
    """RSI 시리즈 (Wilder 스무딩)."""
    rsi_list: List[Optional[float]] = [None] * len(closes)
    if len(closes) < period + 1:
        return rsi_list
    deltas = [closes[i] - closes[i - 1] for i in range(1, len(closes))]
    gains = [max(d, 0) for d in deltas]
    losses = [max(-d, 0) for d in deltas]
    avg_gain = sum(gains[:period]) / period
    avg_loss = sum(losses[:period]) / period
    for i in range(period, len(closes)):
        idx = i - 1
        if i > period:
            avg_gain = (avg_gain * (period - 1) + gains[idx]) / period
            avg_loss = (avg_loss * (period - 1) + losses[idx]) / period
        rs = avg_gain / avg_loss if avg_loss > 0 else float("inf")
        rsi_val = 100 - (100 / (1 + rs)) if avg_loss > 0 else 100.0
        rsi_list[i] = rsi_val
    return rsi_list


def compute_sma_series(closes: List[float], period: int = 20) -> List[Optional[float]]:
    """단순 이동평균(SMA) 계산기 (엔진 내부용)."""
    sma_list: List[Optional[float]] = [None] * len(closes)
    if len(closes) < period:
        return sma_list
    running_sum = sum(closes[:period])
    sma_list[period - 1] = running_sum / period
    for i in range(period, len(closes)):
        running_sum += closes[i] - closes[i - period]
        sma_list[i] = running_sum / period
    return sma_list


def _t2dt(t: str) -> datetime:
    """candle_time / 실매 buy_time → datetime (공통 파서)."""
    from kis_trader.utils.trade_time import parse_trade_datetime
    return parse_trade_datetime(t)


def _pct_to_frac(v: Any, default_pct: float) -> float:
    """퍼센트 숫자(1.5=1.5%%) 또는 소수(0.015) → 비율 소수."""
    if v is None or v == "":
        v = default_pct
    x = float(v)
    return x / 100.0 if x >= 0.2 else x


def _read_tail_pct_from_row(r: Dict[str, Any], db_key: str, env_key: str, default_pct: float) -> float:
    """env_config 행 → 비율 소수. 없으면 get_env_float 폴백."""
    raw = r.get(db_key) if r else None
    if raw not in (None, ""):
        return _pct_to_frac(raw, default_pct)
    try:
        from kis_trader.utils.env import get_env_float
        return _pct_to_frac(get_env_float(env_key, default_pct), default_pct)
    except Exception:
        return _pct_to_frac(default_pct, default_pct)


def compute_tail_atr_prices(
    entry_price: float,
    atr_value: Optional[float],
    params: Dict[str, Any],
) -> Tuple[float, float]:
    """
    꼬리잡기 ATR 손절/익절가 — 배수 적용 후 % 상·하한 캡 (UPDOW ``resolve_atr_exit_pcts`` 와 동일 철학).

    :return: (stop_price, target_price)
    """
    if entry_price <= 0:
        return entry_price * 0.97, entry_price * 1.03
    atr = float(atr_value) if atr_value and float(atr_value) > 0 else entry_price * 0.01
    stop_mult = float(params.get("stop_atr_mult", 2.5))
    target_mult = float(params.get("target_atr_mult", 5.0))

    sl_pct = (atr * stop_mult) / entry_price
    tp_pct = (atr * target_mult) / entry_price

    sl_min = _pct_to_frac(params.get("atr_sl_min_pct", 0.8), 0.8)
    sl_max = _pct_to_frac(params.get("atr_sl_max_pct", 6.0), 6.0)
    tp_min = _pct_to_frac(params.get("atr_tp_min_pct", 0.5), 0.5)
    tp_max = _pct_to_frac(params.get("atr_tp_max_pct", 5.0), 5.0)

    sl_pct = min(max(sl_pct, sl_min), sl_max)
    tp_pct = min(max(tp_pct, tp_min), tp_max)

    stop_p = entry_price * (1.0 - sl_pct)
    target_p = entry_price * (1.0 + tp_pct)
    return stop_p, target_p


def _load_tail_pattern_params_from_row(r: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    """반전 패턴 ON/OFF 및 형태 임계값 — env ``TAIL_PATTERN_*`` 단일 소스."""
    row = r or {}
    return {
        "pattern_hammer": tail_env_bool(row, "TAIL_PATTERN_HAMMER", True),
        "pattern_pin": tail_env_bool(row, "TAIL_PATTERN_PIN", False),
        "pattern_engulfing": tail_env_bool(row, "TAIL_PATTERN_ENGULFING", False),
        "pattern_piercing": tail_env_bool(row, "TAIL_PATTERN_PIERCING", False),
        "pattern_harami": tail_env_bool(row, "TAIL_PATTERN_HARAMI", False),
        "pattern_doji": tail_env_bool(row, "TAIL_PATTERN_DOJI", False),
        "pattern_morning_star": tail_env_bool(row, "TAIL_PATTERN_MORNING_STAR", False),
        "pin_close_upper_ratio": tail_env_float(row, "TAIL_PIN_CLOSE_UPPER_RATIO", 0.66),
        "pin_max_upper_tail_ratio": tail_env_float(row, "TAIL_PIN_MAX_UPPER_TAIL_RATIO", 0.35),
        "engulf_min_body_ratio": tail_env_float(row, "TAIL_ENGULF_MIN_BODY_RATIO", 1.0),
        "piercing_penetrate_ratio": tail_env_float(row, "TAIL_PIERCING_PENETRATE_RATIO", 0.5),
        "harami_max_body_ratio": tail_env_float(row, "TAIL_HARAMI_MAX_BODY_RATIO", 0.5),
        "doji_body_max_ratio": tail_env_float(row, "TAIL_DOJI_BODY_MAX_RATIO", 0.15),
        "morning_star_body_max_ratio": tail_env_float(row, "TAIL_MORNING_STAR_BODY_MAX_RATIO", 0.35),
        "candle_lookback": tail_env_int(row, "TAIL_CANDLE_LOOKBACK", 3),
    }


def _bar_ohlc(c: Dict) -> Tuple[float, float, float, float]:
    return float(c["open"]), float(c["high"]), float(c["low"]), float(c["close"])


def _body_parts(op: float, hi: float, lo: float, cl: float) -> Dict[str, float]:
    """봉 몸통·꼬리 분해 (도지: 몸통 0 이면 range 기준 최소값)."""
    body_top = max(op, cl)
    body_bot = min(op, cl)
    body_len = body_top - body_bot
    rng = hi - lo if hi > lo else 0.0
    if body_len <= 0:
        body_len = max(rng * 0.001, 1e-6) if rng > 0 else 1.0
    lower_tail = max(0.0, body_bot - lo) if lo > 0 else 0.0
    upper_tail = max(0.0, hi - body_top) if hi > 0 else 0.0
    return {
        "body_top": body_top,
        "body_bot": body_bot,
        "body_len": body_len,
        "range": rng,
        "lower_tail": lower_tail,
        "upper_tail": upper_tail,
        "is_bullish": cl >= op,
        "is_bearish": cl < op,
    }


def _hammer_tail_metrics(
    op: float, hi: float, lo: float, cl: float, lo_ref: float,
) -> Tuple[float, float, float]:
    parts = _body_parts(op, hi, lo, cl)
    tail_len = parts["lower_tail"]
    body_len = parts["body_len"]
    lo_use = lo if lo > 0 else lo_ref
    tail_ratio = tail_len / body_len if body_len > 0 else 0.0
    tail_pct = tail_len / lo_use if lo_use > 0 and tail_len > 0 else 0.0
    return tail_ratio, tail_pct, lo_use


def _detect_hammer_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """망치(하단 꼬리) — 신호봉 또는 lookback 봉에서 탐색."""
    rsi_period = int(params.get("rsi_period", 14))
    lookback = int(params.get("candle_lookback", 3))
    tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
    tail_pct_min = float(params.get("tail_pct_min", 0.003))
    lo_ref = float(candles[i]["low"]) if float(candles[i]["low"]) > 0 else float(candles[i]["high"])

    for j in range(i, max(i - lookback, rsi_period) - 1, -1):
        op, hi, lo, cl = _bar_ohlc(candles[j])
        body_top = max(op, cl)
        body_bot = min(op, cl)
        body_len = body_top - body_bot if body_top > body_bot else 1.0
        tail_len = body_bot - lo if lo > 0 else 0.0
        lo_use = lo if lo > 0 else lo_ref
        if tail_len <= 0:
            continue
        tail_ratio = tail_len / body_len if body_len > 0 else 0.0
        tail_pct = tail_len / lo_use if lo_use > 0 else 0.0
        if tail_ratio >= tail_ratio_min and tail_pct >= tail_pct_min:
            return True, {
                "pattern": "hammer",
                "tail_ratio": tail_ratio,
                "tail_pct": tail_pct,
                "pattern_bar_idx": j,
            }
    return False, {}


def _detect_pin_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """스트릭트 핀바 — 하단 꼬리 + 종가 상단 1/3 + 윗꼬리 짧음."""
    rsi_period = int(params.get("rsi_period", 14))
    lookback = int(params.get("candle_lookback", 3))
    tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
    tail_pct_min = float(params.get("tail_pct_min", 0.003))
    pin_close_upper = float(params.get("pin_close_upper_ratio", 0.66))
    pin_upper_max = float(params.get("pin_max_upper_tail_ratio", 0.35))
    lo_ref = float(candles[i]["low"]) if float(candles[i]["low"]) > 0 else float(candles[i]["high"])

    for j in range(i, max(i - lookback, rsi_period) - 1, -1):
        op, hi, lo, cl = _bar_ohlc(candles[j])
        parts = _body_parts(op, hi, lo, cl)
        if parts["range"] <= 0:
            continue
        tail_ratio, tail_pct, _ = _hammer_tail_metrics(op, hi, lo, cl, lo_ref)
        if tail_ratio < tail_ratio_min or tail_pct < tail_pct_min:
            continue
        close_pos = (cl - lo) / parts["range"]
        if close_pos < pin_close_upper:
            continue
        if parts["body_len"] > 0 and parts["upper_tail"] / parts["body_len"] > pin_upper_max:
            continue
        return True, {
            "pattern": "pin",
            "tail_ratio": tail_ratio,
            "tail_pct": tail_pct,
            "pattern_bar_idx": j,
            "close_pos": close_pos,
        }
    return False, {}


def _detect_engulfing_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """불리시 장악형 — 전봉 음봉 몸통을 현재 양봉이 완전 삼킴."""
    if i < 1:
        return False, {}
    engulf_min = float(params.get("engulf_min_body_ratio", 1.0))
    po, ph, pl, pc = _bar_ohlc(candles[i - 1])
    co, ch, cl, cc = _bar_ohlc(candles[i])
    prev = _body_parts(po, ph, pl, pc)
    curr = _body_parts(co, ch, cl, cc)
    if not prev["is_bearish"] or not curr["is_bullish"]:
        return False, {}
    if co > pc or cc < po:
        return False, {}
    if curr["body_len"] < prev["body_len"] * engulf_min:
        return False, {}
    tail_ratio, tail_pct, _ = _hammer_tail_metrics(co, ch, cl, cc, pl if pl > 0 else ch)
    return True, {
        "pattern": "engulfing",
        "tail_ratio": tail_ratio,
        "tail_pct": tail_pct,
        "pattern_bar_idx": i,
    }


def _detect_piercing_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """관통형 — 갭다운 후 전봉 몸통 중간 이상 회복(완전 장악 전)."""
    if i < 1:
        return False, {}
    penetrate = float(params.get("piercing_penetrate_ratio", 0.5))
    po, ph, pl, pc = _bar_ohlc(candles[i - 1])
    co, ch, cl, cc = _bar_ohlc(candles[i])
    prev = _body_parts(po, ph, pl, pc)
    curr = _body_parts(co, ch, cl, cc)
    if not prev["is_bearish"] or not curr["is_bullish"]:
        return False, {}
    if co >= pc:
        return False, {}
    midpoint = prev["body_bot"] + prev["body_len"] * penetrate
    if cc <= midpoint:
        return False, {}
    if cc >= po:
        return False, {}
    tail_ratio, tail_pct, _ = _hammer_tail_metrics(co, ch, cl, cc, pl if pl > 0 else ch)
    return True, {
        "pattern": "piercing",
        "tail_ratio": tail_ratio,
        "tail_pct": tail_pct,
        "pattern_bar_idx": i,
    }


def _detect_harami_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """불리시 하라미 — 큰 음봉 안에 작은 양봉(몸통 포함)."""
    if i < 1:
        return False, {}
    harami_max = float(params.get("harami_max_body_ratio", 0.5))
    po, ph, pl, pc = _bar_ohlc(candles[i - 1])
    co, ch, cl, cc = _bar_ohlc(candles[i])
    prev = _body_parts(po, ph, pl, pc)
    curr = _body_parts(co, ch, cl, cc)
    if not prev["is_bearish"]:
        return False, {}
    if curr["body_top"] > prev["body_top"] or curr["body_bot"] < prev["body_bot"]:
        return False, {}
    if curr["body_len"] > prev["body_len"] * harami_max:
        return False, {}
    if not curr["is_bullish"]:
        return False, {}
    tail_ratio, tail_pct, _ = _hammer_tail_metrics(co, ch, cl, cc, pl if pl > 0 else ch)
    return True, {
        "pattern": "harami",
        "tail_ratio": tail_ratio,
        "tail_pct": tail_pct,
        "pattern_bar_idx": i,
    }


def _detect_doji_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """저점 도지 — 몸통 극소 + 하단 꼬리(망치형) + 종가 중상단."""
    doji_max = float(params.get("doji_body_max_ratio", 0.15))
    tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
    tail_pct_min = float(params.get("tail_pct_min", 0.003))
    pin_close_upper = float(params.get("pin_close_upper_ratio", 0.66))
    op, hi, lo, cl = _bar_ohlc(candles[i])
    parts = _body_parts(op, hi, lo, cl)
    if parts["range"] <= 0:
        return False, {}
    real_body = abs(cl - op)
    if real_body / parts["range"] > doji_max:
        return False, {}
    tail_ratio, tail_pct, _ = _hammer_tail_metrics(op, hi, lo, cl, lo if lo > 0 else hi)
    if tail_ratio < tail_ratio_min or tail_pct < tail_pct_min:
        return False, {}
    close_pos = (cl - lo) / parts["range"]
    if close_pos < pin_close_upper:
        return False, {}
    return True, {
        "pattern": "doji",
        "tail_ratio": tail_ratio,
        "tail_pct": tail_pct,
        "pattern_bar_idx": i,
        "close_pos": close_pos,
    }


def _detect_morning_star_pattern(
    candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
    """샛별형(3봉) — 장음봉 + 작은 별 + 양봉이 1봉 몸통 중간 돌파."""
    if i < 2:
        return False, {}
    star_max = float(params.get("morning_star_body_max_ratio", 0.35))
    o0, h0, l0, c0 = _bar_ohlc(candles[i - 2])
    o1, h1, l1, c1 = _bar_ohlc(candles[i - 1])
    o2, h2, l2, c2 = _bar_ohlc(candles[i])
    bear = _body_parts(o0, h0, l0, c0)
    star = _body_parts(o1, h1, l1, c1)
    bull = _body_parts(o2, h2, l2, c2)
    if not bear["is_bearish"] or bear["body_len"] <= 0:
        return False, {}
    if star["body_len"] > bear["body_len"] * star_max:
        return False, {}
    midpoint = (bear["body_top"] + bear["body_bot"]) / 2.0
    if not bull["is_bullish"] or c2 <= midpoint:
        return False, {}
    tail_ratio, tail_pct, _ = _hammer_tail_metrics(o2, h2, l2, c2, l2 if l2 > 0 else h2)
    return True, {
        "pattern": "morning_star",
        "tail_ratio": tail_ratio,
        "tail_pct": tail_pct,
        "pattern_bar_idx": i,
    }


def eval_tail_reversal_pattern(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, str, Dict[str, Any]]:
    """
    TRIGGER 반전 패턴 OR 평가.

    ``TAIL_PATTERN_*`` 가 모두 OFF 이면 망치만(기존 동작) 검사.
    """
    checks = []
    if _to_bool(params.get("pattern_hammer"), True):
        checks.append(("hammer", _detect_hammer_pattern))
    if _to_bool(params.get("pattern_pin"), False):
        checks.append(("pin", _detect_pin_pattern))
    if _to_bool(params.get("pattern_engulfing"), False):
        checks.append(("engulfing", _detect_engulfing_pattern))
    if _to_bool(params.get("pattern_piercing"), False):
        checks.append(("piercing", _detect_piercing_pattern))
    if _to_bool(params.get("pattern_harami"), False):
        checks.append(("harami", _detect_harami_pattern))
    if _to_bool(params.get("pattern_doji"), False):
        checks.append(("doji", _detect_doji_pattern))
    if _to_bool(params.get("pattern_morning_star"), False):
        checks.append(("morning_star", _detect_morning_star_pattern))
    if not checks:
        checks.append(("hammer", _detect_hammer_pattern))

    for name, fn in checks:
        ok, metrics = fn(candles, i, params)
        if ok:
            return True, name, metrics

    active = ",".join(n for n, _ in checks)
    return False, active, {}


def _tail_volume_spike_ok(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[bool, str]:
    """
    TRIGGER — 신호봉 거래량이 직전 N봉 평균 대비 배수 이상인지 (모멘텀 mom_vol_mult 와 동일 패턴).
    ``tail_vol_mult`` ≤ 0 이면 OFF (기존 백테·실매 동작 유지).
    """
    vol_mult = float(params.get("tail_vol_mult", 0) or 0)
    vol_win = int(params.get("tail_vol_win", 5) or 5)
    if vol_mult <= 0:
        return True, ""
    vol = float(candles[i].get("volume", 0) or 0)
    win = max(1, min(vol_win, i))
    vols = [float(candles[k].get("volume", 0) or 0) for k in range(i - win, i)]
    if not vols or sum(vols) <= 0:
        return False, "거래량창없음"
    avg = sum(vols) / len(vols)
    if avg <= 0 or vol < avg * vol_mult:
        ratio = vol / avg if avg > 0 else 0.0
        return False, "%.2fx < %.1fx" % (ratio, vol_mult)
    return True, ""


def _eval_tail_buy_at_index(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """신호봉 인덱스 ``i`` 에서 꼬리잡기 매수 조건 평가 (백테스트 루프와 동일 시점)."""
    if i < 19 or i >= len(candles):
        return ("탈락-데이터", f"인덱스 부적절 (i={i})", None)

    c = candles[i]
    day = c["candle_time"][:8]
    hm = int(c["candle_time"][8:12])
    op = float(c["open"])
    hi = float(c["high"])
    lo = float(c["low"])
    cl = float(c["close"])

    # 엔진 파라미터 로드
    min_drop_rate = float(params.get("min_drop_rate", 0.03))
    min_recovery_ratio = float(params.get("min_recovery_ratio", 0.5))
    max_rec_3m = float(params.get("max_rec_3m", 0.8))
    tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
    tail_pct_min = float(params.get("tail_pct_min", 0.003))
    rsi_period = int(params.get("rsi_period", 14))
    rsi_threshold = float(params.get("rsi_threshold", 78))
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    time_start_hm = int(params.get("time_start_hm", 930))
    time_end_hm = int(params.get("time_end_hm", 1500))
    cooldown_min = float(params.get("cooldown_min", 15))
    max_daily = int(params.get("max_daily", 3))
    
    # 방어 로직 파라미터
    min_price = float(params.get("min_price", 1000.0))
    max_daily_change = float(params.get("max_daily_change", 20.0))
    ma20_max_above = float(params.get("ma20_max_above", 3.0))

    # SCAN/TRIGGER — skip_hts=true 이면 아래 3분 봉등락(BAR_CHG) 재검사만 생략
    if params.get("skip_hts_scan_dupes") is None:
        skip_hts = resolve_tail_skip_hts_scan_dupes()
    else:
        skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), resolve_tail_skip_hts_scan_dupes())
    use_intraday_drop = _to_bool(params.get("use_intraday_drop"), False)
    use_ma20 = _to_bool(params.get("use_ma20_filter"), False)
    use_rsi = _to_bool(params.get("use_rsi_filter"), True)
    use_daily_range = _to_bool(params.get("use_daily_range_filter"), True)
    use_high_chase = _to_bool(params.get("use_high_chase_filter"), True)
    bar_chg_min_pct = float(params.get("bar_chg_min_pct", -10.0))
    bar_chg_max_pct = float(params.get("bar_chg_max_pct", -1.5))

    if hm < time_start_hm or hm > time_end_hm:
        return (None, None, None)  # 시간대 탈락
    if state.get("daily_cnt", 0) >= max_daily:
        return (
            "탈락-일일한도",
            f"daily={state.get('daily_cnt', 0)}/{max_daily}",
            None,
        )
    _reentry_ok, _reentry_detail = _tail_symbol_reentry_gate(
        int(state.get("daily_cnt", 0) or 0),
        float(state.get("daily_pnl_krw", 0.0) or 0.0),
        float(params.get("slot_money", 0) or 0),
        params,
    )
    if not _reentry_ok:
        return ("탈락-종목일일손익게이트", _reentry_detail, None)
    last_exit_dt = state.get("last_exit_dt")
    if last_exit_dt is not None:
        elapsed = (_t2dt(c["candle_time"]) - last_exit_dt).total_seconds() / 60
        if elapsed < cooldown_min:
            return (None, None, None)

    if cl <= 0 or cl < min_price:
        return ("탈락-가격", f"현재가 부적절 (현재 {cl:,.0f}원, 최소 {min_price:,.0f}원)", None)

    # 당일 누적 OHLC 및 피뢰침 검사
    running_open = op
    running_high = hi
    running_low = lo if lo > 0 else hi
    for j in range(i - 1, -1, -1):
        if candles[j]["candle_time"][:8] != day:
            break
        running_open = float(candles[j]["open"])
        running_high = max(running_high, float(candles[j]["high"]))
        lj = float(candles[j]["low"])
        if lj > 0:
            running_low = min(running_low, lj)

    if cl <= 0 or running_open <= 0:
        return (None, None, None)

    # 일일 변동폭(피뢰침) 검사 — TRIGGER 선택 필터
    if use_daily_range and running_low > 0:
        range_change_pct = (running_high - running_low) / running_low * 100
        if range_change_pct > max_daily_change:
            return ("탈락-피뢰침 급등주", f"일일 변동폭 {range_change_pct:.1f}% > {max_daily_change:.0f}%", None)

    # 3분봉 직전대비 등락 (BAR_CHG) — HTS 일봉A/1분G 와 축이 다름. skip_hts=true 면 생략
    if not skip_hts and i >= 1:
        prev_cl = float(candles[i - 1]["close"])
        if prev_cl > 0:
            bar_chg = (cl - prev_cl) / prev_cl * 100.0
            if bar_chg < bar_chg_min_pct or bar_chg > bar_chg_max_pct:
                return (
                    "탈락-봉등락",
                    f"봉등락 {bar_chg:.2f}% (3분 직전대비: {bar_chg_min_pct:.1f}~{bar_chg_max_pct:.1f}%)",
                    None,
                )

    # 당일 시가→저점 낙폭 — BAR_CHG 와 다른 축; ``TAIL_USE_INTRADAY_DROP=true`` 일 때만
    if use_intraday_drop:
        drop = (running_open - running_low) / running_open
        if drop < min_drop_rate:
            return (
                "탈락-낙폭",
                f"낙폭 {drop*100:.2f}% < {min_drop_rate*100:.1f}% (시가 {running_open:,.0f} → 저점 {running_low:,.0f})",
                None,
            )
    
    # 회복률 검사
    day_range = running_high - running_low
    rec_day = (cl - running_low) / day_range if day_range > 0 else 0
    if rec_day < min_recovery_ratio:
        return (
            "탈락-회복률",
            f"회복률 {rec_day*100:.1f}% < {min_recovery_ratio*100:.0f}% (저점 {running_low:,.0f} → 현재 {cl:,.0f})",
            None,
        )

    # 반전 패턴 OR (망치·핀바·장악·관통·하라미·도지·샛별) — env TAIL_PATTERN_* 로 ON/OFF
    pat_ok, pat_active, pat_metrics = eval_tail_reversal_pattern(candles, i, params)
    if not pat_ok:
        return (
            "탈락-패턴",
            f"반전패턴 미충족 (활성: {pat_active})",
            None,
        )
    tail_ratio = float(pat_metrics.get("tail_ratio", 0.0))
    tail_pct = float(pat_metrics.get("tail_pct", 0.0))
    pattern_name = str(pat_metrics.get("pattern", pat_active))

    # TRIGGER — 신호봉 거래량 폭증 (HTS C는 스캔 시점; 매수 직전 재확인)
    vol_ok, vol_msg = _tail_volume_spike_ok(candles, i, params)
    if not vol_ok:
        return ("탈락-거래량", f"신호봉 거래량 {vol_msg}", None)

    # 3분봉 내 회복 위치 상한 검사
    c_range = hi - lo if hi > lo else 0
    rec_3m = (cl - lo) / c_range if c_range > 0 else 0
    if not (min_recovery_ratio <= rec_3m <= max_rec_3m):
        return (
            "탈락-회복3분",
            f"3분봉 회복률 {rec_3m*100:.1f}% (기준 {min_recovery_ratio*100:.0f}~{max_rec_3m*100:.0f}%)",
            None,
        )

    # 고점 추격 방지 — TRIGGER 선택 필터
    if use_high_chase and cl >= running_high * high_chase_thr:
        return (
            "탈락-피뢰침 고점추격",
            f"현재가 {cl:,.0f} ≥ 고점대비 {high_chase_thr*100:.0f}%",
            None,
        )

    # RSI 검사 — TRIGGER 선택 필터 (HTS B 체결강도와 별개)
    closes = [float(x["close"]) for x in candles]
    ic = params.get("_indicator_cache")
    if ic is not None and hasattr(ic, "rsi_at"):
        rsi_val = ic.rsi_at(i, rsi_period)
    else:
        rsis = compute_rsi_series(closes, rsi_period)
        rsi_val = rsis[i] if i < len(rsis) else None
    if use_rsi and (rsi_val is None or rsi_val >= rsi_threshold):
        return (
            "탈락-RSI",
            (f"RSI {rsi_val:.1f}" if rsi_val is not None else "RSI None") + f" ≥ {rsi_threshold:.0f}",
            None,
        )

    # MA20 방어 — TRIGGER 선택 필터 (기본 OFF: 역배열 컷이 신호를 과하게 줄임)
    if use_ma20:
        ma20 = sum(closes[i - 19:i + 1]) / 20.0
        if cl < ma20:
            return ("탈락-MA20", f"현재가 {cl:,.0f} < MA20 {ma20:,.0f} (역배열)", None)
        if ma20 > 0 and cl > ma20 * (1 + ma20_max_above / 100):
            return ("탈락-MA20초과", f"MA20 대비 {ma20_max_above:.0f}% 이격 초과", None)

    sig = {
        "signal": True,
        "pattern": pattern_name,
        "tail_ratio": tail_ratio,
        "tail_pct": tail_pct,
        "recovery_pos": rec_3m,
        "rsi_val": rsi_val,
        "atr_calc_val": None,
        "signal_candle_time": c.get("candle_time"),
    }
    return _tail_signal_with_whipsaw(params, candles, i, sig)


def _tail_signal_with_whipsaw(
    params: Dict[str, Any],
    candles: List[Dict],
    i: int,
    sig: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """꼬리 신호 dict 반환 직전 — 휩쏘(TRIGGER) 공통 (실매·백테·파라서치)."""
    c = candles[i]
    cl = float(c.get("close") or 0)
    ws_rej, ws_msg = whipsaw_reject_for_signal(
        params, "TAIL",
        signal_bar=c,
        current_price=cl,
    )
    if ws_rej:
        return (ws_rej, ws_msg, None)
    ob_rej, ob_msg = orderbook_reject_for_entry(
        params, "TAIL", current_price=cl,
    )
    if ob_rej:
        return (ob_rej, ob_msg, None)
    prog_rej, prog_msg = program_reject_for_entry(
        params, "TAIL", current_price=cl,
    )
    if prog_rej:
        return (prog_rej, prog_msg, None)
    
    ob_snap = params.get("_backtest_orderbook_snapshot")
    if ob_snap is not None:
        sig["backtest_ob_source"] = str(getattr(ob_snap, "source", "") or "").strip()
        
    return (None, None, sig)


def check_buy_signal_live(
    candles: List[Dict],
    params: Dict[str, Any],
    state: Dict[str, Any],
) -> tuple:
    """
    실시간 꼬리잡기 매수 신호.

    - ``live_backtest_align=True`` (기본): 백테와 동일 — **직전 확정봉=신호봉**,
      **현재 확정봉=진입봉** (신호 직후 봉 시가·실매 시장가 정합).
    - ``live_backtest_align=False``: 구버전 — 마지막 확정봉 1개만 검사.

    candles: 3분봉 (candle_time, open, high, low, close, volume)
    state:   { "last_exit_dt": datetime|None, "daily_cnt": int }
  """
    live_align = _to_bool(params.get("live_backtest_align", True), True)
    lookback = max(1, int(params.get("live_signal_lookback_bars", 1)))
    confirmed = _confirmed_candles_only(candles)

    if len(confirmed) < 20:
        return ("탈락-데이터", f"확정봉 부족 (len={len(confirmed)} < 20)", None)

    last_reject: Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]] = (
        None, None, None,
    )

    if live_align:
        entry_i = len(confirmed) - 1
        return _eval_live_align_lookback(
            confirmed, entry_i, params, state, lookback=lookback,
        )

    i = len(confirmed) - 1
    return _eval_tail_buy_at_index(confirmed, i, params, state)


def _eval_live_align_lookback(
    candles: List[Dict],
    entry_i: int,
    params: Dict[str, Any],
    state: Dict[str, Any],
    *,
    lookback: int = 1,
) -> tuple:
    """실매 align: entry_i=진입봉, entry_i-1-k=신호봉 (lookback개까지)."""
    lb = max(1, int(lookback))
    last_reject: Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]] = (
        None, None, None,
    )
    for k in range(lb):
        signal_i = entry_i - 1 - k
        if signal_i < 19:
            break
        reject, msg, sig = _eval_tail_buy_at_index(
            candles, signal_i, params, state,
        )
        if reject:
            if k == 0:
                last_reject = (reject, msg, None)
            continue
        if sig:
            ent = candles[entry_i]
            entry_open = float(ent.get("open", 0) or 0)
            if entry_open <= 0:
                entry_open = float(ent.get("close", 0) or 0)
            sig["entry_price"] = entry_open
            sig["entry_bar_key"] = ent.get("candle_time")
            sig["signal_bar_key"] = candles[signal_i].get("candle_time")
            sig["signal_i"] = signal_i
            sig["entry_i"] = entry_i
            return (None, None, sig)
    return last_reject


def _last_closed_bar_index(
    candles: List[Dict],
    as_of_hhmm: str,
    tf_min: int,
) -> int:
    """as_of(YYYYMMDDHHMM) 시점까지 **이미 마감**된 마지막 봉 인덱스. 없으면 -1.

    봉 시작 + tf_min ≤ as_of 이면 확정(실매: 진행 중 봉은 신호/진입봉으로 안 씀).
    """
    from kis_trader.engine.candle_rollup import add_candle_minutes

    as_of = str(as_of_hhmm or "")[:12]
    if len(as_of) < 12 or not candles:
        return -1
    best = -1
    for i, c in enumerate(candles):
        ct = str(c.get("candle_time") or "")[:12]
        if len(ct) < 12:
            continue
        end_t = add_candle_minutes(ct, int(tf_min))
        if end_t and end_t <= as_of:
            best = i
        elif ct > as_of:
            break
    return best


def _first_universe_minute_by_code(
    universe_by_slot: Optional[Dict[str, List[str]]],
    universe_timeline: Optional[Any],
    codes: Optional[set] = None,
) -> Dict[str, str]:
    """종목별 최초 편입 분키(YYYYMMDDHHMM)."""
    out: Dict[str, str] = {}
    want = codes
    if universe_timeline is not None and hasattr(universe_timeline, "_keys"):
        try:
            keys = list(getattr(universe_timeline, "_keys") or [])
            code_lists = list(getattr(universe_timeline, "_codes") or [])
            for sk, snap_codes in zip(keys, code_lists):
                minute = str(sk)[:12]
                for c in snap_codes or []:
                    cs = str(c).strip()
                    if not cs or (want is not None and cs not in want):
                        continue
                    if cs not in out:
                        out[cs] = minute
            if out:
                return out
        except Exception:
            pass
    if not universe_by_slot:
        return out
    for slot in sorted(universe_by_slot.keys()):
        minute = str(slot)[:12]
        for c in universe_by_slot.get(slot) or []:
            cs = str(c).strip()
            if not cs or (want is not None and cs not in want):
                continue
            if cs not in out:
                out[cs] = minute
    return out


def _universe_enter_minutes(
    universe_by_slot: Optional[Dict[str, List[str]]],
    universe_timeline: Optional[Any],
    codes: Optional[set] = None,
) -> Set[str]:
    """유니버스 ENTER 분키 집합 (재편입 포함) — 실매 REAL I(편입) 시각 근사.

    최초 편입만 쓰는 ``_first_universe_minute_by_code`` 와 달리,
    EXIT 후 다시 들어온 분도 시계에 넣어 Phase 0-incl 이 재검사한다.
    """
    out: Set[str] = set()
    want = codes
    prev: Set[str] = set()

    def _iter_snaps():
        if universe_timeline is not None and hasattr(universe_timeline, "_keys"):
            try:
                keys = list(getattr(universe_timeline, "_keys") or [])
                code_lists = list(getattr(universe_timeline, "_codes") or [])
                for sk, snap_codes in zip(keys, code_lists):
                    yield str(sk)[:12], {
                        str(c).strip() for c in (snap_codes or []) if str(c).strip()
                    }
                return
            except Exception:
                pass
        if not universe_by_slot:
            return
        for slot in sorted(universe_by_slot.keys()):
            yield str(slot)[:12], {
                str(c).strip() for c in (universe_by_slot.get(slot) or []) if str(c).strip()
            }

    for minute, cur in _iter_snaps():
        if want is not None:
            cur = {c for c in cur if c in want}
        entered = cur - prev
        if entered and minute:
            out.add(minute)
        prev = cur
    return out


def _tail_ratchet_tiers(params: Dict[str, Any]) -> List[Tuple[float, float]]:
    """래칫 단계 트레일 — 돌파 ``_breakout_ratchet_tiers`` 와 동일 형식.

    env/params ``"2:1.5,5:1.0"`` → +2% 수익부터 고점 대비 1.5% 되돌림컷.
    비어 있으면 [] (= OFF, 단일 어깨컷 사용).
    """
    raw = params.get("ratchet_tiers")
    if raw is None:
        from kis_trader.utils.env import get_env_from_db
        raw = get_env_from_db("TAIL_RATCHET_TIERS", "")
    if isinstance(raw, (list, tuple)):
        pairs = list(raw)
    else:
        s = str(raw or "").strip()
        if not s:
            return []
        pairs = []
        for chunk in s.split(","):
            chunk = chunk.strip()
            if not chunk or ":" not in chunk:
                continue
            g, c = chunk.split(":", 1)
            pairs.append((g, c))
    tiers: List[Tuple[float, float]] = []
    for g, c in pairs:
        try:
            gain = abs(float(g)) / 100.0
            cut = abs(float(c)) / 100.0
        except (TypeError, ValueError):
            continue
        if gain <= 0 or cut <= 0:
            continue
        tiers.append((gain, cut))
    tiers.sort(key=lambda x: x[0])
    return tiers


def _tail_max_hold_minutes(params: Dict[str, Any]) -> int:
    """시간컷: 최대 보유 **분** (돌파 BREAKOUT_MAX_HOLD_BARS 와 동일 — 이름은 bars, 단위는 분). 0=OFF."""
    v = params.get("max_hold_bars")
    if v is None:
        from kis_trader.utils.env import get_env_int
        v = get_env_int("TAIL_MAX_HOLD_BARS", 0)
    try:
        return max(0, int(float(v)))
    except (TypeError, ValueError):
        return 0


def _tail_minutes_held(position: Dict[str, Any], current_candle: Dict[str, Any]) -> Optional[int]:
    """진입 이후 경과 분 — 시간컷 판정."""
    et = position.get("entry_time") or position.get("buy_time")
    ct = current_candle.get("candle_time")
    if not et or not ct:
        return None
    try:
        # 실매 buy_time 은 "YYYY-MM-DD HH:MM:SS" — [:12] 자르면 파싱 실패
        e_dt = _t2dt(str(et))
        c_dt = _t2dt(str(ct))
        return max(0, int((c_dt - e_dt).total_seconds() // 60))
    except Exception:
        return None


def check_sell_signal_live(
    position: Dict[str, Any],
    current_candle: Dict[str, Any],
    params: Dict[str, Any],
    is_eod: bool = False,
) -> Optional[tuple]:
    """
    실시간 및 백테스트 공통 청산 조건.
    [V4]: 트레일(어깨) 저가 기준 · 수익 구간 우선 · ATR 캡 익절/손절 · 금액손실컷(트레일 미발동만).

    position: entry_price, entry_time(YYYYMMDDHHMM), stop, target, max_price, qty
    current_candle: high, low, close, candle_time
    반환: (reason_str, exit_price) 또는 None
    """
    shoulder_min_high = float(params.get("shoulder_min_high", 0.005))
    shoulder_cut_pct = float(params.get("shoulder_cut_pct", 0.003))
    max_loss_krw = int(params.get("max_loss_krw", 200000))
    min_hold_sec = float(params.get("min_hold_sec", 30.0))

    hi = float(current_candle.get("high", current_candle["close"]))
    lo = float(current_candle.get("low", current_candle["close"]))
    cl = float(current_candle["close"])
    candle_time = current_candle.get("candle_time", "")

    max_p = max(float(position.get("max_price", 0) or 0), hi)
    ep = float(position["entry_price"])
    stop = float(position["stop"])
    target = float(position["target"])
    qty = int(position.get("qty", 1) or 1)

    # 최소 보유 시간 검사 (너무 짧으면 청산 무시). EOD는 무조건 청산.
    if (not is_eod) and candle_time and position.get("entry_time"):
        try:
            entry_dt = _t2dt(position["entry_time"])
            curr_dt = _t2dt(candle_time)
            if (curr_dt - entry_dt).total_seconds() < min_hold_sec:
                return None
        except Exception:
            pass

    reason = None
    exit_price = cl
    profit_val = (lo - ep) * qty
    drop_pct = (ep - lo) / ep if ep > 0 else 0.0
    min_drop_pct = float(params.get("min_drop_pct_for_loss_cut", 0.015))

    # 1순위: 트레일 — 돌파와 동일: 래칫 문자열 있으면 **단일 어깨 대체**, 없으면 shoulder_min/cut
    ratchet_tiers = _tail_ratchet_tiers(params)
    trail_armed = False
    if ratchet_tiers and ep > 0:
        peak_gain = (max_p - ep) / ep
        cut_ratio = 0.0
        for gain_thr, cut in ratchet_tiers:
            if peak_gain >= gain_thr:
                cut_ratio = cut
        if cut_ratio > 0.0:
            trail_armed = True
            ratchet_line = max_p * (1.0 - cut_ratio)
            if lo > 0 and lo <= ratchet_line:
                reason = "래칫컷"
                exit_price = ratchet_line
    else:
        trail_armed = ep > 0 and max_p >= ep * (1.0 + shoulder_min_high)
        trail_stop_px = max_p * (1.0 - shoulder_cut_pct) if trail_armed else 0.0
        if trail_armed and lo > 0 and lo <= trail_stop_px:
            reason = "어깨컷"
            exit_price = trail_stop_px

    # 2순위: ATR 캡 익절
    if not reason and hi >= target:
        reason = "익절"
        exit_price = target
    # 3순위: ATR 캡 손절
    if not reason and lo > 0 and lo <= stop:
        reason = "손절"
        exit_price = stop
    # 4순위: 보조 트레일링 — 래칫/어깨(1순위)와 별도. TAIL_TRAIL_PCT>0 일 때만 (돌파 BREAKOUT_TRAIL_* 와 동일)
    #   trail_arm_pct: 고점이 진입×(1+arm) 도달 후에만 무장 (0=즉시 무장). 어깨 미발동 구간의 팝 후 되돌림 보호.
    if not reason:
        trail_pct = abs(float(params.get("trail_pct", 0.0) or 0.0))
        trail_arm_pct = abs(float(params.get("trail_arm_pct", 0.0) or 0.0))
        if trail_pct > 0 and ep > 0:
            trail_arm_line = ep * (1.0 + trail_arm_pct)
            if max_p > ep and max_p >= trail_arm_line:
                trail_line = max_p * (1.0 - trail_pct)
                if lo > 0 and lo <= trail_line:
                    reason = "트레일컷"
                    exit_price = trail_line
    # 5순위: 금액 손실컷 — 1순위 어깨/래칫 미발동(수익% 문턱 못 넘음)일 때만
    if (
        not reason
        and not trail_armed
        and profit_val <= -max_loss_krw
        and drop_pct >= min_drop_pct
    ):
        reason = "금액손실컷"
        exit_price = ep - (max_loss_krw / qty) if qty > 0 else lo
    # 6순위: 시간컷 — 돌파와 동일(분 단위), 기본 OFF. 어깨·손절·익절 뒤, 장마감 앞
    if not reason:
        max_hold_min = _tail_max_hold_minutes(params)
        if max_hold_min > 0:
            held = _tail_minutes_held(position, current_candle)
            if held is not None and held >= max_hold_min:
                reason = "시간컷"
                exit_price = cl
    # 7순위: 장 마감 강제 청산
    if not reason and is_eod:
        reason = "장마감"
        exit_price = cl

    if reason:
        return (reason, exit_price)
    return None


def resolve_tail_invest_amount_krw(params: Dict[str, Any]) -> int:
    """
    실매 ``resolve_invest_amount_krw`` + ``TailCatchStrategy`` 종목당 상한과 동일.
    1회 매수 금액 = slot_money (MAX_LOSS/sl 로 투자금을 줄이지 않음).
    """
    slot = int(float(params.get("slot_money") or params.get("invest_amount") or 0))
    if slot <= 0:
        try:
            from kis_trader.utils.env import get_env_int
            slot = get_env_int("TAIL_SLOT_MONEY", 3_000_000)
        except Exception:
            slot = 3_000_000
    hard_cap = int(float(params.get("short_max_buy_amount") or params.get("per_stock_cap") or 0))
    if hard_cap <= 0:
        try:
            from kis_trader.utils.env import get_env_int
            hard_cap = get_env_int("TAIL_MAX_BUY_AMOUNT", 0)
        except Exception:
            hard_cap = 0
    if hard_cap > 0 and slot > hard_cap:
        return hard_cap
    return slot


def _portfolio_exposure_krw(portfolio: Dict[str, Dict[str, Any]]) -> float:
    """동시 보유 매입금 합 (실매 UPDOW 노출·SHORT 총한도와 동일 개념)."""
    total = 0.0
    for p in portfolio.values():
        total += float(p.get("entry_price") or 0) * int(p.get("qty") or 1)
    return total


def _last_close_at_or_before(
    candles: List[Dict],
    t: str,
    default: float,
) -> float:
    """시계 ``t`` 이하 마지막 확정 종가 — EOD 틱 없을 때 실매 장마감가 근사."""
    key = str(t or "")[:12]
    last = float(default or 0)
    if not key:
        return last
    for c in candles:
        ct = str(c.get("candle_time") or "")[:12]
        if not ct:
            continue
        if ct > key:
            break
        px = float(c.get("close") or 0)
        if px > 0:
            last = px
    return last


def _inject_tail_eod_timeline_minutes(
    all_times_set: Set[str],
    params: Dict[str, Any],
    period_ymd: str,
) -> None:
    """
    실매 DB ``TAIL_EOD_ENABLED`` / ``TAIL_EOD_HM`` (params eod_*) 토글을 백테 시계에 반영.
    3분봉 격자에 EOD 분이 없어도, 실매 벽시계처럼 해당 HHMM에 청산 판정 가능.
    """
    enabled, eod_hm = resolve_strategy_eod_params(params, "TAIL")
    if not enabled:
        return
    hh, mm = parse_eod_hm(eod_hm)
    days: Set[str] = set()
    py = str(period_ymd or "")[:8]
    if len(py) == 8 and py.isdigit():
        days.add(py)
    for t in all_times_set:
        d = str(t or "")[:8]
        if len(d) == 8 and d.isdigit():
            if py and d < py:
                continue
            days.add(d)
    for d in days:
        all_times_set.add(f"{d}{hh:02d}{mm:02d}")



def _tail_max_stocks_from_params(params: Dict[str, Any]) -> int:
    for key in ("max_stocks", "short_max_stocks"):
        v = params.get(key)
        if v not in (None, "", 0):
            return max(1, int(v))
    try:
        from kis_trader.utils.env import get_env_int
        n = get_env_int("TAIL_MAX_STOCKS", 3)
        return max(1, int(n))
    except Exception:
        return 3


def _tail_total_budget_from_params(params: Dict[str, Any]) -> float:
    """0 이하면 호출측에서 max_stocks×slot_money 로 유도."""
    for key in ("total_budget_krw", "short_total_budget_krw"):
        v = params.get(key)
        if v not in (None, ""):
            try:
                return float(v)
            except (TypeError, ValueError):
                pass
    try:
        from kis_trader.utils.env import get_env_int
        cap = get_env_int("TAIL_TOTAL_BUDGET_KRW", 0)
        if cap > 0:
            return float(cap)
    except Exception:
        pass
    return 0.0


def _tail_min_invest_ratio_of_slot(params: Dict[str, Any]) -> float:
    """1회 slot 대비 최소 투입 비율 — 미만이면 매수 스킵 (실매: 풀 slot 주문 후 예수금 부족 시만 축소)."""
    v = params.get("min_invest_ratio_of_slot")
    if v not in (None, ""):
        return max(0.01, min(1.0, float(v)))
    try:
        from kis_trader.utils.env import get_env_float
        return max(0.01, min(1.0, get_env_float("TAIL_MIN_INVEST_RATIO_OF_SLOT", 0.90)))
    except Exception:
        return 0.90


def _tail_target_qty_and_cost(entry_price: float, slot_money: float) -> Tuple[int, float]:
    """실매와 동일: 1회 투자금(slot) 기준 목표 수량·매입금."""
    from kis_trader.utils.position_sizing import invest_qty_for_price

    qty = invest_qty_for_price(entry_price, slot_money)
    if qty < 1:
        return 0, 0.0
    return qty, qty * entry_price


def _vol_capped_qty(qty: int, candle: Dict[str, Any], cap_pct: float) -> int:
    """진입봉 거래량×cap_pct% 로 체결 가능 주수 제한 (실매 유동성 제약 근사).

    cap_pct<=0 → 제한 없음. 거래량 정보 없으면(0) 기존 동작 유지(보수적). 반환 0=미체결.
    """
    if cap_pct <= 0:
        return qty
    try:
        vol = int(float(candle.get("volume") or 0))
    except (TypeError, ValueError):
        vol = 0
    if vol <= 0:
        return qty
    fillable = int(vol * cap_pct / 100.0)
    return min(qty, fillable)


def _buy_priority_key(
    code: str,
    uni_codes: Optional[List[str]],
) -> Tuple[int, str]:
    """실매 후보 순회 순서 근사 — 유니버스 편입 순서(index), 없으면 code 정렬."""
    if uni_codes is None:
        return (0, code)
    try:
        return (uni_codes.index(code), code)
    except ValueError:
        return (999999, code)


def _universe_codes_at(
    t: str,
    slot_key: str,
    universe_timeline: Optional[Any],
    universe_by_slot: Optional[Dict[str, List[str]]],
) -> Optional[List[str]]:
    """그 시각(봉 마감초) 유효 유니버스 코드 리스트 (돌파·모멘텀 공통 로직).

    - ``universe_timeline`` (초단위, 실매 get_universe_at 정합) 우선 — 봉 마감(HH:MM:59)
      직전 최신 조건검색 스냅샷. strict lag(1분 지연) 없이 실매와 동일 시점 조회.
    - 없으면 1분 슬롯(``universe_by_slot``) 폴백. 둘 다 없으면 None(전종목·필터없음).
    """
    if universe_timeline is not None:
        return universe_timeline.codes_at(str(t)[:12] + "59")
    if universe_by_slot is not None:
        return universe_by_slot.get(slot_key, [])
    return None


def _fill_portfolio_align_entry(
    *,
    portfolio: Dict[str, Dict[str, Any]],
    ctx: Dict[str, Any],
    code: str,
    pe: Dict[str, Any],
    t: str,
    ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]],
    tick_tf: int,
    use_ticks: bool,
    max_stocks: int,
    slot_money: float,
    total_budget: float,
    min_invest_ratio: float,
    vol_fill_cap_pct: float,
    tick_fill_stats: Dict[str, int],
) -> Tuple[bool, int, int]:
    """align 예약 진입 체결 — 실매: 진입봉 확정 직후 시장가."""
    skipped_micro = 0
    skipped_vol = 0
    if code in portfolio or not pe:
        return False, skipped_micro, skipped_vol
    if len(portfolio) >= max_stocks:
        return False, skipped_micro, skipped_vol

    entry_price = float(pe.get("entry_price") or 0)
    entry_src = "ohlc_open"
    # 중분 편입 시가 애매 — pe 생성 시 enroll_et 기록, 여기서 최종 가드
    from kis_trader.engine.mid_enroll_entry_gate import is_entry_open_ambiguous
    _ebk = str(pe.get("entry_bar_key") or t)[:12]
    _en = pe.get("enroll_et")
    if _en and is_entry_open_ambiguous(
        _ebk, _en, tf_min=max(1, int(tick_tf or 1)),
    ):
        return False, skipped_micro, skipped_vol
    if use_ticks:
        bar_key = str(pe.get("entry_bar_key") or t)[:12]
        bar_ticks = collect_bar_ticks(ticks_by_code, code, bar_key, tick_tf)
        if bar_ticks:
            entry_price, align_src = align_entry_price_from_ticks(bar_ticks, entry_price)
            entry_src = str(align_src or "ws_ticks")
            if align_src in tick_fill_stats:
                tick_fill_stats[align_src] += 1
            else:
                tick_fill_stats["ws_ticks"] = tick_fill_stats.get("ws_ticks", 0) + 1
        else:
            tick_fill_stats["ohlc_open"] = tick_fill_stats.get("ohlc_open", 0) + 1
    if entry_price <= 0:
        return False, skipped_micro, skipped_vol

    exposure = _portfolio_exposure_krw(portfolio)
    remaining = max(0.0, total_budget - exposure)
    target_qty, target_cost = _tail_target_qty_and_cost(entry_price, float(slot_money))
    min_required = target_cost * min_invest_ratio
    if target_qty < 1 or remaining < min_required:
        return False, 1, 0

    invest = min(float(slot_money), remaining, target_cost)
    qty = int(invest / entry_price)
    if qty < 1:
        return False, 1, 0

    if vol_fill_cap_pct > 0:
        _eidx = ctx["time_index"].get(str(pe.get("entry_bar_key") or t)[:12])
        if _eidx is None:
            _eidx = ctx["time_index"].get(t)
        _ebar = ctx["candles"][_eidx] if _eidx is not None else {}
        capped = _vol_capped_qty(qty, _ebar, vol_fill_cap_pct)
        if capped < qty:
            if capped < 1:
                return False, 0, 1
            qty = capped

    cost = qty * entry_price
    if cost < min_required:
        return False, 1, 0
    if exposure + cost > total_budget + 1e-6:
        return False, 1, 0

    portfolio[code] = {
        "entry_price": entry_price,
        "entry_time": t,
        "stop": pe["stop"],
        "target": pe["target"],
        "max_price": entry_price,
        "session_low": entry_price,
        "qty": qty,
        "entry_source": entry_src,
        "ob_source": str(pe.get("backtest_ob_source") or ""),
    }
    return True, 0, 0


def run_tail_backtest_portfolio(
    candles_by_code: Dict[str, List[Dict]],
    params: Dict[str, Any],
    universe_by_slot: Optional[Dict[str, List[str]]] = None,
    ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]] = None,
    orderbook_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
    program_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
) -> List[Dict]:
    """
    시각순 포트폴리오 백테스트 — 실매 BaseStrategy 제약 근사.

    - 모든 종목 봉을 ``candle_time`` 순으로 처리 (종목별 독립 합산 아님)
    - ``max_stocks``: 동시 보유 종목 수 (SHORT_MAX_STOCKS / MAX_STOCKS)
    - ``total_budget_krw``: 동시 보유 매입금 합 상한 (0 → max_stocks×slot_money)
    - 1시각(봉)당 신규 매수 1건 (실매 1루프 1매수)
    - 1회 투자금 = ``slot_money`` (실매 resolve_invest_amount_krw)
    """
    rsi_period = int(params.get("rsi_period", 14))
    min_bars = rsi_period + 5
    max_stocks = _tail_max_stocks_from_params(params)
    slot_money = resolve_tail_invest_amount_krw(params)
    total_budget = _tail_total_budget_from_params(params)
    if total_budget <= 0:
        total_budget = float(max_stocks * slot_money)
    min_invest_ratio = _tail_min_invest_ratio_of_slot(params)
    # 종목 일일 손익 게이트용 — 실매 realized_pnl 과 동일 net 기준으로 온라인 누적 (수수료/세금 반영)
    gate_fee_rate = float(params.get("fee_rate", 0.00015) or 0.00015)
    gate_sell_tax = float(params.get("sell_tax", 0.0018) or 0.0018)
    skipped_micro_buys = 0
    # 체결량 상한(유동성 제약 근사) — 진입봉 거래량×N% 까지만 체결. 0=OFF(동작 불변).
    # 실매 시장가 IOC 가 호가에 있는 만큼만 체결되는 것을 봉단위로 근사 (소형주 미체결↑).
    vol_fill_cap_pct = float(params.get("backtest_vol_fill_cap_pct", 0) or 0)
    skipped_vol_unfilled = 0
    tick_tf = tail_timeframe_min(params)
    use_ticks = bool(ticks_by_code) and tail_backtest_wants_tick_replay(params)
    use_tick_exit = bool(ticks_by_code) and tail_backtest_use_tick_exit(params)
    tick_fallback_ohlc = tail_backtest_tick_fallback_ohlc(params)
    tick_poll_ms = backtest_tick_poll_ms(params, strategy_env="TAIL_BACKTEST_POLL_MS")
    tick_sell_slip = backtest_sell_slip_pct(params, strategy_env="TAIL_BACKTEST_SELL_SLIP_PCT")
    tick_fill_stats = {"ws_ticks": 0, "ohlc_low": 0, "ohlc_open": 0}
    tick_exit_count = 0
    ohlc_exit_count = 0
    attach_indicator_caches_to_params(params, candles_by_code)

    # 종목별 컨텍스트
    ctx_by_code: Dict[str, Dict[str, Any]] = {}
    all_times_set = set()
    period_ymd = str(params.get("_bt_period_start_ymd") or "")[:8]
    universe_timeline = params.get("_universe_timeline")
    for code, raw_rows in candles_by_code.items():
        if len(raw_rows) < min_bars:
            continue
        candles = [dict(r) for r in raw_rows]
        closes = [float(c["close"]) for c in candles]
        ctx_by_code[code] = {
            "code": code,
            "candles": candles,
            "closes": closes,
            "rsis": compute_rsi_series(closes, rsi_period),
            "ma20s": compute_sma_series(closes, 20),
            "atrs": compute_atr_series(candles, 14),
            "time_index": {c["candle_time"]: idx for idx, c in enumerate(candles)},
            "last_exit_dt": {},
            "daily_cnt": {},
            "daily_pnl": {},
            "pending_entry": None,
            "pending_limit": None,
        }
        for c in candles:
            ct = c["candle_time"]
            # 웜업(전일) 봉은 지표용 — 매매 시계에는 기간일만
            if period_ymd and str(ct)[:8] < period_ymd:
                continue
            all_times_set.add(ct)
            # align: 진입 체결은 봉 **마감** 시각(실매 확정봉 직후) — 시계에 마감분 추가
            _bar_close = align_entry_execute_time(ct, tick_tf)
            if _bar_close:
                if not period_ymd or str(_bar_close)[:8] >= period_ymd:
                    all_times_set.add(_bar_close)

    # 편입·재편입 시각을 시계에 넣어, 봉 경계가 아니어도 실매처럼 즉시 매수 판정
    enter_mins = _universe_enter_minutes(
        universe_by_slot, universe_timeline, set(ctx_by_code.keys()),
    )
    for _m in enter_mins:
        if _m and (not period_ymd or str(_m)[:8] >= period_ymd):
            all_times_set.add(_m)

    # 실매 EOD 토글·시각(DB) → 백테 공유 시계에 EOD 분 주입 (종목별 3분봉 키 유무와 무관)
    _inject_tail_eod_timeline_minutes(all_times_set, params, period_ymd)

    all_times = sorted(all_times_set)
    portfolio: Dict[str, Dict[str, Any]] = {}
    all_trades: List[Dict] = []

    # 초단위 유니버스 타임라인 (실매 get_universe_at 정합, 돌파·모멘텀 공통). 없으면 1분 슬롯 폴백.
    # 실매 후보 하드캡(SHORT_CAND_LIMIT) — 백테도 그 시각 유니버스 상위 N개만 검사해야
    # "실매는 20개만 보는데 백테는 그 시각 풀 전체를 본다" 는 정합 오차가 사라진다.
    cand_limit = resolve_tail_cand_limit(params)
    inclusion_lookback_on = _to_bool(
        params.get("backtest_inclusion_lookback", True), True,
    )
    if "backtest_inclusion_lookback" not in params:
        from kis_trader.utils.env import get_env_bool
        inclusion_lookback_on = get_env_bool("TAIL_BT_INCLUSION_LOOKBACK", True)
    live_sig_lookback = max(1, int(params.get("live_signal_lookback_bars", 1)))
    inclusion_entries = 0
    prev_uni_set: Optional[Set[str]] = None

    from kis_trader.backtest.backtest_env_timeline import apply_env_timeline_at

    # 백테 CLI/웹 잡 진행률 (매매 로직 불변 — 파일에 pct만 기록)
    _prog_path = str(params.get("_bt_progress_file") or "").strip()
    _prog_base = float(params.get("_bt_progress_base_pct") or 45)
    _prog_span = float(params.get("_bt_progress_span_pct") or 50)
    _prog_n = max(1, len(all_times))
    _prog_last_i = -1

    def _emit_bt_progress(i: int) -> None:
        nonlocal _prog_last_i
        if not _prog_path:
            return
        # 과도한 디스크 쓰기 방지
        if i != 0 and i != _prog_n - 1 and (i - _prog_last_i) < max(25, _prog_n // 40):
            return
        _prog_last_i = i
        try:
            pct = _prog_base + _prog_span * (float(i + 1) / float(_prog_n))
            payload = {
                "pct": int(max(0, min(99, round(pct)))),
                "phase": "engine",
                "message": f"시계 {i + 1}/{_prog_n}",
                "step": int(i + 1),
                "total": int(_prog_n),
                "ts": __import__("time").time(),
            }
            from pathlib import Path as _P
            _p = _P(_prog_path)
            _tmp = _p.with_suffix(".tmp")
            _tmp.write_text(
                __import__("json").dumps(payload, ensure_ascii=False),
                encoding="utf-8",
            )
            _tmp.replace(_p)
        except Exception:
            pass

    for _ti, t in enumerate(all_times):
        _emit_bt_progress(_ti)
        if apply_env_timeline_at(params, t, "SHORT"):
            max_stocks = _tail_max_stocks_from_params(params)
            slot_money = resolve_tail_invest_amount_krw(params)
            total_budget = _tail_total_budget_from_params(params)
            if total_budget <= 0:
                total_budget = float(max_stocks * slot_money)
            cand_limit = resolve_tail_cand_limit(params)

        slot_key = _slot_key(t, params.get("scan_interval_min", 1))
        uni_codes = _universe_codes_at(t, slot_key, universe_timeline, universe_by_slot)
        if uni_codes is not None and cand_limit > 0 and len(uni_codes) > cand_limit:
            uni_codes = uni_codes[:cand_limit]
        uni_set = set(uni_codes) if uni_codes is not None else None

        # ── Phase 0-incl: 유니버스 신규·재편입 직후 lookback (실매 check_buy_signal_live) ──
        if (
            inclusion_lookback_on
            and uni_set is not None
            and not is_limit_atr_entry(short_entry_mode(params))
        ):
            # 실매: EXIT 후 RE-ENTER 때도 즉시 매수 검사. 최초 편입만 보면 샘표류 누락.
            if prev_uni_set is None:
                newly = set(uni_set)  # 첫 스냅샷 = 전원 신규 편입으로 취급
            else:
                newly = uni_set - prev_uni_set
            if newly and len(portfolio) < max_stocks:
                incl_cands: List[Tuple[Tuple[int, str], str, Dict[str, Any]]] = []
                for code in newly:
                    if code in portfolio or code not in ctx_by_code:
                        continue
                    ctx = ctx_by_code[code]
                    if ctx.get("pending_entry") or ctx.get("pending_limit"):
                        continue
                    entry_i = _last_closed_bar_index(ctx["candles"], t, tick_tf)
                    if entry_i < 19:
                        continue
                    day = str(t)[:8]
                    if ctx["daily_cnt"].get(day, 0) >= int(params.get("max_daily", 3)):
                        continue
                    eval_params = dict(params)
                    ic = get_indicator_cache_from_params(params, code)
                    if ic is not None:
                        eval_params["_indicator_cache"] = ic
                    state = {
                        "daily_cnt": ctx["daily_cnt"].get(day, 0),
                        "last_exit_dt": ctx["last_exit_dt"].get(day),
                        "daily_pnl_krw": ctx["daily_pnl"].get(day, 0.0),
                    }
                    reject, _msg, sig = _eval_live_align_lookback(
                        ctx["candles"], entry_i, eval_params, state,
                        lookback=live_sig_lookback,
                    )
                    if reject or not sig:
                        continue
                    ent = ctx["candles"][entry_i]
                    entry_price = float(sig.get("entry_price") or ent.get("open") or 0)
                    if entry_price <= 0:
                        continue
                    atr = (
                        ctx["atrs"][entry_i]
                        if entry_i < len(ctx["atrs"]) and ctx["atrs"][entry_i] is not None
                        else entry_price * 0.01
                    )
                    stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)
                    exec_t = align_entry_execute_time(ent.get("candle_time"), tick_tf) or t
                    from kis_trader.engine.mid_enroll_entry_gate import (
                        bt_resolve_enroll,
                        is_entry_open_ambiguous,
                    )
                    _ebk = str(ent.get("candle_time") or "")[:12]
                    _enroll = bt_resolve_enroll(
                        code, t,
                        universe_timeline=universe_timeline,
                        universe_by_slot=universe_by_slot,
                        params=params,
                    )
                    if _enroll and is_entry_open_ambiguous(
                        _ebk, _enroll, tf_min=tick_tf, params=params,
                    ):
                        continue
                    pri = _buy_priority_key(code, uni_codes)
                    incl_cands.append((pri, code, {
                        "entry_time": exec_t,
                        "entry_price": entry_price,
                        "stop": stop_p,
                        "target": target_p,
                        "entry_bar_key": ent.get("candle_time"),
                        "enroll_et": _enroll,
                        "from_inclusion": True,
                        "backtest_ob_source": sig.get("backtest_ob_source"),
                    }))
                if incl_cands:
                    incl_cands.sort(key=lambda x: x[0])
                    _pri, pick_code, pe = incl_cands[0]
                    # 편입·재편입 직후 체결:
                    # - entry_time == t : 진입봉 마감 시각과 동일 → 즉시 체결
                    # - entry_time < t  : 재편입 lookback 이 과거 신호봉을 찾은 경우
                    #   (실매: EXIT 후 RE-ENTER 시 과거 align 진입가로 즉시 시장가).
                    #   과거 entry_time 을 pending 으로 남기면 Phase0b 가 영구 미체결 → 금지.
                    # - entry_time > t  : 아직 진입봉 미도래 → pending
                    _ctx_pick = ctx_by_code[pick_code]
                    _pe_t = str(pe.get("entry_time") or "")[:12]
                    _now_t = str(t)[:12]
                    _fill_now = (not _pe_t) or (_pe_t <= _now_t)
                    if _fill_now:
                        _filled, _sm, _sv = _fill_portfolio_align_entry(
                            portfolio=portfolio,
                            ctx=_ctx_pick,
                            code=pick_code,
                            pe=pe,
                            t=t,
                            ticks_by_code=ticks_by_code,
                            tick_tf=tick_tf,
                            use_ticks=use_ticks,
                            max_stocks=max_stocks,
                            slot_money=float(slot_money),
                            total_budget=float(total_budget),
                            min_invest_ratio=min_invest_ratio,
                            vol_fill_cap_pct=vol_fill_cap_pct,
                            tick_fill_stats=tick_fill_stats,
                        )
                        skipped_micro_buys += _sm
                        skipped_vol_unfilled += _sv
                        if _filled:
                            inclusion_entries += 1
                        else:
                            # 잔여한도 등으로 즉시 실패 시에만 pending (시각은 현재 t 로 맞춤)
                            pe = dict(pe)
                            pe["entry_time"] = t
                            _ctx_pick["pending_entry"] = pe
                    else:
                        _ctx_pick["pending_entry"] = pe
                        inclusion_entries += 1

        if uni_set is not None:
            prev_uni_set = set(uni_set)
        # ── Phase 0a: ATR 지정가 체결 (유효 봉 low ≤ limit) ──
        for code, ctx in ctx_by_code.items():
            pl = ctx.get("pending_limit")
            if not pl or code in portfolio:
                continue
            idx = ctx["time_index"].get(t)
            if idx is None:
                continue
            sig_i = int(pl.get("signal_i", -1))
            if idx <= sig_i:
                continue
            vu = str(pl.get("valid_until") or "")[:12]
            if vu and str(t)[:12] > vu:
                ctx["pending_limit"] = None
                continue
            c = ctx["candles"][idx]
            bar_ticks = (
                collect_bar_ticks(ticks_by_code, code, t, tick_tf) if use_ticks else []
            )
            fill, fill_src = try_limit_fill_on_bar_with_ticks(
                c,
                float(pl.get("limit_price") or 0),
                float(pl.get("fill_slip") or 0),
                ticks=bar_ticks,
                params=params,
            )
            if not fill or fill <= 0:
                continue
            if fill_src in tick_fill_stats:
                tick_fill_stats[fill_src] += 1
            if len(portfolio) >= max_stocks:
                continue
            exposure = _portfolio_exposure_krw(portfolio)
            remaining = max(0.0, total_budget - exposure)
            target_qty, target_cost = _tail_target_qty_and_cost(fill, float(slot_money))
            min_required = target_cost * min_invest_ratio
            if target_qty < 1 or remaining < min_required:
                skipped_micro_buys += 1
                continue
            invest = min(float(slot_money), remaining, target_cost)
            qty = int(invest / fill)
            if qty < 1:
                skipped_micro_buys += 1
                continue
            # 유동성 제약: 진입봉 거래량×cap% 까지만 체결 (실매 IOC 미체결 근사)
            capped = _vol_capped_qty(qty, c, vol_fill_cap_pct)
            if capped < qty:
                if capped < 1:
                    skipped_vol_unfilled += 1
                    continue
                qty = capped
            cost = qty * fill
            if cost < min_required or exposure + cost > total_budget + 1e-6:
                skipped_micro_buys += 1
                continue
            portfolio[code] = {
                "entry_price": fill,
                "entry_time": t,
                "stop": pl["stop"],
                "target": pl["target"],
                "max_price": fill,
                "session_low": fill,
                "qty": qty,
                "entry_source": str(fill_src or "ohlc_low"),
                "ob_source": str(pl.get("backtest_ob_source") or ""),
            }
            ctx["pending_limit"] = None
            break

        # ── Phase 0b: 예약 진입 (align — 진입봉 확정 직후 첫 틱 / 편입 lookback) ──
        pending_codes = [
            code for code, ctx in ctx_by_code.items()
            if ctx.get("pending_entry") and ctx["pending_entry"].get("entry_time") == t
        ]
        pending_codes.sort(
            key=lambda c: _buy_priority_key(c, uni_codes),
        )
        for code in pending_codes:
            ctx = ctx_by_code[code]
            pe = ctx.pop("pending_entry", None)
            if not pe or code in portfolio:
                continue
            if len(portfolio) >= max_stocks:
                break
            _filled, _sm, _sv = _fill_portfolio_align_entry(
                portfolio=portfolio,
                ctx=ctx,
                code=code,
                pe=pe,
                t=t,
                ticks_by_code=ticks_by_code,
                tick_tf=tick_tf,
                use_ticks=use_ticks,
                max_stocks=max_stocks,
                slot_money=float(slot_money),
                total_budget=float(total_budget),
                min_invest_ratio=min_invest_ratio,
                vol_fill_cap_pct=vol_fill_cap_pct,
                tick_fill_stats=tick_fill_stats,
            )
            skipped_micro_buys += _sm
            skipped_vol_unfilled += _sv
            if _filled:
                break  # 1시각 1매수

        # ── Phase 1: 보유 종목 청산 ──
        for code in list(portfolio.keys()):
            ctx = ctx_by_code.get(code)
            if ctx is None:
                continue
            idx = ctx["time_index"].get(t)
            candles = ctx["candles"]
            day = t[:8]
            pos = portfolio[code]

            # 편입 lookback 등으로 타임라인에 3분봉 키가 아닌 분(예: 13:23)이
            # 들어오면 time_index miss. 실매는 WS 틱마다 청산하므로, 해당 분
            # 틱만으로 청산 검사(봉 OHLC 폴백 없음 — 유령 청산 방지).
            # 단 EOD는 실매 벽시계와 동일하게 ``t`` 기준(DB eod_enabled/eod_hm)으로 추적.
            is_eod = is_strategy_eod_bar(t, params, "TAIL")
            ep0 = float(pos.get("entry_price") or 0)
            if idx is None:
                if not use_tick_exit and not is_eod:
                    continue
                bar_ticks = (
                    collect_minute_ticks(ticks_by_code, code, t)
                    if use_tick_exit else None
                )
                if not bar_ticks and not is_eod:
                    continue
                # 직전 종가 = EOD 플랫 체결가용. max_price 에는 넣지 않음(실매=틱/WS만).
                last_px = _last_close_at_or_before(candles, t, ep0)
                if last_px <= 0:
                    last_px = ep0
                cur_c_info = {
                    "open": last_px,
                    "high": float(pos.get("max_price") or last_px),
                    "low": float(pos.get("session_low") or last_px),
                    "close": last_px,
                    "candle_time": t,
                }
            else:
                c = candles[idx]
                hi = float(c["high"])
                lo = float(c["low"])
                cl = float(c["close"])
                op = float(c["open"])
                last_px = cl if cl > 0 else ep0
                cur_c_info = {
                    "open": op,
                    "high": hi,
                    "low": lo,
                    "close": cl,
                    "candle_time": t,
                }
                bar_ticks = (
                    collect_bar_ticks(ticks_by_code, code, t, tick_tf)
                    if use_tick_exit else None
                )
                # 틱 청산 ON: 봉 high/holding_peak 로 max_price 선반영 금지
                # (3분봉 high 에 미래 분 고점 포함 → 래칫 look-ahead, 실매와 분기).
                # 틱 OFF(레거시 OHLC 청산)일 때만 봉 high 로 고점 갱신.
                if not use_tick_exit:
                    max_p = max(float(pos.get("max_price", 0) or 0), hi)
                    hp = float(c.get("holding_peak") or 0)
                    if hp > 0:
                        max_p = max(max_p, hp)
                    pos["max_price"] = max_p

            # OHLC 폴백: env/params 명시 ON 만. EOD 로 강제하지 않음(정합 깨짐).
            res = resolve_backtest_sell(
                pos,
                cur_c_info,
                params,
                is_eod=is_eod,
                sell_fn=check_sell_signal_live,
                low_mode="session_low",
                ticks=bar_ticks,
                use_tick_exit=use_tick_exit,
                tick_fallback_ohlc=bool(tick_fallback_ohlc),
                poll_ms=tick_poll_ms,
                slip_pct=tick_sell_slip,
            )
            if not res:
                # 틱으로 미청산 + EOD: 실매처럼 현재가(직전종가)로 장마감만 — 봉 고저로 래칫/손절 재계산 금지
                if is_eod and last_px > 0:
                    reason, exit_price, sell_time, exit_src = (
                        "장마감", last_px, t, "eod_flat",
                    )
                else:
                    continue
            else:
                reason, exit_price, sell_time, _hold_min, exit_src = res
            if exit_src == "ws_ticks":
                tick_exit_count += 1
            elif exit_src == "eod_flat":
                pass
            else:
                ohlc_exit_count += 1
            peak_px = float(pos.get("max_price") or exit_price or 0)
            all_trades.append({
                "code": code,
                "entry_time": pos["entry_time"],
                "exit_time": sell_time or t,
                "entry": round(pos["entry_price"]),
                "exit": round(exit_price),
                "pnl": 0,
                "reason": reason,
                "hold_min": 0,
                "peak_price": round(peak_px),
                "qty": pos.get("qty", 1),
                "entry_source": str(pos.get("entry_source") or ""),
                "exit_source": str(exit_src or ""),
                "ob_source": str(pos.get("ob_source") or ""),
            })
            ctx["last_exit_dt"][day] = _t2dt(sell_time or t)
            ctx["daily_cnt"][day] = ctx["daily_cnt"].get(day, 0) + 1
            # 종목 일일 손익 게이트용 net pnl 누적 — attach_tail_trade_pnl 과 동일 공식(수수료/세금)
            _qty = int(pos.get("qty", 1) or 1)
            _ep, _xp = float(pos["entry_price"]), float(exit_price)
            _net_pnl = (
                (_xp - _ep) * _qty
                - (_ep + _xp) * _qty * gate_fee_rate
                - _xp * _qty * gate_sell_tax
            )
            ctx["daily_pnl"][day] = ctx["daily_pnl"].get(day, 0.0) + _net_pnl
            del portfolio[code]

        # ── Phase 2: 신규 매수 신호 (align=진입봉 확정 직후 / limit=신호봉 이후 지정가) ──
        if len(portfolio) >= max_stocks:
            continue
        exposure = _portfolio_exposure_krw(portfolio)
        if exposure >= total_budget - 1e-6:
            continue

        time_start_hm = int(params.get("time_start_hm", 930))
        time_end_hm = int(params.get("time_end_hm", 1500))
        hm = int(t[8:12])

        candidates: List[Tuple[Tuple[int, str], str, Dict[str, Any]]] = []
        for code, ctx in ctx_by_code.items():
            if code in portfolio or ctx.get("pending_entry") or ctx.get("pending_limit"):
                continue
            candles = ctx["candles"]
            day = t[:8]

            if uni_set is not None:
                if code not in uni_set:
                    continue
            if hm < time_start_hm or hm > time_end_hm:
                continue
            if ctx["daily_cnt"].get(day, 0) >= int(params.get("max_daily", 3)):
                continue
            if day in ctx["last_exit_dt"]:
                elapsed = (_t2dt(t) - ctx["last_exit_dt"][day]).total_seconds() / 60
                if elapsed < float(params.get("cooldown_min", 15)):
                    continue

            eval_params = dict(params)
            ic = get_indicator_cache_from_params(params, code)
            if ic is not None:
                eval_params["_indicator_cache"] = ic
            if "skip_hts_scan_dupes" not in eval_params:
                eval_params["skip_hts_scan_dupes"] = resolve_tail_skip_hts_scan_dupes()
            state = {
                "daily_cnt": ctx["daily_cnt"].get(day, 0),
                "last_exit_dt": ctx["last_exit_dt"].get(day),
                "daily_pnl_krw": ctx["daily_pnl"].get(day, 0.0),
            }

            if is_limit_atr_entry(short_entry_mode(params)):
                idx = ctx["time_index"].get(t)
                if idx is None:
                    continue
                c = candles[idx]
                cl = float(c["close"])
                if cl <= 0:
                    continue
                inject_whipsaw_ticks_into_params(
                    eval_params,
                    ticks_by_code=ticks_by_code,
                    code=code,
                    bar_candle_time=t,
                    strategy="TAIL",
                    tf_min=tick_tf,
                )
                inject_trigger_snapshots_into_params(
                    eval_params,
                    orderbook_by_code=orderbook_by_code,
                    program_by_code=program_by_code,
                    code=code,
                    bar_candle_time=t,
                )
                reject, _msg, sig = _eval_tail_buy_at_index(candles, idx, eval_params, state)
                if reject or not sig:
                    continue
                atr = ctx["atrs"][idx] if ctx["atrs"][idx] is not None else cl * 0.01
                pri = _buy_priority_key(code, uni_codes)
                lp_cfg = tail_limit_params(params)
                sig_bar = candles[idx]
                anchor_px = resolve_limit_anchor_price(
                    lp_cfg["anchor"], sig_bar, candles, idx,
                )
                min_px = float(params.get("min_price", 1000.0))
                limit_px = compute_atr_limit_price(
                    anchor_px, atr, lp_cfg["mult"], min_price=min_px,
                )
                if limit_px <= 0:
                    continue
                stop_p, target_p = compute_tail_atr_prices(limit_px, atr, params)
                vu = limit_valid_until_bar_key(candles, idx, lp_cfg["valid_bars"])
                candidates.append((pri, code, {
                    "pending_limit": True,
                    "signal_i": idx,
                    "limit_price": limit_px,
                    "valid_until": vu,
                    "fill_slip": lp_cfg["fill_slip_pct"],
                    "stop": stop_p,
                    "target": target_p,
                    "backtest_ob_source": sig.get("backtest_ob_source"),
                }))
                continue

            # align — 진입봉 **마감** 시각에만 검사 (실매: 확정봉 직후 시장가)
            entry_i = _last_closed_bar_index(candles, t, tick_tf)
            if entry_i < 19:
                continue
            ent = candles[entry_i]
            close_t = align_entry_execute_time(ent.get("candle_time"), tick_tf)
            if not close_t or str(t)[:12] != str(close_t)[:12]:
                continue
            ent_ct = str(ent.get("candle_time") or "")
            inject_whipsaw_ticks_into_params(
                eval_params,
                ticks_by_code=ticks_by_code,
                code=code,
                bar_candle_time=ent_ct,
                strategy="TAIL",
                tf_min=tick_tf,
            )
            inject_trigger_snapshots_into_params(
                eval_params,
                orderbook_by_code=orderbook_by_code,
                program_by_code=program_by_code,
                code=code,
                bar_candle_time=ent_ct,
            )
            reject, _msg, sig = _eval_live_align_lookback(
                candles, entry_i, eval_params, state,
                lookback=live_sig_lookback,
            )
            if reject or not sig:
                continue
            cl = float(ent.get("close") or 0)
            atr = (
                ctx["atrs"][entry_i]
                if entry_i < len(ctx["atrs"]) and ctx["atrs"][entry_i] is not None
                else cl * 0.01
            )
            pri = _buy_priority_key(code, uni_codes)
            entry_price = float(sig.get("entry_price") or ent.get("open") or cl)
            if entry_price <= 0:
                continue
            stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)
            from kis_trader.engine.mid_enroll_entry_gate import (
                bt_resolve_enroll,
                is_entry_open_ambiguous,
            )
            _ebk = str(ent.get("candle_time") or "")[:12]
            _enroll = bt_resolve_enroll(
                code, t,
                universe_timeline=universe_timeline,
                universe_by_slot=universe_by_slot,
                params=params,
            )
            if _enroll and is_entry_open_ambiguous(
                _ebk, _enroll, tf_min=tick_tf, params=params,
            ):
                continue
            candidates.append((pri, code, {
                "entry_time": close_t,
                "entry_price": entry_price,
                "stop": stop_p,
                "target": target_p,
                "entry_bar_key": ent.get("candle_time"),
                "enroll_et": _enroll,
                "backtest_ob_source": sig.get("backtest_ob_source"),
            }))

        if not candidates:
            continue
        candidates.sort(key=lambda x: x[0])
        _pri, pick_code, pe = candidates[0]
        if pe.get("pending_limit"):
            ctx_by_code[pick_code]["pending_limit"] = pe
        else:
            _ctx_pick = ctx_by_code[pick_code]
            if str(pe.get("entry_time") or "")[:12] == str(t)[:12]:
                _filled, _sm, _sv = _fill_portfolio_align_entry(
                    portfolio=portfolio,
                    ctx=_ctx_pick,
                    code=pick_code,
                    pe=pe,
                    t=t,
                    ticks_by_code=ticks_by_code,
                    tick_tf=tick_tf,
                    use_ticks=use_ticks,
                    max_stocks=max_stocks,
                    slot_money=float(slot_money),
                    total_budget=float(total_budget),
                    min_invest_ratio=min_invest_ratio,
                    vol_fill_cap_pct=vol_fill_cap_pct,
                    tick_fill_stats=tick_fill_stats,
                )
                skipped_micro_buys += _sm
                skipped_vol_unfilled += _sv
                if not _filled:
                    _ctx_pick["pending_entry"] = pe
            else:
                _ctx_pick["pending_entry"] = pe

    skip_stats: Dict[str, Any] = {}
    if skipped_micro_buys:
        skip_stats["skipped_micro_buys"] = skipped_micro_buys
    if skipped_vol_unfilled:
        skip_stats["skipped_vol_unfilled"] = skipped_vol_unfilled
    if use_ticks:
        skip_stats["tick_entry_sources"] = dict(tick_fill_stats)
    if tick_exit_count or ohlc_exit_count:
        skip_stats["tick_exit_count"] = tick_exit_count
        skip_stats["ohlc_exit_count"] = ohlc_exit_count
    if inclusion_entries:
        skip_stats["inclusion_lookback_entries"] = inclusion_entries
    if skip_stats:
        params["_portfolio_skip_stats"] = skip_stats
    return all_trades


def run_tail_backtest_rust_experimental(
    codes_candles: Dict[str, List[Dict]],
    params: Dict[str, Any]
) -> List[Dict]:
    """꼬리잡기 매매 Rust 엔진 실험용 브릿지"""
    try:
        import kis_rust_core
    except ImportError:
        raise RuntimeError("kis_rust_core module is not available")
        
    def get_float(k, default=0.0):
        try: return float(params.get(k, default))
        except: return float(default)
        
    def get_int(k, default=0):
        try: return int(params.get(k, default))
        except: return int(default)
        
    def get_bool(k, default=False):
        v = params.get(k, default)
        if isinstance(v, bool): return v
        s = str(v).lower()
        if s in ("1", "true", "yes", "t", "y"): return True
        return False
        
    rust_params = kis_rust_core.TailParams(
        get_int("time_start_hm", 930),
        get_int("time_end_hm", 1500),
        get_float("cooldown_min", 15.0),
        get_int("max_daily", 3),
        get_float("min_price", 1000.0),
        get_float("rsi_threshold", 78.0),
        get_float("min_drop_rate", 0.03),
        get_float("min_recovery_ratio", 0.5),
        get_float("tail_pct_min", 0.003),
        get_bool("skip_hts_scan_dupes", True),
        abs(get_float("sl_pct", 0.03)),
        get_float("tp_pct", 0.05),
        get_float("trail_pct", 0.0),
        get_float("trail_arm_pct", 0.0),
        get_float("shoulder_min_high", 0.005),
        get_float("shoulder_cut_pct", 0.003),
        get_int("max_hold_bars", 0),
        get_float("max_loss_krw", 200000.0),
        get_float("min_drop_pct_for_loss_cut", 0.015),
    )
    
    import uuid
    import json
    
    session_id = f"exp_tail_{uuid.uuid4().hex[:8]}"
    
    try:
        candles_json = json.dumps(codes_candles)
        kis_rust_core.init_backtest_session_json(session_id, candles_json)
    except Exception as e:
        print(f"[ERROR] Failed to serialize or init rust session: {e}")
        return []
    
    all_trades = []
    try:
        r_trades = kis_rust_core.run_engine_trial_tail(session_id, rust_params)
        all_trades = [
            {
                "code": t.code,
                "buy_time": t.buy_time,
                "sell_time": t.sell_time,
                "buy_price": t.buy_price,
                "sell_price": t.sell_price,
                "entry_time": t.buy_time,
                "exit_time": t.sell_time,
                "entry": t.buy_price,
                "exit": t.sell_price,
                "exit_reason": t.reason,
                "profit_rate": t.pnl_pct,
                "qty": 1,
                "pnl": 0,
                "sell_reason": t.reason,
                "max_price": t.max_price,
                "rsi_entry": t.rsi_entry,
                "strategy": "SHORT"
            }
            for t in r_trades
        ]
    except Exception as e:
        print(f"[ERROR] run_tail_backtest_rust_experimental failed: {e}")
    finally:
        try:
            kis_rust_core.clear_backtest_session(session_id)
        except:
            pass
            
    return all_trades


def run_tail_backtest(
    candles_by_code: Dict[str, List[Dict]],
    params: Dict[str, Any],
    universe_by_slot: Optional[Dict[str, List[str]]] = None,
    ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]] = None,
    orderbook_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
    program_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
) -> List[Dict]:
    """
    백테스트 1회 실행. (backtest_web 및 tail_param_search 호출용)

    기본: ``portfolio_mode=true`` → 시각순 포트폴리오 (실매 MAX_STOCKS·총한도·slot_money).
    ``portfolio_mode=false`` → 레거시 종목별 독립 루프.
    """
    if params.get("use_rust", False):
        return run_tail_backtest_rust_experimental(candles_by_code, params)

    if _to_bool(params.get("portfolio_mode"), True):
        return run_tail_backtest_portfolio(
            candles_by_code, params, universe_by_slot,
            ticks_by_code=ticks_by_code,
            orderbook_by_code=orderbook_by_code,
            program_by_code=program_by_code,
        )

    # ── 레거시: 종목별 독립 시뮬 ──
    tick_tf = tail_timeframe_min(params)
    use_ticks = bool(ticks_by_code) and tail_backtest_wants_tick_replay(params)
    use_tick_exit = bool(ticks_by_code) and tail_backtest_use_tick_exit(params)
    tick_fallback_ohlc = tail_backtest_tick_fallback_ohlc(params)
    tick_poll_ms = backtest_tick_poll_ms(params, strategy_env="TAIL_BACKTEST_POLL_MS")
    tick_sell_slip = backtest_sell_slip_pct(params, strategy_env="TAIL_BACKTEST_SELL_SLIP_PCT")

    # 파라미터 준비
    min_drop_rate = float(params.get("min_drop_rate", 0.03))
    min_recovery_ratio = float(params.get("min_recovery_ratio", 0.5))
    max_rec_3m = float(params.get("max_rec_3m", 0.8))
    tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
    tail_pct_min = float(params.get("tail_pct_min", 0.003))
    shoulder_min_high = float(params.get("shoulder_min_high", 0.005))
    shoulder_cut_pct = float(params.get("shoulder_cut_pct", 0.003))
    rsi_period = int(params.get("rsi_period", 14))
    rsi_threshold = float(params.get("rsi_threshold", 78))
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    time_start_hm = int(params.get("time_start_hm", 930))
    time_end_hm = int(params.get("time_end_hm", 1500))
    cooldown_min = float(params.get("cooldown_min", 15))
    max_daily = int(params.get("max_daily", 3))

    # 방어 로직 (동적 계산용)
    stop_atr_mult = float(params.get("stop_atr_mult", 2.5))
    target_atr_mult = float(params.get("target_atr_mult", 8.0))
    max_loss_krw = int(params.get("max_loss_krw", 200000))
    risk_pct = float(params.get("risk_pct", 0.01))
    kelly_mult = float(params.get("kelly_mult", 0.25))
    capital = float(params.get("capital", 100000000.0))
    static_sl_pct = abs(float(params.get("sl_pct", 0.03)))
    gate_fee_rate = float(params.get("fee_rate", 0.00015) or 0.00015)
    gate_sell_tax = float(params.get("sell_tax", 0.0018) or 0.0018)

    all_trades: List[Dict] = []
    
    for code, candles in candles_by_code.items():
        if len(candles) < rsi_period + 5:
            continue
            
        # 벡터 연산으로 지표 선행 계산 (백테스트 속도 최적화)
        closes = [float(c["close"]) for c in candles]
        rsis = compute_rsi_series(closes, rsi_period)
        ma20s = compute_sma_series(closes, 20)
        atrs = compute_atr_series(candles, 14)

        position = None
        last_exit_dt: Dict[str, datetime] = {}
        daily_cnt: Dict[str, int] = {}
        daily_pnl: Dict[str, float] = {}
        cur_day = None
        running_open, running_high, running_low = 0.0, 0.0, 0.0

        i = rsi_period + 1
        while i < len(candles):
            c = candles[i]
            day = c["candle_time"][:8]
            hm = int(c["candle_time"][8:12])
            op = float(c["open"])
            hi = float(c["high"])
            lo = float(c["low"])
            cl = float(c["close"])

            # 일일 변수 초기화 및 갱신
            if day != cur_day:
                cur_day = day
                running_open = op
                running_high = hi
                running_low = lo if lo > 0 else hi
            else:
                running_high = max(running_high, hi)
                if lo > 0:
                    running_low = min(running_low, lo)

            is_eod = is_strategy_eod_bar(c["candle_time"], params, "TAIL")

            # ── 1. 청산 검사 (포지션 보유 중일 때) ──
            if position is not None:
                cur_c_info = {
                    "open": op,
                    "high": hi,
                    "low": lo,
                    "close": cl,
                    "candle_time": c["candle_time"],
                }
                bar_ticks = (
                    collect_bar_ticks(
                        ticks_by_code, code, c["candle_time"], tick_tf,
                    ) if use_tick_exit else None
                )
                # 틱 청산 ON: 봉 high 로 max_price 선반영 금지 (실매=틱/WS만)
                if not use_tick_exit:
                    max_p = max(position["max_price"], hi)
                    hp = float(c.get("holding_peak") or 0)
                    if hp > 0:
                        max_p = max(max_p, hp)
                    position["max_price"] = max_p
                res = resolve_backtest_sell(
                    position,
                    cur_c_info,
                    params,
                    is_eod=is_eod,
                    sell_fn=check_sell_signal_live,
                    low_mode="session_low",
                    ticks=bar_ticks,
                    use_tick_exit=use_tick_exit,
                    tick_fallback_ohlc=bool(tick_fallback_ohlc),
                    poll_ms=tick_poll_ms,
                    slip_pct=tick_sell_slip,
                )
                if not res and is_eod and cl > 0:
                    res = ("장마감", cl, c["candle_time"], 0.0, "eod_flat")
                
                if res:
                    reason, exit_price, sell_time, _hold_min, _exit_src = res
                    all_trades.append({
                        "code": code,
                        "entry_time": position["entry_time"],
                        "exit_time": sell_time or c["candle_time"],
                        "entry": round(position["entry_price"]),
                        "exit": round(exit_price),
                        "pnl": 0,
                        "reason": reason,
                        "hold_min": 0,
                        "peak_price": round(float(position.get("max_price") or exit_price)),
                        "qty": position.get("qty", 1),
                        "entry_source": str(position.get("entry_source") or ""),
                        "exit_source": str(_exit_src or ""),
                        "ob_source": str(position.get("ob_source") or ""),
                    })
                    last_exit_dt[day] = _t2dt(sell_time or c["candle_time"])
                    daily_cnt[day] = daily_cnt.get(day, 0) + 1
                    # 종목 일일 손익 게이트용 net pnl 누적 (포트폴리오 루프와 동일 공식)
                    _qty = int(position.get("qty", 1) or 1)
                    _ep, _xp = float(position["entry_price"]), float(exit_price)
                    _net_pnl = (
                        (_xp - _ep) * _qty
                        - (_ep + _xp) * _qty * gate_fee_rate
                        - _xp * _qty * gate_sell_tax
                    )
                    daily_pnl[day] = daily_pnl.get(day, 0.0) + _net_pnl
                    position = None
                i += 1
                continue

            # ── 2. 매수 검사 (포지션 없을 때, 유니버스 시뮬레이션 시 해당 슬롯 후보만) ──
            if universe_by_slot is not None:
                # 신봇 기본: 1분봉 == 슬롯 키 (TradeDBExt.get_universe_by_candle_time 키 포맷).
                slot_key = _slot_key(c["candle_time"], params.get("scan_interval_min", 1))
                if code not in universe_by_slot.get(slot_key, []):
                    i += 1
                    continue
            if cl <= 0 or running_open <= 0 or hm < time_start_hm or hm > time_end_hm:
                i += 1
                continue
            if daily_cnt.get(day, 0) >= max_daily:
                i += 1
                continue
            if day in last_exit_dt:
                elapsed = (_t2dt(c["candle_time"]) - last_exit_dt[day]).total_seconds() / 60
                if elapsed < cooldown_min:
                    i += 1
                    continue

            eval_params = dict(params)
            if "skip_hts_scan_dupes" not in eval_params:
                eval_params["skip_hts_scan_dupes"] = resolve_tail_skip_hts_scan_dupes()
            state = {
                "daily_cnt": daily_cnt.get(day, 0),
                "last_exit_dt": last_exit_dt.get(day),
                "daily_pnl_krw": daily_pnl.get(day, 0.0),
            }
            live_sig_lb = max(1, int(params.get("live_signal_lookback_bars", 1)))

            # ── 3. 매수 — limit_atr: 신호봉 i / align: 진입봉 확정(다음 봉 시각) ──
            if is_limit_atr_entry(short_entry_mode(params)):
                reject, _msg, sig = _eval_tail_buy_at_index(candles, i, eval_params, state)
                if reject or not sig:
                    i += 1
                    continue
                atr = atrs[i] if atrs[i] is not None else cl * 0.01
                lp_cfg = tail_limit_params(params)
                sig_bar = candles[i]
                anchor_px = resolve_limit_anchor_price(
                    lp_cfg["anchor"], sig_bar, candles, i,
                )
                min_px = float(params.get("min_price", 1000.0))
                limit_px = compute_atr_limit_price(
                    anchor_px, atr, lp_cfg["mult"], min_price=min_px,
                )
                if limit_px <= 0:
                    i += 1
                    continue
                vu = limit_valid_until_bar_key(candles, i, lp_cfg["valid_bars"])
                filled = False
                for j in range(i + 1, min(i + 1 + lp_cfg["valid_bars"], len(candles))):
                    if candles[j]["candle_time"][:8] != day:
                        break
                    bar_ticks = (
                        collect_bar_ticks(
                            ticks_by_code, code, candles[j]["candle_time"], tick_tf,
                        )
                        if use_ticks else []
                    )
                    fp, _src = try_limit_fill_on_bar_with_ticks(
                        candles[j], limit_px, lp_cfg["fill_slip_pct"],
                        ticks=bar_ticks, params=params,
                    )
                    if fp and fp > 0:
                        entry_price = fp
                        entry_time = candles[j]["candle_time"]
                        stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)
                        from_risk = (capital * risk_pct * kelly_mult) / static_sl_pct if static_sl_pct > 0 else capital
                        from_cap = max_loss_krw / static_sl_pct if static_sl_pct > 0 else capital
                        invest_amount = min(from_risk, from_cap)
                        calc_qty = max(1, int(invest_amount / entry_price))
                        position = {
                            "entry_price": entry_price,
                            "entry_time": entry_time,
                            "stop": stop_p,
                            "target": target_p,
                            "max_price": entry_price,
                            "session_low": entry_price,
                            "qty": calc_qty,
                        }
                        filled = True
                        i = j + 1
                        break
                if not filled:
                    i += 1
                continue

            # align — 진입봉 i-1 마감 시각 = 현재 봉 i 시작 (연속 봉 가정)
            if i < 20:
                i += 1
                continue
            entry_i = i - 1
            ent = candles[entry_i]
            if (
                align_entry_execute_time(ent.get("candle_time"), tick_tf)[:12]
                != str(c.get("candle_time") or "")[:12]
            ):
                i += 1
                continue
            reject, _msg, sig = _eval_live_align_lookback(
                candles, entry_i, eval_params, state, lookback=live_sig_lb,
            )
            if reject or not sig:
                i += 1
                continue

            atr = (
                atrs[entry_i]
                if atrs[entry_i] is not None
                else float(ent.get("close") or cl) * 0.01
            )
            entry_price = float(sig.get("entry_price") or ent.get("open") or cl)
            if entry_price <= 0:
                i += 1
                continue
            if use_ticks:
                bar_ticks = collect_bar_ticks(
                    ticks_by_code, code, ent["candle_time"], tick_tf,
                )
                entry_price, _align_src = align_entry_price_from_ticks(bar_ticks, entry_price)

            entry_time = str(c["candle_time"])[:12]
            stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)

            # 포지션 사이징 로직 (Risk % 및 Max Loss 반영)
            from_risk = (capital * risk_pct * kelly_mult) / static_sl_pct if static_sl_pct > 0 else capital
            from_cap = max_loss_krw / static_sl_pct if static_sl_pct > 0 else capital
            invest_amount = min(from_risk, from_cap)
            calc_qty = max(1, int(invest_amount / entry_price))

            position = {
                "entry_price": entry_price,
                "entry_time": entry_time,
                "stop": stop_p,
                "target": target_p,
                "max_price": entry_price,
                "session_low": entry_price,
                "qty": calc_qty,
            }
            i += 1
            continue

    return all_trades
def init_rust_session(
    session_id: str,
    candles_by_code: Dict[str, List[Dict]],
    ticks_by_code: Optional[Dict[str, List[Dict]]] = None,
):
    """
    Rust 인메모리 엔진에 파이썬 딕셔너리 데이터를 한 번에 밀어넣는 래퍼 함수.
    """
    try:
        import kis_rust_core
    except ImportError:
        print("[WARNING] kis_rust_core 모듈을 찾을 수 없습니다. Rust 백테스트가 불가능합니다.")
        return

    import json
    
    # 딕셔너리를 JSON 문자열로 직렬화하여 Rust로 한 번에 전달 (Zero-copy 파싱)
    try:
        candles_json = json.dumps(candles_by_code)
    except Exception as e:
        print(f"[ERROR] Failed to serialize candles to JSON: {e}")
        return
    
    rust_ticks = {}
    # Phase 2 틱 데이터 지원 확장 대비 (현재는 빈 딕셔너리 전달)
    if ticks_by_code:
        pass

    try:
        kis_rust_core.init_backtest_session_json(session_id, candles_json)
    except Exception as e:
        print(f"[ERROR] Failed to init_rust_session: {e}")

def clear_rust_session(session_id: str):
    """
    Rust 인메모리 엔진의 메모리를 해제하는 래퍼 함수.
    """
    try:
        import kis_rust_core
        kis_rust_core.clear_backtest_session(session_id)
    except Exception as e:
        pass