279 lines
10 KiB
Python
279 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
kis_trader/backtest/optuna_mode_combo.py — Optuna Top-N 최빈 조합 추출·실측 백테
|
|
================================================================================
|
|
파람서치 종료 후 JSON/로그에 넣기 위한 공통 유틸.
|
|
|
|
기준:
|
|
1) results 중 total_pnl 있는 행만
|
|
2) PnL 내림차순 Top-N (기본 20, env OPTUNA_MODE_TOP_N)
|
|
3) 축별 단순 최빈(표수, PnL 가중 없음) → mode_combo
|
|
4) evaluate_fn(mode_combo) 로 1회 실측 백테 (게이트는 호출측 min_* 에 따름)
|
|
|
|
apply 는 하지 않음 — 확인용 리포트만.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections import Counter
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
from kis_trader.utils.env import get_env_int
|
|
from kis_trader.backtest.optuna_tpe_common import finalize_ratchet_combo
|
|
|
|
logger = logging.getLogger("optuna_mode_combo")
|
|
|
|
EvalFn = Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]
|
|
|
|
|
|
def resolve_mode_top_n(default: int = 20) -> int:
|
|
"""Top-N — env OPTUNA_MODE_TOP_N (기본 20)."""
|
|
n = int(get_env_int("OPTUNA_MODE_TOP_N", int(default)))
|
|
return max(1, n)
|
|
|
|
|
|
def mode_combo_from_results(
|
|
results: List[Dict[str, Any]],
|
|
*,
|
|
top_n: int = 20,
|
|
grid_keys: Optional[List[str]] = None,
|
|
params_key: str = "params",
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Top-N(PnL) 축별 최빈 → mode_combo + 빈도 메타.
|
|
|
|
Returns:
|
|
{
|
|
"top_n": int,
|
|
"pool_size": int,
|
|
"params": {축: 최빈값},
|
|
"freq": {축: {"value": ..., "count": n, "of": pool}},
|
|
"top_pnls": [...],
|
|
}
|
|
"""
|
|
rows = [
|
|
r for r in (results or [])
|
|
if r.get("total_pnl") is not None and abs(float(r.get("total_pnl") or 0)) < 1e15
|
|
]
|
|
rows.sort(
|
|
key=lambda r: (
|
|
-float(r.get("total_pnl") or 0),
|
|
-float(r.get("win_rate") or 0),
|
|
-int(r.get("total_trades") or 0),
|
|
)
|
|
)
|
|
pool = rows[: max(1, int(top_n))]
|
|
if not pool:
|
|
return {
|
|
"top_n": int(top_n),
|
|
"pool_size": 0,
|
|
"params": {},
|
|
"freq": {},
|
|
"top_pnls": [],
|
|
}
|
|
|
|
# 축 집합: grid_keys 우선, 없으면 Top pool params 합집합
|
|
keys: List[str] = []
|
|
if grid_keys:
|
|
keys = [k for k in grid_keys if k]
|
|
if not keys:
|
|
seen = set()
|
|
for r in pool:
|
|
for k in (r.get(params_key) or {}).keys():
|
|
if k not in seen:
|
|
seen.add(k)
|
|
keys.append(k)
|
|
|
|
params: Dict[str, Any] = {}
|
|
freq: Dict[str, Any] = {}
|
|
for k in keys:
|
|
c: Counter = Counter()
|
|
samples: Dict[str, Any] = {}
|
|
for r in pool:
|
|
v = (r.get(params_key) or {}).get(k)
|
|
if v is None and params_key != "merged_params":
|
|
v = (r.get("merged_params") or {}).get(k)
|
|
s = str(v)
|
|
c[s] += 1
|
|
samples.setdefault(s, v)
|
|
if not c:
|
|
continue
|
|
best_s, cnt = c.most_common(1)[0]
|
|
params[k] = samples[best_s]
|
|
freq[k] = {"value": params[k], "count": int(cnt), "of": len(pool)}
|
|
|
|
return {
|
|
"top_n": int(top_n),
|
|
"pool_size": len(pool),
|
|
"params": params,
|
|
"freq": freq,
|
|
"top_pnls": [float(r.get("total_pnl") or 0) for r in pool[:10]],
|
|
}
|
|
|
|
|
|
def _bt_summary(result: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
|
if not result:
|
|
return {
|
|
"ok": False,
|
|
"total_pnl": None,
|
|
"total_trades": None,
|
|
"win_rate": None,
|
|
"pf": None,
|
|
"note": "evaluate returned None (게이트·0건·invalid)",
|
|
}
|
|
return {
|
|
"ok": True,
|
|
"total_pnl": float(result.get("total_pnl") or 0),
|
|
"total_trades": int(result.get("total_trades") or 0),
|
|
"win_rate": float(result.get("win_rate") or 0),
|
|
"pf": float(result.get("pf") or 0) if result.get("pf") is not None else None,
|
|
"score": float(result.get("score") or 0) if result.get("score") is not None else None,
|
|
}
|
|
|
|
|
|
def enrich_out_data_with_mode_combo(
|
|
out_data: Dict[str, Any],
|
|
*,
|
|
evaluate_fn: Optional[EvalFn] = None,
|
|
top_n: Optional[int] = None,
|
|
grid_keys: Optional[List[str]] = None,
|
|
params_key: str = "params",
|
|
log: Optional[logging.Logger] = None,
|
|
on_partial_save: Optional[Callable[[Dict[str, Any]], None]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
out_data['results'] 기준 최빈 추출 → (선택) 실측 백테 → out_data['mode_combo'] 기록 + 로그.
|
|
|
|
evaluate_fn: mode params → evaluate_*_param_combo 결과 dict 또는 None.
|
|
on_partial_save: 최빈 params 기록 직후(실측 전) 호출 — JSON에 mode_combo가 남도록.
|
|
"""
|
|
lg = log or logger
|
|
n = int(top_n) if top_n is not None else resolve_mode_top_n(20)
|
|
keys = grid_keys or list(out_data.get("grid_keys") or [])
|
|
mode_meta = mode_combo_from_results(
|
|
list(out_data.get("results") or []),
|
|
top_n=n,
|
|
grid_keys=keys or None,
|
|
params_key=params_key,
|
|
)
|
|
# 래칫 숫자축 최빈 → 엔진용 ratchet_tiers 재조립 (불일치 방지)
|
|
strat = str(out_data.get("strategy") or "").strip().lower()
|
|
off_tok = "off" if strat == "tail" else ""
|
|
mode_params = finalize_ratchet_combo(
|
|
dict(mode_meta.get("params") or {}),
|
|
off_token=off_tok,
|
|
)
|
|
report: Dict[str, Any] = {
|
|
"method": "top_n_per_axis_mode",
|
|
"top_n": mode_meta["top_n"],
|
|
"pool_size": mode_meta["pool_size"],
|
|
"params": mode_params,
|
|
"freq": mode_meta["freq"],
|
|
"top_pnls": mode_meta["top_pnls"],
|
|
"backtest": None,
|
|
"vs_best": None,
|
|
"note": "trial 번호 없음(축별 최빈 조립). optuna_best_trial_number 와 별개.",
|
|
}
|
|
|
|
best_pnl = None
|
|
best_tr = None
|
|
res0 = (out_data.get("results") or [None])[0]
|
|
if res0:
|
|
best_pnl = float(res0.get("total_pnl") or 0)
|
|
best_tr = int(res0.get("total_trades") or 0)
|
|
|
|
lg.info(
|
|
"📊 [mode] Top-%d 최빈 추출 | pool=%d | top_pnls=%s",
|
|
mode_meta["top_n"],
|
|
mode_meta["pool_size"],
|
|
mode_meta["top_pnls"][:5],
|
|
)
|
|
if mode_params:
|
|
# 축별 빈도 요약 (짧게)
|
|
bits = []
|
|
for k, meta in list(mode_meta["freq"].items())[:12]:
|
|
bits.append(f"{k}={meta['value']}({meta['count']}/{meta['of']})")
|
|
lg.info("📊 [mode] params(일부): %s", " | ".join(bits))
|
|
if "ratchet_tiers" in mode_params:
|
|
lg.info("📊 [mode] ratchet_tiers 재조립=%r", mode_params.get("ratchet_tiers"))
|
|
|
|
# 실측 전에 먼저 JSON에 박아 둠 (실측 중 죽어도 mode_combo.params 는 남음)
|
|
out_data["mode_combo"] = report
|
|
if on_partial_save is not None:
|
|
try:
|
|
on_partial_save(out_data)
|
|
except Exception as exc:
|
|
lg.warning("⚠️ mode_combo 부분저장 실패: %s", exc)
|
|
|
|
if evaluate_fn is not None and mode_params:
|
|
try:
|
|
bt = evaluate_fn(dict(mode_params))
|
|
report["backtest"] = _bt_summary(bt)
|
|
if report["backtest"].get("ok"):
|
|
lg.info(
|
|
"🧪 [mode] 실측 백테 | pnl=%s | trades=%s | wr=%.1f%% | pf=%s",
|
|
report["backtest"]["total_pnl"],
|
|
report["backtest"]["total_trades"],
|
|
float(report["backtest"]["win_rate"] or 0),
|
|
report["backtest"].get("pf"),
|
|
)
|
|
else:
|
|
lg.warning("🧪 [mode] 실측 백테 실패/게이트: %s", report["backtest"].get("note"))
|
|
except Exception as exc:
|
|
report["backtest"] = {"ok": False, "error": str(exc)}
|
|
lg.warning("🧪 [mode] 실측 백테 예외: %s", exc)
|
|
|
|
if best_pnl is not None and report.get("backtest") and report["backtest"].get("ok"):
|
|
mode_pnl = float(report["backtest"]["total_pnl"] or 0)
|
|
best_wr = float(res0.get("win_rate") or 0) if res0 else None
|
|
best_pf = float(res0.get("pf") or 0) if res0 and res0.get("pf") is not None else None
|
|
mode_wr = report["backtest"].get("win_rate")
|
|
mode_pf = report["backtest"].get("pf")
|
|
report["vs_best"] = {
|
|
"best_pnl": best_pnl,
|
|
"best_trades": best_tr,
|
|
"best_wr": best_wr,
|
|
"best_pf": best_pf,
|
|
"mode_pnl": mode_pnl,
|
|
"mode_trades": report["backtest"].get("total_trades"),
|
|
"mode_wr": float(mode_wr) if mode_wr is not None else None,
|
|
"mode_pf": float(mode_pf) if mode_pf is not None else None,
|
|
"delta_pnl": round(mode_pnl - best_pnl, 2),
|
|
"delta_wr": (
|
|
round(float(mode_wr) - float(best_wr), 2)
|
|
if mode_wr is not None and best_wr is not None
|
|
else None
|
|
),
|
|
"delta_pf": (
|
|
round(float(mode_pf) - float(best_pf), 2)
|
|
if mode_pf is not None and best_pf is not None
|
|
else None
|
|
),
|
|
}
|
|
lg.info(
|
|
"📐 [mode vs #1] best_pnl=%s (%s건 wr=%.1f%% pf=%s) | "
|
|
"mode_pnl=%s (%s건 wr=%.1f%% pf=%s) | Δpnl=%+.0f Δwr=%+.1f",
|
|
best_pnl,
|
|
best_tr,
|
|
float(best_wr or 0),
|
|
best_pf,
|
|
mode_pnl,
|
|
report["backtest"].get("total_trades"),
|
|
float(mode_wr or 0),
|
|
mode_pf,
|
|
mode_pnl - best_pnl,
|
|
(float(mode_wr) - float(best_wr)) if mode_wr is not None and best_wr is not None else 0.0,
|
|
)
|
|
|
|
out_data["mode_combo"] = report
|
|
# 다단트레일 추천 (탐색 축 아님 — 미리보기/apply 전용)
|
|
try:
|
|
from kis_trader.backtest.optuna_daily_trail_recommend import (
|
|
attach_daily_trail_recommend,
|
|
)
|
|
|
|
attach_daily_trail_recommend(out_data, log=lg)
|
|
except Exception as exc:
|
|
lg.warning("⚠️ daily_trail_recommend 첨부 실패: %s", exc)
|
|
return out_data
|