#!/usr/bin/env python3
"""
kis_trader/backtest/param_search_breakout.py — 돌파매매 백테스트 파라미터 자동 탐색 (Grid Search)
========================================================================================================
[전략 = BREAKOUT]
- 키움 HTS 'breakout' 조건검색(직전봉 등락률 + 고가돌파 + 거래량 spike) 기반 추세 돌파.
- 실매매 BreakoutStrategy.check_buy → ``BREAKOUT_ENTRY_MODE=intrabar`` (B, HTS E) / ``align`` (A).
- 본 CLI 는 ``run_breakout_backtest`` + ``breakout_scan_buy_at_bar`` (라이브·백테 동일 진입 함수).

[관련 CLI (전략별 파일 분리, 2026-05 정리)]
    SCALP(Reversal): kis_trader/backtest/param_search_scalping.py
    MOMENTUM       : kis_trader/backtest/param_search_momentum.py
    BREAKOUT       : kis_trader/backtest/param_search_breakout.py   ← 이 파일
    SHORT(꼬리잡기) : kis_trader/backtest/tail_param_search.py
    UPDOW(하락매수): kis_trader/backtest/param_search_updow.py

실행:
    cd /home/hoon/kis_bot
    python3 kis_trader/backtest/param_search_breakout.py --start 2026-04-21 --end 2026-05-08
    python3 -m kis_trader.backtest.param_search_breakout --mode coarse --top 30

옵션:
    --start         시작일 (기본: 오늘-7일)
    --end           종료일 (기본: 오늘)
    --mode          탐색 모드: fast(기본, ~768조합) / coarse / fine / full
    --top           상위 N개 출력·JSON 저장 (기본: 1000)
    --min_trades    최소 거래 건수 필터 (기본: 1)
    --apply         N번째 결과를 DB(BREAKOUT_* env)에 자동 적용 (총손익 > 0 일 때만)
    --fallback-universe  저장 이력 무시, 시뮬레이션 유니버스 사용

⚠️ SCAN(HTS) vs TRIGGER(코드) — 운영 가이드:
    [HTS 널넬게]  A 등락 0.3~20% 또는 삭제 · E 거래량 150~200% · F 돌파 유지 또는 완화
    [코드 엄격]   그리드 vol_mult(2~5배)·prev_chg·tp/sl·time_end — ``param_search_breakout`` coarse 1축
    웹/DB ``BREAKOUT_VOL_MULT`` = 그리드 1위 적용값 (HTS 300%와 별개로 더 세게 잡을 수 있음).

⚠️ 청산 순서 (돌파 전용 — 스캘핑 V4 와 다름, ``breakout.check_sell_signal_breakout_live``):
    EOD → **익절(1순위)** → 어깨컷(2) → 손절(3) → 트레일(4).
    그리드 어깨축은 익절 미달 구간의 **트레일 보호**용(0.3%급 타이트값 아님).

⚠️ 그리드 축 오버라이드: ``BREAKOUT_GRID_COARSE_VOL_MULT=2,2.5,3,4,5`` 등.
"""

import sys, os, json, time, argparse, signal
import heapq
from datetime import datetime, timedelta
from itertools import product
from concurrent.futures import as_completed
from typing import Optional, List, Dict, Any, Tuple


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 logging
logging.getLogger("TradeDB").setLevel(logging.WARNING)

from database import TradeDB
from kis_trader.backtest import breakout_backtest_common as bbc
from kis_trader.backtest.backtest_portfolio_common import (
    merge_param_search_apply_source,
    portfolio_env_patch,
    session_env_patch,
)
from kis_trader.backtest.param_search_cli_common import (
    add_portfolio_cli_args,
    add_search_filter_cli_args,
    apply_session_to_fixed,
    combo_passes_search_filters,
    format_session_hm,
    search_json_meta,
)
from kis_trader.backtest.param_search_pool import (
    ParamSearchSharedPayload,
    ParamSearchProgressETA,
    assert_parent_alive,
    cap_combos_uniform,
    cap_combos_uniform_lazy,
    iter_pool_chunk_results,
    managed_process_pool,
    param_search_chunk_plan,
    param_search_worker_budget_line,
    try_acquire_run_lock,
    worker_shared_get,
)
from kis_trader.backtest.breakout_tick_loader import (
    load_breakout_ticks_by_code,
    tick_coverage_stats,
)
from kis_trader.strategies.breakout import (
    breakout_backtest_wants_tick_replay,
    breakout_entry_mode,
    breakout_invest_amount_krw,
    breakout_min_bars_required,
    breakout_ui_to_engine_params,
    normalize_breakout_max_loss_krw,
)
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_float, get_env_int  # noqa: E402


def _parse_csv_floats(env_key: str, fallback: List[float]) -> List[float]:
    raw = str(get_env_from_db(env_key, "") or "").strip()
    if not raw:
        return list(fallback)
    out: List[float] = []
    for part in raw.split(","):
        part = part.strip()
        if not part:
            continue
        try:
            out.append(float(part))
        except ValueError:
            continue
    return out if out else list(fallback)


def _parse_csv_ints(env_key: str, fallback: List[int]) -> List[int]:
    raw = str(get_env_from_db(env_key, "") or "").strip()
    if not raw:
        return list(fallback)
    out: List[int] = []
    for part in raw.split(","):
        part = part.strip()
        if not part:
            continue
        try:
            out.append(int(float(part)))
        except ValueError:
            continue
    return out if out else list(fallback)


def _parse_csv_strs(env_key: str, fallback: List[str]) -> List[str]:
    """콤마(,) 구분 문자열 후보 축 (예: sl_mode=fixed,atr). 빈 값이면 fallback."""
    raw = str(get_env_from_db(env_key, "") or "").strip()
    if not raw:
        return list(fallback)
    out = [part.strip().lower() for part in raw.split(",") if part.strip()]
    return out if out else list(fallback)


def _parse_csv_tiers(env_key: str, fallback: List[str]) -> List[str]:
    """세미콜론(;) 구분 래칫 티어 후보 → 그리드 문자열 축.

    래칫 티어 1개는 ``"gain:cut,gain:cut"`` 형태(% 단위, 예 ``"2:1.5,5:1.0"``
    = +2%부터 1.5% 되돌림컷, +5%부터 1.0%). 티어 내부 구분이 콤마(,)이므로
    후보(조합 축)끼리는 세미콜론(;)으로 나눈다.
    빈 후보("")는 래칫 OFF(= 단일 어깨컷, 기존 동작) — 비교 기준선으로 항상 포함 권장.
    예: BREAKOUT_GRID_FAST_RATCHET_TIERS=";2:1.5,5:1.0;3:1.5,6:1.0,10:0.7"
    """
    raw = get_env_from_db(env_key, "")
    if raw is None or str(raw).strip() in ("", "None"):
        return list(fallback)
    out: List[str] = [part.strip() for part in str(raw).split(";")]
    return out if out else list(fallback)


def _count_sell_reasons(trades: List[Dict[str, Any]]) -> Dict[str, int]:
    """청산 사유별 건수 — 익절·어깨 순위 변경 후 1위 조합 검증용."""
    out: Dict[str, int] = {}
    for t in trades or []:
        k = str(t.get("sell_reason") or "unknown")
        out[k] = out.get(k, 0) + 1
    return out


def _avg_profit_rate_pct(trades: List[Dict[str, Any]]) -> float:
    if not trades:
        return 0.0
    return round(
        sum(float(t.get("profit_rate", 0) or 0) for t in trades) / len(trades),
        2,
    )


# ──────────────────────────────────────────────────────────────────────────────
# 돌파매매 기본값 (DB 우선)
# ──────────────────────────────────────────────────────────────────────────────

def _bo_fixed_defaults() -> Dict[str, Any]:
    """env_config 에서 BREAKOUT_* 비-그리드 파라미터 기본값 로드.

    ※ UI/CLI 표시 단위(%)로 반환. 워커 _ui_to_engine_params 가 비율로 변환.
    """
    db = TradeDB()
    try:
        row = db.conn.execute(
            "SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
        ).fetchone()
        env = dict(row) if row else {}
    finally:
        db.close()

    def _pick(*keys, default=None, cast=float):
        for k in keys:
            v = env.get(k)
            if v not in (None, "", "None"):
                try:
                    return cast(v)
                except (ValueError, TypeError):
                    continue
        return default

    def _ratio_to_ui_pct(ratio: float) -> float:
        av = abs(float(ratio))
        if av == 0:
            return 0.0
        return round(av * 100, 3) if av < 0.5 else round(av, 3)

    return {
        # 진입 룰 — max_daily_chg 는 fast 그리드 축(콤보가 덮어씀). 미포함 모드 폴백용.
        "lookback_min":   int(_pick("BREAKOUT_LOOKBACK_MIN", default=1, cast=lambda v: int(float(v)))),
        "vol_window":     int(_pick("BREAKOUT_VOL_WIN", default=1, cast=lambda v: int(float(v)))),
        "vol_mult":       float(_pick("BREAKOUT_VOL_MULT", default=0.0)),
        "min_turnover_1m_pct": float(_pick("BREAKOUT_MIN_TURNOVER_1M_PCT", default=0.05)),
        "max_daily_chg":  float(_pick("BREAKOUT_MAX_DAILY_CHG", default=15.0)),
        "min_price":      int(_pick("BREAKOUT_MIN_PRICE", "MIN_STOCK_PRICE", default=1000, cast=lambda v: int(float(v)))),
        # 시간대 (골든타임 09:00~10:30 기본)
        "time_start_hm":  int(_pick("BREAKOUT_TIME_START", default=900, cast=lambda v: int(float(v)))),
        "time_end_hm":    int(_pick("BREAKOUT_GOLDEN_END_HM", "BREAKOUT_TIME_END", default=1030, cast=lambda v: int(float(v)))),
        # 청산 (UI % 단위)
        "trail_pct":      _ratio_to_ui_pct(float(_pick("BREAKOUT_TRAIL_PCT", default=0.015))),
        "shoulder_min_high_pct": _ratio_to_ui_pct(
            float(_pick("BREAKOUT_SHOULDER_MIN_HIGH_PCT", default=0.02)),
        ),
        "shoulder_cut_pct": _ratio_to_ui_pct(
            float(_pick("BREAKOUT_SHOULDER_CUT_PCT", default=0.01)),
        ),
        # 슬롯/수수료/쿨다운
        "slot_money":     int(_pick("BREAKOUT_SLOT_MONEY", "SLOT_MONEY_DEFAULT", default=2_000_000, cast=lambda v: int(float(v)))),
        "fee_rate":       0.015,        # %
        "sell_tax":       0.18,         # %
        "cooldown_min":   30,
        "max_daily":      int(_pick("BREAKOUT_MAX_DAILY", default=1, cast=lambda v: int(float(v)))),
        "max_loss_krw":   normalize_breakout_max_loss_krw(_pick(
            "BREAKOUT_MAX_LOSS_PER_TRADE_KRW", "MAX_LOSS_PER_TRADE_KRW",
            default=200_000, cast=lambda v: int(float(v)),
        )),
        "entry_mode":     str(_pick("BREAKOUT_ENTRY_MODE", default="intrabar") or "intrabar").strip().lower(),
        "intrabar_slippage_pct": float(_pick("BREAKOUT_INTRABAR_SLIPPAGE_PCT", default=0.0)),
        # ── ATR 동적 손절 (sl_mode 는 fast 그리드 축. 미포함 모드 폴백용 기본값) ──
        "sl_mode":        str(_pick("BREAKOUT_SL_MODE", default="fixed", cast=str) or "fixed").strip().lower(),
        "atr_period":     int(_pick("BREAKOUT_ATR_PERIOD", default=14, cast=lambda v: int(float(v)))),
        "atr_sl_mult":    float(_pick("BREAKOUT_ATR_SL_MULT", default=2.0)),
        "atr_sl_min_pct": float(_pick("BREAKOUT_ATR_SL_MIN_PCT", default=0.8)),
        "atr_sl_max_pct": float(_pick("BREAKOUT_ATR_SL_MAX_PCT", default=6.0)),
        "eod_enabled": str(_pick("BREAKOUT_EOD_ENABLED", default="1")).strip().lower() in (
            "1", "true", "t", "y", "yes", "on",
        ),
        "eod_hm": (
            str(_pick("BREAKOUT_EOD_HM", default="15:15")).strip() or "15:15"
        ),
    }


# 결과 디렉터리 공유
def _results_dir_for_write() -> str:
    d = os.path.join(HERE, "results")
    os.makedirs(d, exist_ok=True)
    return d


def _results_dirs_for_read() -> List[str]:
    return [
        os.path.join(HERE, "results"),
        os.path.join(os.path.expanduser("~"), ".kis_bot_search_results"),
    ]


def _latest_json(prefix: str) -> Optional[str]:
    best_path, best_mtime = None, -1.0
    for d in _results_dirs_for_read():
        if not os.path.isdir(d):
            continue
        for f in os.listdir(d):
            if not (f.startswith(prefix) and f.endswith(".json")):
                continue
            p = os.path.join(d, f)
            try:
                m = os.path.getmtime(p)
            except OSError:
                continue
            if m > best_mtime:
                best_mtime = m
                best_path = p
    return best_path


def _apply_from_latest_json(rank: int) -> None:
    latest_path = _latest_json("search_breakout_")
    if not latest_path:
        print("⚠️  search_breakout_*.json 파일이 없습니다.")
        return
    with open(latest_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    top = data.get("top") or []
    if rank < 1 or rank > len(top):
        print(f"⚠️  순번 {rank}이(가) 유효하지 않습니다. (1~{len(top)})")
        return
    item = top[rank - 1]
    merged = merge_param_search_apply_source(item, data)
    if not merged:
        print("⚠️  해당 항목에 merged_params/params가 없습니다.")
        return
    if item.get("total_pnl", 0) <= 0:
        print(f"⚠️ {rank}번째 결과는 총손익 ≤ 0 → DB 미적용. 기존 설정 유지.")
        return
    print(f"📂 {latest_path} 에서 {rank}번째 적용합니다.")
    _apply_to_db(merged)


# ──────────────────────────────────────────────────────────────────────────────
# 돌파매매 그리드
# ──────────────────────────────────────────────────────────────────────────────

def _breakout_grids() -> Dict[str, Dict[str, List]]:
    """탐색 모드별 돌파매매 그리드.

    청산: EOD → 익절 → 어깨 → 손절 → 트레일 (돌파 전용).
    어깨축은 익절(7~15%) 미도달 시 고점 추적·되돌림 컷 — 스캘핑용 0.3%급과 구분.
    """
    return {
        # [FAST] TRIGGER 거래량 = 유통주식 1분 회전율(%) 스윕 · 급등주 한도 스윕.
        # ※ vol_mult(직전봉평균×배수)는 그리드 밖 — FIXED vol_mult=0(OFF). HTS 조건 1.2배는 SCAN 전용.
        # env: BREAKOUT_GRID_FAST_MIN_TURNOVER_1M_PCT=0.05,0.1,0.15
        "fast": {
            # ★ 당일시가 대비 누적 상승 한도(%). HTS 전일대비 등락과 다름.
            "max_daily_chg": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_MAX_DAILY_CHG", [15.0, 20.0, 25.0, 30.0],
            ),
            # ★ TRIGGER 1차 거래량 — 1분 거래량÷유통주식×100 (ka10001). vol_mult 와 별개.
            "min_turnover_1m_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_MIN_TURNOVER_1M_PCT", [0.05, 0.1, 0.15],
            ),
            "prev_chg_min": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_PREV_CHG_MIN", [0.5],
            ),
            # ★ 과열컷 — 직전봉 등락 상한(%). 너무 달군 봉에 추격 진입 회피(휩쏘 방어).
            "prev_chg_max": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_PREV_CHG_MAX", [5.0, 10.0, 15.0],
            ),
            # ★ 익절(%) — R:R 탐색축. 7/10/15.
            "tp_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_TP_PCT", [7.0, 10.0, 15.0],
            ),
            # ★ 손절(%) — PF 직격 축. 2/3/4 (멘토 지적: 고정 3% 단일값 → 탐색).
            "sl_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_SL_PCT", [2.0, 3.0, 4.0],
            ),
            "shoulder_min_high_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_SHOULDER_SMIN", [1.0, 2.0, 3.0],
            ),
            # ★ 어깨컷(%) — 익절 미달 구간 고점 되돌림 컷. 0.5/0.8/1.2.
            "shoulder_cut_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_SHOULDER_CUT", [0.5, 0.8, 1.2],
            ),
            # ★ 트레일(%) — 고점 추적 되돌림 폭. arm 은 고정(아래) 유지.
            "trail_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_TRAIL_PCT", [1.0, 1.5, 2.0],
            ),
            "trail_arm_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_TRAIL_ARM", [1.5],
            ),
            "lookback_min": _parse_csv_ints(
                "BREAKOUT_GRID_FAST_LOOKBACK_MIN", [1],
            ),
            "confirm_margin_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_CONFIRM_MARGIN", [0.0],
            ),
            "body_min_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_BODY_MIN", [0.0],
            ),
            # ★ 매매종료 시각 비교축 — 골든타임(10:30)만 vs 오전연장(11:30) vs 오후추격(13:00).
            #   "오전장만 좋다" 가설을 데이터로 검증(오후 추격이 PF 를 갉아먹는지).
            "time_end_hm": _parse_csv_ints(
                "BREAKOUT_GRID_FAST_TIME_END_HM", [1030, 1130, 1300],
            ),
            # ── 래칫(단계식 트레일) 축 ─────────────────────────────────────
            #   ""(OFF·단일 어깨컷, 기준선) vs 2단/3단 티어. fast 는 cap 균등샘플이라
            #   조합이 늘어도 max-combos 안에서 OFF/ON 이 골고루 섞인다.
            #   ※ 래칫 ON 이면 엔진에서 shoulder_min_high/cut 은 무시(else 분기)된다.
            "ratchet_tiers": _parse_csv_tiers(
                "BREAKOUT_GRID_FAST_RATCHET_TIERS",
                ["", "2:1.5,5:1.0", "3:1.5,6:1.0,10:0.7"],
            ),
            # ── ATR 동적 손절 비교축 ─────────────────────────────────────
            #   sl_mode='fixed'(기존 고정 sl_pct%) vs 'atr'(변동성 비례). atr 일 때만
            #   atr_sl_mult 적용(손절%=ATR×배수/진입가, atr_sl_min/max_pct 캡).
            #   fixed 조합에선 atr_sl_mult 가 무의미(중복) → cap 균등샘플이 흡수.
            "sl_mode": _parse_csv_strs(
                "BREAKOUT_GRID_FAST_SL_MODE", ["fixed", "atr"],
            ),
            "atr_sl_mult": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_ATR_SL_MULT", [1.5, 2.0, 2.5],
            ),
            # ── 호가필터(2단계 조인트) 축 — 기본 단일값=동작 불변 ──────────
            #   max_spread_pct      : 스프레드 상한(%)              ↑완화
            #   min_bid_ask_ratio   : 매수/매도 잔량비 하한          ↓완화
            #   ask_wall_max_qty    : 돌파 저항 근처 매도벽 수량 상한  ↑완화 (돌파 전용)
            "max_spread_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_MAX_SPREAD_PCT", [0.45],
            ),
            "min_bid_ask_ratio": _parse_csv_floats(
                "BREAKOUT_GRID_FAST_MIN_BID_ASK_RATIO", [0.85],
            ),
            "ask_wall_max_qty": _parse_csv_ints(
                "BREAKOUT_GRID_FAST_ASK_WALL_MAX_QTY", [5000],
            ),
        },
        # 5×4×4×5×4×3×3 ≈ 14,400 (무효 prev_chg 조합 제외)
        "coarse": {
            # 급등주(당일시가 대비) — fast 와 동일 축. env BREAKOUT_GRID_COARSE_MAX_DAILY_CHG
            "max_daily_chg": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_MAX_DAILY_CHG", [15.0, 20.0, 25.0, 30.0],
            ),
            # 거래량 배수 — HTS 300%(3.0)보다 코드에서 2~5배 탐색 (핵심 축)
            "vol_mult": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_VOL_MULT", [1.5, 2.0, 2.5, 3.0, 3.5, 4.0],
            ),
            "prev_chg_min": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_PREV_CHG_MIN", [0.5, 1.0, 1.5, 2.0],
            ),
            "prev_chg_max": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_PREV_CHG_MAX", [5.0, 8.0, 10.0, 15.0],
            ),
            "tp_pct": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_TP_PCT", [3.0, 5.0, 7.0, 10.0, 15.0],
            ),
            "sl_pct": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_SL_PCT", [1.5, 2.0, 3.0, 4.0],
            ),
            "shoulder_min_high_pct": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_SHOULDER_SMIN", [1.0, 2.0, 3.0],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_SHOULDER_CUT", [0.5, 0.8, 1.2],
            ),
            "lookback_min": _parse_csv_ints(
                "BREAKOUT_GRID_COARSE_LOOKBACK_MIN", [1, 10],
            ),
            # 가짜돌파(휩쏘) 필터
            "confirm_margin_pct": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_CONFIRM_MARGIN", [0.0, 0.3],
            ),
            "body_min_pct": _parse_csv_floats(
                "BREAKOUT_GRID_COARSE_BODY_MIN", [0.0, 0.3],
            ),
            # 10:30 골든타임 vs 11:30·13:00 (오후 추격 리스크)
            "time_end_hm": _parse_csv_ints(
                "BREAKOUT_GRID_COARSE_TIME_END_HM", [1030],
            ),
        },
        "fine": {
            "vol_mult": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_VOL_MULT", [1.0, 1.5, 2.0, 2.5, 3.0, 4.0],
            ),
            "prev_chg_min": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_PREV_CHG_MIN", [0.3, 0.5, 1.0, 1.5, 2.5],
            ),
            "prev_chg_max": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_PREV_CHG_MAX", [4.0, 6.0, 8.0, 12.0, 15.0],
            ),
            "tp_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_TP_PCT", [2.0, 3.0, 5.0, 7.0, 10.0],
            ),
            "sl_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_SL_PCT", [1.0, 1.5, 2.0, 3.0],
            ),
            "trail_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_TRAIL_PCT", [1.0, 1.5, 2.0],
            ),
            "shoulder_min_high_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_SHOULDER_SMIN", [1.0, 2.0, 3.0, 5.0],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_SHOULDER_CUT", [0.5, 0.8, 1.2, 1.5],
            ),
            # 래칫(단계식 트레일) 후보 — ""(OFF·단일 어깨컷) vs 2단/3단 티어.
            # 래칫 ON 이면 shoulder_* 축은 무시되어 중복 조합이 생기지만 cap 샘플로 흡수됨.
            "ratchet_tiers": _parse_csv_tiers(
                "BREAKOUT_GRID_FINE_RATCHET_TIERS",
                ["", "2:1.5,5:1.0", "3:1.5,6:1.0,10:0.7"],
            ),
            "lookback_min": _parse_csv_ints(
                "BREAKOUT_GRID_FINE_LOOKBACK_MIN", [1, 5, 10, 20],
            ),
            # 가짜돌파(휩쏘) 필터
            "confirm_margin_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_CONFIRM_MARGIN", [0.0, 0.2, 0.5],
            ),
            "body_min_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FINE_BODY_MIN", [0.0, 0.3],
            ),
            # 시간컷(분) — 0=OFF · 옆으로 기는 실패 돌파 정리
            "max_hold_bars": _parse_csv_ints(
                "BREAKOUT_GRID_FINE_MAX_HOLD_BARS", [0, 60],
            ),
            "time_end_hm": _parse_csv_ints(
                "BREAKOUT_GRID_FINE_TIME_END_HM", [1030, 1130, 1300],
            ),
        },
        "full": {
            "vol_mult": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_VOL_MULT", [1.0, 1.5, 2.0, 2.5, 3.0, 4.0, 5.0],
            ),
            "prev_chg_min": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_PREV_CHG_MIN", [0.3, 0.5, 1.0, 1.5, 2.0, 3.0],
            ),
            "prev_chg_max": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_PREV_CHG_MAX", [4.0, 6.0, 8.0, 10.0, 12.0, 15.0, 20.0],
            ),
            "tp_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_TP_PCT", [2.0, 3.0, 5.0, 7.0, 10.0, 15.0, 20.0],
            ),
            "sl_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_SL_PCT", [1.0, 1.5, 2.0, 3.0, 5.0],
            ),
            "trail_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_TRAIL_PCT", [1.0, 1.5, 2.0, 3.0],
            ),
            "shoulder_min_high_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_SHOULDER_SMIN", [1.0, 2.0, 3.0, 5.0],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_SHOULDER_CUT", [0.5, 0.8, 1.2, 1.5],
            ),
            "max_hold_bars": _parse_csv_ints(
                "BREAKOUT_GRID_FULL_MAX_HOLD_BARS", [0, 30, 60],
            ),
            "lookback_min": _parse_csv_ints(
                "BREAKOUT_GRID_FULL_LOOKBACK_MIN", [1, 5, 10, 20],
            ),
            "vol_window": _parse_csv_ints(
                "BREAKOUT_GRID_FULL_VOL_WINDOW", [1, 3, 5],
            ),
            # 가짜돌파(휩쏘) 필터
            "confirm_margin_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_CONFIRM_MARGIN", [0.0, 0.2, 0.5],
            ),
            "body_min_pct": _parse_csv_floats(
                "BREAKOUT_GRID_FULL_BODY_MIN", [0.0, 0.3, 0.5],
            ),
            "time_end_hm": _parse_csv_ints(
                "BREAKOUT_GRID_FULL_TIME_END_HM", [1030, 1130, 1300, 1500],
            ),
        },
    }


# ──────────────────────────────────────────────────────────────────────────────
# DB 적용
# ──────────────────────────────────────────────────────────────────────────────

def _get_breakout_field_map() -> Dict[str, Tuple[str, Any]]:
    return {
        # 진입 룰
        "lookback_min":   ("BREAKOUT_LOOKBACK_MIN",  lambda v: str(int(float(v)))),
        "vol_window":     ("BREAKOUT_VOL_WIN",       lambda v: str(int(float(v)))),
        "vol_mult":       ("BREAKOUT_VOL_MULT",      lambda v: str(float(v))),
        "min_turnover_1m_pct": ("BREAKOUT_MIN_TURNOVER_1M_PCT", lambda v: str(float(v))),
        "prev_chg_min":   ("BREAKOUT_PREV_CHG_MIN",  lambda v: str(float(v))),
        "prev_chg_max":   ("BREAKOUT_PREV_CHG_MAX",  lambda v: str(float(v))),
        # 청산 룰 (% → 비율)
        "sl_pct":         ("BREAKOUT_STOP_LOSS_PCT",   lambda v: str(float(v) / 100)),
        "tp_pct":         ("BREAKOUT_TAKE_PROFIT_PCT", lambda v: str(float(v) / 100)),
        # ATR 동적 손절 (sl_min/max 는 % 단위 그대로 저장 — 엔진이 /100 처리)
        "sl_mode":        ("BREAKOUT_SL_MODE",        lambda v: str(v).strip().lower()),
        "atr_period":     ("BREAKOUT_ATR_PERIOD",     lambda v: str(int(float(v)))),
        "atr_sl_mult":    ("BREAKOUT_ATR_SL_MULT",    lambda v: str(float(v))),
        "atr_sl_min_pct": ("BREAKOUT_ATR_SL_MIN_PCT", lambda v: str(float(v))),
        "atr_sl_max_pct": ("BREAKOUT_ATR_SL_MAX_PCT", lambda v: str(float(v))),
        "trail_pct":      ("BREAKOUT_TRAIL_PCT",       lambda v: str(float(v) / 100)),
        "trail_arm_pct":  ("BREAKOUT_TRAIL_ARM_PCT",   lambda v: str(float(v) / 100)),
        "shoulder_min_high_pct": ("BREAKOUT_SHOULDER_MIN_HIGH_PCT", lambda v: str(float(v) / 100)),
        "shoulder_cut_pct": ("BREAKOUT_SHOULDER_CUT_PCT", lambda v: str(float(v) / 100)),
        "max_hold_bars":  ("BREAKOUT_MAX_HOLD_BARS",  lambda v: str(int(float(v)))),
        # 시간/방어
        "time_start_hm":  ("BREAKOUT_TIME_START",      lambda v: str(int(float(v)))),
        "time_end_hm":    ("BREAKOUT_GOLDEN_END_HM",   lambda v: "%d:%02d" % (
            int(float(v)) // 100, int(float(v)) % 100,
        )),
        "max_daily_chg":  ("BREAKOUT_MAX_DAILY_CHG", lambda v: str(float(v))),
        "min_price":      ("BREAKOUT_MIN_PRICE",     lambda v: str(int(float(v)))),
        # 가짜돌파(휩쏘) 필터 — % 그대로 저장 (_eval 가 % 로 읽음)
        "confirm_margin_pct": ("BREAKOUT_CONFIRM_MARGIN_PCT", lambda v: str(float(v))),
        "body_min_pct":   ("BREAKOUT_BODY_MIN_PCT",   lambda v: str(float(v))),
        "max_loss_krw":   ("BREAKOUT_MAX_LOSS_PER_TRADE_KRW", lambda v: str(int(float(v)))),
        "slot_money":     ("BREAKOUT_SLOT_MONEY",    lambda v: str(int(float(v)))),
        "entry_mode":     ("BREAKOUT_ENTRY_MODE",    lambda v: str(v).strip().lower()),
        "intrabar_slippage_pct": ("BREAKOUT_INTRABAR_SLIPPAGE_PCT", lambda v: str(float(v))),
    }


def _apply_to_db(best_params: dict) -> None:
    """1위 파라미터 → env_config INSERT (최신 행 복사 후 BREAKOUT_* 패치).

    ※ 구버전 UPDATE 방식은 DB에 없는 컬럼명으로 실패함(1054).
    ``param_search_apply_snapshot._patch_from_breakout_merged`` 와 동일 경로.
    """
    from kis_trader.backtest.param_search_apply_snapshot import (  # noqa: WPS433
        _patch_from_breakout_merged,
    )

    patch = _patch_from_breakout_merged(best_params)
    patch.update(portfolio_env_patch("BREAKOUT", best_params))
    patch.update(session_env_patch("BREAKOUT", best_params))
    if not patch:
        print("DB 적용할 파라미터가 없습니다.")
        return

    db = TradeDB()
    try:
        latest = db.get_latest_env()
        snap = dict(latest["snapshot"]) if latest else {}
        snap.update(patch)
        eid = db.insert_env_snapshot(snap)
        if not eid:
            print("❌ env_config INSERT 실패 (컬럼 마이그레이션 후 재시도)")
            return
    finally:
        db.close()

    print("\n✅ DB env_config INSERT 완료 (BREAKOUT_* 키, id=%s):" % eid)
    for db_col, val in sorted(patch.items()):
        print(f"   {db_col:<35s} = {val}")


def apply_params_to_db(best_params: dict) -> None:
    """웹·CLI 공통 — 파라서치 merged → DB."""
    _apply_to_db(best_params)


# ──────────────────────────────────────────────────────────────────────────────
# UI(%) → 엔진(비율) 변환
# ──────────────────────────────────────────────────────────────────────────────

def _ui_to_engine_params(ui_params: dict) -> dict:
    """``breakout_ui_to_engine_params`` 와 동일 (웹 백테·파서치 공용)."""
    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],
    keys: List[str],
    codes_candles: Optional[Dict[str, List[Dict]]],
    min_trades: int,
    min_win_rate: float,
    min_pf: float,
    top_n: int,
    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,
) -> List[Tuple[float, float, int, Dict]]:
    """워커: 청크 내 조합 평가 — 시각순 포트폴리오 (breakout_backtest_common)."""
    shared = worker_shared_get()
    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 {}
        if universe_by_slot is None:
            universe_by_slot = shared.get("universe_by_slot")
        ticks_preloaded = shared.get("ticks_by_code")
        # ws_ticks 공유메모리(opt-in): descriptor 로 read-only attach (워커당 1회 재사용).
        if not ticks_preloaded:
            _desc = shared.get("ticks_shared_descriptor")
            if _desc:
                _tm = shared.get("_ticks_mapping_cache")
                if _tm is None:
                    from kis_trader.backtest.shared_ticks import SharedTicksMapping
                    _tm = SharedTicksMapping(_desc)
                    shared["_ticks_mapping_cache"] = _tm
                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] = {}
    attach_indicator_caches_to_params(cache_holder, codes_candles)
    local_heap: List[Tuple[float, float, int, Dict]] = []
    for combo in param_chunk:
        assert_parent_alive()
        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,
            log_verdict_by_code=log_verdict_preloaded,
            share_denom_by_code=share_denom_preloaded,
        )
        if result_pkg is None:
            continue

        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)
        elif total_pnl > local_heap[0][0]:
            heapq.heapreplace(local_heap, item_t)
    return local_heap


# ──────────────────────────────────────────────────────────────────────────────
# 캔들 로드
# ──────────────────────────────────────────────────────────────────────────────

def _load_candles_for_search(
    start: str,
    end: str,
    lookback_min: int = 1,
    fixed: Optional[Dict[str, Any]] = None,
) -> dict:
    db = TradeDB()
    codes_candles: Dict[str, List[Dict]] = {}
    try:
        start_key = (start.replace("-", "") + "0000") if start else "20260101"
        end_key   = (end.replace("-", "") + "2359")   if end   else "99991231"

        codes_raw = db.conn.execute(
            "SELECT DISTINCT code FROM ws_candles WHERE timeframe=1 "
            "AND candle_time >= %s AND candle_time <= %s ORDER BY code",
            [start_key, end_key]
        ).fetchall()
        codes = [r["code"] for r in codes_raw]

        probe = dict(fixed or {})
        probe.setdefault("lookback_min", lookback_min)
        probe.setdefault("vol_window", 7)
        need_n = breakout_min_bars_required(probe)
        for code in codes:
            rows = db.conn.execute(
                "SELECT candle_time, open, high, low, close, volume "
                "FROM ws_candles "
                "WHERE timeframe=1 AND code=%s "
                "AND candle_time >= %s AND candle_time <= %s "
                "AND is_confirmed=1 "
                "ORDER BY candle_time ASC",
                [code, start_key, end_key]
            ).fetchall()
            if len(rows) < need_n:
                continue
            codes_candles[code] = [dict(r) for r in rows]
    finally:
        db.close()
    return codes_candles


# ──────────────────────────────────────────────────────────────────────────────
# 메인 탐색 루틴
# ──────────────────────────────────────────────────────────────────────────────

def run_search(
    start: str, end: str, mode: str, top_n: int,
    min_trades: int, min_win_rate: float, min_pf: float,
    apply_rank: Optional[int],
    from_file_only: bool = False,
    use_fallback_universe: bool = False,
    slot_money: Optional[float] = None,
    max_stocks: Optional[int] = None,
    total_budget_krw: Optional[float] = None,
    time_start_hm: Optional[int] = None,
    time_end_hm: Optional[int] = None,
    max_combos: Optional[int] = None,
    orderbook_filter: str = "off",
) -> bool:
    if from_file_only and apply_rank is not None and apply_rank >= 1:
        _apply_from_latest_json(apply_rank)
        return True

    grid = _breakout_grids()[mode]
    keys = list(grid.keys())
    axes = [grid[k] for k in keys]
    # 데카르트곱 전체를 RAM 에 펼치지 않는다 (호가 스윕 시 수천만 → OOM 방지).
    dict_combos, total_grid, max_combos_cap, dropped_by_cap = cap_combos_uniform_lazy(
        keys,
        axes,
        mode,
        strategy_env_prefix="BREAKOUT",
        default_fast=200,
        max_combos_override=max_combos,
    )
    total = len(dict_combos)
    print(f"\n[BREAKOUT {mode.upper()} 모드] 그리드: {total_grid:,} → 백테: {total:,}  |  기간: {start} ~ {end}")
    if mode == "fast":
        print(f"📌 [fast] {total_grid:,}→{max_combos_cap}균등샘플 · 돌파 TRIGGER 축")
    if dropped_by_cap:
        print(f"   (max-combos={max_combos_cap} 균등 샘플, 제외 {dropped_by_cap:,}개)")
    print(f"📌 1위 정렬 기준: 총손익 최대")
    print(f"📌 필터: 승률≥{min_win_rate}% · PF≥{min_pf} · 거래≥{min_trades}건")
    print("📌 청산순: EOD → 익절 → 어깨 → 손절 → 트레일 (돌파 전용)")
    _em = breakout_entry_mode()
    print(f"📌 진입 모드: {_em} (intrabar=B·HTS E / align=A·다음봉 시가)")
    print("=" * 70)

    FIXED_DEFAULTS = _bo_fixed_defaults()
    apply_session_to_fixed(
        FIXED_DEFAULTS,
        time_start_hm=time_start_hm,
        time_end_hm=time_end_hm,
    )

    # ── 호가필터 ON/OFF (2단계 워크플로) ───────────────────────────────
    # off(기본): 1단계 — 호가 게이트 없이 코어 파라미터만 순수 탐색.
    # on       : 2단계 — kiwoom_0d 본체로 스프레드·잔량비·돌파매도벽 실제 적용.
    # auto     : env/DB 의 *_ORDERBOOK_FILTER_ENABLED 값을 그대로 따름.
    _ob_mode = (orderbook_filter or "off").strip().lower()
    if _ob_mode == "off":
        FIXED_DEFAULTS["_orderbook_filter_enabled"] = False
    elif _ob_mode == "on":
        FIXED_DEFAULTS["_orderbook_filter_enabled"] = True
    _ob_filter_on = bool(FIXED_DEFAULTS.get("_orderbook_filter_enabled")) or _ob_mode == "auto"
    print(
        f"📌 호가필터: {_ob_mode.upper()} "
        f"({'적용' 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_v = float(portfolio["slot_money"])
    max_stocks_v = int(portfolio["max_stocks"])
    total_budget_v = float(portfolio["total_budget_krw"])
    period_days = max(
        1,
        (datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days + 1,
    )
    print(
        f"💼 포트폴리오: 1회 {slot_money_v:,.0f}원 | 동시 {max_stocks_v}종 | "
        f"총한도 {total_budget_v:,.0f}원 | 매매 {format_session_hm(FIXED_DEFAULTS)}"
    )
    if portfolio.get("budget_warning"):
        print(f"💰 {portfolio['budget_warning']}")

    print("⏳ DB에서 캔들 데이터를 메모리로 불러오는 중...")
    codes_candles = _load_candles_for_search(
        start, end, FIXED_DEFAULTS.get("lookback_min", 1), FIXED_DEFAULTS,
    )
    print(f"✅ 데이터 로드 완료: {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())
        n_ok = sum(1 for v in share_denom_by_code.values() if v > 0)
        print(f"✅ stock_share_meta: {n_ok}/{len(codes_candles)}종목 유통주식수 로드")
    finally:
        _share_db.close()

    ticks_by_code: Dict[str, Dict[str, List[Dict]]] = {}
    tick_backtest_meta: Dict[str, Any] = {}
    engine_probe = _ui_to_engine_params(FIXED_DEFAULTS)
    engine_probe["_orderbook_filter_enabled"] = FIXED_DEFAULTS.get("_orderbook_filter_enabled")
    if breakout_backtest_wants_tick_replay(engine_probe):
        start_key = (start.replace("-", "") + "0000") if start else "20260101"
        end_key = (end.replace("-", "") + "2359") if end else "99991231"
        _tick_db = TradeDB()
        try:
            ticks_by_code, tick_rows = load_breakout_ticks_by_code(
                _tick_db, start_key, end_key, set(codes_candles.keys()),
            )
            tick_backtest_meta = tick_coverage_stats(codes_candles, ticks_by_code)
            tick_backtest_meta["ws_tick_rows_loaded"] = tick_rows
            cov = tick_backtest_meta.get("tick_bar_coverage_pct", 0)
            print(
                f"✅ ws_ticks {tick_rows:,}건 | 분봉 커버리지 {cov}% "
                f"({tick_backtest_meta.get('tick_codes_with_data', 0)}/"
                f"{tick_backtest_meta.get('tick_codes_total', 0)}종목)"
            )
            if tick_rows <= 0:
                print("⚠️  ws_ticks 없음 — B안은 1분봉 high 폴백 (WS_TICK_SAVE_ENABLED 후 재탐색)")
        finally:
            _tick_db.close()

    orderbook_by_code: Dict[str, Dict[str, List[Dict]]] = {}
    program_by_code: Dict[str, Dict[str, List[Dict]]] = {}
    log_verdict_by_code: Dict[str, Dict[str, List[Dict]]] = {}
    trigger_snap_meta: Dict[str, Any] = {}
    # kiwoom_0d 본체 재계산 — 필터 ON + 호가필터 축 **실제 스윕**(값 2개↑) 일 때만.
    _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:
        engine_probe["backtest_use_kiwoom_body_snapshot"] = True
        engine_probe["_backtest_use_kiwoom_body"] = True
        _ob_axis_vals = {k: grid.get(k) for k in _ob_axes if len(set(grid.get(k) or [])) > 1}
        print(
            f"📌 호가필터 스윕 활성 → kiwoom_0d 본체 재계산 "
            f"(축 {_ob_axis_vals}, 본체 없는 날짜는 log_backfill 폴백)"
        )
    elif _ob_filter_on:
        print("📌 호가필터 ON · 스윕 없음 → 본체 재계산 생략(판정 재생 경로)")
    if codes_candles:
        start_key = (start.replace("-", "") + "0000") if start else "20260101"
        end_key = (end.replace("-", "") + "2359") if end else "99991231"
        _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 {}
            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)
            lv_rows = int(trigger_snap_meta.get("log_verdict_rows") or 0)
            print(
                f"✅ TRIGGER 스냅샷 ws_orderbook {ob_rows:,}건 | ws_program {pg_rows:,}건 "
                f"| log_backfill 판정 {lv_rows:,}건 "
                f"(호가종목 {trigger_snap_meta.get('orderbook_codes_with_data', 0)} / "
                f"프로그램종목 {trigger_snap_meta.get('program_codes_with_data', 0)})"
            )
            if ob_rows <= 0 and pg_rows <= 0:
                print("⚠️  TRIGGER 스냅샷 없음 — 호가·프로그램 필터 스킵 (실매 수집 후 재탐색)")
        finally:
            _snap_db.close()

    # ── 유니버스 ──
    universe_by_slot = None
    fallback_sim_interval = 5
    start_ymd = start.replace("-", "") if start else ""
    end_ymd = end.replace("-", "") if end else ""
    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  # type: ignore
            _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)
                print(f"✅ 유니버스: 신봇 BREAKOUT 이력 사용 | {n_bins:,}분봉 · 평균 {avg:.1f}종목")
            else:
                print("ℹ️  BREAKOUT 이력 없음 → 시뮬레이션 fallback 자동 사용")
        except Exception as _e:
            logging.getLogger("param_search_breakout").debug(
                "신봇 유니버스 이력 조회 스킵: %s", _e,
            )

    if universe_by_slot is None:
        # 모멘텀과 동일하게 scalping_engine.build_universe_simulation 재사용
        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,
        )
        n_slots = len(universe_by_slot)
        avg_per_slot = sum(len(c) for c in universe_by_slot.values()) / max(1, n_slots)
        print(f"✅ 유니버스: 시뮬레이션 사용 | {fallback_sim_interval}분 슬롯 {n_slots}개 · "
              f"슬롯당 평균 {avg_per_slot:.1f}종목")

    # ── ws_ticks 공유메모리 — 워커별 사본 대신 1벌 공유 (momentum·tail 과 동일) ────────
    # 기본 ON(검증 완료). ON 이면 컬럼(numpy)로 shared_memory 에 1벌 올리고 워커는 read-only
    # attach → 메모리 N배 제거 + 워커 상한 해제. breakout 의 반복 핫루프(whipsaw)와 intrabar
    # 진입 경로 모두 TickColumnView 로 배열 직접 읽어 dict 경로와 bit-identical(E2E 667거래).
    # 끄려면 BREAKOUT_PARAM_SEARCH_SHARED_TICKS=0. numpy/shm 미지원·빌드 실패 시 자동 폴백.
    shared_tick_store = None
    if get_env_bool("BREAKOUT_PARAM_SEARCH_SHARED_TICKS", True) and ticks_by_code:
        from kis_trader.backtest.shared_ticks import (
            build_shared_ticks,
            shared_ticks_available,
        )
        if shared_ticks_available():
            shared_tick_store = build_shared_ticks(ticks_by_code)
            if shared_tick_store is not None:
                import atexit as _atexit
                _atexit.register(shared_tick_store.unlink)
                print("📦 ws_ticks 공유메모리 ON — 워커 attach(read-only), 사본 제거")
                ticks_by_code = {}
                import gc as _gc
                _gc.collect()
                try:
                    import ctypes as _ctypes
                    _ctypes.CDLL("libc.so.6").malloc_trim(0)
                except Exception:
                    pass
            else:
                print("⚠️  ws_ticks 공유메모리 build 실패 — 기존 경로 폴백")
        else:
            print("⚠️  numpy/shared_memory 미지원 — 기존 경로 폴백")

    shared = ParamSearchSharedPayload({
        "codes_candles": codes_candles,
        "universe_by_slot": universe_by_slot,
        "ticks_by_code": ticks_by_code,
        "ticks_shared_descriptor": (shared_tick_store.descriptor() if shared_tick_store else None),
        "tick_backtest_meta": tick_backtest_meta,
        "orderbook_by_code": orderbook_by_code,
        "program_by_code": program_by_code,
        "log_verdict_by_code": log_verdict_by_code,
        "trigger_snapshot_meta": trigger_snap_meta,
        "share_denom_by_code": share_denom_by_code,
    })
    payload_bytes = shared.estimate_bytes()
    n_cpu = os.cpu_count() or 4
    _cpu_frac = get_env_float("PARAM_SEARCH_CPU_FRAC", 0.8)
    max_workers, chunk_size, _ = param_search_chunk_plan(total, payload_bytes)
    # ── 틱재생(ws_ticks) payload 시 워커 상한 (OOM 방지) ─────────────────
    # 틱재생 기본 ON 이후, 워커별로 ws_ticks 등 사적 구조가 payload 추정치보다
    # 크게 부풀어(모멘텀에서 워커당 anon-RSS ~3.3GB 관측) 다수 워커 동시 실행 시
    # 저용량 머신이 OOM-kill 로 죽는다. 모멘텀·꼬리잡기와 동일 가드 적용.
    # (하드코딩 금지 — DB/Env 로 조정, 기본 2)
    if ticks_by_code:
        tick_cap = get_env_int("BREAKOUT_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS", 2)
        if tick_cap > 0 and max_workers > tick_cap:
            max_workers = tick_cap
            print(f"📌 ws_ticks payload — 워커 상한 {max_workers} (BREAKOUT_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS)")
    # 공유메모리 틱은 사본이 없어 OOM 위험이 없다 → 기본 상한 없음(CPU/메모리 계획대로).
    if shared_tick_store is not None:
        shared_cap = get_env_int("BREAKOUT_PARAM_SEARCH_MAX_WORKERS_WITH_SHARED_TICKS", 0)
        if shared_cap > 0 and max_workers > shared_cap:
            max_workers = shared_cap
            print(f"📌 ws_ticks 공유메모리 — 워커 상한 {max_workers} (BREAKOUT_PARAM_SEARCH_MAX_WORKERS_WITH_SHARED_TICKS)")
    chunks = [dict_combos[i:i + chunk_size] for i in range(0, len(dict_combos), chunk_size)]
    print(param_search_worker_budget_line(payload_bytes))
    print(f"⚙️  멀티프로세싱 시작 (코어: {n_cpu}, 워커: {max_workers}, CPU {_cpu_frac*100:.0f}%) | 청크: {len(chunks):,}개 (청크당 ~{chunk_size}조합)")

    start_time = time.time()
    global_heap: List[Tuple[float, float, Tuple[int, int], Dict]] = []
    progress_eta = ParamSearchProgressETA(len(chunks), max_workers)
    with managed_process_pool(max_workers, shared_payload=shared) as executor:
        def _submit(chunk: List[Dict[str, Any]]):
            return executor.submit(
                _evaluate_breakout_chunk, chunk, FIXED_DEFAULTS, keys, None,
                min_trades, min_win_rate, min_pf, top_n, None,
                slot_money_v, max_stocks_v, total_budget_v, fee_rate, sell_tax, period_days,
            )

        processed = 0
        use_carriage_return = sys.stdout.isatty()
        for local_results in iter_pool_chunk_results(
            executor, chunks, _submit, max_workers=max_workers,
        ):
            processed += 1
            for idx, item in enumerate(local_results):
                pnl, wr, _, result_pkg = item
                tie = (processed, idx)
                entry = (pnl, wr, tie, result_pkg)
                if len(global_heap) < top_n:
                    heapq.heappush(global_heap, entry)
                elif pnl > global_heap[0][0]:
                    heapq.heapreplace(global_heap, entry)

            progress = (processed / len(chunks)) * 100
            elapsed_so_far = time.time() - start_time
            eta_str = ParamSearchProgressETA.format_sec(
                progress_eta.remaining_sec(processed, elapsed_so_far),
            )
            elapsed_str = ParamSearchProgressETA.format_elapsed(elapsed_so_far)
            line = (f"⏳ 진행률: {progress:.1f}% ({processed:,}/{len(chunks):,}) | "
                    f"경과 {elapsed_str} | 남은 {eta_str}")
            if use_carriage_return:
                print(f"\r{line}", end="", flush=True)
            else:
                print(line, flush=True)
        if use_carriage_return:
            print(flush=True)

    # 워커 종료 후 공유메모리 즉시 해제 (atexit 는 크래시 대비 이중 안전장치).
    if shared_tick_store is not None:
        shared_tick_store.unlink()
        shared_tick_store = None

    elapsed = time.time() - start_time
    if not global_heap:
        print("\n⚠️  조건을 만족하는 조합이 없습니다. (--min_trades 를 낮추거나 기간을 늘려보세요.)")
        return False

    results = [heapq.heappop(global_heap)[3] for _ in range(len(global_heap))]
    results.sort(key=lambda r: (-r["total_pnl"], -r["win_rate"]))

    print(f"\n완료: {elapsed:.1f}초  |  유효 결과: {len(results):,}건")
    print(f"\n{'='*100}")
    print(f" 🏆  BREAKOUT TOP {min(top_n, len(results))} (총손익 기준)")
    print(f"{'='*100}")
    hdr_keys = list(keys)
    col_w = max(len(k) for k in hdr_keys) + 2
    hdr = "  ".join(f"{k:>{col_w}}" for k in hdr_keys)
    print(f"{hdr}  | {'손익(원)':>12}  {'승률':>6}  {'거래':>5}  {'PF':>5}")
    print("-" * (len(hdr) + 50))
    for r in results[:top_n]:
        p = r["params"]
        row = "  ".join(
            (f"{str(p[k]):>{col_w}}" if isinstance(p[k], (bool, str)) else f"{p[k]:>{col_w}.4g}")
            for k in hdr_keys
        )
        print(f"{row}  | {r['total_pnl']:>+12,.0f}  {r['win_rate']:>5.1f}%  "
              f"{r['total_trades']:>5}  {r['pf']:>5.2f}")

    trade_counts = sorted({r["total_trades"] for r in results})
    print(f"\n📊 거래수 다양성: {len(trade_counts)}가지 → {trade_counts[:20]}{'...' if len(trade_counts) > 20 else ''}")

    if results:
        r0 = results[0]
        reasons = r0.get("sell_reasons") or {}
        if reasons:
            reason_line = ", ".join(f"{k}={v}" for k, v in sorted(reasons.items(), key=lambda x: -x[1]))
            print(f"🏅 1위 청산분포: {reason_line} | 건당평균수익률 {r0.get('avg_profit_rate', 0)}%")

    out_dir = _results_dir_for_write()
    ts      = datetime.now().strftime("%Y%m%d_%H%M%S")
    out_path = os.path.join(out_dir, f"search_breakout_{mode}_{ts}.json")
    payload = {
        "strategy": "BREAKOUT",
        "mode": mode,
        "start": start,
        "end": end,
        "grid_combos": total_grid,
        "max_combos_cap": max_combos_cap,
        "min_win_rate": min_win_rate,
        "min_pf": min_pf,
        "exit_priority": "eod → take_profit → shoulder_cut → stop_loss → trailing",
        "top": [
            {
                "rank": i + 1,
                "params": r["params"],
                "merged_params": r.get("merged_params", r["params"]),
                "total_pnl": r["total_pnl"],
                "win_rate": r["win_rate"],
                "total_trades": r["total_trades"],
                "pf": r["pf"],
                "avg_hold": r["avg_hold"],
                "mdd": r["mdd"],
                "bot_pct": r.get("bot_pct"),
                "daily_avg_pct": r.get("daily_avg_pct"),
                "avg_profit_rate": r.get("avg_profit_rate"),
                "sell_reasons": r.get("sell_reasons"),
                "skipped_micro_buys": r.get("skipped_micro_buys", 0),
            }
            for i, r in enumerate(results)
        ],
    }
    payload.update(search_json_meta(portfolio, FIXED_DEFAULTS))
    try:
        with open(out_path, "w", encoding="utf-8") as f:
            json.dump(payload, f, ensure_ascii=False, indent=2)
        print(f"\n💾 결과 저장: {out_path}")
    except (PermissionError, OSError) as _e:
        fallback_dir = os.path.join(os.path.expanduser("~"), ".kis_bot_search_results")
        os.makedirs(fallback_dir, exist_ok=True)
        out_path = os.path.join(fallback_dir, f"search_breakout_{mode}_{ts}.json")
        with open(out_path, "w", encoding="utf-8") as f:
            json.dump(payload, f, ensure_ascii=False, indent=2)
        print(f"\n⚠️  기본 경로 쓰기 실패({type(_e).__name__}). 폴백 저장: {out_path}")

    if apply_rank is not None and 1 <= apply_rank <= len(results):
        cand = results[apply_rank - 1]
        if cand.get("total_pnl", 0) <= 0:
            print(f"⚠️ {apply_rank}번째 결과 총손익 ≤ 0 → DB 미적용. 기존 설정 유지.")
        else:
            merged_apply = merge_param_search_apply_source(cand, payload)
            _apply_to_db(merged_apply)
            print(f"✅ {apply_rank}번째 결과 적용 완료")

    return True


# ──────────────────────────────────────────────────────────────────────────────
# CLI 진입점
# ──────────────────────────────────────────────────────────────────────────────

def main():
    today    = datetime.now().strftime("%Y-%m-%d")
    week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")

    parser = argparse.ArgumentParser(description="돌파매매 백테스트 파라미터 Grid Search (BreakoutStrategy + run_breakout_backtest)")
    parser.add_argument("--start",    default=week_ago, help="시작일 (YYYY-MM-DD)")
    parser.add_argument("--end",      default=today,    help="종료일 (YYYY-MM-DD)")
    parser.add_argument("--mode",     default="fast", choices=["fast", "coarse", "fine", "full"],
                        help="탐색 모드: fast(그리드→균등200·~10–15분) / coarse / fine / full")
    parser.add_argument(
        "--max-combos", type=int, default=None, dest="max_combos",
        help="백테 조합 상한 (fast 기본 env BREAKOUT_FAST_MAX_COMBOS 또는 PARAM_SEARCH_FAST_MAX_COMBOS=200, 0=무제한)",
    )
    parser.add_argument("--top",         default=1000, type=int)
    parser.add_argument("--min_trades",  default=1,    type=int)
    add_search_filter_cli_args(parser)
    parser.add_argument("--apply",       nargs="?", const=1, type=int, default=None, metavar="N",
                        help="N번째 결과 DB 적용 (기본 1, 총손익>0 일 때만). --from-file 시 최근 JSON에서 적용")
    parser.add_argument("--from-file",   action="store_true",
                        help="--apply N 과 함께: 탐색 생략, 최근 search_breakout_*.json 에서만 DB 적용")
    parser.add_argument("--fallback-universe", action="store_true", dest="fallback_universe")
    parser.add_argument(
        "--orderbook-filter", default="off", choices=["off", "on", "auto"],
        dest="orderbook_filter",
        help="호가필터: off=1단계(코어만·기본) / on=2단계(kiwoom 본체 스프레드·잔량비·돌파매도벽) / auto=DB값",
    )
    add_portfolio_cli_args(parser)
    args = parser.parse_args()

    def _sigterm_to_kbd(_sig, _frm):
        raise KeyboardInterrupt("SIGTERM 수신 → 워커 정리 후 종료")
    try:
        signal.signal(signal.SIGTERM, _sigterm_to_kbd)
    except Exception:
        pass

    run_lock = None
    if not (args.from_file and args.apply is not None):
        run_lock = try_acquire_run_lock("param_search_breakout")
        if run_lock is None:
            print(
                "⛔ 이미 실행 중인 param_search_breakout 이 있습니다.\n"
                "   ps -ef | grep param_search_breakout\n"
                "   pkill -f 'param_search_breakout.py'  후 재실행하세요.",
                flush=True,
            )
            sys.exit(2)

    try:
        run_search(
            start          = args.start,
            end            = args.end,
            mode           = args.mode,
            top_n          = args.top,
            min_trades     = args.min_trades,
            min_win_rate   = args.min_win_rate,
            min_pf         = args.min_pf,
            apply_rank     = args.apply,
            from_file_only = args.from_file,
            use_fallback_universe = args.fallback_universe,
            slot_money      = args.slot_money,
            max_stocks      = args.max_stocks,
            total_budget_krw = args.total_budget,
            time_start_hm   = args.time_start,
            time_end_hm     = args.time_end,
            max_combos      = args.max_combos,
            orderbook_filter = args.orderbook_filter,
        )
    except KeyboardInterrupt as e:
        print(f"\n⛔ {e} — 미완료 결과 없이 종료합니다.", flush=True)
        sys.exit(130)
    finally:
        if run_lock is not None:
            run_lock.release()


if __name__ == "__main__":
    main()
