diff --git a/kis_trader/backtest/optuna_breakout.py b/kis_trader/backtest/optuna_breakout.py new file mode 100644 index 0000000..5fcafa5 --- /dev/null +++ b/kis_trader/backtest/optuna_breakout.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +"""kis_trader/backtest/optuna_breakout.py — 돌파 Optuna (Grid add-on).""" +from __future__ import annotations + +import json +import logging +import os +import time +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional + +import optuna +from optuna.samplers import RandomSampler, TPESampler + +from database import TradeDB +from kis_trader.backtest import breakout_backtest_common as bbc +from kis_trader.backtest.optuna_search_space import breakout_grid_axis_keys, suggest_breakout_params +from kis_trader.backtest.param_search_breakout import ( + _bo_fixed_defaults, + _load_candles_for_search, + _breakout_grids, + _ui_to_engine_params, + apply_params_to_db, + evaluate_breakout_param_combo, +) +from kis_trader.backtest.param_search_cli_common import ( + apply_session_to_fixed, + combo_passes_search_filters, + format_session_hm, +) +from kis_trader.backtest.tail_param_search import _results_dir_for_write +from kis_trader.strategies.breakout import breakout_backtest_wants_tick_replay, breakout_entry_mode +from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params +from kis_trader.backtest.breakout_tick_loader import load_breakout_ticks_by_code + +logger = logging.getLogger("param_search_optuna") + +_FAIL_OBJECTIVE = -1e18 + + +@dataclass +class BreakoutSearchContext: + start: str + end: str + mode: str + base_fixed: Dict[str, Any] + codes_candles: Dict[str, List[Dict]] + universe_by_slot: Optional[Dict[str, List[str]]] + ticks_by_code: Any + orderbook_by_code: Dict[str, Any] + program_by_code: Dict[str, Any] + log_verdict_by_code: Dict[str, Any] + share_denom_by_code: Dict[str, float] + 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] + start_key: str + end_key: str + cache_holder: Dict[str, Any] = field(default_factory=dict) + + +def prepare_breakout_search_context( + start: str, + end: str, + mode: str, + *, + 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", +) -> Optional[BreakoutSearchContext]: + grids = _breakout_grids() + if mode not in grids: + logger.error("❌ 돌파 mode: %s (fast/coarse/fine/full)", mode) + return None + + base_fixed = _bo_fixed_defaults() + apply_session_to_fixed(base_fixed, time_start_hm=time_start_hm, time_end_hm=time_end_hm) + + _ob_mode = (orderbook_filter or "off").strip().lower() + if _ob_mode == "off": + base_fixed["_orderbook_filter_enabled"] = False + elif _ob_mode == "on": + base_fixed["_orderbook_filter_enabled"] = True + ob_filter_on = bool(base_fixed.get("_orderbook_filter_enabled")) or _ob_mode == "auto" + logger.info( + "📌 호가필터: %s (%s)", + _ob_mode.upper(), + "적용" if ob_filter_on else "스킵 — 코어 파라미터 순수 탐색", + ) + + db = TradeDB() + try: + row = db.conn.execute("SELECT * FROM env_config ORDER BY id DESC LIMIT 1").fetchone() + env_row = dict(row) if row else {} + finally: + db.close() + + fee_rate, sell_tax, slot_from_env = bbc.fee_and_slot_from_env(env_row) + portfolio = bbc.resolve_breakout_portfolio_params( + env_row, None, + slot_money=slot_money if slot_money is not None else slot_from_env, + 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"]) + period_days = max( + 1, + (datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days + 1, + ) + logger.info( + f"💼 포트폴리오: 1회 {slot_money_f:,.0f}원 | 동시 {max_stocks_i}종 | " + f"총한도 {total_budget_f:,.0f}원 | 매매 {format_session_hm(base_fixed)}" + ) + logger.info("📌 진입 모드: %s", breakout_entry_mode()) + + codes_candles = _load_candles_for_search( + start, end, base_fixed.get("lookback_min", 1), base_fixed, + ) + if not codes_candles: + logger.error("❌ 캔들 데이터 없음") + return None + logger.info("✅ 데이터 로드: %s종목", len(codes_candles)) + + share_denom_by_code: Dict[str, float] = {} + _share_db = TradeDB() + try: + from kis_trader.share.stock_share import load_share_denom_map + share_denom_by_code = load_share_denom_map(_share_db, codes_candles.keys()) + finally: + _share_db.close() + + start_key = (start.replace("-", "") + "0000") if start else "202601010000" + end_key = (end.replace("-", "") + "2359") if end else "999912312359" + start_ymd = start.replace("-", "") if start else "" + end_ymd = end.replace("-", "") if end else "" + + ticks_by_code: Dict[str, Any] = {} + engine_probe = _ui_to_engine_params(base_fixed) + engine_probe["_orderbook_filter_enabled"] = base_fixed.get("_orderbook_filter_enabled") + if breakout_backtest_wants_tick_replay(engine_probe): + _tick_db = TradeDB() + try: + ticks_by_code, tick_rows = load_breakout_ticks_by_code( + _tick_db, start_key, end_key, set(codes_candles.keys()), + ) + logger.info("✅ ws_ticks %s건", f"{tick_rows:,}") + finally: + _tick_db.close() + + grid = grids[mode] + _ob_axes = ("max_spread_pct", "min_bid_ask_ratio", "ask_wall_max_qty") + _ob_sweeping = any(len(set(grid.get(k) or [])) > 1 for k in _ob_axes) + if ob_filter_on and _ob_sweeping: + base_fixed["backtest_use_kiwoom_body_snapshot"] = True + base_fixed["_backtest_use_kiwoom_body"] = True + + orderbook_by_code: Dict[str, Any] = {} + program_by_code: Dict[str, Any] = {} + log_verdict_by_code: Dict[str, Any] = {} + _snap_db = TradeDB() + 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( + _snap_db, start_key, end_key, set(codes_candles.keys()), + engine_params=engine_probe, strategy="BREAKOUT", + ) + log_verdict_by_code = trigger_snap_meta.get("log_verdict_by_code") or {} + finally: + _snap_db.close() + + universe_by_slot = None + fallback_sim_interval = 5 + if not use_fallback_universe and start_ymd and end_ymd: + try: + from kis_trader.database.db_manager import get_db as _get_ext_db + _ext = _get_ext_db() + history = _ext.get_universe_by_candle_time( + strategy_id="BREAKOUT", start_ymd=start_ymd, end_ymd=end_ymd, + ) + if history: + universe_by_slot = history + n_bins = len(history) + avg = sum(len(v) for v in history.values()) / max(1, n_bins) + logger.info("✅ 유니버스: BREAKOUT 이력 | %s분봉 · 평균 %.1f종목", n_bins, avg) + except Exception as exc: + logger.debug("유니버스 이력 스킵: %s", exc) + + if universe_by_slot is None: + from kis_trader.engine import scalping_engine as se + universe_top_n = int(os.environ.get("UPDATE_UNIVERSE_TOP_N", "20")) + universe_min_score = float(os.environ.get("UPDATE_UNIVERSE_MIN_SCORE", "4.0")) + universe_by_slot = se.build_universe_simulation( + codes_candles, + top_n=universe_top_n, + min_score=universe_min_score, + scan_interval_min=fallback_sim_interval, + ) + base_fixed["scan_interval_min"] = fallback_sim_interval + logger.info("📌 유니버스: 시뮬 fallback (%d분)", fallback_sim_interval) + else: + base_fixed["scan_interval_min"] = 1 + + cache_holder: Dict[str, Any] = {} + attach_indicator_caches_to_params(cache_holder, codes_candles) + + return BreakoutSearchContext( + start=start, + end=end, + mode=mode, + base_fixed=base_fixed, + codes_candles=codes_candles, + universe_by_slot=universe_by_slot, + ticks_by_code=ticks_by_code, + orderbook_by_code=orderbook_by_code, + program_by_code=program_by_code, + log_verdict_by_code=log_verdict_by_code, + share_denom_by_code=share_denom_by_code, + 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=breakout_grid_axis_keys(mode), + start_key=start_key, + end_key=end_key, + cache_holder=cache_holder, + ) + + +def _make_sampler(name: str, seed: Optional[int]): + n = (name or "tpe").strip().lower() + if n == "random": + return RandomSampler(seed=seed) + return TPESampler(seed=seed, multivariate=True) + + +def run_breakout_optuna( + ctx: BreakoutSearchContext, + *, + n_trials: int, + storage_url: str, + study_name: str, + min_trades: int, + min_win_rate: float, + min_pf: float, + sort_by: str = "pnl", + sampler_name: str = "tpe", + seed: Optional[int] = None, + n_jobs: int = 1, + show_progress: bool = True, +) -> optuna.Study: + study = optuna.create_study( + study_name=study_name, + storage=storage_url, + load_if_exists=True, + direction="maximize", + sampler=_make_sampler(sampler_name, seed), + ) + + def objective(trial: optuna.Trial) -> float: + combo = suggest_breakout_params(trial, ctx.mode) + result = evaluate_breakout_param_combo( + combo, + base_fixed=ctx.base_fixed, + grid_keys=ctx.grid_keys, + codes_candles=ctx.codes_candles, + 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, + fee_rate=ctx.fee_rate, + sell_tax=ctx.sell_tax, + 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, + share_denom_by_code=ctx.share_denom_by_code, + ) + if result is None: + trial.set_user_attr("gates_ok", False) + return _FAIL_OBJECTIVE + obj = float(result["win_rate"]) if sort_by == "win_rate" else float(result["total_pnl"]) + 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("total_trades", int(result["total_trades"])) + trial.set_user_attr("merged_json", json.dumps(result.get("merged_params") or {}, ensure_ascii=False)) + return obj + + logger.info("🔬 Optuna BREAKOUT | study=%s | trials=%d", study_name, n_trials) + t0 = time.time() + study.optimize(objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress) + elapsed = time.time() - t0 + + 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 + merged_raw = trial.user_attrs.get("merged_json") or "{}" + try: + merged = json.loads(merged_raw) + except json.JSONDecodeError: + merged = dict(trial.params) + passing.append({ + "params": dict(trial.params), + "merged_params": merged, + "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), + "optuna_trial_number": trial.number, + }) + + if sort_by == "win_rate": + passing.sort(key=lambda r: (-r["win_rate"], -r["total_pnl"])) + else: + passing.sort(key=lambda r: (-r["total_pnl"], -r["win_rate"])) + profitable = [r for r in passing if r["total_pnl"] > 0] + if profitable: + passing = profitable + + hints: Dict[str, str] = {} + out_data = { + "engine": "optuna", + "strategy": "breakout", + "mode": ctx.mode, + "start": ctx.start, + "end": ctx.end, + "slot_money": int(ctx.slot_money), + "max_stocks": ctx.max_stocks, + "total_budget_krw": int(ctx.total_budget_krw), + "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: hints[k] for k in ctx.grid_keys if k in hints}, + "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), + "results": passing[:5000], + } + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + out_path = os.path.join(_results_dir_for_write(), f"optuna_breakout_{ctx.mode}_{ts}.json") + with open(out_path, "w", encoding="utf-8") as f: + json.dump(out_data, f, indent=2, ensure_ascii=False) + logger.info("💾 Optuna 결과 저장: %s", out_path) + study._kis_export_path = out_path # type: ignore[attr-defined] + return study + + +def apply_best_breakout_trial(study: optuna.Study) -> bool: + if not study.best_trial or study.best_value <= _FAIL_OBJECTIVE + 1: + logger.warning("⚠️ 적용할 best trial 없음") + return False + pnl = float(study.best_trial.user_attrs.get("total_pnl") or 0) + if pnl <= 0: + logger.warning("⚠️ Best trial 총손익 ≤ 0 — DB 미적용") + return False + merged = json.loads(study.best_trial.user_attrs.get("merged_json") or "{}") + apply_params_to_db(merged) + logger.info("🚀 [Optuna apply-best] breakout trial #%d → env_config", study.best_trial.number) + return True diff --git a/kis_trader/backtest/optuna_common.py b/kis_trader/backtest/optuna_common.py new file mode 100644 index 0000000..848df45 --- /dev/null +++ b/kis_trader/backtest/optuna_common.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +""" +kis_trader/backtest/optuna_common.py — Optuna storage·DB 공통 (MariaDB 141) +========================================================================= +TradeDB(database.py) 와 동일 호스트·계정, 전용 DB kis_optuna 에 study 저장. +Win11·VM 양쪽에서 같은 storage 로 trial 공유·재개 가능. +""" +from __future__ import annotations + +import logging +import os +from typing import Optional +from urllib.parse import quote_plus + +from kis_trader.utils.env import get_env_from_db + +logger = logging.getLogger("optuna_common") + +# Optuna 전용 MariaDB (매매 DB kis_quant_db 와 분리) +DEFAULT_OPTUNA_DB_NAME = "kis_optuna" +OPTUNA_STRATEGIES = ("tail", "momentum", "breakout") + + +def mariadb_creds() -> dict: + """TradeDB(database.py) 와 동일 우선순위 — env > 기본 141.""" + return { + "host": os.environ.get("DB_HOST", "192.168.0.141"), + "port": int(os.environ.get("DB_PORT", "3306")), + "user": os.environ.get("DB_USER", "jae"), + "password": os.environ.get("DB_PASS", "1234"), + } + + +def resolve_optuna_db_name() -> str: + """ + Optuna storage DB — 기본 kis_optuna (매매 kis_quant_db 와 분리). + env OPTUNA_DB_NAME 로 오버라이드 가능. + """ + raw = get_env_from_db("OPTUNA_DB_NAME", "") + if raw and str(raw).strip() not in ("", "None"): + return str(raw).strip() + env = os.environ.get("OPTUNA_DB_NAME", "") + if env and str(env).strip(): + return str(env).strip() + return DEFAULT_OPTUNA_DB_NAME + + +def build_mariadb_storage_url(db_name: Optional[str] = None) -> str: + """mysql+pymysql://…@141/optuna 형식 storage URL.""" + creds = mariadb_creds() + name = (db_name or resolve_optuna_db_name()).strip() + user = quote_plus(creds["user"]) + passwd = quote_plus(creds["password"]) + return ( + f"mysql+pymysql://{user}:{passwd}@{creds['host']}:{creds['port']}/{name}" + f"?charset=utf8mb4" + ) + + +def ensure_optuna_database(db_name: Optional[str] = None) -> str: + """ + MariaDB 141 — kis_optuna 존재 확인 (없으면 CREATE 시도). + """ + name = (db_name or resolve_optuna_db_name()).strip() + creds = mariadb_creds() + + try: + import pymysql + except ImportError as exc: + raise ImportError( + "Optuna MariaDB storage 는 pymysql 필요: pip install PyMySQL" + ) from exc + + # DB 존재 여부만 확인 (이미 있으면 CREATE 생략) + conn = pymysql.connect( + host=creds["host"], + port=creds["port"], + user=creds["user"], + password=creds["password"], + charset="utf8mb4", + autocommit=True, + connect_timeout=10, + ) + try: + with conn.cursor() as cur: + cur.execute("SHOW DATABASES LIKE %s", (name,)) + exists = cur.fetchone() is not None + if not exists: + cur.execute( + f"CREATE DATABASE IF NOT EXISTS `{name}` " + "DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" + ) + logger.info( + "📦 Optuna DB 생성: %s@%s:%s/%s", + creds["user"], creds["host"], creds["port"], name, + ) + else: + logger.info( + "📦 Optuna storage DB: %s@%s:%s/%s", + creds["user"], creds["host"], creds["port"], name, + ) + except Exception as exc: + logger.error("❌ Optuna DB '%s' 접속/확인 실패: %s", name, exc) + raise + finally: + conn.close() + return name + + +def resolve_optuna_storage_url(cli_override: Optional[str] = None) -> str: + """ + Storage URL 우선순위: + 1) CLI --storage + 2) OPTUNA_STORAGE_URL (DB/env) + 3) MariaDB 141 / kis_optuna (TradeDB 동일 계정) + """ + if cli_override and str(cli_override).strip(): + return str(cli_override).strip() + from_db = get_env_from_db("OPTUNA_STORAGE_URL", "") + if from_db and str(from_db).strip() not in ("", "None"): + return str(from_db).strip() + db_name = ensure_optuna_database() + return build_mariadb_storage_url(db_name) + + +def resolve_study_name( + *, + strategy: str, + mode: str, + start: str, + end: str, + cli_override: Optional[str] = None, +) -> str: + """Study 이름 — 전략·기간·모드 포함.""" + if cli_override and str(cli_override).strip(): + return str(cli_override).strip() + env_key = f"OPTUNA_{strategy.upper()}_STUDY_NAME" + from_db = get_env_from_db(env_key, "") + if from_db and str(from_db).strip() not in ("", "None"): + return str(from_db).strip() + legacy = get_env_from_db("OPTUNA_TAIL_STUDY_NAME", "") + if strategy == "tail" and legacy and str(legacy).strip() not in ("", "None"): + return str(legacy).strip() + return f"{strategy}_{mode}_{start}_{end}" + + +def optuna_run_lock_name(strategy: str) -> str: + return f"{strategy}_param_search_optuna" diff --git a/kis_trader/backtest/optuna_momentum.py b/kis_trader/backtest/optuna_momentum.py new file mode 100644 index 0000000..c01c205 --- /dev/null +++ b/kis_trader/backtest/optuna_momentum.py @@ -0,0 +1,409 @@ +#!/usr/bin/env python3 +"""kis_trader/backtest/optuna_momentum.py — 모멘텀 Optuna (Grid add-on).""" +from __future__ import annotations + +import json +import logging +import os +import time +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional + +import optuna +from optuna.samplers import RandomSampler, TPESampler + +from database import TradeDB +from kis_trader.backtest import momentum_backtest_common as mbc +from kis_trader.backtest import scalping_backtest_common as sbc +from kis_trader.backtest.optuna_search_space import momentum_grid_axis_keys, suggest_momentum_params +from kis_trader.backtest.param_search_cli_common import ( + apply_session_to_fixed, + combo_passes_search_filters, + format_session_hm, +) +from kis_trader.backtest.param_search_momentum import ( + MOMENTUM_GRID_AXIS_HINTS_KO, + _load_candles_for_search, + _mom_fixed_defaults, + _momentum_grids, + apply_params_to_db, + evaluate_momentum_param_combo, +) +from kis_trader.backtest.tail_param_search import _results_dir_for_write +from kis_trader.engine import momentum_engine as me +from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params +from kis_trader.utils.env import get_env_float + +logger = logging.getLogger("param_search_optuna") + +_FAIL_OBJECTIVE = -1e18 + + +@dataclass +class MomentumSearchContext: + start: str + end: str + mode: str + base_fixed: Dict[str, Any] + codes_candles: Dict[str, List[Dict]] + universe_by_slot: Optional[Dict[str, List[str]]] + ticks_by_code: Any + orderbook_by_code: Dict[str, Any] + program_by_code: Dict[str, Any] + log_verdict_by_code: 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] + start_key: str + end_key: str + cache_holder: Dict[str, Any] = field(default_factory=dict) + + +def prepare_momentum_search_context( + start: str, + end: str, + mode: str, + *, + 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", +) -> Optional[MomentumSearchContext]: + grids = _momentum_grids() + if mode not in grids: + logger.error("❌ 모멘텀 mode: %s (fast/rr/coarse/fine/full)", mode) + return None + + base_fixed = _mom_fixed_defaults() + apply_session_to_fixed(base_fixed, time_start_hm=time_start_hm, time_end_hm=time_end_hm) + + from kis_trader.engine.momentum_tick_replay import ( + momentum_backtest_use_tick_entry as _te, + momentum_backtest_use_tick_exit as _tx, + ) + base_fixed["backtest_use_tick_entry"] = _te(None) + base_fixed["backtest_use_tick_exit"] = _tx(None) + + _ob_mode = (orderbook_filter or "off").strip().lower() + if _ob_mode == "off": + base_fixed["_orderbook_filter_enabled"] = False + elif _ob_mode == "on": + base_fixed["_orderbook_filter_enabled"] = True + ob_filter_on = bool(base_fixed.get("_orderbook_filter_enabled")) or _ob_mode == "auto" + logger.info( + "📌 호가필터: %s (%s)", + _ob_mode.upper(), + "적용" if ob_filter_on else "스킵 — 코어 파라미터 순수 탐색", + ) + + db = TradeDB() + try: + row = db.conn.execute("SELECT * FROM env_config ORDER BY id DESC LIMIT 1").fetchone() + env_row = dict(row) if row else {} + finally: + db.close() + + fee_rate, sell_tax, slot_from_env = sbc.fee_and_slot_from_env(env_row, strategy="MOMENTUM") + portfolio = sbc.resolve_scalp_portfolio_params( + env_row, None, strategy="MOMENTUM", + slot_money=slot_money if slot_money is not None else slot_from_env, + 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"]) + period_days = max( + 1, + (datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days + 1, + ) + logger.info( + f"💼 포트폴리오: 1회 {slot_money_f:,.0f}원 | 동시 {max_stocks_i}종 | " + f"총한도 {total_budget_f:,.0f}원 | 매매 {format_session_hm(base_fixed)}" + ) + + codes_candles = _load_candles_for_search(start, end, base_fixed.get("rsi_period", 3)) + if not codes_candles: + logger.error("❌ 캔들 데이터 없음") + return None + logger.info("✅ 데이터 로드: %s종목", len(codes_candles)) + + start_key = (start.replace("-", "") + "0000") if start else "202601010000" + end_key = (end.replace("-", "") + "2359") if end else "999912312359" + start_ymd = start.replace("-", "") if start else "" + end_ymd = end.replace("-", "") if end else "" + + universe_by_slot = None + fallback_sim_interval = 5 + if not use_fallback_universe and start_ymd and end_ymd: + try: + from kis_trader.backtest.momentum_backtest_common import resolve_momentum_universe + history, src, n_bins, _scan_iv, timing = resolve_momentum_universe( + start_ymd, end_ymd, use_saved_history=True, strategy_id="MOMENTUM", + ) + if history: + universe_by_slot = history + avg = sum(len(v) for v in history.values()) / max(1, n_bins) + logger.info( + "✅ 유니버스: MOMENTUM 이력 | %s분봉 · 평균 %.1f종목", n_bins, avg, + ) + except Exception as exc: + logger.debug("유니버스 이력 스킵: %s", exc) + + if universe_by_slot is None: + universe_top_n = int(os.environ.get("UPDATE_UNIVERSE_TOP_N", "20")) + universe_min_score = float(os.environ.get("UPDATE_UNIVERSE_MIN_SCORE", "4.0")) + universe_by_slot = me.build_universe_simulation_momentum( + codes_candles, + top_n=universe_top_n, + min_score=universe_min_score, + scan_interval_min=fallback_sim_interval, + ) + base_fixed["scan_interval_min"] = fallback_sim_interval + logger.info("📌 유니버스: 모멘텀 시뮬 fallback (%d분)", fallback_sim_interval) + else: + base_fixed["scan_interval_min"] = 1 + + grid = grids[mode] + _ob_axes = ("max_spread_pct", "min_bid_ask_ratio", "ask_max_mult") + _ob_sweeping = any(len(set(grid.get(k) or [])) > 1 for k in _ob_axes) + if ob_filter_on and _ob_sweeping: + base_fixed["backtest_use_kiwoom_body_snapshot"] = True + base_fixed["_backtest_use_kiwoom_body"] = True + + orderbook_by_code: Dict[str, Any] = {} + program_by_code: Dict[str, Any] = {} + log_verdict_by_code: Dict[str, Any] = {} + ticks_by_code: Dict[str, Any] = {} + _snap_db = TradeDB() + try: + from kis_trader.backtest.trigger_snapshot_loader import ( + backtest_needs_trigger_snapshot_load, + load_trigger_snapshots_by_code, + ) + from kis_trader.engine.momentum_tick_replay import ( + momentum_backtest_use_tick_entry, + momentum_backtest_use_tick_exit, + ) + if backtest_needs_trigger_snapshot_load(base_fixed, strategy="MOMENTUM"): + orderbook_by_code, program_by_code, trigger_snap_meta = load_trigger_snapshots_by_code( + _snap_db, start_key, end_key, set(codes_candles.keys()), + engine_params=base_fixed, strategy="MOMENTUM", + ) + log_verdict_by_code = trigger_snap_meta.get("log_verdict_by_code") or {} + if momentum_backtest_use_tick_exit(base_fixed) or momentum_backtest_use_tick_entry(base_fixed): + from kis_trader.backtest.momentum_tick_loader import load_momentum_ticks_by_code + ticks_by_code, tick_rows = load_momentum_ticks_by_code( + _snap_db, start_key, end_key, set(codes_candles.keys()), + ) + logger.info("✅ ws_ticks %s건", f"{tick_rows:,}") + finally: + _snap_db.close() + + cache_holder: Dict[str, Any] = {} + attach_indicator_caches_to_params(cache_holder, codes_candles) + + return MomentumSearchContext( + start=start, + end=end, + mode=mode, + base_fixed=base_fixed, + codes_candles=codes_candles, + universe_by_slot=universe_by_slot, + ticks_by_code=ticks_by_code, + orderbook_by_code=orderbook_by_code, + program_by_code=program_by_code, + log_verdict_by_code=log_verdict_by_code, + 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=momentum_grid_axis_keys(mode), + start_key=start_key, + end_key=end_key, + cache_holder=cache_holder, + ) + + +def _make_sampler(name: str, seed: Optional[int]): + n = (name or "tpe").strip().lower() + if n == "random": + return RandomSampler(seed=seed) + return TPESampler(seed=seed, multivariate=True) + + +def _momentum_objective_value(result: Dict[str, Any], sort_by: str) -> float: + pnl = float(result["total_pnl"]) + if sort_by == "score": + mdd_floor = get_env_float("MOMENTUM_SCORE_MDD_FLOOR", 10000.0) + mdd = float(result.get("mdd") or 0) + return pnl / max(mdd, mdd_floor) + if sort_by == "win_rate": + return float(result["win_rate"]) + return pnl + + +def run_momentum_optuna( + ctx: MomentumSearchContext, + *, + 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: + study = optuna.create_study( + study_name=study_name, + storage=storage_url, + load_if_exists=True, + direction="maximize", + sampler=_make_sampler(sampler_name, seed), + ) + + def objective(trial: optuna.Trial) -> float: + combo = suggest_momentum_params(trial, ctx.mode) + result = evaluate_momentum_param_combo( + combo, + base_fixed=ctx.base_fixed, + grid_keys=ctx.grid_keys, + codes_candles=ctx.codes_candles, + 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, + fee_rate=ctx.fee_rate, + sell_tax=ctx.sell_tax, + 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, + start_key=ctx.start_key, + end_key=ctx.end_key, + ) + if result is None: + trial.set_user_attr("gates_ok", False) + return _FAIL_OBJECTIVE + obj = _momentum_objective_value(result, sort_by) + 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("score", float(obj if sort_by == "score" else _momentum_objective_value(result, "score"))) + trial.set_user_attr("total_trades", int(result["total_trades"])) + trial.set_user_attr("merged_json", json.dumps(result.get("merged_params") or {}, ensure_ascii=False)) + return float(obj) + + logger.info( + "🔬 Optuna MOMENTUM | study=%s | trials=%d | sort=%s", + study_name, n_trials, sort_by, + ) + t0 = time.time() + study.optimize(objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress) + elapsed = time.time() - t0 + + 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 + merged_raw = trial.user_attrs.get("merged_json") or "{}" + try: + merged = json.loads(merged_raw) + except json.JSONDecodeError: + merged = dict(trial.params) + row = { + "params": dict(trial.params), + "merged_params": merged, + "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), + "score": float(trial.user_attrs.get("score") or 0), + "optuna_trial_number": trial.number, + } + passing.append(row) + + if sort_by == "score": + passing.sort(key=lambda r: (-r["score"], -r["total_pnl"], -r["win_rate"])) + elif sort_by == "win_rate": + passing.sort(key=lambda r: (-r["win_rate"], -r["total_pnl"])) + else: + passing.sort(key=lambda r: (-r["total_pnl"], -r["win_rate"])) + + profitable = [r for r in passing if r["total_pnl"] > 0] + if profitable: + passing = profitable + + out_data = { + "engine": "optuna", + "strategy": "momentum", + "mode": ctx.mode, + "start": ctx.start, + "end": ctx.end, + "slot_money": int(ctx.slot_money), + "max_stocks": ctx.max_stocks, + "total_budget_krw": int(ctx.total_budget_krw), + "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: MOMENTUM_GRID_AXIS_HINTS_KO[k] for k in ctx.grid_keys if k in MOMENTUM_GRID_AXIS_HINTS_KO}, + "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), + "results": passing[:5000], + } + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + out_path = os.path.join(_results_dir_for_write(), f"optuna_momentum_{ctx.mode}_{ts}.json") + with open(out_path, "w", encoding="utf-8") as f: + json.dump(out_data, f, indent=2, ensure_ascii=False) + logger.info("💾 Optuna 결과 저장: %s", out_path) + study._kis_export_path = out_path # type: ignore[attr-defined] + return study + + +def apply_best_momentum_trial(study: optuna.Study) -> bool: + if not study.best_trial or study.best_value <= _FAIL_OBJECTIVE + 1: + logger.warning("⚠️ 적용할 best trial 없음") + return False + pnl = float(study.best_trial.user_attrs.get("total_pnl") or 0) + if pnl <= 0: + logger.warning("⚠️ Best trial 총손익 ≤ 0 — DB 미적용") + return False + merged_raw = study.best_trial.user_attrs.get("merged_json") or "{}" + merged = json.loads(merged_raw) + apply_params_to_db(merged) + logger.info("🚀 [Optuna apply-best] momentum trial #%d → env_config", study.best_trial.number) + return True diff --git a/kis_trader/backtest/optuna_search_space.py b/kis_trader/backtest/optuna_search_space.py new file mode 100644 index 0000000..85e6cdd --- /dev/null +++ b/kis_trader/backtest/optuna_search_space.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +""" +kis_trader/backtest/optuna_search_space.py — Optuna 탐색 공간 (Grid 축 재사용) +============================================================================== +각 전략 Grid 와 동일한 이산 축을 trial.suggest_categorical 로 샘플링. +""" +from __future__ import annotations + +from typing import Any, Dict, List + +import optuna + +from kis_trader.backtest.param_search_breakout import _breakout_grids +from kis_trader.backtest.param_search_momentum import ( + _momentum_combo_grid_valid, + _momentum_grids, +) +from kis_trader.backtest.tail_param_search import _tail_grids + + +def _dedupe_preserve_order(values: List[Any]) -> List[Any]: + seen = set() + out: List[Any] = [] + for v in values: + key = v if isinstance(v, (int, float, str, bool)) else repr(v) + if key in seen: + continue + seen.add(key) + out.append(v) + return out + + +def _suggest_from_grid(trial: optuna.Trial, grid: Dict[str, List[Any]]) -> Dict[str, Any]: + combo: Dict[str, Any] = {} + for key, values in grid.items(): + if not values: + continue + choices = _dedupe_preserve_order(list(values)) + combo[key] = trial.suggest_categorical(key, choices) + return combo + + +def suggest_tail_params(trial: optuna.Trial, mode: str) -> Dict[str, Any]: + return _suggest_from_grid(trial, _tail_grids(mode)) + + +def suggest_momentum_params(trial: optuna.Trial, mode: str) -> Dict[str, Any]: + combo = _suggest_from_grid(trial, _momentum_grids()[mode]) + if not _momentum_combo_grid_valid(combo): + raise optuna.TrialPruned("momentum invalid combo") + return combo + + +def suggest_breakout_params(trial: optuna.Trial, mode: str) -> Dict[str, Any]: + combo = _suggest_from_grid(trial, _breakout_grids()[mode]) + if "prev_chg_min" in combo and "prev_chg_max" in combo: + if float(combo["prev_chg_min"]) >= float(combo["prev_chg_max"]): + raise optuna.TrialPruned("breakout prev_chg invalid") + return combo + + +def tail_grid_axis_keys(mode: str) -> List[str]: + return list(_tail_grids(mode).keys()) + + +def momentum_grid_axis_keys(mode: str) -> List[str]: + return list(_momentum_grids()[mode].keys()) + + +def breakout_grid_axis_keys(mode: str) -> List[str]: + return list(_breakout_grids()[mode].keys()) diff --git a/kis_trader/backtest/param_search_breakout.py b/kis_trader/backtest/param_search_breakout.py index 5194a71..0a8be66 100644 --- a/kis_trader/backtest/param_search_breakout.py +++ b/kis_trader/backtest/param_search_breakout.py @@ -623,6 +623,11 @@ def _apply_to_db(best_params: dict) -> None: print(f" {db_col:<35s} = {val}") +def apply_params_to_db(best_params: dict) -> None: + """웹·CLI 공통 — 파라서치 merged → DB.""" + _apply_to_db(best_params) + + # ────────────────────────────────────────────────────────────────────────────── # UI(%) → 엔진(비율) 변환 # ────────────────────────────────────────────────────────────────────────────── @@ -632,6 +637,116 @@ def _ui_to_engine_params(ui_params: dict) -> dict: return breakout_ui_to_engine_params(ui_params) +def evaluate_breakout_param_combo( + combo: Dict[str, Any], + *, + base_fixed: Dict[str, Any], + grid_keys: List[str], + codes_candles: Dict[str, List[Dict]], + min_trades: int, + min_win_rate: float, + min_pf: float, + universe_by_slot: Optional[Dict[str, List[str]]] = None, + slot_money: float = 2_000_000.0, + max_stocks: int = 3, + total_budget_krw: float = 6_000_000.0, + fee_rate: float = 0.00015, + sell_tax: float = 0.0018, + period_days: int = 1, + cache_holder: Optional[Dict[str, Any]] = None, + ticks_by_code: Any = None, + orderbook_by_code: Any = None, + program_by_code: Any = None, + log_verdict_by_code: Any = None, + share_denom_by_code: Optional[Dict[str, float]] = None, +) -> Optional[Dict[str, Any]]: + """단일 돌파 조합 백테 — Grid 워커·Optuna objective 공통.""" + if "prev_chg_min" in combo and "prev_chg_max" in combo: + if float(combo["prev_chg_min"]) >= float(combo["prev_chg_max"]): + return None + + ui_params = dict(base_fixed) + ui_params.update(combo) + engine_params = _ui_to_engine_params(ui_params) + if "max_spread_pct" in ui_params and ui_params.get("max_spread_pct") is not None: + engine_params["_ob_max_spread_pct"] = float(ui_params["max_spread_pct"]) + if "min_bid_ask_ratio" in ui_params and ui_params.get("min_bid_ask_ratio") is not None: + engine_params["_ob_min_bid_ask_ratio"] = float(ui_params["min_bid_ask_ratio"]) + if "ask_wall_max_qty" in ui_params and ui_params.get("ask_wall_max_qty") is not None: + engine_params["_ob_ask_wall_max_qty"] = float(ui_params["ask_wall_max_qty"]) + if ui_params.get("_orderbook_filter_enabled") is not None: + engine_params["_orderbook_filter_enabled"] = bool(ui_params["_orderbook_filter_enabled"]) + if cache_holder: + engine_params.update(cache_holder) + engine_params["slot_money"] = float(slot_money) + engine_params["max_stocks"] = int(max_stocks) + engine_params["total_budget_krw"] = float(total_budget_krw) + engine_params["portfolio_mode"] = True + if share_denom_by_code: + engine_params["share_denom_by_code"] = share_denom_by_code + if log_verdict_by_code: + engine_params["_backtest_log_verdict_by_code"] = log_verdict_by_code + + meta: Dict[str, Any] = {} + trades = bbc.run_breakout_backtest_web_aligned( + codes_candles, engine_params, universe_by_slot, + slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax, + max_stocks=max_stocks, total_budget_krw=total_budget_krw, + ticks_by_code=ticks_by_code, + orderbook_by_code=orderbook_by_code, + program_by_code=program_by_code, + meta_out=meta, + ) + stats = bbc.summarize_breakout_trades( + trades, total_budget_krw=total_budget_krw, period_days=period_days, + ) + total_trades = stats["total_trades"] + if total_trades < min_trades: + return None + + total_pnl = stats["total_pnl"] + win_rate = stats["win_rate"] + pf = float(stats.get("pf") or 0) + if not combo_passes_search_filters( + win_rate=win_rate, pf=pf, + min_win_rate=min_win_rate, min_pf=min_pf, + ): + return None + + avg_hold = stats["avg_hold_min"] + peak, mdd, cum = 0.0, 0.0, 0.0 + for t in trades: + cum += t["pnl"] + if cum > peak: + peak = cum + dd = peak - cum + if dd > mdd: + mdd = dd + + merged = dict(ui_params) + merged["slot_money"] = float(slot_money) + merged["max_stocks"] = int(max_stocks) + merged["total_budget_krw"] = float(total_budget_krw) + + return { + "params": {k: ui_params[k] for k in grid_keys if k in ui_params}, + "total_pnl": int(total_pnl), + "win_rate": round(win_rate, 2), + "total_trades": total_trades, + "pf": round(pf, 2), + "avg_hold": round(avg_hold, 1), + "mdd": round(mdd), + "bot_pct": stats["bot_pct"], + "daily_avg_pct": stats["daily_avg_pct"], + "avg_profit_rate": _avg_profit_rate_pct(trades), + "sell_reasons": _count_sell_reasons(trades), + "skipped_micro_buys": int( + (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0 + ), + "merged_params": merged, + } + + def _evaluate_breakout_chunk( param_chunk: List[Dict[str, Any]], base_fixed: Dict[str, Any], @@ -654,6 +769,8 @@ def _evaluate_breakout_chunk( ticks_preloaded = None orderbook_preloaded = None program_preloaded = None + log_verdict_preloaded = None + share_denom_preloaded = None if shared: if codes_candles is None: codes_candles = shared.get("codes_candles") or {} @@ -672,6 +789,8 @@ def _evaluate_breakout_chunk( ticks_preloaded = _tm orderbook_preloaded = shared.get("orderbook_by_code") program_preloaded = shared.get("program_by_code") + log_verdict_preloaded = shared.get("log_verdict_by_code") + share_denom_preloaded = shared.get("share_denom_by_code") if codes_candles is None: codes_candles = {} cache_holder: Dict[str, Any] = {} @@ -679,96 +798,33 @@ def _evaluate_breakout_chunk( local_heap: List[Tuple[float, float, int, Dict]] = [] for combo in param_chunk: assert_parent_alive() - if "prev_chg_min" in combo and "prev_chg_max" in combo: - if float(combo["prev_chg_min"]) >= float(combo["prev_chg_max"]): - continue - - ui_params = dict(base_fixed) - ui_params.update(combo) - engine_params = _ui_to_engine_params(ui_params) - # 호가필터 임계값 → per-run 오버라이드 (kiwoom_0d 본체 재계산 시 적용) - if "max_spread_pct" in ui_params and ui_params.get("max_spread_pct") is not None: - engine_params["_ob_max_spread_pct"] = float(ui_params["max_spread_pct"]) - if "min_bid_ask_ratio" in ui_params and ui_params.get("min_bid_ask_ratio") is not None: - engine_params["_ob_min_bid_ask_ratio"] = float(ui_params["min_bid_ask_ratio"]) - if "ask_wall_max_qty" in ui_params and ui_params.get("ask_wall_max_qty") is not None: - engine_params["_ob_ask_wall_max_qty"] = float(ui_params["ask_wall_max_qty"]) - # 호가필터 ON/OFF 플래그 전달 (base_fixed → 워커, 변환에서 누락 방지 위해 명시 복사) - if ui_params.get("_orderbook_filter_enabled") is not None: - engine_params["_orderbook_filter_enabled"] = bool(ui_params["_orderbook_filter_enabled"]) - engine_params.update(cache_holder) - engine_params["slot_money"] = float(slot_money) - engine_params["max_stocks"] = int(max_stocks) - engine_params["total_budget_krw"] = float(total_budget_krw) - engine_params["portfolio_mode"] = True - if shared: - sm = shared.get("share_denom_by_code") - if sm: - engine_params["share_denom_by_code"] = sm - lv = shared.get("log_verdict_by_code") - if lv: - engine_params["_backtest_log_verdict_by_code"] = lv - - meta: Dict[str, Any] = {} - trades = bbc.run_breakout_backtest_web_aligned( - codes_candles, engine_params, universe_by_slot, - slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax, - max_stocks=max_stocks, total_budget_krw=total_budget_krw, + result_pkg = evaluate_breakout_param_combo( + combo, + base_fixed=base_fixed, + grid_keys=keys, + codes_candles=codes_candles, + min_trades=min_trades, + min_win_rate=min_win_rate, + min_pf=min_pf, + universe_by_slot=universe_by_slot, + slot_money=slot_money, + max_stocks=max_stocks, + total_budget_krw=total_budget_krw, + fee_rate=fee_rate, + sell_tax=sell_tax, + period_days=period_days, + cache_holder=cache_holder, ticks_by_code=ticks_preloaded, orderbook_by_code=orderbook_preloaded, program_by_code=program_preloaded, - meta_out=meta, + log_verdict_by_code=log_verdict_preloaded, + share_denom_by_code=share_denom_preloaded, ) - stats = bbc.summarize_breakout_trades( - trades, total_budget_krw=total_budget_krw, period_days=period_days, - ) - total_trades = stats["total_trades"] - if total_trades < min_trades: + if result_pkg is None: continue - total_pnl = stats["total_pnl"] - win_rate = stats["win_rate"] - pf = float(stats.get("pf") or 0) - if not combo_passes_search_filters( - win_rate=win_rate, pf=pf, - min_win_rate=min_win_rate, min_pf=min_pf, - ): - continue - avg_hold = stats["avg_hold_min"] - peak, mdd, cum = 0.0, 0.0, 0.0 - for t in trades: - cum += t["pnl"] - if cum > peak: - peak = cum - dd = peak - cum - if dd > mdd: - mdd = dd - - merged = dict(ui_params) - merged["slot_money"] = float(slot_money) - merged["max_stocks"] = int(max_stocks) - merged["total_budget_krw"] = float(total_budget_krw) - - sell_reasons = _count_sell_reasons(trades) - avg_profit_rate = _avg_profit_rate_pct(trades) - - result_pkg = { - "params": {k: ui_params[k] for k in keys}, - "total_pnl": int(total_pnl), - "win_rate": round(win_rate, 2), - "total_trades": total_trades, - "pf": round(pf, 2), - "avg_hold": round(avg_hold, 1), - "mdd": round(mdd), - "bot_pct": stats["bot_pct"], - "daily_avg_pct": stats["daily_avg_pct"], - "avg_profit_rate": avg_profit_rate, - "sell_reasons": sell_reasons, - "skipped_micro_buys": int( - (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0 - ), - "merged_params": merged, - } + total_pnl = result_pkg["total_pnl"] + win_rate = result_pkg["win_rate"] item_t = (total_pnl, win_rate, id(result_pkg), result_pkg) if len(local_heap) < top_n: heapq.heappush(local_heap, item_t) diff --git a/kis_trader/backtest/param_search_momentum.py b/kis_trader/backtest/param_search_momentum.py index 2d46598..c84f940 100644 --- a/kis_trader/backtest/param_search_momentum.py +++ b/kis_trader/backtest/param_search_momentum.py @@ -888,6 +888,111 @@ def _ui_to_engine_params(ui_params: dict) -> dict: return engine_params +def evaluate_momentum_param_combo( + combo: Dict[str, Any], + *, + base_fixed: Dict[str, Any], + grid_keys: List[str], + codes_candles: Dict[str, List[Dict]], + min_trades: int, + min_win_rate: float, + min_pf: float, + universe_by_slot: Optional[Dict[str, List[str]]] = None, + slot_money: float = 3_000_000.0, + max_stocks: int = 3, + total_budget_krw: float = 9_000_000.0, + fee_rate: float = 0.00015, + sell_tax: float = 0.0018, + period_days: int = 1, + cache_holder: Optional[Dict[str, Any]] = None, + ticks_by_code: Any = None, + orderbook_by_code: Any = None, + program_by_code: Any = None, + log_verdict_by_code: Any = None, + start_key: str = "", + end_key: str = "", +) -> Optional[Dict[str, Any]]: + """단일 모멘텀 조합 백테 — Grid 워커·Optuna objective 공통.""" + if not _momentum_combo_grid_valid(combo): + return None + + ui_params = dict(base_fixed) + ui_params.update(combo) + engine_params = _ui_to_engine_params(ui_params) + if cache_holder: + engine_params.update(cache_holder) + engine_params["slot_money"] = float(slot_money) + engine_params["max_stocks"] = int(max_stocks) + engine_params["total_budget_krw"] = float(total_budget_krw) + engine_params["portfolio_mode"] = True + if log_verdict_by_code: + engine_params["_backtest_log_verdict_by_code"] = log_verdict_by_code + + meta: Dict[str, Any] = {} + if len(start_key) >= 12: + meta["start_key"] = start_key + engine_params["_backtest_period_start_key"] = start_key[:12] + if len(end_key) >= 12: + meta["end_key"] = end_key + + trades = mbc.run_momentum_backtest_web_aligned( + codes_candles, engine_params, universe_by_slot, + slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax, + max_stocks=max_stocks, total_budget_krw=total_budget_krw, + ticks_by_code=ticks_by_code, + orderbook_by_code=orderbook_by_code, + program_by_code=program_by_code, + meta_out=meta, + ) + stats = mbc.summarize_momentum_trades( + trades, total_budget_krw=total_budget_krw, period_days=period_days, + ) + total_trades = stats["total_trades"] + if total_trades < min_trades: + return None + + total_pnl = stats["total_pnl"] + win_rate = stats["win_rate"] + pf = float(stats.get("pf") or 0) + if not combo_passes_search_filters( + win_rate=win_rate, pf=pf, + min_win_rate=min_win_rate, min_pf=min_pf, + ): + return None + + avg_hold = stats["avg_hold_min"] + peak, mdd, cum = 0.0, 0.0, 0.0 + for t in trades: + cum += t["pnl"] + if cum > peak: + peak = cum + dd = peak - cum + if dd > mdd: + mdd = dd + + merged = dict(ui_params) + merged["slot_money"] = float(slot_money) + merged["max_stocks"] = int(max_stocks) + merged["total_budget_krw"] = float(total_budget_krw) + + return { + "params": {k: ui_params[k] for k in grid_keys if k in ui_params}, + "total_pnl": int(total_pnl), + "win_rate": round(win_rate, 2), + "total_trades": total_trades, + "pf": round(pf, 2), + "avg_hold": round(avg_hold, 1), + "mdd": round(mdd), + "bot_pct": stats["bot_pct"], + "daily_avg_pct": stats["daily_avg_pct"], + "sell_reasons": mbc.count_momentum_sell_reasons(trades), + "skipped_micro_buys": int( + (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0 + ), + "merged_params": merged, + } + + def _evaluate_momentum_chunk( param_chunk: List[Dict[str, Any]], base_fixed: Dict[str, Any], @@ -910,6 +1015,7 @@ def _evaluate_momentum_chunk( orderbook_preloaded = None program_preloaded = None ticks_preloaded = None + log_verdict_preloaded = None if shared: if codes_candles is None: codes_candles = shared.get("codes_candles") or {} @@ -938,89 +1044,34 @@ def _evaluate_momentum_chunk( local_heap: List[Tuple[float, float, int, Dict]] = [] for combo in param_chunk: assert_parent_alive() - if not _momentum_combo_grid_valid(combo): - continue - - ui_params = dict(base_fixed) - ui_params.update(combo) - engine_params = _ui_to_engine_params(ui_params) - engine_params.update(cache_holder) - engine_params["slot_money"] = float(slot_money) - engine_params["max_stocks"] = int(max_stocks) - engine_params["total_budget_krw"] = float(total_budget_krw) - engine_params["portfolio_mode"] = True - if shared: - lv = shared.get("log_verdict_by_code") - if lv: - engine_params["_backtest_log_verdict_by_code"] = lv - - meta: Dict[str, Any] = {} - if shared: - sk = str(shared.get("start_key") or "") - if len(sk) >= 12: - meta["start_key"] = sk - engine_params["_backtest_period_start_key"] = sk[:12] - # 웹 백테와 동일: end_key 있어야 scan_at 유니버스 타임라인 부착 - ek = str(shared.get("end_key") or "") - if len(ek) >= 12: - meta["end_key"] = ek - trades = mbc.run_momentum_backtest_web_aligned( - codes_candles, engine_params, universe_by_slot, - slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax, - max_stocks=max_stocks, total_budget_krw=total_budget_krw, + result_pkg = evaluate_momentum_param_combo( + combo, + base_fixed=base_fixed, + grid_keys=keys, + codes_candles=codes_candles, + min_trades=min_trades, + min_win_rate=min_win_rate, + min_pf=min_pf, + universe_by_slot=universe_by_slot, + slot_money=slot_money, + max_stocks=max_stocks, + total_budget_krw=total_budget_krw, + fee_rate=fee_rate, + sell_tax=sell_tax, + period_days=period_days, + cache_holder=cache_holder, ticks_by_code=ticks_preloaded, orderbook_by_code=orderbook_preloaded, program_by_code=program_preloaded, - meta_out=meta, + log_verdict_by_code=log_verdict_preloaded, + start_key=str(shared.get("start_key") or "") if shared else "", + end_key=str(shared.get("end_key") or "") if shared else "", ) - stats = mbc.summarize_momentum_trades( - trades, total_budget_krw=total_budget_krw, period_days=period_days, - ) - total_trades = stats["total_trades"] - if total_trades < min_trades: + if result_pkg is None: continue - total_pnl = stats["total_pnl"] - win_rate = stats["win_rate"] - pf = float(stats.get("pf") or 0) - if not combo_passes_search_filters( - win_rate=win_rate, pf=pf, - min_win_rate=min_win_rate, min_pf=min_pf, - ): - continue - avg_hold = stats["avg_hold_min"] - peak, mdd, cum = 0.0, 0.0, 0.0 - for t in trades: - cum += t["pnl"] - if cum > peak: - peak = cum - dd = peak - cum - if dd > mdd: - mdd = dd - - merged = dict(ui_params) - merged["slot_money"] = float(slot_money) - merged["max_stocks"] = int(max_stocks) - merged["total_budget_krw"] = float(total_budget_krw) - - result_pkg = { - "params": {k: ui_params[k] for k in keys}, - "total_pnl": int(total_pnl), - "win_rate": round(win_rate, 2), - "total_trades": total_trades, - "pf": round(pf, 2), - "avg_hold": round(avg_hold, 1), - "mdd": round(mdd), - "bot_pct": stats["bot_pct"], - "daily_avg_pct": stats["daily_avg_pct"], - "sell_reasons": mbc.count_momentum_sell_reasons(trades), - "skipped_micro_buys": int( - (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0 - ), - "merged_params": merged, - } - # 청크 내 상위 top_n (총손익 기준): root = 현재 보관 중 최저 손익 → heapreplace 로만 교체 - # (과거 min-heap + pushpop 버그로 워커·머지 단계에서 손익/승률이 뒤바뀌어 순위가 무너짐) + total_pnl = result_pkg["total_pnl"] + win_rate = result_pkg["win_rate"] item_t = (total_pnl, win_rate, id(result_pkg), result_pkg) if len(local_heap) < top_n: heapq.heappush(local_heap, item_t) diff --git a/kis_trader/backtest/param_search_optuna.py b/kis_trader/backtest/param_search_optuna.py new file mode 100644 index 0000000..9f42456 --- /dev/null +++ b/kis_trader/backtest/param_search_optuna.py @@ -0,0 +1,784 @@ +#!/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 + +실행 예: + # 꼬리 + python3 kis_trader/backtest/param_search_optuna.py --strategy tail --mode fast --trials 200 + # 모멘텀 (1위 정렬 기본 score=순익/MDD) + python3 kis_trader/backtest/param_search_optuna.py --strategy momentum --mode fast --trials 200 + # 돌파 + python3 kis_trader/backtest/param_search_optuna.py --strategy breakout --mode fast --trials 200 + +Win11 + VM 동시 분산: 같은 study-name · 같은 storage(141/kis_optuna) 로 각각 --trials 실행. + +DB 적용: + --apply-best (1위 trial → env_config, 총손익≤0 이면 스킵) + +Env (선택): + OPTUNA_DB_NAME=kis_optuna # 기본. 변경 시에만 설정 + OPTUNA_STORAGE_URL=... # 전체 URL 직접 지정 시 위보다 우선 + OPTUNA_TAIL_STUDY_NAME=... # study 이름 고정 +""" +from __future__ import annotations + +import argparse +import json +import logging +import os +import signal +import sys +import time +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, + optuna_run_lock_name, + resolve_optuna_storage_url, + resolve_study_name, +) +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_search_space import suggest_tail_params, tail_grid_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_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", "full", "massive"], + "momentum": ["fast", "rr", "coarse", "fine", "full"], + "breakout": ["fast", "coarse", "fine", "full"], +} + + +@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) + + +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", +) -> Optional[TailSearchContext]: + """ + run_search 와 동일한 데이터·base_params 1회 로드 (Grid 중복 최소화). + 데이터 없으면 None. + """ + db = TradeDB() + try: + base_params = te.get_tail_defaults_from_db(db) + 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 + ob_filter_on = bool(base_params.get("_orderbook_filter_enabled")) or _ob_mode == "auto" + logger.info( + "📌 호가필터: %s (%s)", + _ob_mode.upper(), + "적용" if ob_filter_on else "스킵 — 코어 파라미터 순수 탐색", + ) + + row = db.conn.execute("SELECT * FROM env_config ORDER BY id DESC LIMIT 1").fetchone() + r = dict(row) if row else {} + 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 + 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) + ) + if use_fallback_universe: + print("📌 [유니버스] --fallback-universe: 저장 이력 무시 → ws_candles 전 종목") + elif universe_source == "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 저장 이력 (웹백테 동일) | " + 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 + base_params.setdefault("force_eod_exit", False) + base_params.setdefault("backtest_tick_fallback_ohlc", True) + from kis_trader.engine.tail_tick_replay import tail_backtest_use_tick_db as _tail_use_tick + base_params.setdefault("backtest_use_tick_db", _tail_use_tick(None)) + if base_params.get("backtest_use_tick_db"): + logger.info("📌 틱재생(ws_ticks): ON — 실매 체결 정합 모드 (OHLC 낙관편향 제거)") + + 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}" + ) + + 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) + _tick_probe["entry_mode"] = "align" + 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 후 재탐색)") + + from kis_trader.backtest.tail_param_search import _tail_grids + 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 "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_grid_axis_keys(mode), + ob_filter_on=ob_filter_on, + cache_holder=cache_holder, + ) + finally: + db.close() + + +def _make_sampler(name: str, seed: Optional[int]): + n = (name or "tpe").strip().lower() + if n == "random": + return RandomSampler(seed=seed) + return TPESampler(seed=seed, multivariate=True) + + +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 = "pnl", + 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, + ) + + def objective(trial: optuna.Trial) -> float: + combo = suggest_tail_params(trial, ctx.mode) + 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 + + 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("total_trades", int(result["total_trades"])) + trial.set_user_attr("params_json", json.dumps(result["params"], ensure_ascii=False)) + + if sort_by == "win_rate": + return float(result["win_rate"]) + return float(result["total_pnl"]) + + 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() + study.optimize( + objective, + n_trials=n_trials, + n_jobs=n_jobs, + show_progress_bar=show_progress, + ) + elapsed = time.time() - t0 + logger.info("✅ Optuna 완료 | %.1f초 | 완료 trial %d", elapsed, len(study.trials)) + + # JSON export — study.user_attrs 기준 (n_jobs>1 에도 안전) + 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) + passing.append({ + "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), + "optuna_trial_number": trial.number, + }) + if sort_by == "pnl": + passing.sort(key=lambda r: (-float(r["total_pnl"]), -float(r["win_rate"]))) + else: + passing.sort(key=lambda r: (-float(r["win_rate"]), -float(r["total_pnl"]))) + + profitable = [r for r in passing if float(r.get("total_pnl") or 0) > 0] + if profitable: + passing = profitable + + out_data = { + "engine": "optuna", + "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), + "results": passing[:5000], + } + + 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) + + logger.info("💾 Optuna 결과 저장: %s", out_path) + + 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 + + +def apply_best_trial(study: optuna.Study, ctx: TailSearchContext) -> bool: + """Best trial → env_config (총손익≤0 스킵).""" + if not study.best_trial or study.best_value <= _FAIL_OBJECTIVE + 1: + logger.warning("⚠️ 적용할 best trial 없음 — DB 미적용") + return False + pnl = float(study.best_trial.user_attrs.get("total_pnl") or 0) + if pnl <= 0: + logger.warning("⚠️ Best trial 총손익 ≤ 0 — DB 미적용. 기존 설정 유지.") + return False + params_raw = study.best_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] trial #%d → env_config", study.best_trial.number) + logger.info("적용된 값: %s", json.dumps(env_map, indent=2, ensure_ascii=False)) + return True + + +def main() -> None: + today = datetime.now().strftime("%Y-%m-%d") + week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") + + 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", + ) + 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( + "--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="목적함수: tail/breakout pnl|win_rate, momentum score|pnl|win_rate (미지정=전략 기본)", + ) + add_search_filter_cli_args(parser) + parser.add_argument("--min_trades", default=3, type=int, help="최소 거래 건수") + 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( + "--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 끄기", + ) + args = parser.parse_args() + + 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)) + + n_jobs = args.n_jobs + if n_jobs is None: + n_jobs = get_env_int("PARAM_SEARCH_OPTUNA_N_JOBS", 1) + 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/breakout 중 하나") + sys.exit(2) + + 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: + sort_by = "score" if strategy == "momentum" else "pnl" + momentum_sort = {"score", "pnl", "win_rate"} + basic_sort = {"pnl", "win_rate"} + if strategy == "momentum" and sort_by not in momentum_sort: + logger.error("❌ momentum --sort-by 는 score|pnl|win_rate") + sys.exit(2) + if strategy in ("tail", "breakout") and sort_by not in basic_sort: + logger.error("❌ %s --sort-by 는 pnl|win_rate", strategy) + 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_name = resolve_study_name( + strategy=strategy, + mode=mode, + start=args.start, + end=args.end, + cli_override=args.study_name, + ) + + 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, + ) + 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 == "momentum": + ctx_m = prepare_momentum_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, + ) + 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: + apply_best_momentum_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, + ) + 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) + + except KeyboardInterrupt as e: + print(f"\n⛔ {e} — 중단", flush=True) + sys.exit(130) + finally: + run_lock.release() + + +if __name__ == "__main__": + main() diff --git a/kis_trader/backtest/tail_param_search.py b/kis_trader/backtest/tail_param_search.py index 1893327..67a2358 100644 --- a/kis_trader/backtest/tail_param_search.py +++ b/kis_trader/backtest/tail_param_search.py @@ -516,6 +516,107 @@ TAIL_GRID_AXIS_HINTS_KO: Dict[str, str] = { } +def evaluate_tail_param_combo( + combo: Dict[str, Any], + *, + base_params: Dict[str, Any], + candles_by_code: Dict[str, List[Dict]], + fee_rate: float, + sell_tax: float, + min_trades: int, + min_win_rate: float, + min_pf: float, + universe_by_slot: Optional[Dict[str, List[str]]] = None, + slot_money: float = 3_000_000.0, + max_stocks: int = 3, + total_budget_krw: float = 9_000_000.0, + period_days: int = 1, + cache_holder: Optional[Dict[str, Any]] = None, + ticks_by_code: Any = None, + orderbook_by_code: Any = None, + program_by_code: Any = None, + log_verdict_by_code: Any = None, +) -> Optional[Dict[str, Any]]: + """ + 단일 파라미터 조합 백테 — Grid 워커·Optuna objective 공통. + min_trades·승률·PF 게이트 미통과 시 None. + """ + test_params = dict(base_params) + test_params.update(combo) + if "max_spread_pct" in combo and combo.get("max_spread_pct") is not None: + test_params["_ob_max_spread_pct"] = float(combo["max_spread_pct"]) + if "min_bid_ask_ratio" in combo and combo.get("min_bid_ask_ratio") is not None: + test_params["_ob_min_bid_ask_ratio"] = float(combo["min_bid_ask_ratio"]) + # 당일손익 트레일 익절 — tier(우선) 또는 단일 drop 이 켜졌을 때만 시뮬 게이트 ON. + _tier_combo = str(combo.get("trail_tiers") or "").strip().lower() + _tier_on = _tier_combo not in ("", "off") + if _tier_on: + test_params["_backtest_daily_profit_trail"] = True + test_params.setdefault("daily_profit_mode", "trailing") + test_params["daily_trail_tiers"] = str(combo["trail_tiers"]).strip() + elif float(combo.get("trail_drop_pct") or 0) > 0: + test_params["_backtest_daily_profit_trail"] = True + test_params.setdefault("daily_profit_mode", "trailing") + test_params["daily_trail_drop_pct"] = float(combo["trail_drop_pct"]) + if combo.get("trail_arm_krw") is not None: + test_params["daily_trail_arm_krw"] = float(combo["trail_arm_krw"]) + else: + _arm_env = os.environ.get("TAIL_GRID_TRAIL_ARM_KRW", "") + if str(_arm_env).strip() != "": + test_params["daily_trail_arm_krw"] = float(_arm_env) + if cache_holder: + test_params.update(cache_holder) + if log_verdict_by_code: + test_params["_backtest_log_verdict_by_code"] = log_verdict_by_code + + meta: Dict[str, Any] = {} + trades = tbc.run_tail_backtest_web_aligned( + candles_by_code, test_params, universe_by_slot, + slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax, + max_stocks=max_stocks, total_budget_krw=total_budget_krw, + ticks_by_code=ticks_by_code, + orderbook_by_code=orderbook_by_code, + program_by_code=program_by_code, + meta_out=meta, + ) + + stats = tbc.summarize_tail_trades( + trades, + total_budget_krw=total_budget_krw, + period_days=period_days, + ) + total_trades = stats["total_trades"] + if total_trades < min_trades: + return None + + win_rate = stats["win_rate"] + total_pnl = stats["total_pnl"] + pf = float(stats.get("pf") or 0) + if not combo_passes_search_filters( + win_rate=win_rate, pf=pf, + min_win_rate=min_win_rate, min_pf=min_pf, + ): + return None + + return { + "params": combo, + "apply_cfg": {**base_params, **combo}, + "total_trades": total_trades, + "win_rate": win_rate, + "total_pnl": total_pnl, + "wins": stats["wins"], + "losses": stats["losses"], + "pf": stats["pf"], + "bot_pct": stats["bot_pct"], + "daily_avg_pct": stats["daily_avg_pct"], + "avg_hold_min": stats["avg_hold_min"], + "sell_reasons": _count_tail_sell_reasons(trades), + "skipped_micro_buys": int( + (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0 + ), + } + + def evaluate_param_chunk( param_chunk: List[Dict[str, Any]], base_params: Dict[str, Any], @@ -541,6 +642,7 @@ def evaluate_param_chunk( ticks_preloaded = None orderbook_preloaded = None program_preloaded = None + log_verdict_preloaded = None if shared: if candles_by_code is None: candles_by_code = shared.get("candles_by_code") or {} @@ -559,6 +661,7 @@ def evaluate_param_chunk( ticks_preloaded = _tm orderbook_preloaded = shared.get("orderbook_by_code") program_preloaded = shared.get("program_by_code") + log_verdict_preloaded = shared.get("log_verdict_by_code") if candles_by_code is None: candles_by_code = {} cache_holder: Dict[str, Any] = {} @@ -567,86 +670,31 @@ def evaluate_param_chunk( for combo in param_chunk: assert_parent_alive() - test_params = dict(base_params) - test_params.update(combo) - if "max_spread_pct" in combo and combo.get("max_spread_pct") is not None: - test_params["_ob_max_spread_pct"] = float(combo["max_spread_pct"]) - if "min_bid_ask_ratio" in combo and combo.get("min_bid_ask_ratio") is not None: - test_params["_ob_min_bid_ask_ratio"] = float(combo["min_bid_ask_ratio"]) - # 당일손익 트레일 익절 — tier(우선) 또는 단일 drop 이 켜졌을 때만 시뮬 게이트 ON. - # tier='off' & drop=0 → 게이트 OFF(기존 백테 동작 불변·기준군). - _tier_combo = str(combo.get("trail_tiers") or "").strip().lower() - _tier_on = _tier_combo not in ("", "off") - if _tier_on: - test_params["_backtest_daily_profit_trail"] = True - test_params.setdefault("daily_profit_mode", "trailing") - # tier 우선 — 전용키로 주입(꼬리 개별포지션 키와 분리). drop/arm 은 무시됨. - test_params["daily_trail_tiers"] = str(combo["trail_tiers"]).strip() - elif float(combo.get("trail_drop_pct") or 0) > 0: - test_params["_backtest_daily_profit_trail"] = True - test_params.setdefault("daily_profit_mode", "trailing") - # 단일 drop 폴백 — 전용키로 명시 주입. - test_params["daily_trail_drop_pct"] = float(combo["trail_drop_pct"]) - # arm(발동 최소수익) — 그리드 축 우선 → env → DB. (전용키 → DB 미변경) - if combo.get("trail_arm_krw") is not None: - test_params["daily_trail_arm_krw"] = float(combo["trail_arm_krw"]) - else: - _arm_env = os.environ.get("TAIL_GRID_TRAIL_ARM_KRW", "") - if str(_arm_env).strip() != "": - test_params["daily_trail_arm_krw"] = float(_arm_env) - test_params.update(cache_holder) - if shared: - lv = shared.get("log_verdict_by_code") - if lv: - test_params["_backtest_log_verdict_by_code"] = lv - - meta: Dict[str, Any] = {} - trades = tbc.run_tail_backtest_web_aligned( - candles_by_code, test_params, universe_by_slot, - slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax, - max_stocks=max_stocks, total_budget_krw=total_budget_krw, + result_pkg = evaluate_tail_param_combo( + combo, + base_params=base_params, + candles_by_code=candles_by_code, + fee_rate=fee_rate, + sell_tax=sell_tax, + min_trades=min_trades, + min_win_rate=min_win_rate, + min_pf=min_pf, + universe_by_slot=universe_by_slot, + slot_money=slot_money, + max_stocks=max_stocks, + total_budget_krw=total_budget_krw, + period_days=period_days, + cache_holder=cache_holder, ticks_by_code=ticks_preloaded, orderbook_by_code=orderbook_preloaded, program_by_code=program_preloaded, - meta_out=meta, + log_verdict_by_code=log_verdict_preloaded, ) - - stats = tbc.summarize_tail_trades( - trades, - total_budget_krw=total_budget_krw, - period_days=period_days, - ) - total_trades = stats["total_trades"] - if total_trades < min_trades: + if result_pkg is None: continue - win_rate = stats["win_rate"] - total_pnl = stats["total_pnl"] - pf = float(stats.get("pf") or 0) - if not combo_passes_search_filters( - win_rate=win_rate, pf=pf, - min_win_rate=min_win_rate, min_pf=min_pf, - ): - continue - - result_pkg = { - "params": combo, - "apply_cfg": {**base_params, **combo}, - "total_trades": total_trades, - "win_rate": win_rate, - "total_pnl": total_pnl, - "wins": stats["wins"], - "losses": stats["losses"], - "pf": stats["pf"], - "bot_pct": stats["bot_pct"], - "daily_avg_pct": stats["daily_avg_pct"], - "avg_hold_min": stats["avg_hold_min"], - "sell_reasons": _count_tail_sell_reasons(trades), - "skipped_micro_buys": int( - (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0 - ), - } - + win_rate = result_pkg["win_rate"] + total_pnl = result_pkg["total_pnl"] if len(local_heap) < top_n: heapq.heappush(local_heap, (win_rate, total_pnl, id(result_pkg), result_pkg)) else: diff --git a/requirements.txt b/requirements.txt index 702cf44..f201f2d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -91,6 +91,7 @@ multitasking==0.0.12 mypy_extensions==1.1.0 networkx==3.4.2 numpy==1.26.4 +optuna==4.2.1 openai==2.20.0 orjson==3.11.7 oslex==0.1.3