kis_bot/kis_trader/backtest/param_search_optuna.py

#!/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)

    # 2026-09-06 (docs/rust_engine_parity_port_plan.md · 룰 29): env 게이트로 Rust 세션 초기화 제한.
    _use_rust_env = os.environ.get("BACKTEST_USE_RUST") == "1"
    rust_session_id: Optional[str] = None
    if _use_rust_env:
        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  # noqa: F401
            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}")
            rust_session_id = None
    else:
        logger.info("🐍 [Python 엔진] Rust 세션 미초기화 (BACKTEST_USE_RUST != 1)")

    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 세션이 없으면 세팅하지 않음 (Python 경로)
        if rust_session_id is not None:
            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)
        # rust_session_id 는 use_rust=True 로 초기화된 경우에만 clear 필요
        if rust_session_id is not None:
            try:
                from kis_trader.engine.tail_engine import clear_rust_session
                clear_rust_session(rust_session_id)
            except Exception:
                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()