kis_bot/scripts/verify_optuna_tpe_apply_dryrun.py

#!/usr/bin/env python3
"""
verify_optuna_tpe_apply_dryrun.py — TPE 결과 JSON → apply 패치 dry-run
DB INSERT 없음. 소수·키·setup_* 누락만 검사. (모멘텀 DB 오류 재발 방지)

  .venv/bin/python scripts/verify_optuna_tpe_apply_dryrun.py
  .venv/bin/python scripts/verify_optuna_tpe_apply_dryrun.py --json path/to.json
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))


def _latest_tpe_json(strategy: str) -> Optional[Path]:
    d = ROOT / "kis_trader" / "backtest" / "results"
    files = sorted(d.glob(f"optuna_{strategy}_tpe_*.json"), key=lambda p: p.stat().st_mtime)
    return files[-1] if files else None


def _patch_for(strategy: str, merged: Dict[str, Any]) -> Dict[str, str]:
    from kis_trader.backtest.backtest_portfolio_common import (
        session_env_patch,
        strip_portfolio_keys_from_apply_patch,
    )
    if strategy == "momentum":
        from kis_trader.backtest.param_search_apply_snapshot import _patch_from_momentum_merged
        patch = _patch_from_momentum_merged(merged)
        patch.update(session_env_patch("MOMENTUM", merged))
        return strip_portfolio_keys_from_apply_patch(patch, "MOMENTUM")
    if strategy == "breakout":
        from kis_trader.backtest.param_search_apply_snapshot import _patch_from_breakout_merged
        patch = _patch_from_breakout_merged(merged)
        patch.update(session_env_patch("BREAKOUT", merged))
        return strip_portfolio_keys_from_apply_patch(patch, "BREAKOUT")
    if strategy == "scalp":
        from kis_trader.backtest.param_search_scalping import _params_to_db_snapshot
        return _params_to_db_snapshot(merged)
    if strategy == "tail":
        from kis_trader.backtest.tail_param_search import _tail_params_to_env_map
        patch = _tail_params_to_env_map(merged)
        return strip_portfolio_keys_from_apply_patch(patch, "TAIL", "SHORT")
    raise ValueError(strategy)


def _check_hm(key: str, val: Any) -> Optional[str]:
    try:
        hm = int(float(val))
    except (TypeError, ValueError):
        return f"{key}={val!r} not int"
    h, m = divmod(hm, 100)
    if h < 0 or h > 23 or m < 0 or m > 59:
        return f"{key}={hm} invalid HHMM (분>59 등)"
    return None


def verify_one(path: Path) -> Tuple[bool, List[str]]:
    msgs: List[str] = []
    data = json.loads(path.read_text(encoding="utf-8"))
    strategy = str(data.get("strategy") or "")
    mode = str(data.get("mode") or "")
    results = data.get("results") or []
    if not results:
        return False, [f"{path.name}: results empty"]
    top = results[0]
    merged = dict(top.get("merged_params") or top.get("params") or {})
    pnl = float(top.get("total_pnl") or 0)
    msgs.append(
        f"{path.name} | {strategy}/{mode} trial#{top.get('optuna_trial_number')} "
        f"pnl={pnl} trades={top.get('total_trades')} wr={top.get('win_rate')}"
    )
    # HHMM sanity on params
    for k in (
        "time_start_hm", "time_end_hm", "mom_time_end_hm",
    ):
        if k in merged and merged[k] not in (None, ""):
            err = _check_hm(k, merged[k])
            if err:
                msgs.append(f"  ❌ {err}")
                return False, msgs
    try:
        patch = _patch_for(strategy, merged)
    except Exception as e:
        msgs.append(f"  ❌ patch build failed: {e}")
        return False, msgs
    if not patch:
        msgs.append("  ❌ empty patch")
        return False, msgs
    # float string should be finite; ratio keys ending _PCT often /100 already
    bad_float = []
    for k, v in patch.items():
        if any(x in k for x in ("_PCT", "_MULT", "RATIO", "THR")):
            try:
                float(v)
            except (TypeError, ValueError):
                bad_float.append(k)
    if bad_float:
        msgs.append(f"  ❌ non-float patch values: {bad_float[:8]}")
        return False, msgs
    # momentum setup keys must be present if in merged
    if strategy == "momentum":
        for src, envk in (
            ("setup_vol_max_mult", "MOMENTUM_SETUP_VOL_MAX_MULT"),
            ("setup_bear_bars_min", "MOMENTUM_SETUP_BEAR_BARS_MIN"),
            ("mom_time_end_hm", "MOMENTUM_TIME_END_HM"),
        ):
            if src in merged and envk not in patch:
                msgs.append(f"  ❌ missing apply key {envk} (from {src})")
                return False, msgs
    msgs.append(f"  ✅ dry-run patch keys={len(patch)} (DB 미적용)")
    sample = sorted(patch.items())[:8]
    for k, v in sample:
        msgs.append(f"     {k}={v}")
    return True, msgs


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--json", action="append", default=[])
    ap.add_argument(
        "--strategies",
        default="momentum,tail,breakout,scalp",
        help="comma list when --json 미지정",
    )
    args = ap.parse_args()
    paths: List[Path] = [Path(p) for p in args.json]
    if not paths:
        for s in [x.strip() for x in args.strategies.split(",") if x.strip()]:
            p = _latest_tpe_json(s)
            if p:
                paths.append(p)
            else:
                print(f"⏳ {s}: tpe JSON 아직 없음")
    if not paths:
        print("검증할 JSON 없음")
        return 1
    ok_all = True
    for p in paths:
        ok, msgs = verify_one(p)
        for m in msgs:
            print(m)
        ok_all = ok_all and ok
    return 0 if ok_all else 2


if __name__ == "__main__":
    raise SystemExit(main())