kis_bot/kis_trader/engine/scalping_engine.py

#!/usr/bin/env python3
"""
kis_trader/engine/scalping_engine.py — 스캘핑(SCALP reversal) 백테·파서치·실매 공통 엔진
====================================================================
SCALP reversal 전용. 모멘텀(MOMENTUM)은 ``momentum_engine.py`` 로 완전 분리됨.

■ 라이브·백테 검증 원칙 (뇌동 분기 금지)
  - SCALP 진입/청산: 웹·CLI·실매가 ``check_buy_signal_*`` / ``check_sell_signal_live`` 동일 경로.
  - MOMENTUM: ``momentum_engine`` 전용 (SCALP 와 분리).

■ SCALP reversal 공통 로직 (A안 — RSI V자 + scalp_re HTS SCAN)
  - SCAN: 키움 ``CONDITION_SCALP_KIWOOM_NAME`` (기본 scalp_re).
  - ``SCALP_SKIP_HTS_SCAN_DUPES=true`` (kiwoom_condition 기본): TRIGGER 에 낙폭·RSI·되돌림 중복 검사 생략.
  - 매수(reversal): 시간대, 쿨다운, 일일 진입 횟수, RSI 과매도/과매수, 되돌림(음봉->양봉), 낙폭, 거래량.
  - 매수(macd): HTS C — MACD+Stochastic [12,26,5,3,3] 골든크로스 (``SCALP_USE_MACD_CROSS=true``).
  - 매수 방어: 고점추격 방지(high_chase_thr), 급등주(max_daily_chg), 최소 가격(min_price).
  - 매도(V4, tail_engine 동일): 1순위 어깨컷, 2순위 익절, 3순위 손절, 4순위 금액손실컷, 5순위 장마감청산.
  - 백테 SCALP 청산: 1분 OHLC → N회 intrabar (``check_sell_signal_backtest_bar``).

캔들 형식: list of dict with keys candle_time(YYYYMMDDHHMI), open, high, low, close, volume
"""

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

from kis_trader.utils.env import get_env_int


def _frac_from_env_keys(r: Optional[dict], keys: Tuple[str, ...], default: float) -> float:
    """env_config 행에서 비율(0.003=0.3%) 값 로드. 키 순서대로 fallback."""
    if r:
        for k in keys:
            v = r.get(k)
            if v not in (None, ""):
                try:
                    return float(v)
                except (ValueError, TypeError):
                    continue
    return float(default)


def resolve_effective_tp_pct(tp_pct: float, tp_max_pct: float) -> float:
    """익절 % — tp_max_pct 상한 적용 (꼬리 TAIL_ATR_TP_MAX_PCT 와 동일 개념).

    tp_pct=2.5%, tp_max=2% → 실제 2% 목표가.
    tp_max=0 이면 상한 미적용(OFF).
    """
    tp = abs(float(tp_pct))
    cap = abs(float(tp_max_pct))
    if cap > 0:
        return min(tp, cap)
    return tp


def effective_tp_pct_from_params(params: Dict[str, Any]) -> float:
    """params['tp_pct'] + params['tp_max_pct'] → 진입·청산용 실효 익절 비율."""
    return resolve_effective_tp_pct(
        params.get("tp_pct", 0.015),
        params.get("tp_max_pct", 0.02),
    )


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 resolve_scalp_skip_hts_scan_dupes(r: Optional[Dict[str, Any]] = None) -> bool:
    """
    HTS scalp_re SCAN(kiwoom_condition) 사용 시 TRIGGER 중복 필터 생략 여부.

    - ``SCALP_SKIP_HTS_SCAN_DUPES`` 명시(true/false) → 그대로
    - 미설정 → ``SCALP_UNIVERSE_SOURCE`` 가 condition/kiwoom_condition 이면 True
    - ranking 등 전종목 소스 → False (TRIGGER 에서 HTS A/B/C 재검사)
    """
    if r is None:
        try:
            from kis_trader.utils.env import get_strategy_env_dict
            r = get_strategy_env_dict("SCALP") or {}
        except Exception:
            r = {}
    raw = r.get("SCALP_SKIP_HTS_SCAN_DUPES")
    if raw is not None and str(raw).strip() != "":
        return _to_bool(raw, True)
    # 엔진 defaults dict(이미 skip_hts_scan_dupes 해석됨)를 넘긴 경우 —
    # env 키 없음 → universe fallback(True) 로 덮어쓰지 않음 (실매 false 고정 재발 방지)
    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("SCALP_UNIVERSE_SOURCE") or "condition").strip().lower()
    return src in ("kiwoom_condition", "condition")


# DB 기본값 로드 (백테스트/param_search가 동일한 값 사용하도록 단일 소스)
def get_scalping_defaults_from_db(*, env_row: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    config_scalp + env_config 병합에서 스캘핑 관련 기본값 로드.
    백테스트 API·param_search·실매(get_env_from_db)가 동일 merged 소스를 사용.
    env_row 가 있으면 DB 조회 생략 (웹 백테 env 타임라인용).
    """
    try:
        if env_row is not None:
            r = dict(env_row)
        else:
            from kis_trader.utils.env import get_strategy_env_dict
            r = get_strategy_env_dict("SCALP")
            if not r:
                from database import TradeDB
                db = TradeDB()
                if hasattr(db, "get_merged_env_snapshot"):
                    r = db.get_merged_env_snapshot()
                else:
                    latest = db.get_latest_env() if hasattr(db, "get_latest_env") else None
                    r = dict((latest or {}).get("snapshot") or {})
                db.close()
        if r:
            # SCALP_COOLDOWN_SEC(초) → cooldown_min(분). 실매매와 동일 키 사용
            sec = r.get("SCALP_COOLDOWN_SEC") or r.get("REENTRY_COOLDOWN_SEC") or "600"
            cooldown_min = max(0, int(float(sec)) // 60)
            fee_pct = float(r.get("FEE_RATE_PCT") or 0.015)
            tax_pct = float(r.get("SELL_TAX_RATE_PCT") or 0.18)
            slot = float(r.get("SLOT_MONEY_DEFAULT") or 3_000_000)
            
            # ──────── 방어 로직 (SCALP_* 전용 키 우선 → 공용/타전략 fallback) ────────
            # HIGH_CHASE_THR·MOMENTUM_* 를 먼저 읽으면 Optuna/웹 apply 값이 무시됨.
            high_chase_thr = float(
                r.get("SCALP_HIGH_PRICE_CHASE_THRESHOLD")
                or r.get("HIGH_CHASE_THR")
                or r.get("HIGH_PRICE_CHASE_THRESHOLD")
                or 0.96,
            )
            max_daily_chg = float(
                r.get("SCALP_MAX_DAILY_CHANGE_PCT")
                or r.get("MAX_DAILY_CHG")
                or r.get("MAX_DAILY_CHANGE_PCT")
                or 20.0,
            )
            min_price = float(
                r.get("SCALP_MIN_PRICE")
                or r.get("MOMENTUM_MIN_PRICE")
                or r.get("MIN_PRICE_TAIL")
                or 1000.0,
            )
            max_loss_krw = int(
                float(
                    r.get("SCALP_MAX_LOSS_PER_TRADE_KRW")
                    or r.get("MOMENTUM_MAX_LOSS_PER_TRADE_KRW")
                    or r.get("MAX_LOSS_PER_TRADE_KRW")
                    or 200000,
                ),
            )
            _min_drop_loss = r.get("SCALP_MIN_DROP_PCT_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
            min_margin = float(
                r.get("SCALP_MIN_PROFIT_PCT")
                or r.get("MOMENTUM_MIN_PROFIT_PCT")
                or 0.2,
            )
            rsi_overbought = float(r.get("SCALP_RSI_OVERBOUGHT") or 75.0)
            use_defense_filters = _to_bool(r.get("SCALP_USE_DEFENSE_FILTERS"), True)
            use_macd_cross = _to_bool(r.get("SCALP_USE_MACD_CROSS"), False)
            macd_fast = int(float(r.get("SCALP_MACD_FAST") or 12))
            macd_slow = int(float(r.get("SCALP_MACD_SLOW") or 26))
            macd_signal = int(float(r.get("SCALP_MACD_SIGNAL") or 5))
            stoch_k_period = int(float(r.get("SCALP_STOCH_K_PERIOD") or 5))
            stoch_d_period = int(float(r.get("SCALP_STOCH_D_PERIOD") or 3))
            stoch_slow = int(float(r.get("SCALP_STOCH_SLOW") or 3))
            skip_hts_scan_dupes = resolve_scalp_skip_hts_scan_dupes(r)
            shoulder_min_high = _frac_from_env_keys(
                r,
                ("SCALP_SHOULDER_MIN_HIGH_PCT", "SHOULDER_MIN_HIGH_PCT"),
                0.005,
            )
            shoulder_cut_pct = _frac_from_env_keys(
                r,
                ("SCALP_SHOULDER_CUT_PCT", "SHOULDER_CUT_PCT"),
                0.003,
            )
            _mhs = r.get("SCALP_MIN_HOLD_SEC") or r.get("MIN_HOLD_SEC")
            min_hold_sec = float(_mhs) if _mhs not in (None, "") else 30.0
            tp_max_pct = _frac_from_env_keys(r, ("SCALP_TP_MAX_PCT",), 0.02)
            max_stocks = int(float(r.get("SCALP_MAX_STOCKS") or r.get("MAX_STOCKS") or 3))
            total_budget_krw = float(r.get("SCALP_TOTAL_BUDGET_KRW") or 0)
            portfolio_mode = _to_bool(r.get("SCALP_PORTFOLIO_MODE"), True)
            rsi_period = int(float(
                r.get("SCALP_RSI_PERIOD") or r.get("RSI_PERIOD") or 3,
            ))
            sl_pct = abs(float(r.get("SCALP_STOP_LOSS_PCT") or 0.015))
            tp_pct = abs(float(r.get("SCALP_TAKE_PROFIT_PCT") or 0.015))
            rsi_oversold = float(r.get("SCALP_RSI_OVERSOLD") or 25.0)
            drop_rate = float(r.get("SCALP_MIN_DROP_RATE") or 0.015)
            require_reversal_candle = _to_bool(r.get("SCALP_REQUIRE_REVERSAL_CANDLE"), True)
            _vm = r.get("VOL_MULTIPLIER")
            vol_mult = float(_vm) if _vm not in (None, "") else 0.0
            time_start_hm = int(float(r.get("SCALP_TIME_START") or r.get("TIME_START") or 900))
            time_end_hm = int(float(r.get("SCALP_TIME_END") or r.get("TIME_END") or 1530))
            max_daily = int(float(r.get("SCALP_MAX_DAILY") or r.get("MAX_DAILY") or 3))
            # EOD — 실매 기존 15:25 장마감청산과 동일 (strategy_eod SCALP)
            eod_enabled = _to_bool(r.get("SCALP_EOD_ENABLED"), True)
            if r.get("SCALP_EOD_ENABLED") in (None, "", "None") and r.get("force_eod_exit") not in (None, "", "None"):
                eod_enabled = _to_bool(r.get("force_eod_exit"), True)
            eod_hm = str(r.get("SCALP_EOD_HM") or "15:25").strip() or "15:25"
            # 신호=T−1 확정봉, 진입=형성 중 봉 T 시가(첫 틱) — 모멘텀/꼬리 BT 정합
            # DB 신규 컬럼 NULL→'' 이면 _to_bool('')=False 가 되므로 빈값은 미설정 취급
            _la = r.get("SCALP_LIVE_BACKTEST_ALIGN")
            live_backtest_align = (
                True if _la in (None, "") else _to_bool(_la, True)
            )
            _lb = r.get("SCALP_LIVE_SIGNAL_LOOKBACK_BARS")
            live_signal_lookback_bars = int(float(_lb or 1))
            _lf = r.get("SCALP_LIVE_ALIGN_USE_FORMING_BAR")
            live_align_use_forming_bar = (
                True if _lf in (None, "") else _to_bool(_lf, True)
            )
            backtest_use_tick_entry = _to_bool(
                r.get("SCALP_BACKTEST_USE_TICK_ENTRY"), True,
            )
            backtest_tick_fallback_ohlc = _to_bool(
                r.get("SCALP_BACKTEST_TICK_FALLBACK_OHLC"), False,
            )
        else:
            cooldown_min, fee_pct, tax_pct, slot = 10, 0.015, 0.18, 300_000.0
            high_chase_thr, max_daily_chg, min_price = 0.96, 20.0, 1000.0
            max_loss_krw, min_margin, rsi_overbought = 200000, 0.2, 75.0
            min_drop_pct_for_loss_cut = 0.015
            use_defense_filters = True
            use_macd_cross = False
            macd_fast, macd_slow, macd_signal = 12, 26, 5
            stoch_k_period, stoch_d_period, stoch_slow = 5, 3, 3
            skip_hts_scan_dupes = True
            shoulder_min_high, shoulder_cut_pct, min_hold_sec = 0.005, 0.003, 30.0
            tp_max_pct = 0.02
            max_stocks, total_budget_krw, portfolio_mode = 3, 0.0, True
            rsi_period = 3
            sl_pct, tp_pct = 0.015, 0.015
            rsi_oversold, drop_rate = 25.0, 0.015
            require_reversal_candle = True
            vol_mult = 0.0
            time_start_hm, time_end_hm, max_daily = 900, 1530, 3
            eod_enabled, eod_hm = True, "15:25"
            live_backtest_align, live_signal_lookback_bars = True, 1
            live_align_use_forming_bar = True
            backtest_use_tick_entry, backtest_tick_fallback_ohlc = True, False
            
    except Exception:
        cooldown_min, fee_pct, tax_pct, slot = 10, 0.015, 0.18, 300_000.0
        high_chase_thr, max_daily_chg, min_price = 0.96, 20.0, 1000.0
        max_loss_krw, min_margin, rsi_overbought = 200000, 0.2, 75.0
        min_drop_pct_for_loss_cut = 0.015
        use_defense_filters = True
        use_macd_cross = False
        macd_fast, macd_slow, macd_signal = 12, 26, 5
        stoch_k_period, stoch_d_period, stoch_slow = 5, 3, 3
        skip_hts_scan_dupes = True
        shoulder_min_high, shoulder_cut_pct, min_hold_sec = 0.005, 0.003, 30.0
        tp_max_pct = 0.02
        max_stocks, total_budget_krw, portfolio_mode = 3, 0.0, True
        rsi_period = 3
        sl_pct, tp_pct = 0.015, 0.015
        rsi_oversold, drop_rate = 25.0, 0.015
        require_reversal_candle = True
        vol_mult = 0.0
        time_start_hm, time_end_hm, max_daily = 900, 1530, 3
        eod_enabled, eod_hm = True, "15:25"
        live_backtest_align, live_signal_lookback_bars = True, 1
        live_align_use_forming_bar = True
        backtest_use_tick_entry, backtest_tick_fallback_ohlc = True, False

    return {
        "cooldown_min":   cooldown_min,
        "time_start_hm":  time_start_hm,
        "time_end_hm":    time_end_hm,
        "time_start":     time_start_hm,
        "time_end":       time_end_hm,
        "fee_rate":       fee_pct / 100,
        "sell_tax":       tax_pct / 100,
        "slot_money":     slot,
        "rsi_period":     rsi_period,
        "rsi_oversold":   rsi_oversold,
        "sl_pct":         sl_pct,
        "tp_pct":         tp_pct,
        "drop_rate":      drop_rate,
        "require_reversal_candle": require_reversal_candle,
        "vol_mult":       vol_mult,
        "trail_trigger":  0.007,
        "trail_stop":     0.004,
        "max_daily":      max_daily,
        "high_chase_thr": high_chase_thr,
        "max_daily_chg":  max_daily_chg,
        "min_price":      min_price,
        "max_loss_krw":   max_loss_krw,
        "min_drop_pct_for_loss_cut": min_drop_pct_for_loss_cut,
        "min_margin":     min_margin / 100,
        "rsi_overbought": rsi_overbought,
        "use_defense_filters": use_defense_filters,
        "use_macd_cross": use_macd_cross,
        "macd_fast": macd_fast,
        "macd_slow": macd_slow,
        "macd_signal": macd_signal,
        "stoch_k_period": stoch_k_period,
        "stoch_d_period": stoch_d_period,
        "stoch_slow": stoch_slow,
        "skip_hts_scan_dupes": skip_hts_scan_dupes,
        "shoulder_min_high": shoulder_min_high,
        "shoulder_cut_pct": shoulder_cut_pct,
        "min_hold_sec": min_hold_sec,
        "tp_max_pct": tp_max_pct,
        "max_stocks": max_stocks,
        "total_budget_krw": total_budget_krw,
        "portfolio_mode": portfolio_mode,
        "eod_enabled": eod_enabled,
        "eod_hm": eod_hm,
        # 레거시 별칭 — 포트폴리오/단건 엔진 force_eod 경로 호환
        "force_eod_exit": eod_enabled,
        "live_backtest_align": live_backtest_align,
        "live_signal_lookback_bars": live_signal_lookback_bars,
        "live_align_use_forming_bar": live_align_use_forming_bar,
        "backtest_use_tick_entry": backtest_use_tick_entry,
        "backtest_tick_fallback_ohlc": backtest_tick_fallback_ohlc,
    }


def compute_rsi_series(closes: list, period: int = 3) -> list:
    """RSI 시리즈 계산 (Wilder 스무딩). backtest_web과 동일."""
    rsi_list = [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 _ema_series(values: List[float], period: int) -> List[Optional[float]]:
    """지수이동평균(EMA) 시리즈. 초기값은 SMA."""
    n = len(values)
    out: List[Optional[float]] = [None] * n
    if period <= 0 or n < period:
        return out
    mult = 2.0 / (period + 1)
    ema = sum(values[:period]) / period
    out[period - 1] = ema
    for i in range(period, n):
        ema = (values[i] - ema) * mult + ema
        out[i] = ema
    return out


def _sma_optional_series(values: List[Optional[float]], period: int) -> List[Optional[float]]:
    """None 을 건너뛰지 않고, 윈도우 내 유효값만으로 SMA."""
    n = len(values)
    out: List[Optional[float]] = [None] * n
    if period <= 0:
        return out
    for i in range(period - 1, n):
        window = [v for v in values[i - period + 1 : i + 1] if v is not None]
        if len(window) == period:
            out[i] = sum(window) / period
    return out


def compute_macd_stoch_hts_lines(
    candles: List[Dict],
    macd_fast: int = 12,
    macd_slow: int = 26,
    macd_signal: int = 5,
    stoch_k_period: int = 5,
    stoch_d_period: int = 3,
    stoch_slow: int = 3,
) -> Tuple[
    List[Optional[float]],
    List[Optional[float]],
    List[Optional[float]],
    List[Optional[float]],
    List[Optional[float]],
    List[Optional[float]],
]:
    """
    HTS 조건식 C — MACD+Stochastic [12,26,5,3,3] 복합선.

    - fast_line = MACD + Slow%K
    - slow_line = Signal + Slow%D
    - 골든크로스: fast 가 slow 를 0봉전(현재봉) 상향돌파

    Returns:
        macd_line, signal_line, slow_k, slow_d, fast_combined, slow_combined
    """
    n = len(candles)
    closes = [float(c["close"]) for c in candles]
    highs = [float(c["high"]) for c in candles]
    lows = [float(c["low"]) for c in candles]

    ema_fast = _ema_series(closes, macd_fast)
    ema_slow = _ema_series(closes, macd_slow)
    macd_line: List[Optional[float]] = [None] * n
    for i in range(n):
        if ema_fast[i] is not None and ema_slow[i] is not None:
            macd_line[i] = ema_fast[i] - ema_slow[i]

    macd_vals = [v for v in macd_line if v is not None]
    signal_line: List[Optional[float]] = [None] * n
    if len(macd_vals) >= macd_signal:
        sig_sub = _ema_series(macd_vals, macd_signal)
        idx = 0
        for i in range(n):
            if macd_line[i] is not None:
                if idx < len(sig_sub):
                    signal_line[i] = sig_sub[idx]
                idx += 1

    raw_k: List[Optional[float]] = [None] * n
    for i in range(stoch_k_period - 1, n):
        hh = max(highs[i - stoch_k_period + 1 : i + 1])
        ll = min(lows[i - stoch_k_period + 1 : i + 1])
        if hh == ll:
            raw_k[i] = 50.0
        else:
            raw_k[i] = (closes[i] - ll) / (hh - ll) * 100.0

    slow_k = _sma_optional_series(raw_k, stoch_slow)
    slow_d = _sma_optional_series(slow_k, stoch_d_period)

    fast_combined: List[Optional[float]] = [None] * n
    slow_combined: List[Optional[float]] = [None] * n
    for i in range(n):
        if macd_line[i] is not None and slow_k[i] is not None:
            fast_combined[i] = macd_line[i] + slow_k[i]
        if signal_line[i] is not None and slow_d[i] is not None:
            slow_combined[i] = signal_line[i] + slow_d[i]

    return macd_line, signal_line, slow_k, slow_d, fast_combined, slow_combined


def _is_macd_stoch_golden_cross(
    fast_combined: List[Optional[float]],
    slow_combined: List[Optional[float]],
    i: int,
) -> bool:
    """0봉전 골든크로스: fast 가 slow 를 상향돌파."""
    if i < 1:
        return False
    f0, f1 = fast_combined[i], fast_combined[i - 1]
    s0, s1 = slow_combined[i], slow_combined[i - 1]
    if None in (f0, f1, s0, s1):
        return False
    return f0 > s0 and f1 <= s1


def _macd_min_bars(params: Dict[str, Any]) -> int:
    """MACD+Stoch 계산에 필요한 최소 봉 수."""
    macd_slow = int(params.get("macd_slow", 26))
    macd_signal = int(params.get("macd_signal", 5))
    stoch_k = int(params.get("stoch_k_period", 5))
    stoch_slow = int(params.get("stoch_slow", 3))
    stoch_d = int(params.get("stoch_d_period", 3))
    return macd_slow + macd_signal + stoch_k + stoch_slow + stoch_d + 5


def _apply_scalp_defense_and_vol(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    cl: float,
    lo: float,
    day: str,
) -> Tuple[str, Optional[str], Optional[str]]:
    """
    방어 필터 + 거래량 배수 (reversal·macd 공통).

    Returns:
        ("ok", None, None) — 통과
        ("reject", code, msg) — 탈락
        ("skip", None, None) — 당일 시가 없음 (신호 없음)
    """
    drop_rate = float(params.get("drop_rate", 0.015))
    vol_mult = float(params.get("vol_mult", 0))
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    max_daily_chg = float(params.get("max_daily_chg", 20.0))
    min_price = float(params.get("min_price", 1000.0))
    use_defense_filters = _to_bool(params.get("use_defense_filters", True), True)
    skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), False)

    if use_defense_filters and cl < min_price:
        return ("reject", "탈락-최소가격", "%.0f < %.0f" % (cl, min_price))

    running_open = float(candles[i]["open"])
    running_low = lo
    running_high = float(candles[i]["high"])
    for j in range(i - 1, -1, -1):
        if candles[j]["candle_time"][:8] != day:
            break
        running_open = float(candles[j]["open"])
        running_low = min(running_low, float(candles[j]["low"]))
        running_high = max(running_high, float(candles[j]["high"]))

    if running_open <= 0:
        return ("skip", None, None)

    # HTS scalp_re SCAN(kiwoom) — B/C/F/D 이미 통과. 낙폭·고점·급등·분봉거래량 중복 생략
    if use_defense_filters and not skip_hts:
        dr = (running_open - running_low) / running_open
        if dr < drop_rate:
            return ("reject", "탈락-낙폭", "%.2f%% < %.1f%%(SCALP_MIN_DROP_RATE)" % (dr * 100, drop_rate * 100))
        if cl >= running_high * high_chase_thr:
            return ("reject", "탈락-고점추격", "현재가 %.0f >= 고가 %.0f * %.2f" % (cl, running_high, high_chase_thr))
        if running_low > 0:
            daily_chg_pct = (running_high - running_low) / running_low * 100
            if daily_chg_pct > max_daily_chg:
                return ("reject", "탈락-급등주", "일일변동 %.1f%% > %.0f%%" % (daily_chg_pct, max_daily_chg))

    if not skip_hts and vol_mult > 0:
        volumes = [float(x.get("volume", 0)) for x in candles]
        vol = volumes[i] if i < len(volumes) else 0
        win = max(1, min(20, i))
        vol_avg = sum(volumes[i - win : i]) / win
        if vol_avg > 0 and vol < vol_avg * vol_mult:
            return ("reject", "탈락-거래량", "%.0f < 평균%.0f × %.1f" % (vol, vol_avg, vol_mult))

    return ("ok", None, None)


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

    신봇 기준:
      * 실매매는 10초 주기 REST 폴링 + 변동 tick 마다 초단위 event_time 저장.
      * 백테스트 caller(``param_search``/``tail_param_search``)가
        ``TradeDBExt.get_universe_by_candle_time()`` 으로 **1분 캔들 시각 키**
        (YYYYMMDDHHMM) 를 가진 dict 를 만들어 엔진에 주입한다.
      * 엔진은 그 dict 를 분단위(scan_interval_min=1) 키로 바로 조회.

    과거 호환:
      * ``--fallback-universe`` 시뮬레이션 경로는 여전히 5분 버킷팅을 쓸 수
        있도록 ``scan_interval_min`` 파라미터는 남겨 둔다.
    """
    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 build_universe_simulation(
    codes_candles: Dict[str, List[Dict]],
    top_n: int = 20,
    min_score: float = 4.0,
    scan_interval_min: int = 5,
) -> Dict[str, List[str]]:
    """
    과거 1분봉만으로 '5분마다 강도 순 상위 N종목' 유니버스를 흉내 냄.
    kiwoom_universe_scanner의 개미털기 점수(drop_rate·회복률)를 1분봉 기준으로 근사.
    실매매의 target_candidates 5분 갱신과 동일한 타이밍으로 백테스트 제한용.

    Returns:
        slot_key(YYYYMMDDHHMM) → 해당 슬롯에 매수 검사 허용할 종목 코드 리스트
    """
    slot_codes_scores: Dict[str, List[Tuple[str, float, float]]] = {}  # slot -> [(code, score, vol)]

    for code, rows in codes_candles.items():
        if len(rows) < 2:
            continue
        candles = [dict(r) for r in rows]
        # 날짜별로 묶어서 당일 시가/고/저/종가 누적
        by_day: Dict[str, List[Dict]] = {}
        for c in candles:
            day = c["candle_time"][:8]
            if day not in by_day:
                by_day[day] = []
            by_day[day].append(c)

        for day, day_candles in by_day.items():
            day_candles.sort(key=lambda x: x["candle_time"])
            day_open = float(day_candles[0]["open"])
            running_high = max(float(c["high"]) for c in day_candles)
            running_low = min(float(c["low"]) for c in day_candles)
            vol_sum = sum(float(c.get("volume", 0)) for c in day_candles)

            # 5분 단위 슬롯: 해당 슬롯 시작 시점까지의 데이터로 점수 계산 (슬롯 시작 직전 봉까지)
            market_open_min = 9 * 60  # 09:00 = 540분
            seen_slots = set()
            for j, c in enumerate(day_candles):
                ct = c["candle_time"]
                hm = int(ct[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)
                slot_key = day + str(slot_hm).zfill(4)
                if slot_key in seen_slots:
                    continue
                seen_slots.add(slot_key)
                # as_of: 이 슬롯에 쓰일 마지막 봉 (슬롯 시작 직전 봉, 장시작 09:00 이상)
                slot_min_val = (slot_hm // 100) * 60 + (slot_hm % 100)
                as_of_min = max(market_open_min, slot_min_val - 1)
                as_of_hm = (as_of_min // 60) * 100 + (as_of_min % 60)
                as_of_str = day + str(as_of_hm).zfill(4)
                up_to = [x for x in day_candles if x["candle_time"] <= as_of_str]
                if not up_to:
                    continue
                o = float(up_to[0]["open"])
                hi = max(float(x["high"]) for x in up_to)
                lo = min(float(x["low"]) for x in up_to)
                cl = float(up_to[-1]["close"])
                drop_rate = (o - lo) / o if o > 0 else 0
                rng = hi - lo
                recovery_pos = (cl - lo) / rng if rng > 0 else 0
                score = (drop_rate * 100) if (drop_rate >= 0.03 and recovery_pos >= 0.5) else 0.0
                vol_part = sum(float(x.get("volume", 0)) for x in up_to)

                if slot_key not in slot_codes_scores:
                    slot_codes_scores[slot_key] = []
                slot_codes_scores[slot_key].append((code, score, vol_part))

    # 슬롯별 상위 top_n, min_score 이상만
    universe_by_slot: Dict[str, List[str]] = {}
    for slot_key, lst in slot_codes_scores.items():
        lst = [(c, s, v) for c, s, v in lst if s >= min_score]
        lst.sort(key=lambda x: (-x[1], -x[2]))
        universe_by_slot[slot_key] = [x[0] for x in lst[:top_n]]
    return universe_by_slot


# ──────────────────────────────────────────────────────────────────────────────
# [레거시] 모멘텀 유니버스 시뮬 — momentum_engine.build_universe_simulation_momentum 위임
# ──────────────────────────────────────────────────────────────────────────────
def build_universe_simulation_momentum(
    codes_candles: Dict[str, List[Dict]],
    top_n: int = 20,
    min_score: float = 4.0,
    scan_interval_min: int = 5,
) -> Dict[str, List[str]]:
    """[폐기] ``momentum_engine.build_universe_simulation_momentum`` 사용."""
    from kis_trader.engine.momentum_engine import build_universe_simulation_momentum as _impl
    return _impl(codes_candles, top_n=top_n, min_score=min_score, scan_interval_min=scan_interval_min)



def run_scalping_backtest(
    codes_candles: 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,
) -> List[Dict]:
    """
    종목별 캔들에 대해 스캘핑 백테스트 실행. 실매매와 동일한 규칙(방어로직 포함) 적용.

    universe_by_slot이 주어지면, 5분마다 해당 슬롯의 후보 종목에서만 매수 신호를 검사
    (실매매의 target_candidates 5분 갱신과 동일한 유니버스 시뮬레이션).
    """
    if _to_bool(params.get("use_rust"), False):
        from kis_trader.utils.logger import get_logger
        get_logger("kis_trader.scalping_engine").info("🚀 Rust 엔진(Experimental)으로 스캘핑 시뮬레이션을 실행합니다.")
        return run_scalping_backtest_rust_experimental(codes_candles, params)

    if _to_bool(params.get("portfolio_mode"), True):
        from kis_trader.backtest.scalping_portfolio_backtest import run_scalping_backtest_portfolio
        return run_scalping_backtest_portfolio(
            codes_candles, params, universe_by_slot=universe_by_slot, mode="reversal",
            ticks_by_code=ticks_by_code,
        )

    rsi_period = int(params.get("rsi_period", 3))
    rsi_oversold = float(params.get("rsi_oversold", 25))
    rsi_overbought = float(params.get("rsi_overbought", 75.0))
    # 롱 스캘핑: 손절·익절 폭은 양의 비율이어야 함. 폼/DB에서 음수(-1.2%)가 넘어오면
    # stop 이 진입가 위로 뒤집혀 같은 봉에서 '손절'로 오표시되며 청산가가 역전되는 버그 발생.
    sl_pct = abs(float(params.get("sl_pct", 0.015)))
    tp_pct = effective_tp_pct_from_params(params)
    drop_rate = float(params.get("drop_rate", 0.015))
    slot_money = float(params.get("slot_money", 300_000))
    fee_rate = float(params.get("fee_rate", 0.00015))
    sell_tax = float(params.get("sell_tax", 0.0018))
    cooldown_min = float(params.get("cooldown_min", 10))
    trail_trigger = float(params.get("trail_trigger", 0))
    trail_stop = float(params.get("trail_stop", 0.004))
    time_start_hm = int(params.get("time_start_hm", 900))
    time_end_hm = int(params.get("time_end_hm", 1530))
    max_daily = int(params.get("max_daily", 3))
    vol_mult = float(params.get("vol_mult", 0))
    
    # 방어 로직 파라미터
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    max_daily_chg = float(params.get("max_daily_chg", 20.0))
    min_price = float(params.get("min_price", 1000.0))
    max_loss_krw = float(params.get("max_loss_krw", 200000.0))
    min_margin = float(params.get("min_margin", 0.002))
    use_defense_filters = _to_bool(params.get("use_defense_filters", True), True)
    use_macd_cross = _to_bool(params.get("use_macd_cross", False), False)

    from kis_trader.engine.strategy_eod import is_strategy_eod_bar
    from kis_trader.engine.tick_exit_common import (
        backtest_sell_slip_pct,
        backtest_tick_poll_ms,
        collect_minute_ticks,
        resolve_backtest_sell,
        strategy_tick_fallback_ohlc,
        strategy_use_tick_exit,
    )
    use_tick_exit = bool(ticks_by_code) and strategy_use_tick_exit(
        params, "SCALP_BACKTEST_USE_TICK_EXIT", default=True,
    )
    tick_fallback_ohlc = strategy_tick_fallback_ohlc(
        params, "SCALP_BACKTEST_TICK_FALLBACK_OHLC", default=False,
    )
    tick_poll_ms = backtest_tick_poll_ms(params, strategy_env="SCALP_BACKTEST_POLL_MS")
    tick_sell_slip = backtest_sell_slip_pct(params, strategy_env="SCALP_BACKTEST_SELL_SLIP_PCT")

    all_trades: List[Dict] = []

    for code, rows in codes_candles.items():
        if len(rows) < rsi_period + 5:
            continue
        candles = [dict(r) for r in rows]
        closes = [float(c["close"]) for c in candles]
        volumes = [float(c.get("volume", 0)) for c in candles]
        rsis = compute_rsi_series(closes, rsi_period)
        macd_combined = _macd_lines_from_params(candles, params) if use_macd_cross else None

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

        for i in range(rsi_period + 1, len(candles)):
            c = candles[i]
            day = c["candle_time"][:8]
            hm = int(c["candle_time"][8:12])
            cl = float(c["close"])
            lo = float(c["low"])
            hi = float(c["high"])
            vol = volumes[i] if i < len(volumes) else 0

            # 당일 고가/저가 누적 (피뢰침 방지용)
            if day != cur_day:
                cur_day = day
                running_open = float(c["open"])
                running_high = hi
                running_low = lo
            else:
                running_high = max(running_high, hi)
                running_low = min(running_low, lo)

            # 실매 SCALP_EOD_HM(기본 15:25) 과 동일 — force_eod 마지막봉만 의존 금지
            is_eod = is_strategy_eod_bar(c["candle_time"], params, "SCALP")

            # ── 포지션 보유 중: 청산 체크 ──
            if position is not None:
                # 진입봉(같은 candle_time)에서는 청산 금지:
                # 백테스트가 이미 알고 있는 봉의 고/저를 즉시 사용하는 look-ahead를 방지.
                if c["candle_time"] == position["entry_time"]:
                    continue
                # 틱 있으면 resolve_backtest_sell — 없으면 기존 OHLC N회 청산
                reason = None
                exit_price = cl
                sell_time = c["candle_time"]
                if use_tick_exit:
                    minute_ticks = collect_minute_ticks(
                        ticks_by_code, code, c["candle_time"],
                    )
                    res5 = resolve_backtest_sell(
                        position,
                        c,
                        params,
                        is_eod=is_eod,
                        sell_fn=check_sell_signal_live,
                        low_mode="current",
                        ticks=minute_ticks,
                        use_tick_exit=use_tick_exit,
                        tick_fallback_ohlc=tick_fallback_ohlc,
                        poll_ms=tick_poll_ms,
                        slip_pct=tick_sell_slip,
                    )
                    if res5:
                        reason, exit_price, sell_time, _hold, _src = res5
                else:
                    # check_sell_signal_backtest_bar — 1분봉 N회 청산(실매 current_price 정렬)
                    reason, exit_price = check_sell_signal_backtest_bar(
                        position, c, params, is_eod,
                    ) or (None, cl)
                
                if reason:
                    qty = position["qty"]
                    buy_amt = position["entry_price"] * qty
                    sell_amt = exit_price * qty
                    pnl = (
                        sell_amt
                        - buy_amt
                        - buy_amt * fee_rate
                        - sell_amt * fee_rate
                        - sell_amt * sell_tax
                    )
                    hold_min = int(
                        (_t2dt(sell_time or c["candle_time"]) - _t2dt(position["entry_time"])).total_seconds() / 60
                    )
                    all_trades.append({
                        "code": code,
                        "buy_time": position["entry_time"],
                        "sell_time": sell_time or c["candle_time"],
                        "buy_price": position["entry_price"],
                        "sell_price": round(exit_price, 2),
                        "qty": qty,
                        "pnl": round(pnl),
                        "profit_rate": round(
                            (exit_price - position["entry_price"]) / position["entry_price"] * 100, 2
                        ),
                        "hold_min": hold_min,
                        "sell_reason": reason,
                        "rsi_entry": round(position["rsi"], 1),
                    })
                    last_exit_dt[day] = _t2dt(sell_time or c["candle_time"])
                    position = None
                continue

            # ── 포지션 없음: 매수 신호 (유니버스 시뮬레이션 시 해당 슬롯 후보만 검사) ──
            if universe_by_slot is not None:
                # 신봇 기본: 1분봉 == 슬롯 키.  simulation fallback 은 caller 가
                # ``scan_interval_min`` 을 params 에 명시해 5분 슬롯으로 바꿀 수 있음.
                slot_key = _slot_key(c["candle_time"], params.get("scan_interval_min", 1))
                if code not in universe_by_slot.get(slot_key, []):
                    continue
            bt_state = {
                "last_exit_dt": last_exit_dt.get(day),
                "daily_cnt": daily_cnt.get(day, 0),
            }
            st = _apply_buy_state_filters(
                candles, i, params,
                bt_state,
            )
            if st[2] is None:
                continue

            reject, _, sig = _eval_scalp_buy_at_index(candles, i, params, macd_combined=macd_combined)
            if reject or not sig:
                continue

            rsi = sig.get("rsi")
            if rsi is None:
                continue

            # 다음 봉 시가에 진입
            if i + 1 >= len(candles):
                continue
            next_c = candles[i + 1]
            if next_c["candle_time"][:8] != day:
                continue
            entry_price = float(next_c["open"])
            if entry_price <= 0:
                continue

            # 포지션 크기 계산 (최대 금액 손실 컷 기반)
            invest_amount = slot_money
            if max_loss_krw > 0 and sl_pct > 0:
                invest_limit = max_loss_krw / sl_pct
                invest_amount = min(invest_limit, slot_money)

            qty = max(1, int(invest_amount / entry_price))
            stop = entry_price * (1 - sl_pct)
            target = entry_price * (1 + tp_pct)
            
            position = {
                "entry_price": entry_price,
                "entry_time": next_c["candle_time"],
                "qty": qty,
                "stop": stop,
                "target": target,
                "max_price": entry_price,
                "rsi": rsi,
            }
            daily_cnt[day] = daily_cnt.get(day, 0) + 1

    all_trades.sort(key=lambda x: x["sell_time"])
    return all_trades


# ── 실시간 봇용: 단일 시점 매수/매도 판단 (백테스트와 100% 동일 규칙) ──────────────

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 _apply_buy_state_filters(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """시간대·쿨다운·일일횟수 등 상태 필터 (신호봉 인덱스 i 기준 — 백테스트와 동일)."""
    c = candles[i]
    time_start_hm = int(params.get("time_start_hm", 900))
    time_end_hm = int(params.get("time_end_hm", 1530))
    cooldown_min = float(params.get("cooldown_min", 10))
    max_daily = int(params.get("max_daily", 3))
    hm = int(c["candle_time"][8:12])

    if hm < time_start_hm or hm >= time_end_hm:
        return (None, None, 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 state.get("daily_cnt", 0) >= max_daily:
        return (None, None, None)

    return (None, None, {"ok": True})


def _macd_lines_from_params(
    candles: List[Dict],
    params: Dict[str, Any],
) -> Tuple[List[Optional[float]], List[Optional[float]]]:
    """종목당 1회 MACD+Stoch 복합선 계산 (백테스트 루프용)."""
    _, _, _, _, fast_combined, slow_combined = compute_macd_stoch_hts_lines(
        candles,
        macd_fast=int(params.get("macd_fast", 12)),
        macd_slow=int(params.get("macd_slow", 26)),
        macd_signal=int(params.get("macd_signal", 5)),
        stoch_k_period=int(params.get("stoch_k_period", 5)),
        stoch_d_period=int(params.get("stoch_d_period", 3)),
        stoch_slow=int(params.get("stoch_slow", 3)),
    )
    return fast_combined, slow_combined


def _eval_scalp_buy_at_index(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    macd_combined: Optional[Tuple[List[Optional[float]], List[Optional[float]]]] = None,
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """SCALP 진입 모드 분기: MACD 골든크로스 vs RSI reversal → 호가필터."""
    if _to_bool(params.get("use_macd_cross", False), False):
        reject, msg, sig = _eval_macd_golden_buy_at_index(
            candles, i, params, macd_combined=macd_combined,
        )
    else:
        reject, msg, sig = _eval_reversal_buy_at_index(candles, i, params)
    if reject or not sig:
        return (reject, msg, sig)
    cl = float(candles[i].get("close") or 0)
    # 휩쏘 — 모멘텀과 동일 순서(휩쏘 → 호가). SCALP_WHIPSAW_* / 주입틱·실매 WS.
    from kis_trader.engine.whipsaw_filter import whipsaw_reject_for_signal
    ws_rej, ws_msg = whipsaw_reject_for_signal(
        params, "SCALP",
        signal_bar=candles[i],
        current_price=cl,
    )
    if ws_rej:
        return (ws_rej, ws_msg, None)
    # 호가필터 — ORDERBOOK_FILTER_ENABLED / params['_orderbook_filter_enabled']
    # 실매 기본 OFF 면 차단 없음(수집만). 파람서치 --orderbook-filter on 시 차단.
    from kis_trader.engine.orderbook_filter import orderbook_reject_for_entry
    ob_rej, ob_msg = orderbook_reject_for_entry(params, "SCALP", current_price=cl)
    if ob_rej:
        return (ob_rej, ob_msg, None)
    return (None, None, sig)


def _eval_macd_golden_buy_at_index(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    macd_combined: Optional[Tuple[List[Optional[float]], List[Optional[float]]]] = None,
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """
    HTS C — MACD+Stochastic [12,26,5,3,3] 골든크로스 (0봉전 상향돌파).
    SCAN 은 K/L/제외만 두고, 진입 타이밍은 TRIGGER 에서 검사 (SCAN vs TRIGGER 분리).
    """
    min_bars = _macd_min_bars(params)
    if i < 1 or i >= len(candles) or len(candles) < min_bars:
        return ("탈락-봉부족", "MACD+Stoch 계산 봉 %d/%d" % (len(candles), min_bars), None)

    c = candles[i]
    day = c["candle_time"][:8]
    cl = float(c["close"])
    lo = float(c["low"])

    status, rej_code, rej_msg = _apply_scalp_defense_and_vol(candles, i, params, cl, lo, day)
    if status == "reject":
        return (rej_code, rej_msg, None)
    if status == "skip":
        return (None, None, None)

    if macd_combined is not None:
        fast_combined, slow_combined = macd_combined
    else:
        fast_combined, slow_combined = _macd_lines_from_params(candles, params)

    if not _is_macd_stoch_golden_cross(fast_combined, slow_combined, i):
        f0 = fast_combined[i]
        s0 = slow_combined[i]
        return (
            "탈락-MACD골든X",
            "fast=%s slow=%s (0봉전 상향돌파 아님)" % (
                "%.2f" % f0 if f0 is not None else "N/A",
                "%.2f" % s0 if s0 is not None else "N/A",
            ),
            None,
        )

    rsi_period = int(params.get("rsi_period", 3))
    closes = [float(x["close"]) for x in candles]
    rsis = compute_rsi_series(closes, rsi_period)
    rsi = rsis[i] if i < len(rsis) else None

    return (None, None, {"signal": True, "rsi": rsi, "entry_mode": "macd_golden"})


def _eval_reversal_buy_at_index(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """
    봉 i 에서 reversal 매수 신호 판정 (시간·쿨다운 제외).
    ``run_scalping_backtest`` 의 신호봉 검사와 동일 규칙.
    """
    if i < 1 or i >= len(candles):
        return ("탈락-봉부족", "인덱스 범위 밖", None)

    rsi_period = int(params.get("rsi_period", 3))
    rsi_oversold = float(params.get("rsi_oversold", 25))
    rsi_overbought = float(params.get("rsi_overbought", 75.0))
    drop_rate = float(params.get("drop_rate", 0.015))
    vol_mult = float(params.get("vol_mult", 0))
    high_chase_thr = float(params.get("high_chase_thr", 0.96))
    max_daily_chg = float(params.get("max_daily_chg", 20.0))
    min_price = float(params.get("min_price", 1000.0))
    use_defense_filters = _to_bool(params.get("use_defense_filters", True), True)
    require_reversal_candle = bool(params.get("require_reversal_candle", True))
    skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), False)

    c = candles[i]
    day = c["candle_time"][:8]
    cl = float(c["close"])
    lo = float(c["low"])
    prev_c = candles[i - 1]

    status, rej_code, rej_msg = _apply_scalp_defense_and_vol(candles, i, params, cl, lo, day)
    if status == "reject":
        return (rej_code, rej_msg, None)
    if status == "skip":
        return (None, None, None)

    closes = [float(x["close"]) for x in candles]
    rsis = compute_rsi_series(closes, rsi_period)
    rsi = rsis[i] if i < len(rsis) else None

    # kiwoom scalp_re SCAN — HTS B/C/F/D 통과 후 진입 타이밍만 (RSI V자·되돌림 중복 금지)
    if skip_hts:
        return (None, None, {"signal": True, "rsi": rsi, "entry_mode": "reversal_hts_scan"})

    if rsi is None:
        return ("탈락-RSI없음", "RSI 미계산 (봉 축적 중)", None)
    if rsi <= 0.0:
        return ("탈락-RSI무효", "RSI3=0.0 (봉 부족, 계산 불가)", None)
    if rsi > rsi_overbought:
        return ("탈락-RSI과열", "RSI3=%.1f > %.0f" % (rsi, rsi_overbought), None)
    if rsi > rsi_oversold:
        return ("탈락-RSI조건", "RSI3=%.1f (과매도<%.0f 아님)" % (rsi, rsi_oversold), None)

    if require_reversal_candle:
        prev_bear = float(prev_c["close"]) < float(prev_c["open"])
        curr_bull = cl > float(c["open"])
        if not (prev_bear and curr_bull):
            return ("탈락-되돌림없음", "prev_bear=%s curr_bull=%s" % (prev_bear, curr_bull), None)

    return (None, None, {"signal": True, "rsi": rsi, "entry_mode": "reversal"})


def check_buy_signal_live(
    candles: List[Dict],
    params: Dict[str, Any],
    state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    """
    실시간 reversal 매수 신호.

  - ``live_backtest_align=True`` (기본):
    - 신호봉 = 직전 확정봉 (T−1)
    - 진입봉 = 형성 중 봉 T (``live_align_use_forming_bar`` 기본 True)
      → ``entry_price`` = 진입봉 시가(실매·BT 첫 틱/시가 정합)
    - forming 없으면 구 폴백: 신호=confirmed[-2], 진입=confirmed[-1] open
  - ``live_backtest_align=False``: 구버전 — 마지막 1봉만 검사.
    """
    if len(candles) < 4:
        return ("탈락-봉부족", "확정봉 4개 미만", None)

    live_align = _to_bool(params.get("live_backtest_align", True), True)
    lookback = max(1, int(params.get("live_signal_lookback_bars", 1)))
    use_forming = _to_bool(params.get("live_align_use_forming_bar", True), True)

    forming: Optional[Dict[str, Any]] = None
    if live_align and use_forming and candles:
        last = candles[-1]
        if last.get("is_confirmed") in (0, False, "0", "false"):
            forming = last

    confirmed = _confirmed_candles_only(candles)

    if len(confirmed) < 4:
        return ("탈락-봉부족", "확정봉 4개 미만", None)

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

    if live_align:
        # forming 있으면: 신호=confirmed[-1], 진입=forming open (BT next open 과 동일)
        # forming 없으면: 신호=confirmed[-2], 진입=confirmed[-1] open (구 폴백)
        if forming is not None:
            entry_bar = forming
            signal_base_i = len(confirmed) - 1
        else:
            entry_bar = confirmed[-1]
            signal_base_i = len(confirmed) - 2
        for k in range(lookback):
            signal_i = signal_base_i - k
            if signal_i < 1:
                break
            st = _apply_buy_state_filters(confirmed, signal_i, params, state)
            if st[2] is None:
                continue
            reject, msg, sig = _eval_scalp_buy_at_index(confirmed, signal_i, params)
            if reject:
                if k == 0:
                    last_reject = (reject, msg, None)
                continue
            if sig:
                entry_open = float(entry_bar.get("open", 0) or 0)
                if entry_open <= 0:
                    entry_open = float(entry_bar.get("close", 0) or 0)
                sig["entry_price"] = entry_open
                sig["entry_bar_key"] = entry_bar.get("candle_time")
                return (None, None, sig)
        return last_reject

    # 구버전: 마지막 확정봉 1개만 신호봉으로 검사
    i = len(confirmed) - 1
    st = _apply_buy_state_filters(confirmed, i, params, state)
    if st[2] is None:
        return (None, None, None)
    return _eval_scalp_buy_at_index(confirmed, i, params)


def _intrabar_exit_prices(
    open_: float, high: float, low: float, close: float, n_checks: int,
) -> List[float]:
    """1분 OHLC → 실매 10초 폴링 흉내 가격 경로 (open→high→low→close).

    **high·low 앵커는 반드시 경유** — 선형 보간만 쓰면 고가/저가를 빗나가 어깨컷이 어긋난다.
    """
    n = max(2, int(n_checks))
    anchors: List[float] = []
    for px in (float(open_), float(high), float(low), float(close)):
        if not anchors or px != anchors[-1]:
            anchors.append(px)
    if len(anchors) == 1:
        return [anchors[0]] * n

    if n <= len(anchors):
        return anchors[:n]

    # 앵커(o,h,l,c) 사이에 보조 점 삽입해 n개 맞춤
    seg_count = len(anchors) - 1
    extra_total = n - len(anchors)
    extra_per_seg = [extra_total // seg_count] * seg_count
    for i in range(extra_total % seg_count):
        extra_per_seg[i] += 1

    out: List[float] = [anchors[0]]
    for si in range(seg_count):
        a, b = anchors[si], anchors[si + 1]
        extras = extra_per_seg[si]
        for j in range(1, extras + 1):
            t = j / (extras + 1)
            out.append(a + (b - a) * t)
        if out[-1] != b:
            out.append(b)
    return out[:n] if len(out) >= n else out + [out[-1]] * (n - len(out))


def check_sell_signal_backtest_bar(
    position: Dict[str, Any],
    candle: Dict[str, Any],
    params: Dict[str, Any],
    is_eod: bool = False,
    sell_fn: Optional[Callable[..., Optional[tuple]]] = None,
    low_mode: str = "current",
) -> Optional[tuple]:
    """백테 전용 청산 — 1분·N분 OHLC를 N회 가격 체크로 쪼개 실매와 정렬.

    실매 폴링 흉내:
      - ``low_mode="current"`` (스캘핑·모멘텀): ``low=close=current_price``
      - ``low_mode="session_low"`` (꼬리잡기): ``low=누적 session_low``, ``close=current_price``

    ``sell_fn`` 미지정 시 ``check_sell_signal_live`` (scalping_engine).
    ``BACKTEST_EXIT_CHECKS_PER_BAR`` (기본 6) = 1 이면 구 OHLC 1회 체크.
    """
    if sell_fn is None:
        sell_fn = check_sell_signal_live
    n_checks = get_env_int("BACKTEST_EXIT_CHECKS_PER_BAR", 6)
    if n_checks <= 1:
        return sell_fn(position, candle, params, is_eod)

    o = float(candle.get("open", candle["close"]))
    h = float(candle.get("high", candle["close"]))
    l = float(candle.get("low", candle["close"]))
    c = float(candle["close"])
    ct = candle.get("candle_time", "")

    prices = _intrabar_exit_prices(o, h, l, c, n_checks)
    result: Optional[tuple] = None
    session_low: Optional[float] = None
    if str(low_mode).strip().lower() == "session_low":
        ep = float(position.get("entry_price", 0) or 0)
        session_low = float(position.get("session_low", ep) or ep)

    for idx, px in enumerate(prices):
        mp = float(position.get("max_price", 0) or 0)
        if px > mp:
            position["max_price"] = px
            mp = px
        if session_low is not None:
            session_low = min(session_low, px)
            position["session_low"] = session_low
            lo_sim = session_low
        else:
            lo_sim = px
        sim = {
            "open": o,
            "high": mp,
            "low": lo_sim,
            "close": px,
            "candle_time": ct,
        }
        eod_here = bool(is_eod and idx == len(prices) - 1)
        result = sell_fn(position, sim, params, is_eod=eod_here)
        if result:
            return result
    return result


def check_sell_signal_live(
    position: Dict[str, Any],
    current_candle: Dict[str, Any],
    params: Dict[str, Any],
    is_eod: bool = False,
) -> Optional[tuple]:
    """
    실시간 봇 및 백테스트 공용: 보유 포지션 청산 (tail_engine V4 어깨컷과 동일 우선순위).

    position:  { "entry_price", "entry_time", "qty", "stop", "target", "max_price" }
    current_candle: { "high", "low", "close", "candle_time"(optional) }
    반환:      (reason_str, exit_price) 또는 None

    [청산 우선순위 — 꼬리잡기 tail_engine.check_sell_signal_live 와 동일]
      1순위 어깨컷: max_price 갱신 후 shoulder_cut_pct 되돌림 (저가 lo 기준, 매도선 체결)
      2순위 익절 (target)
      3순위 손절 (stop)
      4순위 금액손실컷 — 어깨 미발동(수익 문턱 미도달)일 때만
      5순위 장마감청산
    """
    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 = float(params.get("max_loss_krw", 200000.0))
    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)
    position["max_price"] = max_p

    ep = float(position["entry_price"])
    stop = float(position["stop"])
    target = float(position["target"])
    qty = int(position.get("qty", 1) or 1)

    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))

    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
    trail_hit = trail_armed and lo > 0 and lo <= trail_stop_px

    if trail_hit:
        reason = "어깨컷"
        exit_price = trail_stop_px
    elif hi >= target:
        reason = "익절"
        exit_price = target
    elif lo > 0 and lo <= stop:
        reason = "손절"
        exit_price = stop
    elif (
        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
    elif is_eod:
        reason = "장마감청산"
        exit_price = cl

    if reason:
        return (reason, exit_price)
    return None

def run_scalping_backtest_rust_experimental(
    codes_candles: Dict[str, List[Dict]],
    params: Dict[str, Any],
) -> List[Dict]:
    """
    Rust 엔진 (kis_rust_core) 을 통한 초고속 스캘핑 백테스트 (실험).
    """
    try:
        import kis_rust_core
        from kis_rust_core import ScalpParams, CandleData
    except ImportError as e:
        from kis_trader.utils.logger import get_logger
        get_logger("kis_trader.scalping_engine").error(f"Rust core import failed: {e}")
        return []

    use_defense_filters = str(params.get("use_defense_filters", True)).strip().lower() in ("1", "true", "t", "y", "yes", "on")
    
    # skip_hts_scan_dupes 는 HTS 조건검색 엔진 쓸 때 낙폭/RSI 중복을 끌지 여부.
    skip_hts = False
    if "skip_hts_scan_dupes" in params:
        skip_hts = str(params.get("skip_hts_scan_dupes")).strip().lower() in ("1", "true", "t", "y", "yes", "on")
    else:
        src = str(params.get("SCALP_UNIVERSE_SOURCE", "condition")).strip().lower()
        skip_hts = src in ("kiwoom_condition", "condition")

    rp = ScalpParams(
        rsi_period=int(params.get("rsi_period", 3)),
        rsi_oversold=float(params.get("rsi_oversold", 25.0)),
        rsi_overbought=float(params.get("rsi_overbought", 75.0)),
        sl_pct=abs(float(params.get("sl_pct", 0.015))),
        tp_pct=effective_tp_pct_from_params(params),
        drop_rate=float(params.get("drop_rate", 0.015)),
        cooldown_min=float(params.get("cooldown_min", 10.0)),
        max_daily=int(params.get("max_daily", 3)),
        high_chase_thr=float(params.get("high_chase_thr", 0.96)),
        max_daily_chg=float(params.get("max_daily_chg", 20.0)),
        min_price=float(params.get("min_price", 1000.0)),
        vol_mult=float(params.get("vol_mult", 0.0)),
        use_defense_filters=use_defense_filters,
        skip_hts=skip_hts,
        time_start_hm=int(params.get("time_start_hm", 900)),
        time_end_hm=int(params.get("time_end_hm", 1530))
    )

    all_trades = []
    
    for code, rows in codes_candles.items():
        if not rows:
            continue
        
        # 캔들 변환
        rust_candles = []
        for r in rows:
            rust_candles.append(CandleData(
                str(r["candle_time"]),
                float(r["open"]),
                float(r["high"]),
                float(r["low"]),
                float(r["close"]),
                float(r.get("volume", 0)),
                float(r.get("rsi", 50.0)) # TODO: rust 내부에서 rsi 계산하도록 변경 필요
            ))
            
        res = kis_rust_core.run_scalp_backtest_fast(code, rust_candles, rp)
        
        for t in res:
            all_trades.append({
                "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,
                "qty": 1, # 임시
                "pnl": 0, # PnL 부착부에서 재계산
                "profit_rate": round(t.pnl_pct, 2),
                "hold_min": 0,
                "sell_reason": t.reason,
                "rsi_entry": round(t.rsi_entry, 1),
                "is_rust_core": True,
            })

    return all_trades