#!/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())