#!/usr/bin/env python3 """ kis_trader/backtest/optuna_grid_narrow.py — 1차 Optuna JSON → 2차 TPE 좁힌 그리드 ================================================================================== 1차 trial pool(PnL 양수 등)의 축별 p25~p75 밴드로 Grid choices 를 줄인다. OPTUNA_GRID_NARROW_JSON 경로가 있으면 optuna_search_space 가 해당 축만 덮어쓴다. """ from __future__ import annotations import json import os from pathlib import Path from typing import Any, Dict, List, Optional from kis_trader.backtest.optuna_mode_combo import ( _build_mode_band_profile, _coerce_numeric, select_mode_pool_rows, ) from kis_trader.backtest.optuna_search_space import _dedupe_preserve_order from kis_trader.utils.env import get_env_float, get_env_int def _strategy_original_grid(strategy: str, mode: str) -> Dict[str, List[Any]]: s = str(strategy or "").strip().lower() m = str(mode or "tpe").strip().lower() or "tpe" if s == "breakout": from kis_trader.backtest.param_search_breakout import _breakout_grids grids = _breakout_grids() # TPE 는 categorical grid 없음 — fast 밴드 envelope 로 2차 narrow grid_mode = m if m in grids else ("fast" if m == "tpe" else m) return dict(grids.get(grid_mode) or {}) if s == "momentum": from kis_trader.backtest.param_search_momentum import _momentum_grids return dict(_momentum_grids(market="KR").get(m) or {}) if s == "us_momentum": from kis_trader.backtest.param_search_momentum import _momentum_grids return dict(_momentum_grids(market="US").get(m) or {}) if s == "scalp": from kis_trader.backtest.param_search_scalping import _scalp_grids return dict(_scalp_grids().get(m) or {}) if s == "tail": from kis_trader.backtest.tail_param_search import _tail_grids return dict(_tail_grids(m) or {}) return {} def _narrow_axis_values( original: List[Any], band: Dict[str, Any], *, min_choices: int = 2, max_choices: int = 8, ) -> List[Any]: if not original: return list(original or []) min_c = max(1, int(min_choices)) max_c = max(min_c, int(max_choices)) if band.get("kind") == "categorical": mode_v = band.get("mode") kept = [v for v in original if str(v) == str(mode_v)] if not kept and mode_v is not None: kept = [mode_v] return _dedupe_preserve_order(kept)[:max_c] if kept else _dedupe_preserve_order(list(original))[:max_c] expand = max(0.0, float(get_env_float("OPTUNA_MODE_REFINE_BAND_EXPAND_IQR", 0.5))) nums_map: Dict[float, Any] = {} for v in original: n = _coerce_numeric(v) if n is not None: nums_map[n] = v if not nums_map: mode_v = band.get("mode") kept = [v for v in original if str(v) == str(mode_v)] return _dedupe_preserve_order(kept or list(original))[:max_c] p25 = float(band["p25"]) p50 = float(band["p50"]) p75 = float(band["p75"]) iqr = float(band["iqr"]) lo = p25 - expand * iqr hi = p75 + expand * iqr kept: List[Any] = [] for n, raw in sorted(nums_map.items()): if lo <= n <= hi: kept.append(raw) if len(kept) < min_c: by_dist = sorted(nums_map.items(), key=lambda t: abs(t[0] - p50)) kept = [raw for _, raw in by_dist[:max_c]] return _dedupe_preserve_order(kept)[:max_c] def build_narrow_grid_from_optuna_data( data: Dict[str, Any], *, mode: Optional[str] = None, strategy: Optional[str] = None, ) -> Dict[str, Any]: """ 1차 JSON → 2차 TPE용 narrow grid dict + 메타. Returns: {"grid": {axis: [choices]}, "meta": {...}} """ strat = str(strategy or data.get("strategy") or "").strip().lower() m = str(mode or data.get("mode") or "tpe").strip().lower() or "tpe" grid_keys = list(data.get("grid_keys") or []) original = _strategy_original_grid(strat, m) if not original: return {"grid": {}, "meta": {"error": f"unknown strategy/mode: {strat}/{m}"}} rows = list(data.get("results_all") or data.get("results") or []) pool = select_mode_pool_rows(rows, data=data) keys = grid_keys or list(original.keys()) profile = _build_mode_band_profile(pool, keys) narrow: Dict[str, List[Any]] = {} stats: Dict[str, Any] = {} for k in keys: orig = list(original.get(k) or []) if not orig: continue band = profile.get(k) if not band: narrow[k] = _dedupe_preserve_order(orig) stats[k] = {"kept": len(narrow[k]), "of": len(orig), "reason": "no_band"} continue narrowed = _narrow_axis_values(orig, band) if len(narrowed) >= 1: narrow[k] = narrowed stats[k] = {"kept": len(narrowed), "of": len(orig), "band": band.get("kind")} else: narrow[k] = _dedupe_preserve_order(orig) stats[k] = {"kept": len(narrow[k]), "of": len(orig), "reason": "fallback_full"} return { "grid": narrow, "meta": { "strategy": strat, "mode": m, "pool_size": len(pool), "pool_kind": str(data.get("mode_pool_kind") or ""), "band_axes": len(profile), "axis_stats": stats, }, } def write_narrow_grid_json(path: str, narrow_grid: Dict[str, List[Any]]) -> str: p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(narrow_grid, indent=2, ensure_ascii=False), encoding="utf-8") return str(p) def build_and_write_narrow_grid( data: Dict[str, Any], out_path: str, *, mode: Optional[str] = None, strategy: Optional[str] = None, ) -> Dict[str, Any]: built = build_narrow_grid_from_optuna_data(data, mode=mode, strategy=strategy) grid = dict(built.get("grid") or {}) write_narrow_grid_json(out_path, grid) meta = dict(built.get("meta") or {}) meta["path"] = str(out_path) meta["n_axes"] = len(grid) return meta def load_narrow_grid_override() -> Dict[str, List[Any]]: """OPTUNA_GRID_NARROW_JSON 파일 → axis choices (없으면 {}).""" raw = str(os.environ.get("OPTUNA_GRID_NARROW_JSON") or "").strip() if not raw or not Path(raw).is_file(): return {} try: obj = json.loads(Path(raw).read_text(encoding="utf-8")) except Exception: return {} if not isinstance(obj, dict): return {} out: Dict[str, List[Any]] = {} for k, v in obj.items(): if isinstance(v, list) and v: out[str(k)] = list(v) return out def resolve_refine_phase2_trials(phase1_trials: int) -> int: """2차 trial 수 — env OPTUNA_MODE_REFINE_PHASE2_TRIALS (0=1차와 동일).""" env_v = int(get_env_int("OPTUNA_MODE_REFINE_PHASE2_TRIALS", 0)) if env_v > 0: return max(1, min(2000, env_v)) return max(1, min(2000, int(phase1_trials or 200)))