kis_bot/kis_trader/backtest/backtest_env_timeline.py

#!/usr/bin/env python3
"""
웹 백테스트 전용 env 타임라인 — 실매·파람서치 경로에는 부착하지 않는다.

``backtest_env_timeline=True`` (웹 bt_meta) 일 때만 ``_env_timeline`` 을 params 에
넣고, 포트폴리오/단일 루프에서 ``apply_env_timeline_at`` 으로 TRIGGER·포트폴리오
숫자를 시점별 갱신한다.
"""
from __future__ import annotations

from typing import Any, Dict, Optional

from kis_trader.backtest.env_timeline import EnvTimeline, build_env_timeline

# params 에 남겨 두는 구조·메타 키 — env 스냅샷이 덮어쓰지 않음
_PRESERVE_PARAM_KEYS = frozenset({
    "_env_timeline",
    "_env_timeline_evt_key",
    "_universe_timeline",
    "_momentum_universe_timeline",
    "_universe_timeline_meta",
    "backtest_env_timeline",
    "portfolio_mode",
    "scan_interval_min",
    "backtest_universe_scan_at",
    "backtest_vol_fill_cap_pct",
    "strict_fill_verify",
    "limit_fill_slip_pct",
    "fee_rate",
    "sell_tax",
    "slot_money",
    "max_stocks",
    "total_budget_krw",
    "_backtest_period_start_key",
    "_backtest_log_verdict_by_code",
    "_portfolio_skip_stats",
    "share_denom_by_code",
})


def _truthy(val: Any) -> bool:
    if val is None:
        return False
    s = str(val).strip().lower()
    return s in ("1", "true", "t", "y", "yes", "on")


def backtest_env_timeline_enabled(params: Optional[Dict[str, Any]] = None) -> bool:
    """웹 백테 env 타임라인 ON 여부 — 기본 False (파람서치·실매 영향 없음)."""
    if params is None:
        return False
    if params.get("backtest_env_timeline") is not None:
        return _truthy(params.get("backtest_env_timeline"))
    return params.get("_env_timeline") is not None


def attach_backtest_env_timeline_to_params(
    engine_params: Dict[str, Any],
    meta_out: Optional[Dict[str, Any]],
    strategy_id: str,
) -> None:
    """웹 백테 구간 env 타임라인 부착 — meta_out['backtest_env_timeline']=True 일 때만."""
    if not backtest_env_timeline_enabled(engine_params):
        if meta_out and _truthy(meta_out.get("backtest_env_timeline")):
            engine_params["backtest_env_timeline"] = True
        else:
            return
    db = (meta_out or {}).get("db")
    start_key = str((meta_out or {}).get("start_key") or "")
    end_key = str((meta_out or {}).get("end_key") or "")
    start_ymd = start_key[:8] if len(start_key) >= 8 else ""
    end_ymd = end_key[:8] if len(end_key) >= 8 else ""
    if not start_ymd or not end_ymd:
        return
    tl = build_env_timeline(db, start_ymd, end_ymd)
    if tl is None:
        return
    engine_params["_env_timeline"] = tl
    engine_params["_strategy_id_for_env_tl"] = str(strategy_id or "").strip().upper()
    if meta_out is not None:
        meta_out["backtest_env_timeline"] = True
        meta_out["env_timeline_snapshots"] = tl.snapshot_count


def engine_defaults_from_env_row(
    strategy_id: str,
    env_row: Dict[str, Any],
) -> Dict[str, Any]:
    """전략별 get_*_defaults_from_db — env_row 스냅샷만 사용 (DB 재조회 없음).

    반환값은 반드시 **엔진 비율**(0.02=2%) 이어야 한다.
    웹이 ``ui% → breakout_ui_to_engine_params`` 한 뒤 타임라인이 UI% 를 다시
    덮어쓰면 trail/shoulder 가 100배 커져 청산·슬롯이 붕괴한다 (BREAKOUT 전용 함정).
    """
    sid = str(strategy_id or "").strip().upper()
    if sid in ("SHORT", "TAIL"):
        from kis_trader.engine.tail_engine import get_tail_defaults_from_db
        return get_tail_defaults_from_db(env_row=env_row)
    if sid == "MOMENTUM":
        from kis_trader.engine.momentum_engine import get_momentum_defaults_from_db
        return get_momentum_defaults_from_db(env_row=env_row)
    if sid == "SCALP":
        from kis_trader.engine.scalping_engine import get_scalping_defaults_from_db
        return get_scalping_defaults_from_db(env_row=env_row)
    if sid == "BREAKOUT":
        from kis_trader.backtest.breakout_backtest_common import get_breakout_defaults_from_env_row
        from kis_trader.strategies.breakout import breakout_ui_to_engine_params

        # get_breakout_defaults_from_env_row = 웹 폼용 UI% → 엔진 비율로 변환 후 적용
        ui = get_breakout_defaults_from_env_row(env_row)
        eng = breakout_ui_to_engine_params(ui)
        # buy_params 동기화·invest_cap 이 읽는 별칭 (값은 엔진 비율)
        eng["sl_pct"] = float(eng.get("stop_loss_pct", 0.0) or 0.0)
        eng["tp_pct"] = float(eng.get("take_profit_pct", 0.0) or 0.0)
        if eng.get("shoulder_min_high") is not None:
            eng["shoulder_min_high_pct"] = float(eng["shoulder_min_high"])
        # ui_to_engine 이 빼먹는 트리거 스냅샷 플래그
        for k in ("ob_filter_enabled", "pg_filter_enabled", "max_spread_pct"):
            if k in ui:
                eng[k] = ui[k]
        return eng
    if sid == "RANGE_BREAK":
        from kis_trader.engine.range_break_engine import get_range_break_defaults_from_db
        return get_range_break_defaults_from_db(env_row=env_row)
    if sid == "DBBAND":
        from kis_trader.engine.dbband_engine import get_dbband_defaults_from_db
        return get_dbband_defaults_from_db(env_row=env_row)
    return {}


def apply_env_timeline_at(
    params: Dict[str, Any],
    candle_time: str,
    strategy_id: Optional[str] = None,
) -> bool:
    """
    봉 시각에 맞는 env 스냅샷을 params 에 반영. 스냅샷 변경 시에만 True.

    포트폴리오 루프 변수(max_stocks, slot_money 등)는 호출측에서 params 재읽기.
    """
    tl: Optional[EnvTimeline] = params.get("_env_timeline")
    if tl is None:
        return False
    snap, evt_key = tl.snapshot_at(candle_time)
    if not snap or not evt_key:
        return False
    if evt_key == params.get("_env_timeline_evt_key"):
        return False
    sid = str(
        strategy_id or params.get("_strategy_id_for_env_tl") or ""
    ).strip().upper()
    defaults = engine_defaults_from_env_row(sid, snap)
    for k, v in defaults.items():
        if k in _PRESERVE_PARAM_KEYS:
            continue
        params[k] = v
    params["_env_timeline_evt_key"] = evt_key
    return True