kis_bot/kis_trader/scripts/compare_momentum_scan_memo.py

#!/usr/bin/env python3
"""
모멘텀 백테 — baseline vs 스캔 eval 메모이즈 동일성·속도 비교 (프로덕션 코드 수정 없음).

제안 최적화: 틱·호가 없을 때 eval_momentum_buy_at_index 를
(code, signal_idx, daily_cnt, last_exit_dt) 키로 캐시 → 10초 스캔큐 내 중복 연산 제거.
유니버스·포트폴리오·청산 루프는 그대로 유지 → 결과 불변 전제 검증용.

사용:
    cd /home/hoon/kis_bot
    python3 kis_trader/scripts/compare_momentum_scan_memo.py
    python3 kis_trader/scripts/compare_momentum_scan_memo.py --start 2026-06-22 --end 2026-06-26
"""
from __future__ import annotations

import argparse
import copy
import json
import sys
import time
from typing import Any, Dict, List, Optional, Tuple

sys.path.insert(0, "/home/hoon/kis_bot")

from kis_trader.backtest import momentum_backtest_common as mbc
from kis_trader.backtest import momentum_portfolio_backtest as mpb
from kis_trader.backtest.momentum_backtest_common import resolve_momentum_universe
from kis_trader.backtest.param_search_momentum import _load_candles_for_search, _ui_to_engine_params

_orig_eval = mpb.eval_momentum_buy_at_index
_eval_cache: Dict[Tuple, Tuple] = {}
_cache_hits = 0
_cache_miss = 0


def _state_key(state: Dict[str, Any]) -> Tuple:
    led = state.get("last_exit_dt")
    led_s = led.isoformat() if led is not None else ""
    return (int(state.get("daily_cnt", 0) or 0), led_s)


def _params_eval_key(params: Dict[str, Any]) -> Tuple:
    """매수신호에 영향 주는 엔진 파라미터 + indicator cache 객체 id."""
    ic = params.get("_indicator_cache")
    keys = (
        "rsi_period", "mom_rsi_min", "mom_rsi_max", "time_start_hm", "mom_time_end_hm",
        "time_end_hm", "cooldown_min", "max_daily", "max_daily_chg", "min_price",
        "use_defense_filters", "use_high_chase_filter", "use_daily_range_filter",
        "use_ema_filter", "use_rsi_max_filter", "ema_fast_period", "ema_slow_period",
        "high_chase_thr", "mom_vol_mult", "mom_vol_win", "pattern_breakout",
        "pattern_pullback", "chase_lookback_min", "pullback_lookback_min",
        "pullback_min_pct", "pullback_max_pct", "mom_max_from_open_pct",
        "mom_min_from_open_pct", "skip_hts_scan_dupes",
        "_ob_max_spread_pct", "_ob_min_bid_ask_ratio", "_ob_ask_max_mult",
        "_backtest_orderbook_snapshot", "_backtest_program_snapshot",
        "_backtest_log_orderbook_verdict",
    )
    parts: List[Any] = [id(ic)]
    for k in keys:
        v = params.get(k)
        if isinstance(v, dict):
            parts.append(id(v))
        elif isinstance(v, (list, tuple)):
            parts.append(tuple(v) if len(v) < 8 else id(v))
        else:
            parts.append(v)
    return tuple(parts)


def _memo_eval(
    candles: List[Dict],
    i: int,
    params: Dict[str, Any],
    state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
    global _cache_hits, _cache_miss
    key = (id(candles), int(i), _state_key(state), _params_eval_key(params))
    if key in _eval_cache:
        _cache_hits += 1
        return _eval_cache[key]
    r = _orig_eval(candles, i, params, state)
    _eval_cache[key] = r
    _cache_miss += 1
    return r


def _install_memo(enabled: bool) -> None:
    global _cache_hits, _cache_miss
    _cache_hits = 0
    _cache_miss = 0
    _eval_cache.clear()
    if enabled:
        mpb.eval_momentum_buy_at_index = _memo_eval
    else:
        mpb.eval_momentum_buy_at_index = _orig_eval


def _trade_key(t: Dict[str, Any]) -> Tuple:
    return (
        str(t.get("code") or ""),
        str(t.get("buy_time") or t.get("entry_time") or ""),
        str(t.get("sell_time") or t.get("exit_time") or ""),
    )


def _norm_trade(t: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "code": str(t.get("code") or ""),
        "buy_time": str(t.get("buy_time") or t.get("entry_time") or ""),
        "sell_time": str(t.get("sell_time") or t.get("exit_time") or ""),
        "buy_price": round(float(t.get("buy_price") or t.get("entry") or 0), 4),
        "sell_price": round(float(t.get("sell_price") or t.get("exit") or 0), 4),
        "qty": int(t.get("qty") or 0),
        "pnl": int(round(float(t.get("pnl") or 0))),
        "sell_reason": str(t.get("sell_reason") or ""),
        "hold_min": round(float(t.get("hold_min") or 0), 1),
    }


def _run_once(
    cc: Dict,
    eng: Dict,
    univ: Optional[Dict],
    meta: Dict,
    *,
    use_memo: bool,
) -> Tuple[List[Dict], Dict[str, Any], float, Dict[str, int]]:
    _install_memo(use_memo)
    t0 = time.time()
    tr = mbc.run_momentum_backtest_web_aligned(
        cc, copy.deepcopy(eng), univ,
        slot_money=300000, fee_rate=0.00015, sell_tax=0.0018,
        max_stocks=20, total_budget_krw=6000000,
        meta_out=copy.deepcopy(meta),
    )
    elapsed = time.time() - t0
    stats = mbc.summarize_momentum_trades(
        tr, total_budget_krw=6000000, period_days=5,
    )
    cache_info = {"hits": _cache_hits, "misses": _cache_miss}
    return tr, stats, elapsed, cache_info


def _compare_trades(base: List[Dict], opt: List[Dict]) -> Dict[str, Any]:
    bn = [_norm_trade(t) for t in sorted(base, key=_trade_key)]
    on = [_norm_trade(t) for t in sorted(opt, key=_trade_key)]
    out: Dict[str, Any] = {
        "baseline_count": len(bn),
        "memo_count": len(on),
        "identical": bn == on,
        "diff_samples": [],
    }
    if bn == on:
        return out
    n = max(len(bn), len(on))
    for i in range(n):
        b = bn[i] if i < len(bn) else None
        o = on[i] if i < len(on) else None
        if b != o:
            out["diff_samples"].append({"idx": i, "baseline": b, "memo": o})
            if len(out["diff_samples"]) >= 10:
                break
    return out


def _load_params_from_json() -> Dict[str, Any]:
    path = "kis_trader/backtest/results/search_momentum_fast_20260627_015211.json"
    with open(path, encoding="utf-8") as f:
        return json.load(f)["top"][0]["merged_params"]


def main() -> int:
    parser = argparse.ArgumentParser(description="모멘텀 baseline vs scan-eval 메모이즈 비교")
    parser.add_argument("--start", default="2026-06-22")
    parser.add_argument("--end", default="2026-06-26")
    args = parser.parse_args()

    sk = args.start.replace("-", "") + "0000"
    ek = args.end.replace("-", "") + "2359"
    print(f"기간: {args.start} ~ {args.end}")
    print("=" * 70)

    t_load = time.time()
    cc = _load_candles_for_search(args.start, args.end, 3)
    univ, src, n_slots, _, _ = resolve_momentum_universe(
        sk[:8], ek[:8], use_saved_history=True,
    )
    mp = _load_params_from_json()
    eng = _ui_to_engine_params(mp)
    eng.update({
        "slot_money": 300000,
        "max_stocks": 20,
        "total_budget_krw": 6000000,
        "portfolio_mode": True,
    })
    meta = {"start_key": sk, "end_key": ek}
    print(f"로드 {time.time() - t_load:.1f}s | 종목 {len(cc)} | 유니버스 {src} {n_slots}슬롯")
    print("파라미터: search_momentum_fast 1위 merged (EMA OFF, vol×3, sl1.5%)")
    print("=" * 70)

    print("\n[1/2] BASELINE (현재 코드 그대로)")
    tr_b, st_b, el_b, _ = _run_once(cc, eng, univ, meta, use_memo=False)
    print(f"  시간 {el_b:.1f}s | 거래 {st_b['total_trades']} | 손익 {int(st_b['total_pnl']):+,} | "
          f"PF {st_b['pf']} | 승률 {st_b['win_rate']}%")

    print("\n[2/2] MEMO (eval_momentum_buy_at_index 메모이즈 — 프로덕션 미적용, 스크립트만)")
    tr_m, st_m, el_m, cache = _run_once(cc, eng, univ, meta, use_memo=True)
    print(f"  시간 {el_m:.1f}s | 거래 {st_m['total_trades']} | 손익 {int(st_m['total_pnl']):+,} | "
          f"PF {st_m['pf']} | 승률 {st_m['win_rate']}%")
    print(f"  eval 캐시 hit {cache['hits']:,} / miss {cache['misses']:,} "
          f"(hit률 {100.0 * cache['hits'] / max(1, cache['hits'] + cache['misses']):.1f}%)")

    cmp = _compare_trades(tr_b, tr_m)
    speedup = el_b / el_m if el_m > 0 else 0.0
    print("\n" + "=" * 70)
    print("비교 결과")
    print(f"  거래건수: baseline {cmp['baseline_count']} vs memo {cmp['memo_count']}")
    print(f"  손익:     baseline {int(st_b['total_pnl']):+,} vs memo {int(st_m['total_pnl']):+,}")
    print(f"  PF:       baseline {st_b['pf']} vs memo {st_m['pf']}")
    print(f"  속도:     baseline {el_b:.1f}s → memo {el_m:.1f}s ({speedup:.2f}x)")
    print(f"  거래내역 동일: {'✅ YES' if cmp['identical'] else '❌ NO'}")
    if not cmp["identical"]:
        print("\n  ⚠️ 차이 샘플 (최대 10건):")
        for d in cmp["diff_samples"]:
            print(f"    #{d['idx']}")
            print(f"      baseline: {d['baseline']}")
            print(f"      memo:     {d['memo']}")
    print("=" * 70)
    return 0 if cmp["identical"] else 1


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