284 lines
10 KiB
Python
284 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
verify_optuna_tpe_parity_checklist.py — 전전략 TPE 수정사항 스모크 검증
|
|
DB/실매 미변경. suggest 1회 + 축/게이트/TIME apply OFF 검사.
|
|
|
|
.venv/bin/python scripts/verify_optuna_tpe_parity_checklist.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Tuple
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
|
|
def _ok(msg: str) -> None:
|
|
print(f" ✅ {msg}")
|
|
|
|
|
|
def _fail(msg: str) -> None:
|
|
print(f" ❌ {msg}")
|
|
|
|
|
|
def _suggest_once(name: str, fn) -> Tuple[bool, Dict[str, Any], str]:
|
|
import optuna
|
|
import optuna.logging as ol
|
|
|
|
ol.set_verbosity(ol.WARNING)
|
|
study = optuna.create_study(direction="maximize")
|
|
combo: Dict[str, Any] = {}
|
|
pruned = False
|
|
|
|
def objective(trial: optuna.Trial) -> float:
|
|
nonlocal combo, pruned
|
|
try:
|
|
combo = fn(trial)
|
|
except optuna.TrialPruned:
|
|
pruned = True
|
|
raise
|
|
return 1.0
|
|
|
|
try:
|
|
study.optimize(objective, n_trials=1, show_progress_bar=False)
|
|
except Exception as exc:
|
|
return False, {}, str(exc)
|
|
if pruned or not combo:
|
|
return False, {}, "pruned_or_empty"
|
|
return True, combo, ""
|
|
|
|
|
|
def main() -> int:
|
|
fails: List[str] = []
|
|
print("=== Optuna TPE 전전략 수정사항 검증 ===")
|
|
|
|
# 1) 게이트 기본값
|
|
print("\n[1] 탐색/사후 게이트")
|
|
from kis_trader.backtest.optuna_common import (
|
|
OPTUNA_REPORT_MIN_PF_DEFAULT,
|
|
OPTUNA_REPORT_MIN_WIN_RATE_DEFAULT,
|
|
optuna_report_gate_defaults,
|
|
optuna_search_gate_defaults,
|
|
)
|
|
|
|
sw, sp, st = optuna_search_gate_defaults()
|
|
if float(sw) != 0.0:
|
|
fails.append(f"search min_win_rate={sw} (want 0)")
|
|
_fail(f"탐색 min_win_rate={sw}")
|
|
else:
|
|
_ok("탐색 min_win_rate 기본 0")
|
|
if float(sp) != 0.0:
|
|
fails.append(f"search min_pf={sp}")
|
|
_fail(f"탐색 min_pf={sp}")
|
|
else:
|
|
_ok("탐색 min_pf 기본 0")
|
|
if int(st) != 1:
|
|
fails.append(f"search min_trades={st}")
|
|
_fail(f"탐색 min_trades={st}")
|
|
else:
|
|
_ok("탐색 min_trades 기본 1")
|
|
rw, rp, rt = optuna_report_gate_defaults()
|
|
if float(rw) != float(OPTUNA_REPORT_MIN_WIN_RATE_DEFAULT) or float(rp) != float(OPTUNA_REPORT_MIN_PF_DEFAULT):
|
|
fails.append(f"report gates wr/pf={rw}/{rp}")
|
|
_fail(f"report gates {rw}/{rp}")
|
|
else:
|
|
_ok(f"사후 게이트 WR≥{rw} PF≥{rp} min_trades≥{rt}")
|
|
|
|
# 2) TIME apply OFF
|
|
print("\n[2] session_env_patch / TIME apply")
|
|
from kis_trader.backtest.backtest_portfolio_common import session_env_patch
|
|
from kis_trader.utils.env import get_env_bool
|
|
|
|
if get_env_bool("PARAM_SEARCH_APPLY_SESSION_TIME", False):
|
|
fails.append("PARAM_SEARCH_APPLY_SESSION_TIME is ON")
|
|
_fail("PARAM_SEARCH_APPLY_SESSION_TIME=ON (기본은 OFF여야 함)")
|
|
else:
|
|
_ok("PARAM_SEARCH_APPLY_SESSION_TIME 기본 OFF")
|
|
for strat in ("MOMENTUM", "BREAKOUT", "SCALP", "TAIL"):
|
|
patch = session_env_patch(strat, {"time_start_hm": 900, "time_end_hm": 1300})
|
|
if patch:
|
|
fails.append(f"session_env_patch({strat})={patch}")
|
|
_fail(f"{strat} session_env_patch 비어있지 않음: {patch}")
|
|
else:
|
|
_ok(f"{strat} session_env_patch → {{}}")
|
|
|
|
# 3) TPE space: no time_*, skip_hts=false, ratchet numeric
|
|
print("\n[3] TPE suggest 스모크 (전략별 1 trial)")
|
|
from kis_trader.backtest.optuna_breakout_tpe_space import (
|
|
breakout_tpe_axis_keys,
|
|
suggest_breakout_params_tpe,
|
|
)
|
|
from kis_trader.backtest.optuna_momentum_tpe_space import (
|
|
momentum_tpe_axis_keys,
|
|
suggest_momentum_params_tpe,
|
|
)
|
|
from kis_trader.backtest.optuna_scalping_tpe_space import (
|
|
scalp_tpe_axis_keys,
|
|
suggest_scalp_params_tpe,
|
|
)
|
|
from kis_trader.backtest.optuna_tail_tpe_space import (
|
|
suggest_tail_params_tpe,
|
|
tail_tpe_axis_keys,
|
|
)
|
|
from kis_trader.backtest.optuna_tpe_common import (
|
|
RATCHET_TPE_AXIS_KEYS,
|
|
format_ratchet_tiers_string,
|
|
finalize_ratchet_combo,
|
|
)
|
|
|
|
specs = [
|
|
("momentum", suggest_momentum_params_tpe, momentum_tpe_axis_keys, True, ""),
|
|
("breakout", suggest_breakout_params_tpe, breakout_tpe_axis_keys, True, ""),
|
|
("tail", suggest_tail_params_tpe, tail_tpe_axis_keys, True, "off"),
|
|
("scalp", suggest_scalp_params_tpe, scalp_tpe_axis_keys, False, ""),
|
|
]
|
|
for name, fn, keys_fn, has_ratchet, off_tok in specs:
|
|
# 래칫 ON/OFF 둘 다 볼 수 있게 여러 번
|
|
seen_on = False
|
|
seen_off = False
|
|
last_combo: Dict[str, Any] = {}
|
|
last_err = ""
|
|
ok_any = False
|
|
for _ in range(20):
|
|
ok, combo, err = _suggest_once(name, fn)
|
|
if not ok:
|
|
last_err = err
|
|
continue
|
|
ok_any = True
|
|
last_combo = combo
|
|
if has_ratchet:
|
|
if combo.get("ratchet_on") is False:
|
|
seen_off = True
|
|
else:
|
|
seen_on = True
|
|
else:
|
|
break
|
|
if seen_on and seen_off:
|
|
break
|
|
if not ok_any:
|
|
fails.append(f"{name} suggest: {last_err}")
|
|
_fail(f"{name} suggest 실패: {last_err}")
|
|
continue
|
|
combo = last_combo
|
|
keys = keys_fn()
|
|
time_keys = [k for k in keys if "time_" in k or k.endswith("_hm")]
|
|
if time_keys:
|
|
fails.append(f"{name} time axes={time_keys}")
|
|
_fail(f"{name} 시간축 잔존: {time_keys}")
|
|
else:
|
|
_ok(f"{name} 시간축 없음")
|
|
if combo.get("skip_hts_scan_dupes") is not False:
|
|
fails.append(f"{name} skip_hts={combo.get('skip_hts_scan_dupes')}")
|
|
_fail(f"{name} skip_hts_scan_dupes!=False")
|
|
else:
|
|
_ok(f"{name} skip_hts_scan_dupes=False")
|
|
if has_ratchet:
|
|
for rk in ("ratchet_on", "ratchet_n", "ratchet_tiers"):
|
|
if rk not in combo:
|
|
fails.append(f"{name} missing {rk}")
|
|
_fail(f"{name} {rk} 누락")
|
|
for rk in RATCHET_TPE_AXIS_KEYS:
|
|
if rk not in keys:
|
|
fails.append(f"{name} axis missing {rk}")
|
|
_fail(f"{name} axis에 {rk} 없음")
|
|
if not seen_off:
|
|
fails.append(f"{name} OFF sample missing")
|
|
_fail(f"{name} OFF 샘플 미관측")
|
|
else:
|
|
_ok(f"{name} OFF 샘플 확인")
|
|
if not seen_on:
|
|
fails.append(f"{name} ON sample missing")
|
|
_fail(f"{name} ON 샘플 미관측")
|
|
else:
|
|
_ok(f"{name} ON 샘플 확인")
|
|
# ON 고정 샘플로 문자열 형식 검사
|
|
on_combo = None
|
|
for _ in range(24):
|
|
ok, c2, _e = _suggest_once(name, fn)
|
|
if ok and c2.get("ratchet_on"):
|
|
on_combo = c2
|
|
break
|
|
if on_combo is None and not seen_on:
|
|
fails.append(f"{name} could not sample ON")
|
|
_fail(f"{name} ON 샘플 확보 실패")
|
|
elif on_combo:
|
|
tiers = str(on_combo.get("ratchet_tiers") or "")
|
|
if ":" not in tiers:
|
|
fails.append(f"{name} ON tiers empty/bad={tiers!r}")
|
|
_fail(f"{name} ON tiers={tiers!r}")
|
|
else:
|
|
_ok(f"{name} ON → ratchet_tiers={tiers!r}")
|
|
fin = finalize_ratchet_combo(dict(on_combo), off_token=off_tok)
|
|
if str(fin.get("ratchet_tiers")) != str(on_combo.get("ratchet_tiers")):
|
|
fails.append(f"{name} finalize mismatch")
|
|
_fail(
|
|
f"{name} finalize {fin.get('ratchet_tiers')!r} "
|
|
f"!= {on_combo.get('ratchet_tiers')!r}"
|
|
)
|
|
else:
|
|
_ok(f"{name} finalize 일치")
|
|
off_combo = {"ratchet_on": False, "ratchet_n": 0}
|
|
fin_off = finalize_ratchet_combo(off_combo, off_token=off_tok)
|
|
if fin_off.get("ratchet_tiers") != off_tok:
|
|
fails.append(f"{name} OFF finalize={fin_off.get('ratchet_tiers')!r}")
|
|
_fail(f"{name} OFF finalize={fin_off.get('ratchet_tiers')!r}")
|
|
else:
|
|
_ok(f"{name} OFF finalize → {off_tok!r}")
|
|
else:
|
|
if "ratchet_tiers" in combo or "ratchet_on" in combo:
|
|
fails.append(f"{name} unexpected ratchet")
|
|
_fail(f"{name} 래칫 축이 있으면 안 됨")
|
|
else:
|
|
_ok(f"{name} 래칫 축 없음(의도)")
|
|
_ok(f"{name} suggest OK keys={len(keys)} combo={len(combo)}")
|
|
|
|
# 조립 헬퍼 단위
|
|
s = format_ratchet_tiers_string([(5.0, 2.0), (10.0, 1.0)])
|
|
if s != "5:2,10:1":
|
|
fails.append(f"format={s}")
|
|
_fail(f"format_ratchet={s}")
|
|
else:
|
|
_ok("format_ratchet_tiers_string")
|
|
|
|
# 4) 브리핑 min_trades 문구
|
|
print("\n[4] 브리핑 min_trades")
|
|
from kis_trader.backtest.optuna_briefing import build_rule_briefing
|
|
|
|
md = build_rule_briefing(
|
|
{
|
|
"strategy": "momentum",
|
|
"mode": "tpe",
|
|
"start": "2026-07-20",
|
|
"end": "2026-07-21",
|
|
"backtest_days": 2,
|
|
"optuna_n_trials_requested": 1,
|
|
"optuna_trials_completed": 1,
|
|
"min_win_rate": 0,
|
|
"min_pf": 0,
|
|
"min_trades": 1,
|
|
"results_all": [],
|
|
"results_gated": [],
|
|
"report_gates": {"min_win_rate": 40, "min_pf": 1, "min_trades": 1},
|
|
},
|
|
)
|
|
if "min_trades" not in md or "최소 거래" not in md:
|
|
fails.append("briefing missing min_trades expl")
|
|
_fail("브리핑에 min_trades 설명 없음")
|
|
else:
|
|
_ok("브리핑에 min_trades 설명 포함")
|
|
|
|
print("\n=== 결과 ===")
|
|
if fails:
|
|
print(f"FAIL {len(fails)}건")
|
|
for f in fails:
|
|
print(f" - {f}")
|
|
return 1
|
|
print("ALL PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|