kis_bot/scripts/momentum_optuna_trades_export_smoke.py

#!/usr/bin/env python3
"""Step1 검증: enrich_out_data_with_mode_combo 가 best/mode _trades 를 JSON에 남기는지.

기존 Optuna JSON의 #1·mode params 만 빌려, 동일 evaluate 경로로 1회 실측.
DB 미변경. 전체 Optuna 재실행 아님.
"""
from __future__ import annotations

import argparse
import json
import traceback
from copy import deepcopy
from pathlib import Path


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument(
        "--json",
        default="kis_trader/backtest/results/optuna_momentum_tpe_20260821_220227.json",
    )
    ap.add_argument("--out", required=True)
    args = ap.parse_args()

    src = Path(args.json)
    d = json.loads(src.read_text())
    results = list(d.get("results") or [])
    if not results:
        print("no results", flush=True)
        return 1

    from kis_trader.backtest.optuna_momentum import prepare_momentum_search_context
    from kis_trader.backtest.optuna_mode_combo import enrich_out_data_with_mode_combo
    from kis_trader.backtest.param_search_momentum import evaluate_momentum_param_combo

    print(
        f"prepare {d['start']}~{d['end']} for trades-export smoke",
        flush=True,
    )
    ctx = prepare_momentum_search_context(
        d["start"],
        d["end"],
        "tpe",
        slot_money=float(d["slot_money"]),
        max_stocks=int(d["max_stocks"]),
        total_budget_krw=float(d["total_budget_krw"]),
        orderbook_filter="off",
        market="KR",
        history_source="kiwoom",
    )
    if ctx is None:
        print("prepare failed", flush=True)
        return 1

    grid_keys = list(d.get("grid_keys") or ctx.grid_keys)

    def _eval(combo):
        return evaluate_momentum_param_combo(
            combo,
            base_fixed=ctx.base_fixed,
            grid_keys=grid_keys,
            codes_candles=ctx.codes_candles,
            min_trades=1,
            min_win_rate=0.0,
            min_pf=0.0,
            universe_by_slot=ctx.universe_by_slot,
            slot_money=ctx.slot_money,
            max_stocks=ctx.max_stocks,
            total_budget_krw=ctx.total_budget_krw,
            fee_rate=ctx.fee_rate,
            sell_tax=ctx.sell_tax,
            period_days=ctx.period_days,
            cache_holder=ctx.cache_holder,
            ticks_by_code=ctx.ticks_by_code,
            orderbook_by_code=ctx.orderbook_by_code,
            program_by_code=ctx.program_by_code,
            log_verdict_by_code=ctx.log_verdict_by_code,
            start_key=ctx.start_key,
            end_key=ctx.end_key,
            include_trades=True,
        )

    out_data = {
        "engine": "optuna_trades_smoke",
        "strategy": "momentum",
        "mode": d.get("mode"),
        "start": d["start"],
        "end": d["end"],
        "grid_keys": grid_keys,
        # enrich 가 results[0] 을 best 로 씀 — 원본 #199 행 유지
        "results": [deepcopy(results[0])],
        "source_json": str(src),
        "db_touched": False,
        # 후처리(호가 TPE)는 스킵 — evaluate 만 검증. attach_topn 은 enrich 안이므로
        # run_ob 을 막으려면 env? 그냥 돌리면 오래 걸림.
        # → enrich 전체가 후처리까지 함. 스모크는 후처리 OFF 필요.
    }

    # 후처리 끄기: attach_topn_postprocess 의 run_ob_whipsaw — enrich 가 True 고정.
    # 스모크 시간 절약: enrich 대신 best/mode 직접 호출과 동일 검증.
    from kis_trader.backtest.optuna_common import slim_trades_for_optuna_json
    from kis_trader.backtest.optuna_mode_combo import _attach_best_trial_trades

    _attach_best_trial_trades(out_data, _eval, log=__import__("logging").getLogger("smoke"))

    mode_params = dict((d.get("mode_combo") or {}).get("params") or {})
    bt = _eval(mode_params) if mode_params else None
    mode_trades = []
    mode_summary = None
    if isinstance(bt, dict):
        mode_trades = slim_trades_for_optuna_json(list(bt.pop("_trades", None) or []))
        mode_summary = {
            "ok": True,
            "total_pnl": bt.get("total_pnl"),
            "total_trades": bt.get("total_trades"),
            "win_rate": bt.get("win_rate"),
            "pf": bt.get("pf"),
            "_trades": mode_trades,
        }
    out_data["mode_combo"] = {
        "params": mode_params,
        "backtest": mode_summary,
        "note": "smoke: mode params from source JSON, re-eval with _trades",
    }

    best_tr = out_data.get("best_trial_trades") or []
    reeval = out_data.get("best_trial_reeval") or {}
    checks = {
        "best_trades_n": len(best_tr),
        "best_trades_eq_reeval_tr": len(best_tr) == int(reeval.get("total_trades") or -1),
        "mode_trades_n": len(mode_trades),
        "mode_trades_eq_summary_tr": (
            len(mode_trades) == int((mode_summary or {}).get("total_trades") or -1)
            if mode_summary
            else False
        ),
        "best_has_aug21": any(
            str(t.get("sell_time") or "").startswith("20260821") for t in best_tr
        ),
        "slim_keys_ok": (
            all(
                set(t.keys())
                >= {"code", "buy_time", "sell_time", "pnl", "sell_reason"}
                for t in best_tr[:3]
            )
            if best_tr
            else False
        ),
    }
    out_data["smoke_checks"] = checks
    out_data["smoke_ok"] = all(
        [
            checks["best_trades_n"] > 0,
            checks["best_trades_eq_reeval_tr"],
            checks["mode_trades_n"] > 0,
            checks["mode_trades_eq_summary_tr"],
            checks["slim_keys_ok"],
        ]
    )

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(out_data, ensure_ascii=False, indent=2))
    print("WROTE", out, flush=True)
    print("CHECKS", json.dumps(checks, ensure_ascii=False), flush=True)
    print("SMOKE_OK", out_data["smoke_ok"], flush=True)
    print(
        "BEST_REEVAL",
        json.dumps(reeval, ensure_ascii=False),
        flush=True,
    )
    return 0 if out_data["smoke_ok"] else 2


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception:
        traceback.print_exc()
        raise SystemExit(1)