#!/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 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, ) report: Dict[str, Any] = { "method": "top_n_per_axis_mode", "top_n": mode_meta["top_n"], "pool_size": mode_meta["pool_size"], "params": mode_meta["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_meta["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)) # 실측 전에 먼저 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_meta["params"]: try: bt = evaluate_fn(dict(mode_meta["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) report["vs_best"] = { "best_pnl": best_pnl, "best_trades": best_tr, "mode_pnl": mode_pnl, "mode_trades": report["backtest"].get("total_trades"), "delta_pnl": round(mode_pnl - best_pnl, 2), } lg.info( "📐 [mode vs #1] best_pnl=%s (%s건) | mode_pnl=%s (%s건) | Δ=%+.0f", best_pnl, best_tr, mode_pnl, report["backtest"].get("total_trades"), mode_pnl - best_pnl, ) out_data["mode_combo"] = report return out_data