#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ verify_three_paths.py — 실매 / param_search / 웹백테 가 '똑같은 엔진 파라미터'로 도는지 검증한다. 목적 ---- JS(웹) 입력값은 표시단위(%)로 받아 쿼리스트링으로 전송되고, 파이썬(api_backtest_*) 에서 ÷100 등으로 엔진 비율(ratio)로 되돌린다. 이 왕복(ratio→표시→ratio)이 손실 없이 카논(실매·param_search 가 쓰는 get_*_defaults_from_db) 과 100% 일치하는지 수치로 확인한다. (소수점/정수 변환 차이 적발) 검증 구조 --------- - 실매(live) : 전략 객체가 get_*_defaults_from_db() 비율값을 그대로 사용. - param_search: base = get_*_defaults_from_db() (코드상 동일 함수 → 자동 일치). - 웹백테(web) : _*_ui_defaults_from_db() (표시%) → (JS는 숫자 그대로 통과) → api_backtest_* 변환(÷100) → 엔진 비율. 따라서 'web 왕복 후 비율' == 'canonical 비율' 이면 세 경로가 동일하다. 이 스크립트는 외부 서버/데이터 없이 변환만 재현해 비교한다. """ from __future__ import annotations import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) # 허용 오차 — 부동소수 반올림(웹 표시 round(x*100,3)) 으로 생길 수 있는 미세 오차 TOL = 1e-9 def _fmt(v): if isinstance(v, float): return f"{v:.10g}" return str(v) def _cmp_rows(rows): """rows: [(name, canonical, web, unit)] → 출력 + 불일치 수 반환.""" bad = 0 print(f" {'필드':28} {'canonical(실매/파서치)':>22} {'web 왕복후':>16} 판정") print(" " + "-" * 78) for name, can, web, unit in rows: if isinstance(can, (int, float)) and isinstance(web, (int, float)): ok = abs(float(can) - float(web)) <= TOL else: ok = str(can) == str(web) mark = "OK " if ok else "❌MISMATCH" if not ok: bad += 1 print(f" {name:28} {_fmt(can):>22} {_fmt(web):>16} {mark} {unit}") return bad # ────────────────────────────────────────────────────────────────────────── # 공통: 웹 표시% → 엔진 비율 (api_backtest_scalping 의 sl/tp 변환과 동일) # ────────────────────────────────────────────────────────────────────────── def _ui_pct_to_ratio(ui_pct) -> float: """웹 입력(%) → 엔진 비율. api_backtest_scalping: float(x)/100 후 abs.""" return abs(float(ui_pct) / 100.0) def verify_momentum() -> int: import backtest_web as bw import kis_trader.engine.momentum_engine as me import kis_trader.engine.scalping_engine as se # noqa: F401 print("\n=== 모멘텀 (MOMENTUM) ===") can = me.get_momentum_defaults_from_db() # 실매 + param_search base (비율) ui = bw._momentum_ui_defaults_from_db(can) # 웹 표시값 (%) # 웹 → 엔진 비율 재현 (api_backtest_scalping 변환) rows = [ ("sl_pct(손절)", can["sl_pct"], _ui_pct_to_ratio(ui["sl_pct"]), "비율"), ("tp_pct(익절)", can["tp_pct"], _ui_pct_to_ratio(ui["tp_pct"]), "비율"), ("tp_max_pct(익절상한)", can["tp_max_pct"], _ui_pct_to_ratio(ui["tp_max_pct"]), "비율"), ("shoulder_min_high", can["shoulder_min_high"], _ui_pct_to_ratio(ui["shoulder_min_high"]), "비율"), ("shoulder_cut_pct", can["shoulder_cut_pct"], _ui_pct_to_ratio(ui["shoulder_cut_pct"]), "비율"), # 정수/그대로 통과 필드 ("mom_rsi_min", can["mom_rsi_min"], float(ui["mom_rsi_min"]), "그대로"), ("mom_rsi_max", can["mom_rsi_max"], float(ui["mom_rsi_max"]), "그대로"), ("mom_vol_mult", can["mom_vol_mult"], float(ui["mom_vol_mult"]), "그대로"), ("mom_vol_win", can["mom_vol_win"], int(float(ui["mom_vol_win"])), "정수"), ("max_daily", can["max_daily"], int(float(ui["max_daily"])), "정수"), ] return _cmp_rows(rows) def verify_breakout() -> int: import backtest_web as bw from kis_trader.strategies.breakout import breakout_ui_to_engine_params print("\n=== 돌파 (BREAKOUT) ===") # 웹과 param_search 는 둘 다 breakout_ui_to_engine_params 사용 (동일 함수). ui = bw._bo_defaults_from_db() web_engine = bw._bo_ui_to_engine_params(ui) # 웹 경로 엔진값 ps_engine = breakout_ui_to_engine_params(dict(ui)) # param_search 경로 (같은 함수) # 실매(live) 가 읽는 DB 비율과도 일치하는지 — breakout strategy 기본 키 from kis_trader.utils.env import get_strategy_env_dict env = get_strategy_env_dict("BREAKOUT") def env_ratio(key, default): v = env.get(key) if v in (None, "", "None"): return float(default) return abs(float(v)) keys = [ ("sl_pct", "stop_loss_pct"), ("tp_pct", "take_profit_pct"), ("trail_pct", "trail_pct"), ("shoulder_min_high_pct", "shoulder_min_high_pct"), ("shoulder_cut_pct", "shoulder_cut_pct"), ] bad = 0 print(" [A] 웹 vs param_search (동일 함수여야 100% 일치)") web_keys = sorted(set(web_engine) & set(ps_engine)) for k in web_keys: a, b = web_engine.get(k), ps_engine.get(k) if isinstance(a, (int, float)) and isinstance(b, (int, float)): if abs(float(a) - float(b)) > TOL: print(f" ❌ {k}: web={a} ps={b}") bad += 1 elif str(a) != str(b): print(f" ❌ {k}: web={a} ps={b}") bad += 1 if bad == 0: print(f" OK — 공통 {len(web_keys)}개 키 전부 일치") print(" [B] 웹 엔진비율 vs 실매 DB 비율") rows = [] for eng_key, _ in keys: if eng_key not in web_engine: continue rows.append((eng_key, web_engine[eng_key], web_engine[eng_key], "비율(웹=엔진)")) # 실제 비교: web_engine 값이 DB 원본 비율과 같은지 rows2 = [] sl_dbf = env_ratio("BREAKOUT_STOP_LOSS_PCT", 0.02) rows2.append(("BREAKOUT_STOP_LOSS_PCT", sl_dbf, abs(float(web_engine.get("stop_loss_pct", web_engine.get("sl_pct", sl_dbf)))), "비율")) bad += _cmp_rows(rows2) return bad def verify_tail() -> int: import backtest_web as bw import kis_trader.engine.tail_engine as te print("\n=== 꼬리 (TAIL/SHORT) ===") can = te.get_tail_defaults_from_db() # 실매 + param_search base ui = bw._tail_ui_defaults_from_db() # 웹 표시값 # 웹 표시(%) → 엔진 비율 재현 후 카논과 비교 (낙폭/손절/익절/어깨 계열) rows = [ ("sl_pct", can.get("sl_pct"), _ui_pct_to_ratio(ui["sl_pct"]), "비율"), ("tp_pct", can.get("tp_pct"), _ui_pct_to_ratio(ui["tp_pct"]), "비율"), ("shoulder_min_high", can.get("shoulder_min_high"), _ui_pct_to_ratio(ui["smin"]), "비율"), ("shoulder_cut_pct", can.get("shoulder_cut_pct"), _ui_pct_to_ratio(ui["scut"]), "비율"), ("rsi_threshold", can.get("rsi_threshold"), float(ui["rsi"]), "그대로"), ("rsi_period", can.get("rsi_period"), int(float(ui["rsi_period"])), "정수"), ("max_daily", can.get("max_daily"), int(float(ui["max_daily"])), "정수"), ("stop_atr_mult", can.get("stop_atr_mult"), float(ui["stop_atr_mult"]), "그대로"), ("target_atr_mult", can.get("target_atr_mult"), float(ui["target_atr_mult"]), "그대로"), ] return _cmp_rows(rows) def verify_scalping() -> int: import backtest_web as bw import kis_trader.engine.scalping_engine as se from kis_trader.backtest.param_search_scalping import _fixed_defaults, _ui_to_engine_params print("\n=== 스캘핑 (SCALP) ===") can = se.get_scalping_defaults_from_db() # 실매 + param_search base (비율) ui = bw._scalp_ui_defaults_from_db() # 웹 표시값 (%) # [A] 웹 표시(%) → ÷100 왕복 == 실매 비율 rows = [ ("sl_pct(손절)", can["sl_pct"], _ui_pct_to_ratio(ui["sl_pct"]), "비율"), ("tp_pct(익절)", can["tp_pct"], _ui_pct_to_ratio(ui["tp_pct"]), "비율"), ("tp_max_pct(익절상한)", can["tp_max_pct"], _ui_pct_to_ratio(ui["tp_max_pct"]), "비율"), ("drop_rate(낙폭)", can["drop_rate"], _ui_pct_to_ratio(ui["drop_rate"]), "비율"), ("shoulder_min_high", can["shoulder_min_high"], _ui_pct_to_ratio(ui["shoulder_min_high"]), "비율"), ("shoulder_cut_pct", can["shoulder_cut_pct"], _ui_pct_to_ratio(ui["shoulder_cut_pct"]), "비율"), ("rsi_oversold", can["rsi_oversold"], float(ui["rsi_oversold"]), "그대로"), ("rsi_period", can["rsi_period"], int(float(ui["rsi_period"])), "정수"), ("max_daily", can["max_daily"], int(float(ui["max_daily"])), "정수"), ("use_macd_cross", bool(can.get("use_macd_cross")), bool(ui.get("use_macd_cross")), "bool"), ("skip_hts_scan_dupes", bool(can.get("skip_hts_scan_dupes")), bool(ui.get("skip_hts_scan_dupes")), "bool"), ] bad = _cmp_rows(rows) # [B] param_search _ui_to_engine == 실매 비율 print(" [B] param_search _ui_to_engine vs 실매 비율") ps_ui = _fixed_defaults() eng = _ui_to_engine_params(dict(ps_ui)) rows_b = [] for k in ( "sl_pct", "tp_pct", "tp_max_pct", "drop_rate", "shoulder_min_high", "shoulder_cut_pct", ): rows_b.append((k, can.get(k), eng.get(k), "비율")) bad += _cmp_rows(rows_b) return bad def main() -> int: total_bad = 0 for fn in (verify_momentum, verify_breakout, verify_tail, verify_scalping): try: total_bad += fn() except Exception as e: # noqa: BLE001 import traceback print(f"\n[ERROR] {fn.__name__}: {e}") traceback.print_exc() total_bad += 1 print("\n" + "=" * 80) if total_bad == 0: print("✅ 검증 통과 — 실매 / param_search / 웹백테 가 동일한 엔진 파라미터로 돕니다.") else: print(f"❌ 불일치 {total_bad}건 — 위 MISMATCH 항목을 확인하세요.") return total_bad if __name__ == "__main__": sys.exit(0 if main() == 0 else 1)