#!/usr/bin/env python3 """ kis_trader/backtest/param_search_optuna.py — Optuna TPE 파라미터 탐색 (전략별) ============================================================================== 기존 Grid CLI(tail_param_search.py 등)는 그대로 두고, Bayesian(TPE) add-on. 현재 구현: --strategy tail | momentum | breakout | scalp 실행 예: # 꼬리 python3 kis_trader/backtest/param_search_optuna.py --strategy tail --mode fast --trials 200 # 모멘텀 (1위 정렬 기본 score=순익/MDD) # --mode fine → 기존 Grid 이산 메뉴 + categorical # --mode tpe → 연속 float/int (TPE 가 구간 축소, Grid 메뉴 미사용) python3 kis_trader/backtest/param_search_optuna.py --strategy momentum --mode tpe --trials 200 # 돌파 python3 kis_trader/backtest/param_search_optuna.py --strategy breakout --mode fast --trials 200 # 스캘핑 RSI V자 (trigger=진입 / exit=청산) python3 kis_trader/backtest/param_search_optuna.py --strategy scalp --mode trigger --trials 100 Win11 + VM 동시 분산: 같은 study-name · 같은 storage(141/kis_optuna) 로 각각 --trials 실행. DB 적용: --apply-best (사후게이트 results_gated 통과 trial → env_config, 총손익≤0 이면 스킵) Env (선택): OPTUNA_DB_NAME=kis_optuna # 기본. 변경 시에만 설정 OPTUNA_STORAGE_URL=... # 전체 URL 직접 지정 시 위보다 우선 OPTUNA_TAIL_STUDY_NAME=... # study 이름 고정 PARAM_SEARCH_OPTUNA_MIN_WIN_RATE / MIN_PF # 탐색 게이트 기본 0 (TPE 학습) PARAM_SEARCH_OPTUNA_REPORT_MIN_WIN_RATE / MIN_PF # 사후 후보·apply (기본 40 / 1.0) PARAM_SEARCH_OPTUNA_BRIEFING_AI=1 # 최종 JSON 시 Claude 보강(키 없으면 규칙만) """ from __future__ import annotations import argparse import json import logging import os import signal import sys import time # 2026-09-06: 쓰레기 스킵·Rust 강제 세팅 제거 (docs/정합성.md §9) # - CANDLE_GARBAGE_FALLBACK: bar_is_garbage 가 wall-clock(recv_ts) 기준으로 정정되어 # 실매 RAM 3초컷과 동일 논리 → 강제 OFF 불필요. DB env(기본 True) 그대로 사용. # - BACKTEST_USE_RUST: 사용자가 웹 UI/CLI 에서 명시적으로 켤 때만 활성화 (기본 Python). # 이전엔 강제 "1" → use_rust=False 잡도 Rust 로 돌아 정합 사고 (rust_engine_parity_port_plan.md §부록 B #1) from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Any, Dict, List, Optional HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(os.path.dirname(HERE)) if ROOT not in sys.path: sys.path.insert(0, ROOT) if HERE not in sys.path: sys.path.insert(0, HERE) import optuna from optuna.samplers import RandomSampler, TPESampler from database import TradeDB from kis_trader.backtest import tail_backtest_common as tbc from kis_trader.backtest.optuna_common import ( OPTUNA_STRATEGIES, announce_optuna_json_path, build_optuna_result_tiers, ensure_optuna_gate_env_defaults, optuna_run_lock_name, optuna_search_gate_defaults, pick_gated_apply_trial, release_shared_tick_store, resolve_optuna_storage_url, resolve_study_name, set_optuna_trial_stability_attrs, stability_fields_from_trial_attrs, ) from kis_trader.backtest.optuna_mode_combo import enrich_out_data_with_mode_combo from kis_trader.backtest.optuna_breakout import ( apply_best_breakout_trial, prepare_breakout_search_context, run_breakout_optuna, ) from kis_trader.backtest.optuna_momentum import ( apply_best_momentum_trial, prepare_momentum_search_context, run_momentum_optuna, ) from kis_trader.backtest.optuna_scalping import ( apply_best_scalp_trial, prepare_scalp_search_context, run_scalp_optuna, ) from kis_trader.backtest.optuna_dart import ( apply_best_dart_trial, prepare_dart_search_context, run_dart_optuna, ) from kis_trader.backtest.optuna_search_space import suggest_tail_params, tail_grid_axis_keys from kis_trader.backtest.optuna_tail_tpe_space import ( normalize_tpe_tail_entry_mode, suggest_tail_params_tpe, tail_tpe_axis_keys, ) from kis_trader.backtest.param_search_cli_common import ( add_portfolio_cli_args, add_search_filter_cli_args, combo_passes_search_filters, ) from kis_trader.backtest.param_search_pool import try_acquire_run_lock from kis_trader.backtest.tail_param_search import ( TAIL_GRID_AXIS_HINTS_KO, _results_dir_for_write, _tail_params_to_env_map, apply_params_to_db, evaluate_tail_param_combo, ) from kis_trader.engine import tail_engine as te from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params from kis_trader.utils.env import get_env_bool, get_env_from_db, get_env_int logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger("param_search_optuna") # 게이트 미통과 trial — Optuna direction=maximize 에서 최하점 _FAIL_OBJECTIVE = -1e18 # 전략별 --mode 허용값 (Grid CLI 와 동일) STRATEGY_MODES: Dict[str, List[str]] = { "tail": ["fast", "coarse", "fine", "wide", "full", "massive", "tpe"], "momentum": ["fast", "exit", "rr", "coarse", "fine", "wide", "full", "tpe"], "us_momentum": ["fast", "exit", "rr", "coarse", "fine", "wide", "full", "tpe"], "breakout": ["fast", "coarse", "fine", "wide", "full", "tpe"], "scalp": ["fast", "trigger", "exit", "coarse", "fine", "full", "wide", "tpe"], "dart": ["fast", "coarse", "fine"], } @dataclass class TailSearchContext: """Optuna objective 1회 로드 — trial 마다 재사용.""" start: str end: str mode: str tail_tf: int base_params: Dict[str, Any] candles_by_code: Dict[str, List[Dict]] total_candles: int has_holding_peak: bool universe_by_slot: Optional[Dict[str, List[str]]] universe_source: str universe_history_slots: int scan_interval_min: int ticks_by_code: Any tick_rows: int orderbook_by_code: Dict[str, Any] program_by_code: Dict[str, Any] log_verdict_by_code: Dict[str, Any] trigger_snap_meta: Dict[str, Any] fee_rate: float sell_tax: float slot_money: float max_stocks: int total_budget_krw: float period_days: int portfolio: Dict[str, Any] grid_keys: List[str] ob_filter_on: bool cache_holder: Dict[str, Any] = field(default_factory=dict) shared_tick_store: Any = None # ws_ticks 공유메모리 핸들 (종료 시 unlink) tpe_entry_mode: str = "align" # TPE 고정 진입모드(탐색 축 아님) def prepare_tail_search_context( start: str, end: str, mode: str, *, timeframe: int = 3, use_fallback_universe: bool = False, time_start_hm: Optional[int] = None, time_end_hm: Optional[int] = None, slot_money: Optional[float] = None, max_stocks: Optional[int] = None, total_budget_krw: Optional[float] = None, orderbook_filter: str = "off", history_source: Optional[str] = None, entry_mode: Optional[str] = None, ) -> Optional[TailSearchContext]: """ run_search 와 동일한 데이터·base_params 1회 로드 (Grid 중복 최소화). 데이터 없으면 None. """ db = TradeDB() try: base_params = te.get_tail_defaults_from_db(db) if os.environ.get("BACKTEST_USE_RUST") == "1": base_params["use_rust"] = True if time_start_hm is not None: base_params["time_start_hm"] = int(time_start_hm) if time_end_hm is not None: base_params["time_end_hm"] = int(time_end_hm) _ob_mode = (orderbook_filter or "off").strip().lower() if _ob_mode == "off": base_params["_orderbook_filter_enabled"] = False elif _ob_mode == "on": base_params["_orderbook_filter_enabled"] = True from kis_trader.backtest.optuna_tpe_common import optuna_tpe_needs_orderbook_feed need_ob_feed = optuna_tpe_needs_orderbook_feed(mode, _ob_mode) ob_filter_on = ( bool(base_params.get("_orderbook_filter_enabled")) or _ob_mode == "auto" or need_ob_feed ) if need_ob_feed: base_params["backtest_use_trigger_snapshot_db"] = True logger.info( "📌 호가필터: %s (%s)%s", _ob_mode.upper(), "스냅로드" if need_ob_feed else ("적용" if ob_filter_on else "스킵 — 코어 파라미터 순수 탐색"), " · TPE 호가축" if need_ob_feed else "", ) from kis_trader.backtest.backtest_portfolio_common import load_portfolio_env_row r = load_portfolio_env_row(db) fee_rate, sell_tax, _slot_from_fee = tbc.fee_and_slot_from_env_row(r) portfolio = tbc.resolve_tail_portfolio_params( r, base_params, slot_money=slot_money if slot_money is not None else _slot_from_fee, max_stocks=max_stocks, total_budget_krw=total_budget_krw, ) slot_money_f = float(portfolio["slot_money"]) max_stocks_i = int(portfolio["max_stocks"]) total_budget_f = float(portfolio["total_budget_krw"]) tbc.merge_tail_portfolio_into_params(base_params, portfolio) base_params["capital"] = float( r.get("BACKTEST_CAPITAL") or base_params.get("capital") or 100_000_000.0 ) period_days = max( 1, (datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days + 1, ) tail_tf = int(timeframe) if tail_tf not in tbc.VALID_TIMEFRAMES: logger.error("❌ timeframe 은 3·5·15·60 중 하나여야 합니다 (backtest_web 과 동일)") return None start_key, end_key, start_ymd, end_ymd = tbc.date_keys(start, end) use_saved_history = not use_fallback_universe from kis_trader.backtest.universe_history_source import ( resolve_backtest_universe_history_source, ) _hs = resolve_backtest_universe_history_source(history_source) universe_by_slot, universe_source, universe_history_slots, scan_interval_min = ( tbc.resolve_tail_universe( start_ymd, end_ymd, use_saved_history=use_saved_history, history_source=_hs, ) ) # scan_at 타임라인과 슬롯 dict 동일 소스 스태시 base_params["_universe_history_source"] = _hs if use_fallback_universe: print("📌 [유니버스] --fallback-universe: 저장 이력 무시 → ws_candles 전 종목") elif str(universe_source or "").startswith("history"): avg = ( sum(len(v) for v in universe_by_slot.values()) / max(1, universe_history_slots) if universe_by_slot else 0 ) print( f"✅ 유니버스: SHORT 저장 이력 src={universe_source} | " f"{universe_history_slots:,}슬롯 · 평균 {avg:.1f}종목" ) else: print("📌 [유니버스] 저장 이력 없음 → ws_candles 전 종목 (웹 폴백과 동일)") base_params = dict(base_params) base_params["scan_interval_min"] = scan_interval_min base_params["timeframe"] = tail_tf from kis_trader.engine.tail_tick_replay import ( tail_backtest_use_tick_db as _tail_use_tick, tail_backtest_use_tick_exit as _tail_use_tick_exit, ) base_params.setdefault("backtest_use_tick_db", _tail_use_tick(None)) base_params.setdefault("backtest_use_tick_exit", _tail_use_tick_exit(None)) # 절대규칙: Optuna/파람은 OHLC 폴백으로 숫자 변조 금지 (DB에 ON이어도 강제 OFF) base_params["backtest_tick_fallback_ohlc"] = False tpe_entry_mode = "align" if mode == "tpe": _raw_em = entry_mode if _raw_em in (None, "", "None"): _raw_em = get_env_from_db("TAIL_PARAM_SEARCH_ENTRY_MODE", "") or "align" tpe_entry_mode = normalize_tpe_tail_entry_mode(_raw_em) base_params["entry_mode"] = tpe_entry_mode logger.info( "📌 TPE 진입모드 고정: %s (한 스터디=한 모드, 탐색 축 아님)", tpe_entry_mode, ) if base_params.get("backtest_use_tick_db") or base_params.get("backtest_use_tick_exit"): logger.info("📌 틱재생(ws_ticks): ON — OHLC 폴백 강제 OFF (정합 절대규칙)") logger.info( f"📅 데이터 로드: {start} ~ {end} | TF={tail_tf} | " f"유니버스={universe_source} | 매수시간 " f"{base_params.get('time_start_hm', 930):04d}-{base_params.get('time_end_hm', 1500):04d}" ) try: from kis_trader.backtest.optuna_feed_trace import log_bt_feed_chain_banner log_bt_feed_chain_banner(context="Optuna-TAIL") except Exception: pass rsi_period = int(base_params.get("rsi_period", 14)) candles_by_code, total_candles, has_holding_peak = tbc.load_tail_candles_by_code( db, start_key, end_key, tail_tf, rsi_period=rsi_period, ) if not candles_by_code: logger.info("❌ 백테스트할 데이터가 없습니다.") return None ticks_by_code: Dict[str, Any] = {} tick_rows = 0 from kis_trader.engine.tail_tick_replay import tail_backtest_wants_tick_replay from kis_trader.backtest.tail_tick_loader import load_tail_ticks_by_code, tick_coverage_stats _tick_probe = dict(base_params) if tail_backtest_wants_tick_replay(_tick_probe): _tick_db = TradeDB() try: ticks_by_code, tick_rows = load_tail_ticks_by_code( _tick_db, start_key, end_key, set(candles_by_code.keys()), ) finally: _tick_db.close() if tick_rows > 0: tick_meta = tick_coverage_stats(candles_by_code, ticks_by_code) cov = tick_meta.get("tick_bar_coverage_pct", 0) logger.info( "✅ ws_ticks %s건 | 3분봉 커버리지 %s%% (%s/%s종목)", f"{tick_rows:,}", cov, tick_meta.get("tick_codes_with_data", 0), tick_meta.get("tick_codes_total", 0), ) elif tail_backtest_wants_tick_replay(_tick_probe): logger.warning("⚠️ ws_ticks 없음 — OHLC 폴백 (WS_TICK_SAVE_ENABLED 후 재탐색)") # ── ws_ticks 공유메모리 (Optuna, opt-in) — dict→numpy 컬럼 shared_memory 로 RAM 절감 ── # 끄려면 OPTUNA_PARAM_SEARCH_SHARED_TICKS=0. numpy/shm 미지원·빌드 실패 시 자동 폴백. shared_tick_store = None if get_env_bool("OPTUNA_PARAM_SEARCH_SHARED_TICKS", True) and ticks_by_code: from kis_trader.backtest.shared_ticks import build_shared_ticks_view _view, shared_tick_store = build_shared_ticks_view(ticks_by_code, enabled=True) if shared_tick_store is not None: import atexit as _atexit _atexit.register(shared_tick_store.unlink) # 크래시 시 /dev/shm 누수 방지 logger.info("📦 ws_ticks 공유메모리 ON (Optuna) — dict 사본 제거, RAM 절감") ticks_by_code = _view import gc as _gc _gc.collect() try: import ctypes as _ctypes _ctypes.CDLL("libc.so.6").malloc_trim(0) except Exception: pass from kis_trader.backtest.tail_param_search import _tail_grids # tpe = 연속 Optuna (Grid 미사용). 알 수 없는 mode 가 fast 로 폴백되면 안 됨. if mode == "tpe": pre_grid: Dict[str, Any] = {} base_params["skip_hts_scan_dupes"] = False logger.info( "📌 mode=tpe — 연속(float/int) 탐색 (Grid categorical 미사용, TPE 가 구간 축소)" ) else: pre_grid = _tail_grids(mode) _ob_axes = ("max_spread_pct", "min_bid_ask_ratio") _ob_sweeping = any(len(set(pre_grid.get(k) or [])) > 1 for k in _ob_axes) if ob_filter_on and _ob_sweeping: base_params["backtest_use_kiwoom_body_snapshot"] = True base_params["_backtest_use_kiwoom_body"] = True logger.info("📌 호가필터 스윕 활성 → kiwoom_0d 본체 재계산") orderbook_by_code: Dict[str, Any] = {} program_by_code: Dict[str, Any] = {} log_verdict_by_code: Dict[str, Any] = {} trigger_snap_meta: Dict[str, Any] = {} try: from kis_trader.backtest.trigger_snapshot_loader import load_trigger_snapshots_by_code orderbook_by_code, program_by_code, trigger_snap_meta = load_trigger_snapshots_by_code( db, start_key, end_key, set(candles_by_code.keys()), engine_params=base_params, strategy="TAIL", ) log_verdict_by_code = trigger_snap_meta.get("log_verdict_by_code") or {} ob_rows = int(trigger_snap_meta.get("ws_orderbook_rows_loaded") or 0) pg_rows = int(trigger_snap_meta.get("ws_program_rows_loaded") or 0) logger.info( "✅ TRIGGER 스냅샷 ws_orderbook %s건 | ws_program %s건", f"{ob_rows:,}", f"{pg_rows:,}", ) except Exception as _snap_ex: logger.debug("trigger snapshot 로드 스킵: %s", _snap_ex) logger.info( f"📦 종목: {len(candles_by_code)}개 | 캔들: {total_candles:,}개 | portfolio_mode=ON" ) logger.info( f"💼 포트폴리오: 1회 {slot_money_f:,.0f}원 | 동시 {max_stocks_i}종 | " f"총한도 {total_budget_f:,.0f}원" ) if portfolio.get("budget_warning"): logger.warning(f"💰 {portfolio['budget_warning']}") if mode != "tpe" and "entry_mode" not in pre_grid: _search_entry = get_env_from_db("TAIL_PARAM_SEARCH_ENTRY_MODE", "") if _search_entry not in (None, "", "None"): base_params["entry_mode"] = str(_search_entry).strip().lower() cache_holder: Dict[str, Any] = {} attach_indicator_caches_to_params(cache_holder, candles_by_code) return TailSearchContext( start=start, end=end, mode=mode, tail_tf=tail_tf, base_params=base_params, candles_by_code=candles_by_code, total_candles=total_candles, has_holding_peak=has_holding_peak, universe_by_slot=universe_by_slot, universe_source=universe_source, universe_history_slots=universe_history_slots, scan_interval_min=scan_interval_min, ticks_by_code=ticks_by_code, tick_rows=tick_rows, orderbook_by_code=orderbook_by_code, program_by_code=program_by_code, log_verdict_by_code=log_verdict_by_code, trigger_snap_meta=trigger_snap_meta, fee_rate=fee_rate, sell_tax=sell_tax, slot_money=slot_money_f, max_stocks=max_stocks_i, total_budget_krw=total_budget_f, period_days=period_days, portfolio=portfolio, grid_keys=tail_tpe_axis_keys(tpe_entry_mode) if mode == "tpe" else tail_grid_axis_keys(mode), ob_filter_on=ob_filter_on, cache_holder=cache_holder, shared_tick_store=shared_tick_store, tpe_entry_mode=tpe_entry_mode, ) finally: db.close() def _make_sampler(name: str, seed: Optional[int]): n = (name or "tpe").strip().lower() if n == "random": return RandomSampler(seed=seed) # multivariate TPE + 조건부 suggest 시 independent sampling 경고가 trial마다 폭주 → 억제 return TPESampler(seed=seed, multivariate=True, warn_independent_sampling=False) def run_tail_optuna( ctx: TailSearchContext, *, n_trials: int, storage_url: str, study_name: str, min_trades: int, min_win_rate: float, min_pf: float, sort_by: str = "score", sampler_name: str = "tpe", seed: Optional[int] = None, n_jobs: int = 1, show_progress: bool = True, ) -> optuna.Study: """Optuna study 실행 — trial.user_attrs 에 상세 결과 저장.""" direction = "maximize" sampler = _make_sampler(sampler_name, seed) study = optuna.create_study( study_name=study_name, storage=storage_url, load_if_exists=True, direction=direction, sampler=sampler, ) from kis_trader.backtest.optuna_study_store import ( bind_study_trials, clamp_optimize_n_trials, finalize_optuna_export, make_study_goal_stop_callback, ) bind_study_trials(study, n_trials=n_trials, log=logger) n_trials = clamp_optimize_n_trials(study, n_trials, log=logger) import uuid rust_session_id = f"optuna_tail_{uuid.uuid4().hex[:8]}" try: from kis_trader.engine.tail_engine import init_rust_session, clear_rust_session init_rust_session(rust_session_id, ctx.candles_by_code, ctx.ticks_by_code) except Exception as e: logger.warning(f"Failed to init_rust_session: {e}") def objective(trial: optuna.Trial) -> float: if ctx.mode == "tpe": combo = suggest_tail_params_tpe(trial, entry_mode=ctx.tpe_entry_mode) else: combo = suggest_tail_params(trial, ctx.mode) # Rust 세션 ID를 파라미터에 넘겨서 하위에서 사용할 수 있게 함 combo["_rust_session_id"] = rust_session_id result = evaluate_tail_param_combo( combo, base_params=ctx.base_params, candles_by_code=ctx.candles_by_code, fee_rate=ctx.fee_rate, sell_tax=ctx.sell_tax, min_trades=min_trades, min_win_rate=min_win_rate, min_pf=min_pf, universe_by_slot=ctx.universe_by_slot, slot_money=ctx.slot_money, max_stocks=ctx.max_stocks, total_budget_krw=ctx.total_budget_krw, period_days=ctx.period_days, cache_holder=ctx.cache_holder, ticks_by_code=ctx.ticks_by_code, orderbook_by_code=ctx.orderbook_by_code, program_by_code=ctx.program_by_code, log_verdict_by_code=ctx.log_verdict_by_code, ) if result is None: trial.set_user_attr("gates_ok", False) return _FAIL_OBJECTIVE from kis_trader.backtest.optuna_common import ( optuna_score_fields_from_trial, optuna_store_trial_score_user_attrs, ) trial.set_user_attr("gates_ok", True) trial.set_user_attr("total_pnl", float(result["total_pnl"])) trial.set_user_attr("win_rate", float(result["win_rate"])) trial.set_user_attr("pf", float(result.get("pf") or 0)) trial.set_user_attr("mdd", float(result.get("mdd") or 0)) trial.set_user_attr("total_trades", int(result["total_trades"])) trial.set_user_attr("params_json", json.dumps(result["params"], ensure_ascii=False)) set_optuna_trial_stability_attrs(trial, result) return optuna_store_trial_score_user_attrs( trial, result, sort_by, start=ctx.start, end=ctx.end, strategy="tail", ) logger.info( "🔬 Optuna 시작 | study=%s | trials=%d | sampler=%s | storage=%s | n_jobs=%d", study_name, n_trials, sampler_name, storage_url, n_jobs, ) t0 = time.time() try: if n_trials <= 0: logger.info("📌 추가 trial 없음 — 기존 study 결과만 정리") else: study.optimize( objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress, callbacks=[make_study_goal_stop_callback(logger)], ) elapsed = time.time() - t0 logger.info("✅ Optuna 완료 | %.1f초 | 완료 trial %d", elapsed, len(study.trials)) # JSON export — study.user_attrs 기준 (n_jobs>1 에도 안전) from kis_trader.backtest.optuna_common import optuna_score_fields_from_trial passing: List[Dict[str, Any]] = [] for trial in study.trials: if trial.state != optuna.trial.TrialState.COMPLETE: continue if not trial.user_attrs.get("gates_ok"): continue params_raw = trial.user_attrs.get("params_json") or "{}" try: combo = json.loads(params_raw) except json.JSONDecodeError: combo = dict(trial.params) row = { "params": combo, "apply_cfg": {**ctx.base_params, **combo}, "total_trades": int(trial.user_attrs.get("total_trades") or 0), "win_rate": float(trial.user_attrs.get("win_rate") or 0), "total_pnl": float(trial.user_attrs.get("total_pnl") or 0), "pf": float(trial.user_attrs.get("pf") or 0), "mdd": float(trial.user_attrs.get("mdd") or 0), **optuna_score_fields_from_trial(trial), "period_daily_avg_pnl": float(trial.user_attrs.get("period_daily_avg_pnl") or 0), "optuna_trial_number": trial.number, } row.update(stability_fields_from_trial_attrs(trial)) passing.append(row) from kis_trader.backtest.optuna_common import _sort_optuna_rows passing = _sort_optuna_rows(passing, sort_by) tiers = build_optuna_result_tiers(passing, sort_by=sort_by) out_data = { "engine": "optuna", "strategy": "tail", "mode": ctx.mode, "start": ctx.start, "end": ctx.end, "timeframe": ctx.tail_tf, "universe_source": ctx.universe_source, "universe_history_slots": ctx.universe_history_slots, "slot_money": int(ctx.slot_money), "max_stocks": ctx.max_stocks, "total_budget_krw": int(ctx.total_budget_krw), "portfolio_mode": True, "budget_warning": ctx.portfolio.get("budget_warning"), "backtest_days": ctx.period_days, "min_trades": min_trades, "min_win_rate": min_win_rate, "min_pf": min_pf, "sort_by": sort_by, "grid_keys": ctx.grid_keys, "grid_axis_hints": {k: TAIL_GRID_AXIS_HINTS_KO[k] for k in ctx.grid_keys if k in TAIL_GRID_AXIS_HINTS_KO}, "holding_peak_in_candles": ctx.has_holding_peak, "optuna_study_name": study_name, "optuna_storage": storage_url, "optuna_n_trials_requested": n_trials, "optuna_trials_completed": len(study.trials), "optuna_best_value": study.best_value if study.best_trial else None, "optuna_best_trial_number": study.best_trial.number if study.best_trial else None, "elapsed_sec": round(elapsed, 1), **tiers, } from kis_trader.backtest.optuna_common import annotate_optuna_period_daily_avg annotate_optuna_period_daily_avg(out_data) ts = datetime.now().strftime("%Y%m%d_%H%M%S") out_name = f"optuna_tail_{ctx.mode}_{ts}.json" out_dir = _results_dir_for_write() out_path = os.path.join(out_dir, out_name) try: with open(out_path, "w", encoding="utf-8") as f: json.dump(out_data, f, indent=2, ensure_ascii=False) except OSError: fb = os.path.join(os.path.expanduser("~"), ".kis_bot_search_results") os.makedirs(fb, exist_ok=True) out_path = os.path.join(fb, out_name) with open(out_path, "w", encoding="utf-8") as f: json.dump(out_data, f, indent=2, ensure_ascii=False) logger.warning("⚠️ results/ 쓰기 권한 없음 → 폴백 저장: %s", out_path) announce_optuna_json_path( out_path, strategy="tail", mode=ctx.mode, note="중간저장(mode 전)", log=logger, ) def _eval_mode(combo: Dict[str, Any]) -> Optional[Dict[str, Any]]: return evaluate_tail_param_combo( combo, base_params=ctx.base_params, candles_by_code=ctx.candles_by_code, fee_rate=ctx.fee_rate, sell_tax=ctx.sell_tax, min_trades=1, min_win_rate=0.0, min_pf=0.0, universe_by_slot=ctx.universe_by_slot, slot_money=ctx.slot_money, max_stocks=ctx.max_stocks, total_budget_krw=ctx.total_budget_krw, period_days=ctx.period_days, cache_holder=ctx.cache_holder, ticks_by_code=ctx.ticks_by_code, orderbook_by_code=ctx.orderbook_by_code, program_by_code=ctx.program_by_code, log_verdict_by_code=ctx.log_verdict_by_code, include_trades=True, ) def _save_partial(_data: Dict[str, Any]) -> None: try: with open(out_path, "w", encoding="utf-8") as f: json.dump(_data, f, indent=2, ensure_ascii=False) except OSError as exc: logger.warning("⚠️ mode_combo 부분저장 실패: %s", exc) return announce_optuna_json_path( out_path, strategy="tail", mode=ctx.mode, note="mode_combo params 저장(실측 전)", log=logger, ) def _enrich() -> None: enrich_out_data_with_mode_combo( out_data, evaluate_fn=_eval_mode, grid_keys=ctx.grid_keys, log=logger, on_partial_save=_save_partial, ) try: with open(out_path, "w", encoding="utf-8") as f: json.dump(out_data, f, indent=2, ensure_ascii=False) except OSError as exc: logger.warning("⚠️ mode_combo 반영 재저장 실패: %s", exc) announce_optuna_json_path( out_path, strategy="tail", mode=ctx.mode, note="최종 JSON", log=logger, ) finalize_optuna_export( study, out_data=out_data, out_path=out_path, strategy="tail", mode=ctx.mode, enrich_fn=_enrich, log=logger, ) if study.best_trial and study.best_value > _FAIL_OBJECTIVE + 1: bt = study.best_trial logger.info( "🏆 Best trial #%d | objective=%.4g | pnl=%s | wr=%.1f%% | trades=%s", bt.number, study.best_value, bt.user_attrs.get("total_pnl"), float(bt.user_attrs.get("win_rate") or 0), bt.user_attrs.get("total_trades"), ) else: logger.info("⚠️ 조건 만족 trial 없음 (min_trades·승률·PF 게이트 확인)") study._kis_export_path = out_path # type: ignore[attr-defined] return study finally: # mode_combo 실측이 ticks 공유뷰를 쓰므로 optimize 직후 unlink 금지 release_shared_tick_store(ctx, log=logger) try: from kis_trader.engine.tail_engine import clear_rust_session clear_rust_session(rust_session_id) except Exception as e: pass def apply_best_trial(study: optuna.Study, ctx: TailSearchContext) -> bool: """사후게이트 통과 trial → env_config (총손익≤0 스킵).""" trial = pick_gated_apply_trial(study, sort_by="score", fail_objective=_FAIL_OBJECTIVE) if trial is None: logger.warning( "⚠️ 사후게이트(results_gated) 통과 trial 없음 — DB 미적용" ) return False pnl = float(trial.user_attrs.get("total_pnl") or 0) if pnl <= 0: logger.warning("⚠️ gated trial 총손익 ≤ 0 — DB 미적용. 기존 설정 유지.") return False params_raw = trial.user_attrs.get("params_json") or "{}" combo = json.loads(params_raw) merged = {**ctx.base_params, **combo} apply_params_to_db(merged) env_map = _tail_params_to_env_map(merged) logger.info("🚀 [Optuna apply-best] gated trial #%d → env_config", trial.number) logger.info("적용된 값: %s", json.dumps(env_map, indent=2, ensure_ascii=False)) try: from kis_trader.backtest.optuna_daily_trail_recommend import ( apply_daily_trail_recommend_from_optuna_json, ) apply_daily_trail_recommend_from_optuna_json( getattr(study, "_kis_export_path", None), strategy="tail", log=logger, ) except Exception as exc: logger.warning("⚠️ 다단트레일 추천 반영 스킵: %s", exc) return True def main() -> None: from kis_trader.backtest.param_search_dates import resolve_param_search_range week_ago, today = resolve_param_search_range("TAIL", lookback_days=7) # Optuna 게이트·브리핑 키 DB 기본값 (없으면 삽입) try: ensure_optuna_gate_env_defaults() except Exception: pass parser = argparse.ArgumentParser( description="Optuna TPE 파라미터 탐색 (Grid CLI add-on, storage=MariaDB 141 기본)", ) parser.add_argument( "--strategy", default="tail", choices=list(OPTUNA_STRATEGIES), help="전략: tail | momentum | breakout | scalp", ) parser.add_argument("--start", default=week_ago, help="시작일 YYYY-MM-DD (거래일 보정)") parser.add_argument("--end", default=today, help="종료일 YYYY-MM-DD (주말·휴장이면 이전 장운영일)") parser.add_argument("--timeframe", "--tf", default=3, type=int, dest="timeframe", help="ws_candles 분봉 3·5·15·60") add_portfolio_cli_args(parser) parser.add_argument( "--mode", default="fast", help="탐색 축 모드 (전략별 Grid 와 동일 — tail:fast/coarse/… momentum:fast/rr/… breakout:fast/coarse/…)", ) parser.add_argument( "--trials", type=int, default=None, help="Optuna trial 수 (미지정 시 PARAM_SEARCH_OPTUNA_N_TRIALS·DB, 기본 200)", ) parser.add_argument( "--study-name", default=None, dest="study_name", help="Study 이름 (미지정 시 OPTUNA_TAIL_STUDY_NAME 또는 tail_{mode}_{start}_{end})", ) parser.add_argument( "--study-trials", default=None, type=int, dest="study_trials", help="이 study 시도 목표(COMPLETE+PRUNED+FAIL). 미지정/0=이번 --trials. 시도 < 목표면 후처리 스킵", ) parser.add_argument( "--storage", default=None, help="Optuna storage URL (미지정 시 MariaDB 141/kis_optuna)", ) parser.add_argument( "--sampler", default=None, choices=["tpe", "random"], help="샘플러 (미지정 시 PARAM_SEARCH_OPTUNA_SAMPLER·DB, 기본 tpe)", ) parser.add_argument("--seed", type=int, default=None, help="재현용 random seed") parser.add_argument( "--n-jobs", type=int, default=None, dest="n_jobs", help="프로세스 내 병렬 trial (기본 1). PC 2대 분산은 각각 실행 + 동일 study-name", ) parser.add_argument( "--sort-by", default=None, dest="sort_by", help="목적함수: score|score_legacy|pnl|daily_avg|win_rate (미지정=score 전 전략 공통)", ) add_search_filter_cli_args(parser) # Optuna: 탐색 중 승률·PF 게이트 OFF(0) — TPE가 PnL 차이를 학습. 사후 results_gated 로 후보 분리. _sw, _sp, _st = optuna_search_gate_defaults() parser.set_defaults(min_win_rate=_sw, min_pf=_sp, min_trades=_st) parser.add_argument( "--min_trades", default=_st, type=int, help=f"최소 거래 건수 (Optuna 기본 {_st}, Grid CLI 와 별개)", ) parser.add_argument("--fallback-universe", action="store_true", dest="fallback_universe") parser.add_argument("--use-universe-history", action="store_true", dest="use_universe_history") parser.add_argument( "--universe-history-source", default=None, choices=["kiwoom", "ls"], dest="universe_history_source", help="이력 테이블: kiwoom=target_candidates_history, ls=ls_candidates_history " "(기본 env BACKTEST_UNIVERSE_HISTORY_SOURCE 또는 kiwoom)", ) parser.add_argument( "--candle-source", default="", choices=["", "kis", "kiwoom"], dest="candle_source", help="캔들 소스: 빈값=실매 LIVE_TICK_PROVIDER 우선 병합, kis|kiwoom=단일 소스", ) parser.add_argument( "--tick-source", default="", choices=["", "kis", "kiwoom"], dest="tick_source", help="틱 소스 필터 (기본 빈문자열 = 전체 검색)", ) parser.add_argument( "--ob-source", default="", choices=["", "kis", "kiwoom", "kiwoom_0d"], dest="ob_source", help="호가 소스 필터 (기본 빈문자열 = 전체 검색)", ) parser.add_argument( "--entry-mode", default=None, dest="entry_mode", choices=["align", "limit_atr"], help="꼬리 TPE만. 한 스터디에 한 모드(미지정=TAIL_PARAM_SEARCH_ENTRY_MODE 또는 align). " "둘 다 보려면 스터디를 나눠 두 번 실행.", ) parser.add_argument( "--sl-mode", default=None, dest="sl_mode", choices=["fixed", "atr"], help="돌파 TPE만. 한 스터디에 한 손절모드(미지정=fixed). " "fixed=sl_pct 탐색, atr=atr_sl_mult 탐색. 둘 다 보려면 스터디를 나눠 두 번.", ) parser.add_argument( "--orderbook-filter", default="off", choices=["off", "on", "auto"], dest="orderbook_filter", ) parser.add_argument( "--apply-best", action="store_true", dest="apply_best", help="탐색 후 best trial 을 env_config 에 반영", ) parser.add_argument( "--no-progress", action="store_true", dest="no_progress", help="Optuna progress bar 끄기", ) parser.add_argument( "--symbol", default="", help="us_momentum 전용: 1종목 유니버스(종목 cfg Optuna). 예: TSLA", ) args = parser.parse_args() if getattr(args, "candle_source", ""): os.environ["CANDLE_SOURCE"] = str(args.candle_source).strip().lower() if getattr(args, "tick_source", ""): os.environ["TICK_SOURCE"] = str(args.tick_source).strip().lower() if getattr(args, "ob_source", ""): os.environ["OB_SOURCE"] = str(args.ob_source).strip().lower() n_trials = args.trials if n_trials is None: n_trials = get_env_int("PARAM_SEARCH_OPTUNA_N_TRIALS", 200) n_trials = max(1, int(n_trials)) from kis_trader.backtest.optuna_study_store import resolve_cli_study_trials _st_goal = resolve_cli_study_trials(getattr(args, "study_trials", None)) if _st_goal > 0: os.environ["KIS_OPTUNA_STUDY_TRIALS"] = str(_st_goal) n_jobs = args.n_jobs if n_jobs is None: n_jobs = get_env_int("PARAM_SEARCH_OPTUNA_N_JOBS", 8) n_jobs = max(1, int(n_jobs)) sampler_name = args.sampler if not sampler_name: sampler_name = str(get_env_from_db("PARAM_SEARCH_OPTUNA_SAMPLER", "tpe") or "tpe").strip().lower() def _sigterm_to_kbd(_sig, _frm): raise KeyboardInterrupt("SIGTERM 수신 → 종료") try: signal.signal(signal.SIGTERM, _sigterm_to_kbd) except Exception: pass strategy = (args.strategy or "tail").strip().lower() if strategy not in OPTUNA_STRATEGIES: logger.error("❌ --strategy 는 tail/momentum/us_momentum/breakout/scalp 중 하나") sys.exit(2) # CLI --min_trades 와 사후 results_gated 거래수 게이트 정렬 (seq: tail=1 / 타전략=18 등) os.environ["PARAM_SEARCH_OPTUNA_REPORT_MIN_TRADES"] = str(max(1, int(args.min_trades))) allowed_modes = STRATEGY_MODES.get(strategy, []) mode = (args.mode or "fast").strip().lower() if mode not in allowed_modes: logger.error("❌ %s --mode '%s' 불가. 허용: %s", strategy, mode, allowed_modes) sys.exit(2) sort_by = (args.sort_by or "").strip().lower() if not sort_by: from kis_trader.backtest.optuna_common import OPTUNA_SORT_BY_DEFAULT sort_by = OPTUNA_SORT_BY_DEFAULT from kis_trader.backtest.optuna_common import ( OPTUNA_SORT_BY_CHOICES, normalize_optuna_sort_by, ) sort_by = normalize_optuna_sort_by(sort_by, web=False) if sort_by not in OPTUNA_SORT_BY_CHOICES: logger.error("❌ --sort-by 는 score|score_legacy|pnl|daily_avg|win_rate") sys.exit(2) lock_name = optuna_run_lock_name(strategy) run_lock = try_acquire_run_lock(lock_name) if run_lock is None: logger.error( "⛔ 이미 실행 중인 %s 가 있습니다.\n" " ps -ef | grep param_search_optuna\n" " pkill -f 'param_search_optuna.py' 후 재실행", lock_name, ) sys.exit(2) use_fallback = bool(args.fallback_universe) if args.use_universe_history: use_fallback = False storage_url = resolve_optuna_storage_url(args.storage) _study_extra = None if strategy == "tail" and mode == "tpe": _study_extra = normalize_tpe_tail_entry_mode( getattr(args, "entry_mode", None) or "align", ) elif strategy == "breakout" and mode == "tpe": from kis_trader.backtest.optuna_breakout_tpe_space import ( breakout_tpe_study_extra, ) _study_extra = breakout_tpe_study_extra( getattr(args, "sl_mode", None) or "fixed", getattr(args, "orderbook_filter", None) or "off", ) study_name = resolve_study_name( strategy=strategy, mode=mode, start=args.start, end=args.end, cli_override=args.study_name, extra=_study_extra, ) study = None try: if strategy == "tail": ctx = prepare_tail_search_context( args.start, args.end, mode, timeframe=args.timeframe, use_fallback_universe=use_fallback, time_start_hm=args.time_start, time_end_hm=args.time_end, slot_money=args.slot_money, max_stocks=args.max_stocks, total_budget_krw=args.total_budget, orderbook_filter=args.orderbook_filter, history_source=args.universe_history_source, entry_mode=getattr(args, "entry_mode", None), ) if ctx is None: sys.exit(1) study = run_tail_optuna( ctx, n_trials=n_trials, storage_url=storage_url, study_name=study_name, min_trades=args.min_trades, min_win_rate=args.min_win_rate, min_pf=args.min_pf, sort_by=sort_by, sampler_name=sampler_name, seed=args.seed, n_jobs=n_jobs, show_progress=not args.no_progress, ) if args.apply_best: apply_best_trial(study, ctx) elif strategy in ("momentum", "us_momentum"): _mom_market = "US" if strategy == "us_momentum" else "KR" _sym = str(getattr(args, "symbol", "") or "").strip().upper() if _sym and strategy != "us_momentum": logger.error("❌ --symbol 은 us_momentum 전용") sys.exit(2) ctx_m = prepare_momentum_search_context( args.start, args.end, mode, use_fallback_universe=use_fallback or (_mom_market == "US"), time_start_hm=args.time_start, time_end_hm=args.time_end, slot_money=args.slot_money, max_stocks=args.max_stocks, total_budget_krw=args.total_budget, orderbook_filter="off" if _mom_market == "US" else args.orderbook_filter, market=_mom_market, symbol=_sym if strategy == "us_momentum" else "", history_source=args.universe_history_source, ) if ctx_m is None: sys.exit(1) study = run_momentum_optuna( ctx_m, n_trials=n_trials, storage_url=storage_url, study_name=study_name, min_trades=args.min_trades, min_win_rate=args.min_win_rate, min_pf=args.min_pf, sort_by=sort_by, sampler_name=sampler_name, seed=args.seed, n_jobs=n_jobs, show_progress=not args.no_progress, ) if args.apply_best: if strategy == "us_momentum": from kis_trader.backtest.optuna_momentum import apply_best_us_momentum_trial apply_best_us_momentum_trial(study, symbol=_sym) else: apply_best_momentum_trial(study) elif strategy == "scalp": ctx_s = prepare_scalp_search_context( args.start, args.end, mode, use_fallback_universe=use_fallback, time_start_hm=args.time_start, time_end_hm=args.time_end, slot_money=args.slot_money, max_stocks=args.max_stocks, total_budget_krw=args.total_budget, orderbook_filter=args.orderbook_filter, history_source=args.universe_history_source, ) if ctx_s is None: sys.exit(1) study = run_scalp_optuna( ctx_s, n_trials=n_trials, storage_url=storage_url, study_name=study_name, min_trades=args.min_trades, min_win_rate=args.min_win_rate, min_pf=args.min_pf, sort_by=sort_by, sampler_name=sampler_name, seed=args.seed, n_jobs=n_jobs, show_progress=not args.no_progress, ) if args.apply_best: apply_best_scalp_trial(study) elif strategy == "dart": ctx_d = prepare_dart_search_context(args.start, args.end, mode) if ctx_d is None: sys.exit(1) study = run_dart_optuna( ctx_d, n_trials=n_trials, storage_url=storage_url, study_name=study_name, min_trades=args.min_trades, sampler_name=sampler_name, seed=args.seed, show_progress=not args.no_progress, ) if args.apply_best: apply_best_dart_trial(study) else: ctx_b = prepare_breakout_search_context( args.start, args.end, mode, use_fallback_universe=use_fallback, time_start_hm=args.time_start, time_end_hm=args.time_end, slot_money=args.slot_money, max_stocks=args.max_stocks, total_budget_krw=args.total_budget, orderbook_filter=args.orderbook_filter, history_source=args.universe_history_source, sl_mode=getattr(args, "sl_mode", None), ) if ctx_b is None: sys.exit(1) study = run_breakout_optuna( ctx_b, n_trials=n_trials, storage_url=storage_url, study_name=study_name, min_trades=args.min_trades, min_win_rate=args.min_win_rate, min_pf=args.min_pf, sort_by=sort_by, sampler_name=sampler_name, seed=args.seed, n_jobs=n_jobs, show_progress=not args.no_progress, ) if args.apply_best: apply_best_breakout_trial(study) # 종료 직전 절대경로 한 번 더 (로그 끝에서 바로 복사) export = getattr(study, "_kis_export_path", None) if study is not None else None if export: announce_optuna_json_path( str(export), strategy=strategy, mode=mode, note="CLI 종료·열기용 경로", log=logger, ) except KeyboardInterrupt as e: print(f"\n⛔ {e} — 중단", flush=True) sys.exit(130) finally: run_lock.release() if __name__ == "__main__": main()