kis_bot/scripts/post_optuna_mid_enroll_verify.py

#!/usr/bin/env python3
"""Optuna apply 이후 — 현재 DB로 웹정렬 백테 1회씩 + 요약 리포트.

  .venv/bin/python -u scripts/post_optuna_mid_enroll_verify.py \\
    --start 2026-07-21 --end 2026-07-23 --report logs/report.txt
"""
from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from pathlib import Path
from typing import Any, Dict, List

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


def _run_kr_cli(strat: str, start: str, end: str) -> Dict[str, Any]:
    py = ROOT / ".venv" / "bin" / "python"
    if not py.exists():
        py = Path(sys.executable)
    out_json = ROOT / "logs" / f"verify_bt_{strat}_{int(time.time())}.json"
    cmd = [
        str(py), "-u", str(ROOT / "scripts" / "run_strategy_backtest_cli.py"),
        "--strategy", strat,
        "--start", start,
        "--end", end,
        "--universe", "history",
    ]
    # CLI may print summary; also try to capture via env if supported
    p = subprocess.run(cmd, cwd=str(ROOT), capture_output=True, text=True, timeout=3600)
    text = (p.stdout or "") + "\n" + (p.stderr or "")
    summary: Dict[str, Any] = {
        "strategy": strat,
        "rc": p.returncode,
        "tail": "\n".join(text.strip().splitlines()[-40:]),
    }
    # parse common lines
    for line in text.splitlines():
        low = line.lower()
        if "total_pnl" in low or "총손익" in line or "trades=" in low or "거래" in line:
            summary.setdefault("hits", []).append(line.strip())
    out_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    summary["log_json"] = str(out_json)
    return summary


def _run_us_web(start: str, end: str, symbol: str = "") -> Dict[str, Any]:
    """US 웹백테 — Flask 핸들러와 동일 경로가 아니면 엔진 params+포트폴리오 요약만."""
    from database import TradeDB
    from kis_trader.utils.env import invalidate_merged_env_cache
    from kis_trader.backtest import scalping_backtest_common as sbc
    from kis_trader.backtest.us_momentum_web_backtest import (
        build_us_momentum_engine_params,
        load_us_candles,
    )
    from kis_trader.engine.momentum_engine import get_momentum_defaults_from_db

    invalidate_merged_env_cache()
    db = TradeDB()
    ui_def = get_momentum_defaults_from_db()
    args: Dict[str, Any] = {"start": start, "end": end, "use_saved_history": True}
    if symbol:
        args["symbol"] = symbol
        args["use_stock_cfg"] = True
    try:
        from kis_trader.backtest.us_momentum_web_backtest import run_us_momentum_web_backtest
        out = run_us_momentum_web_backtest(args, db=db, ui_def=ui_def)
        trades = out.get("trades") or out.get("trade_list") or []
        if not isinstance(trades, list):
            trades = []
        pnl = sum(float(t.get("pnl") or 0) for t in trades)
        wins = sum(1 for t in trades if float(t.get("pnl") or 0) > 0)
        n = len(trades)
        return {
            "strategy": f"us_momentum:{symbol or 'GLOBAL'}",
            "rc": 0,
            "trades": n,
            "wins": wins,
            "pnl": pnl,
            "wr": (wins / n * 100.0) if n else 0.0,
        }
    except Exception as e:
        # 최소 로드 스모크
        filt = [symbol] if symbol else None
        codes_candles, _, _, codes = load_us_candles(db, start, end, codes_filter=filt)
        return {
            "strategy": f"us_momentum:{symbol or 'GLOBAL'}",
            "rc": 1,
            "error": str(e),
            "codes": len(codes or []),
            "candle_codes": len(codes_candles or {}),
            "trades": 0,
            "wins": 0,
            "pnl": 0.0,
            "wr": 0.0,
        }


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


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--start", required=True)
    ap.add_argument("--end", required=True)
    ap.add_argument("--report", required=True)
    ap.add_argument(
        "--us-only",
        action="store_true",
        help="해외 모멘텀 스모크만 (국장 CLI 스킵)",
    )
    args = ap.parse_args()

    lines: List[str] = []
    lines.append(f"# mid-enroll verify {args.start}..{args.end}")
    lines.append(f"ts={time.strftime('%Y-%m-%d %H:%M:%S')}")
    if args.us_only:
        lines.append("scope=us_only")
    lines.append("")

    # 최근 Optuna JSON 경로
    prefs = [
        "us_momentum_tpe_",
        "us_momentum_tpe_fee_",
        "us_momentum_QQQM_",
        "us_momentum_SPCX_",
        "us_momentum_SPYM_",
        "us_momentum_TSLA_",
    ]
    if not args.us_only:
        prefs.extend(
            (
                "momentum_tpe_",
                "tail_tpe_",
                "breakout_tpe_",
                "scalp_tpe_",
            )
        )
    for pref in prefs:
        p = _latest_optuna_json(pref)
        if p:
            lines.append(f"json:{pref} -> {p}")

    if not args.us_only:
        lines.append("")
        lines.append("## KR web-aligned BT (current DB after apply)")
        for strat in ("momentum", "tail", "breakout", "scalp"):
            print(f"[verify] KR {strat} ...", flush=True)
            try:
                s = _run_kr_cli(strat, args.start, args.end)
                lines.append(f"### {strat} rc={s.get('rc')}")
                for h in s.get("hits") or []:
                    lines.append(f"  {h}")
                lines.append(f"  log={s.get('log_json')}")
            except Exception as e:
                lines.append(f"### {strat} ERR {e}")

    lines.append("")
    lines.append("## US web BT (global + cfg symbols)")
    for sym in ("", "QQQM", "SPCX", "SPYM", "TSLA"):
        tag = sym or "GLOBAL"
        print(f"[verify] US {tag} ...", flush=True)
        try:
            s = _run_us_web(args.start, args.end, sym)
            lines.append(
                f"### us:{tag} trades={s.get('trades')} wr={s.get('wr'):.1f}% "
                f"pnl={s.get('pnl'):.4f}"
            )
            if s.get("error"):
                lines.append(f"  error={s.get('error')}")
        except Exception as e:
            lines.append(f"### us:{tag} ERR {e}")

    # 실매 봇 재시작 (apply 반영)
    lines.append("")
    lines.append("## restart kis_trader_main")
    try:
        r = subprocess.run(
            ["sudo", "systemctl", "restart", "kis_trader_main.service"],
            capture_output=True, text=True, timeout=120,
        )
        time.sleep(2)
        a = subprocess.run(
            ["systemctl", "is-active", "kis_trader_main.service"],
            capture_output=True, text=True, timeout=30,
        )
        lines.append(f"restart_rc={r.returncode} active={a.stdout.strip()}")
    except Exception as e:
        lines.append(f"restart ERR {e}")

    # 실매 정합 메모
    lines.append("")
    lines.append("## live match notes")
    lines.append("- ALIGN_DEFER_MID_BAR_ENROLL + LIVE_ALIGN_ENTRY_FROM_TICK_BUFFER 가 실매·백테 공통")
    lines.append("- 장중 실매 체결 로그의 entry_src=ws_ticks|ohlc_open / 탈락-중편입시가애매 로 확인")
    lines.append("- 미장 10시(22:00 KST) 전 us apply 완료 여부: master 로그 US 구간 END 시각 확인")

    report = Path(args.report)
    report.parent.mkdir(parents=True, exist_ok=True)
    report.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print("\n".join(lines), flush=True)
    print(f"REPORT={report}", flush=True)
    return 0


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