#!/usr/bin/env python3
"""
kis_trader/backtest/param_search_momentum.py — 모멘텀 백테스트 파라미터 자동 탐색 (Grid Search)
=======================================================================================================
[전략 = MOMENTUM — momentum_engine 전용, SCALP reversal 과 분리]

- SCAN: 키움/KIS ``scalp`` 조건검색 → ``target_candidates_history``
- TRIGGER: ``momentum_engine.check_buy_signal_momentum_live`` (실매 MomentumStrategy 동일)
- 청산: ``check_sell_signal_momentum_live`` — 어깨·트레일 선행, tp_max 상한 익절은 마지막
- 백테·파서치: ``momentum_backtest_common.run_momentum_backtest_web_aligned``
  → ``momentum_engine.run_momentum_backtest`` → ``check_sell_signal_momentum_backtest_bar``

[관련 CLI (전략별 파일 분리)]
    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
    통합 적용 CLI : kis_trader/backtest/param_search_apply_snapshot.py  (--json · --rank)

실행:
    cd /home/hoon/kis_bot
    python3 kis_trader/backtest/param_search_momentum.py --start 2026-05-11 --end 2026-05-27
    python3 -m kis_trader.backtest.param_search_momentum --mode fast --top 30

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

⚠️ 모멘텀 백테는 ``momentum_engine`` 매수 루프 사용 (SCALP reversal high_chase 등 미적용).
    방어필터는 MOMENTUM_USE_* / 시가·일변동 컷만 반영.

⚠️ 2026-05 이전 JSON 은 워커 힙 병합 버그로 순위 불일치 가능 — 동일 기간 재탐색 권장.
"""

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

# fast: V2 TRIGGER 축 → MOMENTUM_FAST_MAX_COMBOS(기본 200) 균등샘플 (~10–15분·이력유니버스 기준)
DEFAULT_MAX_COMBOS = 5000  # coarse/fine 상한 (env MOMENTUM_PARAM_SEARCH_MAX_COMBOS)


# kis_bot 루트를 경로에 추가 (스크립트·패키지 실행 모두 대응)
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.engine import momentum_engine as me
from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params
from kis_trader.backtest import momentum_backtest_common as mbc
from kis_trader.backtest import scalping_backtest_common as sbc
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,
    grid_total_combinations,
    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.strategies.breakout import normalize_breakout_max_loss_krw  # noqa: E402
from kis_trader.utils.env import (  # noqa: E402 — DB/os 폴백, 하드코딩 금지 원칙
    get_env_bool,
    get_env_float,
    get_env_from_db,
    get_env_int,
)


def _parse_csv_floats(env_key: str, fallback: List[float]) -> List[float]:
    """env 에 콤마 구분 목록이 있으면 사용 (예: `1.0,1.5,2.0`). 없으면 fallback."""
    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)


# ──────────────────────────────────────────────────────────────────────────────
# 모멘텀 기본값 (DB 우선, 없으면 코드 default)
# ──────────────────────────────────────────────────────────────────────────────

def _rr_search_base_from_env() -> Dict[str, float]:
    """``MOMENTUM_RR_SEARCH_BASE_JSON`` — R:R 스윕 1위 청산축 (UI % 단위)."""
    import json as _json

    raw = str(os.environ.get("MOMENTUM_RR_SEARCH_BASE_JSON") or "").strip()
    if not raw:
        raw = str(get_env_from_db("MOMENTUM_RR_SEARCH_BASE_JSON", "") or "").strip()
    if not raw:
        return {}
    try:
        data = _json.loads(raw)
        if isinstance(data, dict):
            return {k: float(v) for k, v in data.items()}
    except (TypeError, ValueError, _json.JSONDecodeError):
        pass
    return {}


def _mom_fixed_defaults() -> Dict[str, Any]:
    """그리드에 안 넣은 축만 채움 + 각 축의 폴백값.

    - 숫자 기본값은 전부 ``get_env_*`` / DB ``_pick`` 으로만 둠 (리터럴 하드코딩 금지).
    - 트레일·쿨다운·일일한도·방어·슬롯 등은 coarse/fine/full 그리드가 덮어씀 (폴백은 JSON 재현용).
    - ``MOMENTUM_RR_SEARCH_BASE_JSON`` 이 있으면 청산 축 폴백을 R:R 스윕 1위로 맞춤.
    """
    rr_base = _rr_search_base_from_env()
    _d = me.get_momentum_defaults_from_db()
    from kis_trader.utils.env import get_strategy_env_dict
    env = get_strategy_env_dict("MOMENTUM")

    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

    _fee_r = float(_d.get("fee_rate", 0.00015))
    _tax_r = float(_d.get("sell_tax", 0.0018))

    return {
        "rsi_period": get_env_int("MOMENTUM_SEARCH_RSI_PERIOD", int(_d.get("rsi_period", 3))),
        # 슬롯: 탐색 시 max_loss/sl 로 _ui_to_engine_params 가 재계산. 폴백만 env/DB.
        "slot_money": float(_pick(
            "MOMENTUM_SLOT_MONEY", "SLOT_MONEY_DEFAULT",
            default=get_env_int("MOMENTUM_SEARCH_SLOT_MONEY_KRW", int(_d.get("slot_money", 3_000_000))),
            cast=lambda v: float(v),
        )),
        "vol_mult": get_env_float("MOMENTUM_SEARCH_VOL_MULT", 0.0),
        # 아래는 그리드에 해당 키가 없을 때만 쓰는 폴백 (모든 모드에서 키가 있으면 조합값이 덮어씀)
        "shoulder_min_high": float(rr_base.get(
            "shoulder_min_high",
            float(_d.get("shoulder_min_high", 0.005)) * 100.0,
        )),
        "shoulder_cut_pct": float(rr_base.get(
            "shoulder_cut_pct",
            float(_d.get("shoulder_cut_pct", 0.003)) * 100.0,
        )),
        "tp_max_pct": float(rr_base.get(
            "tp_max_pct",
            float(_d.get("tp_max_pct", 0.02)) * 100.0,
        )),
        "sl_pct": float(rr_base.get("sl_pct", float(_d.get("sl_pct", 0.02)) * 100.0)),
        "tp_pct": float(rr_base.get("tp_pct", float(_d.get("tp_pct", 0.02)) * 100.0)),
        "trail_trigger": get_env_float("MOMENTUM_SEARCH_TRAIL_TRIGGER_PCT", 1.5),
        "trail_stop": get_env_float("MOMENTUM_SEARCH_TRAIL_STOP_PCT", 1.0),
        "cooldown_min": float(get_env_int("MOMENTUM_SEARCH_COOLDOWN_MIN", 1)),
        "time_start_hm": int(_pick(
            "MOMENTUM_TIME_START", "SCALP_TIME_START", "TIME_START",
            default=get_env_int("MOMENTUM_SEARCH_TIME_START_HM", int(_d.get("time_start_hm", 900))),
            cast=lambda v: int(float(v)),
        )),
        "time_end_hm": int(_pick(
            "MOMENTUM_TIME_END", "SCALP_TIME_END", "TIME_END",
            default=get_env_int("MOMENTUM_SEARCH_TIME_END_HM", int(_d.get("time_end_hm", 1530))),
            cast=lambda v: int(float(v)),
        )),
        "max_daily": float(get_env_int("MOMENTUM_SEARCH_MAX_DAILY", 10)),
        "fee_rate": _fee_r * 100.0,
        "sell_tax": _tax_r * 100.0,
        "high_chase_thr": get_env_float("MOMENTUM_SEARCH_HIGH_CHASE_THR", 0.99),
        "max_daily_chg": get_env_float("MOMENTUM_SEARCH_MAX_DAILY_CHG_PCT", 50.0),
        "min_price": int(_pick(
            "MOMENTUM_MIN_PRICE", "MIN_STOCK_PRICE",
            default=get_env_float("MOMENTUM_SEARCH_MIN_PRICE_KRW", float(_d.get("min_price", 1000))),
            cast=lambda v: int(float(v)),
        )),
        # 99M 등 OFF 값 → 20만 정규화 (breakout·웹 백테와 동일, 포지션 부풀림 방지)
        "max_loss_krw": float(normalize_breakout_max_loss_krw(_pick(
            "MOMENTUM_MAX_LOSS_PER_TRADE_KRW", "MAX_LOSS_PER_TRADE_KRW",
            default=get_env_int("MOMENTUM_SEARCH_MAX_LOSS_KRW", 200_000),
            cast=lambda v: int(float(v)),
        ))),
        # 탐색 기본 0.2% — DB MOMENTUM_MIN_PROFIT_PCT=10 은 TP(3~7%)와 어긋나 본절선이 비현실적
        "min_margin": get_env_float("MOMENTUM_SEARCH_MIN_MARGIN_PCT", 0.2),
        "use_defense_filters": get_env_bool(
            "MOMENTUM_SEARCH_USE_DEFENSE",
            bool(_d.get("use_defense_filters", True)),
        ),
        "mom_vol_win": float(get_env_int(
            "MOMENTUM_SEARCH_VOL_WIN",
            int(_pick("MOMENTUM_VOL_WIN", "SCALP_MOM_VOL_WIN", default=5, cast=lambda v: int(float(v)))),
        )),
        "mom_time_end_hm": int(_pick(
            "MOMENTUM_TIME_END_HM", "SCALP_MOM_TIME_END_HM",
            default=get_env_int("MOMENTUM_SEARCH_TIME_END_BUY_HM", 1430),
            cast=lambda v: int(float(v)),
        )),
        "mom_min_from_open_pct": get_env_float("MOMENTUM_SEARCH_MIN_FROM_OPEN_PCT", -999.0),
        "mom_max_from_open_pct": float(_d.get("mom_max_from_open_pct", 999.0)),
        "pattern_breakout": bool(_d.get("pattern_breakout", True)),
        "pattern_pullback": bool(_d.get("pattern_pullback", True)),
        "chase_lookback_min": int(_d.get("chase_lookback_min", 10)),
        "pullback_lookback_min": int(_d.get("pullback_lookback_min", 15)),
        "pullback_min_pct": float(_d.get("pullback_min_pct", 0.3)),
        "pullback_max_pct": float(_d.get("pullback_max_pct", 3.0)),
        "use_rsi_max_filter": bool(_d.get("use_rsi_max_filter", False)),
        "use_ema_filter": bool(_d.get("use_ema_filter", True)),
        "use_high_chase_filter": bool(_d.get("use_high_chase_filter", False)),
        "use_daily_range_filter": bool(_d.get("use_daily_range_filter", False)),
        "ema_fast_period": int(_d.get("ema_fast_period", 9)),
        "ema_slow_period": int(_d.get("ema_slow_period", 21)),
        "eod_enabled": bool(_d.get("eod_enabled", True)),
        "eod_hm": str(_d.get("eod_hm") or "15:25").strip() or "15:25",
    }


# ──────────────────────────────────────────────────────────────────────────────
# 결과 디렉터리 (param_search_scalping.py 와 공유)
# ──────────────────────────────────────────────────────────────────────────────

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]:
    """읽기용: 신규 results → 홈 폴백 (param_search_scalping 과 동일)."""
    return [
        os.path.join(HERE, "results"),
        os.path.join(os.path.expanduser("~"), ".kis_bot_search_results"),
    ]


def _latest_json(prefix: str) -> Optional[str]:
    """prefix 로 시작하는 가장 최근 search_momentum_*.json 경로."""
    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


# 호가필터 스윕 축 (2단계 정밀에서 '전수'로 돌리는 손잡이) — 나머지는 코어로 간주.
_MOMENTUM_OB_AXES = ("max_spread_pct", "min_bid_ask_ratio", "ask_max_mult")


def _load_stage1_cores(
    json_path: str, top_n: int, core_keys: List[str],
) -> List[Dict[str, Any]]:
    """stage1 결과 JSON의 top[:N] 에서 '코어축만' 추출 (호가 3축 제외).

    2단계 정밀(coarse-to-fine)에서 1단계 우승 코어를 고정하기 위함.
    양수 손익(total_pnl>0) 결과만 채택해, 1단계에서 검증된 코어만 넘긴다.
    """
    with open(json_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    top = data.get("top") or []
    cores: List[Dict[str, Any]] = []
    for item in top:
        if len(cores) >= top_n:
            break
        if float(item.get("total_pnl", 0) or 0) <= 0:
            continue  # 1단계에서 손실난 코어는 정밀 단계에서 제외
        p = item.get("params") or {}
        core = {k: p[k] for k in core_keys if k in p}
        if core:
            cores.append(core)
    return cores


def _resolve_stage1_json_path(explicit: Optional[str] = None) -> str:
    """2단계 코어고정용 stage1 결과 JSON 경로 (명시 없으면 최신 search_momentum_*.json)."""
    if explicit and os.path.isfile(explicit):
        return explicit
    latest = _latest_json("search_momentum_")
    if not latest:
        raise FileNotFoundError(
            "stage1 결과 JSON이 없습니다. 1단계를 먼저 실행하거나 --stage1-json 경로를 지정하세요."
        )
    return latest


def _build_stage2_core_lock_combos(
    grid: Dict[str, List[Any]],
    stage1_json: str,
    core_top_n: int,
) -> Tuple[List[Dict[str, Any]], List[str], Dict[str, Any], List[Dict[str, Any]]]:
    """1단계 Top-N 코어 고정 × 호가 3축 전수(27) 조합 생성.

    coarse-to-fine 2단계: 코어는 1단계 우승 레시피 그대로, 호가필터만 정밀 스윕.
    """
    keys = list(grid.keys())
    core_keys = [k for k in keys if k not in _MOMENTUM_OB_AXES]
    ob_keys = [k for k in keys if k in _MOMENTUM_OB_AXES]

    with open(stage1_json, "r", encoding="utf-8") as f:
        stage1_data = json.load(f)

    cores = _load_stage1_cores(stage1_json, core_top_n, core_keys)
    if not cores:
        raise RuntimeError(
            f"stage1 JSON에서 양수 손익 코어를 {core_top_n}개 찾지 못했습니다: {stage1_json}"
        )

    ob_axes = [grid[k] for k in ob_keys]
    ob_n = grid_total_combinations(ob_axes)
    dict_combos: List[Dict[str, Any]] = []
    for core in cores:
        for ob_t in product(*ob_axes):
            combo = dict(core)
            combo.update(dict(zip(ob_keys, ob_t)))
            if _momentum_combo_grid_valid(combo):
                dict_combos.append(combo)

    # stage1 top 에서 코어와 매칭되는 1단계 성과(비교용)
    stage1_baselines: List[Dict[str, Any]] = []
    for item in (stage1_data.get("top") or [])[:core_top_n * 2]:
        if float(item.get("total_pnl", 0) or 0) <= 0:
            continue
        p = item.get("params") or {}
        stage1_baselines.append({
            "params": {k: p[k] for k in core_keys if k in p},
            "total_pnl": item.get("total_pnl"),
            "win_rate": item.get("win_rate"),
            "total_trades": item.get("total_trades"),
            "pf": item.get("pf"),
        })
        if len(stage1_baselines) >= core_top_n:
            break

    return dict_combos, keys, stage1_data, stage1_baselines


def _apply_from_latest_json(rank: int) -> None:
    """최근 search_momentum_*.json 에서 rank(1-based) merged_params → env_config."""
    latest_path = _latest_json("search_momentum_")
    if not latest_path:
        print("⚠️  search_momentum_*.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)


# ──────────────────────────────────────────────────────────────────────────────
# 모멘텀 그리드 — coarse / fine / full
# ──────────────────────────────────────────────────────────────────────────────

def _momentum_grids() -> Dict[str, Dict[str, List]]:
    """탐색 모드별 모멘텀 그리드.

    각 축은 DB ``env_config`` 문자열 ``MOMENTUM_GRID_{MODE}_{키대문자}`` 로 덮어쓸 수 있다
    (예: ``MOMENTUM_GRID_COARSE_TRAIL_TRIGGER_PCT`` — 키 suffix 는 아래 ``_parse_csv_*`` 호출 첫 인자 참고).

    - coarse: **진입·손익 6축만** 그리드 (≈144조합). 트레일·쿨다운·일일한도·급등·min_margin 은
      ``_mom_fixed_defaults`` 고정 (min_margin 은 모멘텀 매수 루프 미연결 → 그리드 무의미).
    - fine  : coarse 와 동일 축 — RSI·거래량·TP/SL 등 스텝만 촘촘히
    - full  : RSI·볼·손익 범위 넓힘 + 트레일 OFF(0%) 혼합 + 동일 운영축

    청산 축(어깨·tp_max): 실매와 동일 우선순위. tp_max 는 운영 범위(≈1.8~3%) 그대로 탐색.
    ※ fast 는 13축 카테시안 후 ``MOMENTUM_FAST_MAX_COMBOS`` 균등 샘플 (꼬리잡기 fast 와 동일 패턴).
    ※ EMA OFF 시 fast/slow 는 9/21 고정만 유효 — ``_momentum_combo_grid_valid`` 로 중복 조합 스킵.
    """
    return {
        # ─────────────────────────────────────────────────────────────────────
        # [FAST] V2 TRIGGER — 추격 패턴 + RSI하한 + vol + 손익 + 어깨 + 트레일 + EMA
        # → env MOMENTUM_FAST_MAX_COMBOS(기본 512) 균등샘플
        # ─────────────────────────────────────────────────────────────────────
        "fast": {
            "mom_rsi_min": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_MOM_RSI_MIN", [45, 50, 55],
            ),
            "mom_vol_mult": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_MOM_VOL_MULT", [1.5, 2.0, 3.0],
            ),
            "pattern_breakout": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_PATTERN_BREAKOUT", [1],
            ),
            "pattern_pullback": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_PATTERN_PULLBACK", [0, 1],
            ),
            "chase_lookback_min": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_CHASE_LOOKBACK_MIN", [8, 10, 15],
            ),
            "pullback_lookback_min": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_PULLBACK_LOOKBACK_MIN", [10, 15],
            ),
            "pullback_min_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_PULLBACK_MIN_PCT", [0.2, 0.3],
            ),
            "pullback_max_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_PULLBACK_MAX_PCT", [2.0, 3.0],
            ),
            "sl_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_SL_PCT", [1.5, 2.0],
            ),
            # ── 추세추종형 청산 그리드 ────────────────────────────────────
            # 모멘텀은 갭상승·신고가·거래량 폭증 추세를 '끌고 가서' 먹는 전략.
            # tp/tp_max 를 2% 로 캡하면 큰 수익을 잘라 손익비가 1:1 로 깔려 PF<1.
            # → 고정익절(tp_pct)·하드캡(tp_max)·트레일을 넓혀 추세 수익을 살린다.
            #   (어깨·트레일이 먼저 발동해 수익을 보존하고, tp_max 는 마지막 상한)
            "tp_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_TP_PCT", [2.5, 4.0, 6.0],
            ),
            "tp_max_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_TP_MAX_PCT", [4.0, 6.0, 10.0],
            ),
            # 어깨 발동(진입가 대비 최소 수익%) — 추세는 충분히 오른 뒤 고점추적 시작
            "shoulder_min_high": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_SHOULDER_MIN_HIGH_PCT", [1.0, 2.0],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_SHOULDER_CUT_PCT", [0.15, 0.20, 0.25],
            ),
            # 전용 트레일 폭(%) — 추세 추적의 핵심. 0(OFF) 대신 0.5~1.5 로 끌고 간다.
            "trail_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_TRAIL_PCT", [0.5, 1.0, 1.5],
            ),
            "trail_arm_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_TRAIL_ARM_PCT", [0.0, 0.8],
            ),
            "use_ema_filter": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_USE_EMA_FILTER", [0, 1],
            ),
            "ema_fast_period": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_EMA_FAST_PERIOD", [5, 9, 12],
            ),
            "ema_slow_period": _parse_csv_ints(
                "MOMENTUM_GRID_FAST_EMA_SLOW_PERIOD", [15, 21, 34],
            ),
            # ── 호가필터(2단계 조인트) 축 ──────────────────────────────
            # 기본은 모두 단일값(운영값) → 조합 안 늘고 동작 불변.
            # 여러 값 주면 스윕 + kiwoom_0d 본체 재계산 자동 활성(필터 ON 일 때).
            # (본체는 6/25~ 만 존재 → 그 외 날짜는 효과 없음/판정재생)
            #   max_spread_pct    : 호가 스프레드 상한(%)            ↑완화
            #   min_bid_ask_ratio : 매수/매도 잔량비 하한            ↓완화
            #   ask_max_mult      : 매도벽 허용배수(필요수량×N)      ↑완화
            "max_spread_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_MAX_SPREAD_PCT", [0.45],
            ),
            "min_bid_ask_ratio": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_MIN_BID_ASK_RATIO", [0.85],
            ),
            "ask_max_mult": _parse_csv_floats(
                "MOMENTUM_GRID_FAST_ASK_MAX_MULT", [3.0],
            ),
        },
        # rr: 손익비(어깨·SL·TP) 중심 — 진입은 운영값 1점 고정 (~720조합)
        "rr": {
            "mom_rsi_min": _parse_csv_ints(
                "MOMENTUM_GRID_RR_MOM_RSI_MIN",
                [int(get_env_int("MOMENTUM_RSI_MIN", 55))],
            ),
            "mom_rsi_max": _parse_csv_ints(
                "MOMENTUM_GRID_RR_MOM_RSI_MAX",
                [int(get_env_int("MOMENTUM_RSI_MAX", 90))],
            ),
            "mom_vol_mult": _parse_csv_floats(
                "MOMENTUM_GRID_RR_MOM_VOL_MULT",
                [float(get_env_float("MOMENTUM_VOL_MULT", 1.5))],
            ),
            "sl_pct": _parse_csv_floats(
                "MOMENTUM_GRID_RR_SL_PCT", [1.2, 1.5, 1.8, 2.0],
            ),
            "tp_pct": _parse_csv_floats(
                "MOMENTUM_GRID_RR_TP_PCT", [2.0, 2.5, 3.0],
            ),
            "tp_max_pct": _parse_csv_floats(
                "MOMENTUM_GRID_RR_TP_MAX_PCT", [2.0, 2.5, 3.0],
            ),
            "shoulder_min_high": _parse_csv_floats(
                "MOMENTUM_GRID_RR_SHOULDER_MIN_HIGH_PCT", [0.5, 1.0, 50.0],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "MOMENTUM_GRID_RR_SHOULDER_CUT_PCT", [0.02, 0.15, 0.25],
            ),
        },
        # coarse: 2×2×5×3×3×2×2×2 = 960 (RSI min≥max 무효 조합 제외 시 약간 감소)
        #   그리드: RSI·거래량(mom_vol_mult→MOMENTUM_VOL_MULT)·TP/SL·시가과열컷
        "coarse": {
            "mom_rsi_min": _parse_csv_ints(
                "MOMENTUM_GRID_COARSE_MOM_RSI_MIN", [50, 55],
            ),
            "mom_rsi_max": _parse_csv_ints(
                "MOMENTUM_GRID_COARSE_MOM_RSI_MAX", [90, 95],
            ),
            "mom_vol_mult": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_MOM_VOL_MULT", [1.5,3.0],
            ),
            "tp_pct": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_TP_PCT", [3.0, 5.0],
            ),
            "sl_pct": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_SL_PCT", [1.5, 2.0, 3.0],
            ),
            "mom_max_from_open_pct": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_MOM_MAX_FROM_OPEN_PCT", [20.0, 30.0],
            ),
            "mom_max_from_open_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_MOM_MAX_FROM_OPEN_PCT", [20.0, 25.0, 30.0],
            ),
            "shoulder_min_high": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_SHOULDER_MIN_HIGH_PCT", [0.3, 0.5],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_SHOULDER_CUT_PCT", [0.2, 0.3],
            ),
            "tp_max_pct": _parse_csv_floats(
                "MOMENTUM_GRID_COARSE_TP_MAX_PCT", [1.8, 2.0],
            ),
        },
        # fine: 동일 13축 — 중간 스텝 확대 (≈ 1만 × 8 ≈ 10만 조합)
        "fine": {
            "mom_rsi_min": _parse_csv_ints(
                "MOMENTUM_GRID_FINE_MOM_RSI_MIN", [49, 50, 55],
            ),
            "mom_rsi_max": _parse_csv_ints(
                "MOMENTUM_GRID_FINE_MOM_RSI_MAX", [90, 92, 95],
            ),
            "mom_vol_mult": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_MOM_VOL_MULT", [5.0, 7.5, 10.0, 15.0, 20.0],
            ),
            "tp_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_TP_PCT", [3.0, 4.5, 5.5, 7.0],
            ),
            "sl_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_SL_PCT", [1.5, 2.0, 2.5, 3.0],
            ),
            "mom_time_end_hm": _parse_csv_ints(
                "MOMENTUM_GRID_FINE_MOM_TIME_END_HM", [1300, 1400, 1430],
            ),
            "mom_max_from_open_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_MOM_MAX_FROM_OPEN_PCT", [20.0, 25.0, 30.0],
            ),
            "min_margin": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_MIN_MARGIN", [0.2, 10.0],
            ),
            "shoulder_min_high": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_SHOULDER_MIN_HIGH_PCT", [0.2, 0.3, 0.5],
            ),
            "shoulder_cut_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_SHOULDER_CUT_PCT", [0.2, 0.3, 0.4],
            ),
            "trail_trigger": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_TRAIL_TRIGGER_PCT", [1.0, 1.5],
            ),
            "trail_stop": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_TRAIL_STOP_PCT", [1.0],
            ),
            "cooldown_min": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_COOLDOWN_MIN", [1.0, 3.0],
            ),
            "max_daily": _parse_csv_ints(
                "MOMENTUM_GRID_FINE_MAX_DAILY", [5, 10],
            ),
            "max_daily_chg": _parse_csv_floats(
                "MOMENTUM_GRID_FINE_MAX_DAILY_CHG_PCT", [45.0],
            ),
        },
        # full: 가설 폭넓힘 (낮은 vol_mult·짧은 TP 가설 포함) + 트레일 0 = TP 상한 위주 청산
        # 조합 수 대략 수만 단위 — env 로 각 축 목록 줄이는 것 권장.
        "full": {
            "mom_rsi_min": _parse_csv_ints(
                "MOMENTUM_GRID_FULL_MOM_RSI_MIN", [45, 55],
            ),
            "mom_rsi_max": _parse_csv_ints(
                "MOMENTUM_GRID_FULL_MOM_RSI_MAX", [80, 90, 95],
            ),
            "mom_vol_mult": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_MOM_VOL_MULT", [3.0, 10.0, 20.0],
            ),
            "tp_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_TP_PCT", [3.0, 5.0, 7.0],
            ),
            "sl_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_SL_PCT", [1.5, 2.5, 3.0],
            ),
            "mom_time_end_hm": _parse_csv_ints(
                "MOMENTUM_GRID_FULL_MOM_TIME_END_HM", [1330, 1430],
            ),
            "mom_max_from_open_pct": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_MOM_MAX_FROM_OPEN_PCT", [22.0, 30.0],
            ),
            "min_margin": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_MIN_MARGIN", [0.2, 10.0],
            ),
            "trail_trigger": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_TRAIL_TRIGGER_PCT", [0.0, 1.2],
            ),
            "trail_stop": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_TRAIL_STOP_PCT", [0.5, 1.0],
            ),
            "cooldown_min": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_COOLDOWN_MIN", [1.0, 5.0],
            ),
            "max_daily": _parse_csv_ints(
                "MOMENTUM_GRID_FULL_MAX_DAILY", [5, 10],
            ),
            "max_daily_chg": _parse_csv_floats(
                "MOMENTUM_GRID_FULL_MAX_DAILY_CHG_PCT", [35.0, 50.0],
            ),
        },
    }


MOMENTUM_GRID_AXIS_HINTS_KO: Dict[str, str] = {
    "mom_rsi_min": "RSI3 하한 — 약세 제외",
    "mom_rsi_max": "RSI3 상한 — use_rsi_max_filter ON 시만",
    "mom_vol_mult": "현재봉 거래량 ≥ N봉평균 × 배수",
    "pattern_breakout": "N분 고가 돌파 패턴 (0=OFF, 1=ON)",
    "pattern_pullback": "눌림 재돌파 패턴 (0=OFF, 1=ON)",
    "chase_lookback_min": "돌파 패턴 관찰 구간(분)",
    "pullback_lookback_min": "눌림 패턴 스윙고점 탐색 구간(분)",
    "pullback_min_pct": "눌림 최소 깊이(%)",
    "pullback_max_pct": "눌림 최대 깊이(%)",
    "mom_max_from_open_pct": "레거시: 당일 시가 대비 과열 상한(%)",
    "sl_pct": "고정 손절(%)",
    "tp_pct": "고정 익절(%)",
    "tp_max_pct": "익절 상한(%) — 어깨·트레일 이후 하드 캡",
    "shoulder_min_high": "어깨 발동: 진입가 대비 최소 수익(%) 후 고점 추적",
    "shoulder_cut_pct": "고점 대비 되돌림(%) 시 어깨컷",
    "trail_pct": "전용 트레일 폭(%, 0=OFF) — momentum_engine",
    "trail_arm_pct": "트레일 무장: 진입×(1+arm%) 도달 후 (0=즉시)",
    "max_hold_bars": "최대 보유 1분봉 (0=비활성)",
    "ratchet_tiers": "래칫 티어 문자열 (빈칸=OFF)",
    "use_ema_filter": "EMA 추세필터 (0=OFF, 1=ON) — TRIGGER",
    "ema_fast_period": "EMA 빠른 기간 (fast)",
    "ema_slow_period": "EMA 느린 기간 (slow, fast보다 커야 함)",
    "max_spread_pct": "호가 스프레드 상한(%) — kiwoom_0d 본체 재계산(6/25~)",
    "min_bid_ask_ratio": "매수/매도 잔량비 하한 — kiwoom_0d 본체 재계산(6/25~)",
    "ask_max_mult": "매도벽 허용배수(필요수량×N) — kiwoom_0d 본체 재계산(6/25~)",
}


def _momentum_ema_filter_on(raw: Any) -> bool:
    if isinstance(raw, bool):
        return raw
    if raw in (None, ""):
        return False
    return str(raw).strip().lower() in ("1", "true", "t", "y", "yes", "on")


def _momentum_combo_grid_valid(combo: Dict[str, Any]) -> bool:
    """fast 그리드 — 패턴 OR·EMA·눌림폭·RSI 무효 조합 제거."""
    if "mom_rsi_min" in combo and "mom_rsi_max" in combo:
        if float(combo["mom_rsi_min"]) >= float(combo["mom_rsi_max"]):
            return False
    bo = _momentum_ema_filter_on(combo.get("pattern_breakout", 1))
    pb = _momentum_ema_filter_on(combo.get("pattern_pullback", 1))
    if not bo and not pb:
        return False
    if "pullback_min_pct" in combo and "pullback_max_pct" in combo:
        if float(combo["pullback_min_pct"]) >= float(combo["pullback_max_pct"]):
            return False
    ema_on = _momentum_ema_filter_on(combo.get("use_ema_filter", 0))
    fast = int(float(combo.get("ema_fast_period", 9)))
    slow = int(float(combo.get("ema_slow_period", 21)))
    if not ema_on:
        return fast == 9 and slow == 21
    return slow > fast


# ──────────────────────────────────────────────────────────────────────────────
# DB 적용 — MOMENTUM_* env 키
# ──────────────────────────────────────────────────────────────────────────────

def _get_momentum_field_map() -> Dict[str, Tuple[str, Any]]:
    """모멘텀 파라미터 → env_config 컬럼 매핑.

    ※ 새 키 MOMENTUM_* 를 1차로 채움. 기존 SCALP_MOM_*/SCALP_* 도 같이 채울지는
    --apply 사용자 의도에 따라 향후 옵션 추가 가능. 지금은 새 키만.
    """
    return {
        # 진입 룰
        "mom_rsi_min":    ("MOMENTUM_RSI_MIN",         lambda v: str(int(v))),
        "mom_rsi_max":    ("MOMENTUM_RSI_MAX",         lambda v: str(int(v))),
        "mom_vol_mult":   ("MOMENTUM_VOL_MULT",        lambda v: str(float(v))),
        "mom_vol_win":    ("MOMENTUM_VOL_WIN",         lambda v: str(int(float(v)))),
        "mom_time_end_hm":("MOMENTUM_TIME_END_HM",     lambda v: str(int(float(v)))),
        "mom_max_from_open_pct": ("MOMENTUM_MAX_FROM_OPEN_PCT", lambda v: str(float(v))),
        "mom_min_from_open_pct": ("MOMENTUM_MIN_FROM_OPEN_PCT", lambda v: str(float(v))),
        "pattern_breakout": ("MOMENTUM_PATTERN_BREAKOUT", lambda v: "true" if _momentum_ema_filter_on(v) else "false"),
        "pattern_pullback": ("MOMENTUM_PATTERN_PULLBACK", lambda v: "true" if _momentum_ema_filter_on(v) else "false"),
        "chase_lookback_min": ("MOMENTUM_CHASE_LOOKBACK_MIN", lambda v: str(int(float(v)))),
        "pullback_lookback_min": ("MOMENTUM_PULLBACK_LOOKBACK_MIN", lambda v: str(int(float(v)))),
        "pullback_min_pct": ("MOMENTUM_PULLBACK_MIN_PCT", lambda v: str(float(v))),
        "pullback_max_pct": ("MOMENTUM_PULLBACK_MAX_PCT", lambda v: str(float(v))),
        "use_ema_filter": ("MOMENTUM_USE_EMA_FILTER", lambda v: "true" if _momentum_ema_filter_on(v) else "false"),
        "ema_fast_period": ("MOMENTUM_EMA_FAST_PERIOD", lambda v: str(int(float(v)))),
        "ema_slow_period": ("MOMENTUM_EMA_SLOW_PERIOD", lambda v: str(int(float(v)))),
        # 청산 룰
        "sl_pct":         ("MOMENTUM_STOP_LOSS_PCT",   lambda v: str(float(v) / 100)),
        "tp_pct":         ("MOMENTUM_TAKE_PROFIT_PCT", lambda v: str(float(v) / 100)),
        # 청산 보조 — 웹 저장과 동일: UI 퍼센트 → DB 비율 (/100)
        "trail_trigger":  ("SCALP_ATR_UP_MULT",        lambda v: str(abs(float(v)) / 100.0)),
        "trail_stop":     ("SCALP_ATR_DOWN_MULT",      lambda v: str(abs(float(v)) / 100.0)),
        "shoulder_min_high": ("MOMENTUM_SHOULDER_MIN_HIGH_PCT", lambda v: str(float(v) / 100)),
        "shoulder_cut_pct":  ("MOMENTUM_SHOULDER_CUT_PCT",      lambda v: str(float(v) / 100)),
        "tp_max_pct":          ("MOMENTUM_TP_MAX_PCT",            lambda v: str(float(v) / 100)),
        "cooldown_min":   ("SCALP_COOLDOWN_SEC",       lambda v: str(int(float(v)) * 60)),
        # 방어로직 (모멘텀 전용 키가 없으면 SCALP_* 공유)
        "high_chase_thr": ("SCALP_HIGH_PRICE_CHASE_THRESHOLD", lambda v: str(float(v))),
        "max_daily_chg":  ("SCALP_MAX_DAILY_CHANGE_PCT",       lambda v: str(float(v))),
        "min_price":      ("MOMENTUM_MIN_PRICE",       lambda v: str(int(float(v)))),
        "max_loss_krw":   ("MOMENTUM_MAX_LOSS_PER_TRADE_KRW", lambda v: str(int(float(v)))),
        "min_margin":     ("MOMENTUM_MIN_PROFIT_PCT",  lambda v: str(float(v))),
        "use_defense_filters": ("SCALP_USE_DEFENSE_FILTERS",  lambda v: "true" if v else "false"),
        "max_daily":      ("MOMENTUM_MAX_DAILY",       lambda v: str(int(float(v)))),
        "slot_money":     ("MOMENTUM_SLOT_MONEY",      lambda v: str(int(float(v)))),
    }


def _apply_to_db(best_params: dict) -> Optional[int]:
    """1위 파라미터 → insert_env_snapshot (config_momentum + env_config). env_id 반환."""
    from kis_trader.backtest.param_search_apply_snapshot import (
        _patch_from_momentum_merged,
        apply_env_patch,
    )

    patch = _patch_from_momentum_merged(best_params)
    patch.update(portfolio_env_patch("MOMENTUM", best_params))
    patch.update(session_env_patch("MOMENTUM", best_params))
    if not patch:
        print("DB 적용할 파라미터가 없습니다.")
        return None
    eid = apply_env_patch(patch)
    if eid is None:
        print("❌ insert_env_snapshot 실패")
        return None
    print(f"\n✅ config_momentum + env_config INSERT id={eid} (MOMENTUM_* 키):")
    for k, v in sorted(patch.items()):
        print(f"   {k:<35s} = {v}")
    return eid


def apply_params_to_db(best_params: dict) -> Optional[int]:
    """웹·CLI 공통 — 파라서치 merged → DB (tail_param_search.apply_params_to_db 와 동일 UX)."""
    return _apply_to_db(best_params)


# ──────────────────────────────────────────────────────────────────────────────
# UI(%) → 엔진(비율) 변환 + 워커 청크 평가
# ──────────────────────────────────────────────────────────────────────────────

def _ui_to_engine_params(ui_params: dict) -> dict:
    """UI 표시용(% 등) → 엔진용 비율 단위. 워커에서 공통 사용."""
    engine_params = dict(ui_params)
    engine_params["sl_pct"]        = ui_params["sl_pct"] / 100
    engine_params["tp_pct"]        = ui_params["tp_pct"] / 100
    if "tp_max_pct" in ui_params:
        engine_params["tp_max_pct"] = ui_params["tp_max_pct"] / 100
    elif "tp_max_pct" not in engine_params:
        engine_params["tp_max_pct"] = get_env_float("MOMENTUM_TP_MAX_PCT", 0.02)
    if "trail_trigger" in ui_params:
        engine_params["trail_trigger"] = ui_params["trail_trigger"] / 100
    if "trail_stop" in ui_params:
        engine_params["trail_stop"]    = ui_params["trail_stop"] / 100
    if "trail_pct" in ui_params:
        engine_params["trail_pct"] = abs(float(ui_params["trail_pct"])) / 100.0
    if "trail_arm_pct" in ui_params:
        engine_params["trail_arm_pct"] = abs(float(ui_params["trail_arm_pct"])) / 100.0
    if "max_hold_bars" in ui_params:
        engine_params["max_hold_bars"] = int(float(ui_params["max_hold_bars"] or 0))
    if "shoulder_min_high" in ui_params:
        engine_params["shoulder_min_high"] = ui_params["shoulder_min_high"] / 100
    if "shoulder_cut_pct" in ui_params:
        engine_params["shoulder_cut_pct"] = ui_params["shoulder_cut_pct"] / 100
    engine_params["fee_rate"]      = ui_params["fee_rate"] / 100
    engine_params["sell_tax"]      = ui_params["sell_tax"] / 100
    engine_params["min_margin"]    = ui_params.get("min_margin", 0.2) / 100
    if "use_defense_filters" in ui_params:
        engine_params["use_defense_filters"] = bool(ui_params["use_defense_filters"])
    if "use_ema_filter" in ui_params:
        engine_params["use_ema_filter"] = _momentum_ema_filter_on(ui_params["use_ema_filter"])
    if "use_rsi_max_filter" in ui_params:
        engine_params["use_rsi_max_filter"] = _momentum_ema_filter_on(ui_params["use_rsi_max_filter"])
    if "pattern_breakout" in ui_params:
        engine_params["pattern_breakout"] = _momentum_ema_filter_on(ui_params["pattern_breakout"])
    if "pattern_pullback" in ui_params:
        engine_params["pattern_pullback"] = _momentum_ema_filter_on(ui_params["pattern_pullback"])
    if "chase_lookback_min" in ui_params:
        engine_params["chase_lookback_min"] = int(float(ui_params["chase_lookback_min"]))
    if "pullback_lookback_min" in ui_params:
        engine_params["pullback_lookback_min"] = int(float(ui_params["pullback_lookback_min"]))
    if "pullback_min_pct" in ui_params:
        engine_params["pullback_min_pct"] = float(ui_params["pullback_min_pct"])
    if "pullback_max_pct" in ui_params:
        engine_params["pullback_max_pct"] = float(ui_params["pullback_max_pct"])
    if "ema_fast_period" in ui_params:
        engine_params["ema_fast_period"] = int(float(ui_params["ema_fast_period"]))
    if "ema_slow_period" in ui_params:
        engine_params["ema_slow_period"] = int(float(ui_params["ema_slow_period"]))
    # 호가필터 임계값 → per-run 오버라이드 (kiwoom_0d 본체 재계산 시 적용)
    if "max_spread_pct" in ui_params and ui_params["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["min_bid_ask_ratio"] is not None:
        engine_params["_ob_min_bid_ask_ratio"] = float(ui_params["min_bid_ask_ratio"])
    if "ask_max_mult" in ui_params and ui_params["ask_max_mult"] is not None:
        engine_params["_ob_ask_max_mult"] = float(ui_params["ask_max_mult"])

    engine_params["max_loss_krw"] = normalize_breakout_max_loss_krw(
        ui_params.get("max_loss_krw", 200_000),
    )

    # 손절 % 에 맞춰 1회 투자금(slot_money) 자동 계산 (라이브 봇과 동일 공식)
    max_loss = engine_params.get("max_loss_krw", 200_000)
    if engine_params["sl_pct"] > 0 and 0 < max_loss < 10_000_000:
        # max_loss < 1000만일 때만 자동 계산 (라이브 운영 정상 범위)
        engine_params["slot_money"] = max_loss / engine_params["sl_pct"]

    return engine_params


def evaluate_momentum_param_combo(
    combo: Dict[str, Any],
    *,
    base_fixed: Dict[str, Any],
    grid_keys: List[str],
    codes_candles: Dict[str, List[Dict]],
    min_trades: int,
    min_win_rate: float,
    min_pf: float,
    universe_by_slot: Optional[Dict[str, List[str]]] = None,
    slot_money: float = 3_000_000.0,
    max_stocks: int = 3,
    total_budget_krw: float = 9_000_000.0,
    fee_rate: float = 0.00015,
    sell_tax: float = 0.0018,
    period_days: int = 1,
    cache_holder: Optional[Dict[str, Any]] = None,
    ticks_by_code: Any = None,
    orderbook_by_code: Any = None,
    program_by_code: Any = None,
    log_verdict_by_code: Any = None,
    start_key: str = "",
    end_key: str = "",
) -> Optional[Dict[str, Any]]:
    """단일 모멘텀 조합 백테 — Grid 워커·Optuna objective 공통."""
    if not _momentum_combo_grid_valid(combo):
        return None

    ui_params = dict(base_fixed)
    ui_params.update(combo)
    engine_params = _ui_to_engine_params(ui_params)
    if cache_holder:
        engine_params.update(cache_holder)
    engine_params["slot_money"] = float(slot_money)
    engine_params["max_stocks"] = int(max_stocks)
    engine_params["total_budget_krw"] = float(total_budget_krw)
    engine_params["portfolio_mode"] = True
    if log_verdict_by_code:
        engine_params["_backtest_log_verdict_by_code"] = log_verdict_by_code

    meta: Dict[str, Any] = {}
    if len(start_key) >= 12:
        meta["start_key"] = start_key
        engine_params["_backtest_period_start_key"] = start_key[:12]
    if len(end_key) >= 12:
        meta["end_key"] = end_key

    trades = mbc.run_momentum_backtest_web_aligned(
        codes_candles, engine_params, universe_by_slot,
        slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax,
        max_stocks=max_stocks, total_budget_krw=total_budget_krw,
        ticks_by_code=ticks_by_code,
        orderbook_by_code=orderbook_by_code,
        program_by_code=program_by_code,
        meta_out=meta,
    )
    stats = mbc.summarize_momentum_trades(
        trades, total_budget_krw=total_budget_krw, period_days=period_days,
    )
    total_trades = stats["total_trades"]
    if total_trades < min_trades:
        return None

    total_pnl = stats["total_pnl"]
    win_rate = stats["win_rate"]
    pf = float(stats.get("pf") or 0)
    if not combo_passes_search_filters(
        win_rate=win_rate, pf=pf,
        min_win_rate=min_win_rate, min_pf=min_pf,
    ):
        return None

    avg_hold = stats["avg_hold_min"]
    peak, mdd, cum = 0.0, 0.0, 0.0
    for t in trades:
        cum += t["pnl"]
        if cum > peak:
            peak = cum
        dd = peak - cum
        if dd > mdd:
            mdd = dd

    merged = dict(ui_params)
    merged["slot_money"] = float(slot_money)
    merged["max_stocks"] = int(max_stocks)
    merged["total_budget_krw"] = float(total_budget_krw)

    return {
        "params": {k: ui_params[k] for k in grid_keys if k in ui_params},
        "total_pnl": int(total_pnl),
        "win_rate": round(win_rate, 2),
        "total_trades": total_trades,
        "pf": round(pf, 2),
        "avg_hold": round(avg_hold, 1),
        "mdd": round(mdd),
        "bot_pct": stats["bot_pct"],
        "daily_avg_pct": stats["daily_avg_pct"],
        "sell_reasons": mbc.count_momentum_sell_reasons(trades),
        "skipped_micro_buys": int(
            (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0
        ),
        "merged_params": merged,
    }


def _evaluate_momentum_chunk(
    param_chunk: List[Dict[str, Any]],
    base_fixed: Dict[str, Any],
    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 = 3_000_000.0,
    max_stocks: int = 3,
    total_budget_krw: float = 9_000_000.0,
    fee_rate: float = 0.00015,
    sell_tax: float = 0.0018,
    period_days: int = 1,
) -> List[Tuple[float, float, int, Dict]]:
    """워커: 청크 내 조합 평가 — 시각순 포트폴리오 (scalping_backtest_common, mode=momentum)."""
    shared = worker_shared_get()
    orderbook_preloaded = None
    program_preloaded = None
    ticks_preloaded = None
    log_verdict_preloaded = None
    if shared:
        if codes_candles is None:
            codes_candles = shared.get("codes_candles") or {}
        if universe_by_slot is None:
            universe_by_slot = shared.get("universe_by_slot")
        orderbook_preloaded = shared.get("orderbook_by_code")
        program_preloaded = shared.get("program_by_code")
        ticks_preloaded = shared.get("ticks_by_code")
        # ws_ticks 공유메모리(opt-in): descriptor 만 pickle 로 받고, 워커는 이름으로
        # shared_memory 에 read-only attach 한다(사본 없음). SharedTicksMapping 은
        # 워커당 1회만 만들어 재사용(attach lazy·1회) → dict 경로와 산출물 동일.
        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  # 모듈 전역 _WORKER_SHARED 에 유지
                ticks_preloaded = _tm
        log_verdict_preloaded = shared.get("log_verdict_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_momentum_param_combo(
            combo,
            base_fixed=base_fixed,
            grid_keys=keys,
            codes_candles=codes_candles,
            min_trades=min_trades,
            min_win_rate=min_win_rate,
            min_pf=min_pf,
            universe_by_slot=universe_by_slot,
            slot_money=slot_money,
            max_stocks=max_stocks,
            total_budget_krw=total_budget_krw,
            fee_rate=fee_rate,
            sell_tax=sell_tax,
            period_days=period_days,
            cache_holder=cache_holder,
            ticks_by_code=ticks_preloaded,
            orderbook_by_code=orderbook_preloaded,
            program_by_code=program_preloaded,
            log_verdict_by_code=log_verdict_preloaded,
            start_key=str(shared.get("start_key") or "") if shared else "",
            end_key=str(shared.get("end_key") or "") if shared else "",
        )
        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


# ──────────────────────────────────────────────────────────────────────────────
# 캔들 로드 (param_search_scalping.py 와 동일 패턴)
# ──────────────────────────────────────────────────────────────────────────────

def _load_candles_for_search(start: str, end: str, rsi_period: int) -> 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]

        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) < rsi_period + 5:
                continue
            codes_candles[code] = [dict(r) for r in rows]
        if codes_candles:
            from kis_trader.backtest.momentum_backtest_common import (
                prepend_momentum_candle_warmup,
            )
            prepend_momentum_candle_warmup(db, codes_candles, start_key[:12])
    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",
    stage2_core_lock: bool = False,
    stage1_json: Optional[str] = None,
) -> bool:
    if from_file_only and apply_rank is not None and apply_rank >= 1:
        _apply_from_latest_json(apply_rank)
        return True

    grid = _momentum_grids()[mode]
    keys = list(grid.keys())
    stage1_baselines: List[Dict[str, Any]] = []
    stage1_source_path = ""
    stage2_core_lock_meta: Dict[str, Any] = {}

    if stage2_core_lock:
        # 2단계 정밀: 1단계 Top-N 코어 고정 × 호가 3축 전수 (균등샘플 아님).
        stage1_source_path = _resolve_stage1_json_path(stage1_json)
        core_top_n = get_env_int("MOMENTUM_STAGE2_CORE_TOPN", 7)
        dict_combos, keys, stage1_data, stage1_baselines = _build_stage2_core_lock_combos(
            grid, stage1_source_path, core_top_n,
        )
        ob_axes_n = grid_total_combinations([grid[k] for k in _MOMENTUM_OB_AXES])
        total_grid = len(dict_combos)
        total_combos = total_grid
        max_combos_cap = total_grid
        dropped_by_cap = 0
        sampled = len(dict_combos)
        # 2단계는 호가필터 ON + kiwoom 본체 강제
        orderbook_filter = "on"
        stage2_core_lock_meta = {
            "stage2_core_lock": True,
            "stage1_json": stage1_source_path,
            "stage1_start": stage1_data.get("start"),
            "stage1_end": stage1_data.get("end"),
            "core_top_n": core_top_n,
            "ob_combos_per_core": ob_axes_n,
            "stage1_baselines": stage1_baselines,
        }
        print(f"\n[MOMENTUM STAGE2 · 코어고정] 1단계: {stage1_source_path}")
        print(
            f"📌 Top-{core_top_n} 코어(양수) × 호가 {ob_axes_n}전수 "
            f"= {sampled:,}조합 (균등샘플 없음)  |  기간: {start} ~ {end}"
        )
        for k in _MOMENTUM_OB_AXES:
            hint = MOMENTUM_GRID_AXIS_HINTS_KO.get(k)
            if hint:
                print(f"   [{k}] {hint} → {grid.get(k)}")
        print("📌 1단계 코어 성과(필터 OFF · 비교 기준):")
        for i, bl in enumerate(stage1_baselines, 1):
            print(
                f"   코어#{i}: 손익 {bl.get('total_pnl', 0):+,.0f}원 | "
                f"승률 {bl.get('win_rate', 0):.1f}% | "
                f"거래 {bl.get('total_trades', 0)} | PF {bl.get('pf', 0):.2f}"
            )
        # ★ 2단계 = 검증된 코어에 호가필터만 씌워 '비교'하는 단계.
        #   여기에 승률/PF/거래수 합격필터를 적용하면 전부 탈락 → "만족 0개" 가 떠버린다.
        #   필터를 켜서 결과가 '나빠지는 것'도 의미있는 데이터이므로, 컷 없이 전 조합을 출력한다.
        #   (0거래 조합 = "필터가 해당 코어의 진입을 전부 막음" 을 그대로 보여줌)
        min_trades = 0
        min_win_rate = 0.0
        min_pf = 0.0
        print("📌 2단계 출력 정책: 합격필터 없음 — 전 조합 비교 출력(거래0 포함)")
    else:
        axes = [grid[k] for k in keys]
        # 데카르트곱 전체를 RAM 에 펼치지 않는다 (호가 3축 스윕 시 수천만 → OOM 방지).
        total_combos = grid_total_combinations(axes)
        dict_combos, total_grid, max_combos_cap, dropped_by_cap = cap_combos_uniform_lazy(
            keys,
            axes,
            mode,
            strategy_env_prefix="MOMENTUM",
            default_fast=150,
            default_other=DEFAULT_MAX_COMBOS,
            max_combos_override=max_combos,
            valid_fn=_momentum_combo_grid_valid,
        )
        sampled = len(dict_combos)
        print(f"\n[MOMENTUM {mode.upper()} 모드] 그리드: {total_grid:,} → 백테: {sampled:,}  |  기간: {start} ~ {end}")
        if mode == "fast":
            print(
                f"📌 [fast] MOMENTUM_GRID_FAST_* · {len(keys)}축 · "
                f"{total_grid:,}→{max_combos_cap}균등샘플 · momentum_engine 전용청산"
            )
        if dropped_by_cap:
            print(f"   (max-combos={max_combos_cap} 균등 샘플, 제외 {dropped_by_cap:,}개)")

    if not stage2_core_lock:
        for k in keys:
            hint = MOMENTUM_GRID_AXIS_HINTS_KO.get(k)
            if hint:
                print(f"   [{k}] {hint}")
    print(f"📌 1위 정렬 기준: 순수익/MDD 점수 최대 (낙폭 대비 자본효율 — 동점 시 순익)")
    print(f"📌 필터: 승률≥{min_win_rate}% · PF≥{min_pf} · 거래≥{min_trades}건")
    print("=" * 70)

    t0 = time.time()
    FIXED_DEFAULTS = _mom_fixed_defaults()
    apply_session_to_fixed(
        FIXED_DEFAULTS,
        time_start_hm=time_start_hm,
        time_end_hm=time_end_hm,
    )

    # ── 틱재생(ws_ticks) — 기본 ON (실매 체결 정합) ──────────────────────
    # 실매 청산은 초단위 실제 틱(ws_ticks)으로 체결된다. OHLC 경로(open→high→low→close)
    # 는 "고가 먼저" 낙관 가정이라 어깨컷·트레일컷·익절을 실제보다 유리하게 체결해
    # 백테 손익을 부풀린다(검증: 2026-07-03 OHLC 순위 전 조합 흑자 → 틱 순위 전 조합
    # 적자, 실매 -48,742 정합). OHLC 로 뽑은 "최적"은 실매에서 손실나는 파라미터였다.
    # → 파람서치·웹 백테 모두 기본 틱재생. 끄려면 MOMENTUM_BACKTEST_USE_TICK_EXIT=0.
    from kis_trader.engine.momentum_tick_replay import (
        momentum_backtest_use_tick_entry as _mom_use_tick_entry,
        momentum_backtest_use_tick_exit as _mom_use_tick_exit,
    )
    FIXED_DEFAULTS["backtest_use_tick_entry"] = _mom_use_tick_entry(None)
    FIXED_DEFAULTS["backtest_use_tick_exit"] = _mom_use_tick_exit(None)
    if FIXED_DEFAULTS["backtest_use_tick_exit"] or FIXED_DEFAULTS["backtest_use_tick_entry"]:
        print(
            f"📌 틱재생(ws_ticks): 진입={FIXED_DEFAULTS['backtest_use_tick_entry']} "
            f"청산={FIXED_DEFAULTS['backtest_use_tick_exit']} — 실매 체결 정합 모드"
        )

    # ── 호가필터 ON/OFF (2단계 워크플로) ───────────────────────────────
    # off(기본): 1단계 — 호가 게이트 없이 코어 파라미터만 순수 탐색(표본↑).
    #            log_backfill 판정도 무시되어 "필터 없는 세상의 최적" 을 찾는다.
    # 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
    # auto → 키 주입 안 함 (orderbook_filter._orderbook_filter_enabled_for_entry 가 DB값 사용)
    _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 = sbc.fee_and_slot_from_env(env_row, strategy="MOMENTUM")
    portfolio = sbc.resolve_scalp_portfolio_params(
        env_row,
        None,
        strategy="MOMENTUM",
        slot_money=slot_money if slot_money is not None else slot_from_env,
        max_stocks=max_stocks,
        total_budget_krw=total_budget_krw,
    )
    slot_money_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("rsi_period", 3))
    print(f"✅ 데이터 로드 완료: {len(codes_candles):,}종목")

    # ── 유니버스 ──
    # 기본: target_candidates_history(MOMENTUM) 저장 이력 (웹 momentum 탭과 동일).
    # --fallback-universe: 시뮬만 사용.
    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.backtest.momentum_backtest_common import resolve_momentum_universe
            history, src, n_bins, _scan_iv, timing = resolve_momentum_universe(
                start_ymd, end_ymd, use_saved_history=True, strategy_id="MOMENTUM",
            )
            if history:
                universe_by_slot = history
                avg = sum(len(v) for v in history.values()) / max(1, n_bins)
                timing_label = "strict" if timing == "strict" else "분단위"
                print(
                    f"✅ 유니버스: 신봇 MOMENTUM 이력({timing_label}) | "
                    f"{n_bins:,}분봉 · 평균 {avg:.1f}종목"
                )
            else:
                print("ℹ️  MOMENTUM 이력 없음 → 시뮬레이션 fallback 자동 사용")
        except Exception as _e:
            logging.getLogger("param_search_momentum").debug(
                "신봇 유니버스 이력 조회 스킵: %s", _e,
            )

    if universe_by_slot is None:
        universe_top_n = int(os.environ.get("UPDATE_UNIVERSE_TOP_N", "20"))
        universe_min_score = float(os.environ.get("UPDATE_UNIVERSE_MIN_SCORE", "4.0"))
        # ★ 2026-05: 모멘텀 전용 점수 공식 사용 (Reversal 의 drop_rate 점수 → Momentum 의 강세/추세 점수)
        # ``build_universe_simulation_momentum`` 은 시가대비 상승률·고점근접·양봉비율·RSI 모멘텀
        # 기반으로 점수 매겨, 진짜 "모멘텀형" 종목만 universe 에 올림.
        universe_by_slot = me.build_universe_simulation_momentum(
            codes_candles,
            top_n=universe_top_n, min_score=universe_min_score,
            scan_interval_min=fallback_sim_interval,
        )
        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"✅ 유니버스: 모멘텀 시뮬레이션 사용 (Reversal 점수 → Momentum 점수 공식 전환) | "
              f"{fallback_sim_interval}분 슬롯 {n_slots}개 · 슬롯당 평균 {avg_per_slot:.1f}종목")
        FIXED_DEFAULTS["scan_interval_min"] = fallback_sim_interval
    else:
        FIXED_DEFAULTS["scan_interval_min"] = 1

    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] = {}
    ticks_by_code: Dict[str, Dict[str, List[Dict]]] = {}
    # kiwoom_0d 본체 재계산 모드 (6/25~ 유효) — 다음을 **모두** 만족할 때만 켠다.
    #   ① 호가필터 ON (off/1단계에서는 본체가 불필요 → 메모리·시간 절약)
    #   ② 호가필터 축 중 하나라도 **실제 스윕**(값 2개↑) → 단일 운영값이면 의미 없음
    # 단일값 스프레드(0.45 하나)만으로 본체를 끌어와 동작이 바뀌던 버그를 막는다.
    _ob_axes = ("max_spread_pct", "min_bid_ask_ratio", "ask_max_mult")
    _ob_sweeping = stage2_core_lock or any(
        len(set(grid.get(k) or [])) > 1 for k in _ob_axes
    )
    if _ob_filter_on and _ob_sweeping:
        FIXED_DEFAULTS["backtest_use_kiwoom_body_snapshot"] = True
        FIXED_DEFAULTS["_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 (
                backtest_needs_trigger_snapshot_load,
                load_trigger_snapshots_by_code,
            )
            if backtest_needs_trigger_snapshot_load(FIXED_DEFAULTS, strategy="MOMENTUM"):
                orderbook_by_code, program_by_code, trigger_snap_meta = load_trigger_snapshots_by_code(
                    _snap_db, start_key, end_key, set(codes_candles.keys()),
                    engine_params=FIXED_DEFAULTS, strategy="MOMENTUM",
                )
                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 스냅샷 없음 — 호가·프로그램 필터 스킵 (실매 수집 후 재탐색)")
            else:
                print("📌 호가·프로그램 필터 OFF → TRIGGER 스냅샷 DB 로딩 생략")
            from kis_trader.engine.momentum_tick_replay import (
                momentum_backtest_use_tick_entry,
                momentum_backtest_use_tick_exit,
            )
            if momentum_backtest_use_tick_exit(FIXED_DEFAULTS) or momentum_backtest_use_tick_entry(FIXED_DEFAULTS):
                from kis_trader.backtest.momentum_tick_loader import (
                    load_momentum_ticks_by_code,
                    tick_coverage_stats,
                )
                ticks_by_code, tick_rows = load_momentum_ticks_by_code(
                    _snap_db, start_key, end_key, set(codes_candles.keys()),
                )
                tick_meta = tick_coverage_stats(codes_candles, ticks_by_code)
                tick_meta["ws_tick_rows_loaded"] = tick_rows
                print(
                    f"✅ 모멘텀 청산 ws_ticks {tick_rows:,}건 "
                    f"(커버리지 {tick_meta.get('tick_bar_coverage_pct', 0):.1f}% · "
                    f"{tick_meta.get('tick_codes_with_data', 0)}/{tick_meta.get('tick_codes_total', 0)}종목)"
                )
                if tick_rows <= 0:
                    print("⚠️  ws_ticks 없음 — 1분봉 OHLC 청산 폴백")
        finally:
            _snap_db.close()

    # 멀티프로세싱 — start_key·end_key 둘 다 워커에 전달 (scan_at 유니버스 = 웹 백테 정합)
    _ps_start_key = (start.replace("-", "") + "0000") if start else "202601010000"
    _ps_end_key = (end.replace("-", "") + "2359") if end else "999912312359"

    # ── ws_ticks 공유메모리 — 워커별 사본(~3.3GB×N) 대신 1벌만 공유 ────────────
    # 기본 ON(검증 완료: 실측 CPU 184%→1061%·메모리 오히려↓). ON 이면 틱을 컬럼
    # (numpy)로 shared_memory 에 1벌 올리고, 워커는 이름으로 read-only attach(mmap
    # 공유) → 메모리 N배 제거 + 워커 상한 해제 → CPU 포화 회복. 엔진 핫루프는
    # TickColumnView 로 배열을 직접 읽어(속도 회복) dict 경로와 bit-identical.
    # 끄려면 MOMENTUM_PARAM_SEARCH_SHARED_TICKS=0. numpy/shm 미지원·빌드 실패 시 자동 폴백.
    shared_tick_store = None
    if get_env_bool("MOMENTUM_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)  # 크래시 시 /dev/shm 누수 방지
                print("📦 ws_ticks 공유메모리 ON — 워커 attach(read-only), 사본 제거")
                ticks_by_code = {}  # 부모 대용량 dict 해제 (틱은 shm 에 1벌 존재)
                # 해제한 힙을 OS 로 돌려줘 메모리 플래너가 가용 RAM 을 제대로 보고
                # 워커 수를 충분히(≈CPU 상한) 잡게 한다. (안 하면 잔존 힙 때문에 과소산정)
                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 실패 — 기존(디스크 pickle) 경로 폴백")
        else:
            print("⚠️  numpy/shared_memory 미지원 — 기존(디스크 pickle) 경로 폴백")

    shared = ParamSearchSharedPayload({
        "codes_candles": codes_candles,
        "universe_by_slot": universe_by_slot,
        "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,
        "ticks_by_code": ticks_by_code,
        "ticks_shared_descriptor": (shared_tick_store.descriptor() if shared_tick_store else None),
        "start_key": _ps_start_key,
        "end_key": _ps_end_key,
    })
    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(sampled, payload_bytes)
    # ── 틱재생(ws_ticks) payload 시 워커 상한 (OOM 방지) ─────────────────
    # 틱재생 기본 ON 이후, 워커별로 ws_ticks·휩쏘틱 등 사적(private) 구조가
    # payload 추정치(~수백MB)보다 훨씬 크게(관측: anon-RSS ~3.3GB/워커) 부풀어
    # 5워커 동시 실행 시 13GB 머신이 OOM-kill 로 죽는 현상 확인(2026-07-05).
    # 꼬리잡기(tail_param_search)와 동일하게 틱 존재 시 워커 수를 상한한다.
    # (하드코딩 금지 — DB/Env 로 조정, 기본 3)
    if ticks_by_code:
        # 기본 2 — 13GB 머신에서 워커당 anon-RSS ~3.3GB(틱 사적구조) 관측,
        # 3워커는 부모+payload 합산 시 OOM 경계. 여유 있는 머신은 Env 로 상향.
        tick_cap = get_env_int("MOMENTUM_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} (MOMENTUM_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS)")
    # 공유메모리 틱은 사본이 없어 OOM 위험이 없다 → 기본 상한 없음(CPU/메모리 계획대로).
    # 필요 시 Env 로 별도 상한(기본 0=무제한).
    if shared_tick_store is not None:
        shared_cap = get_env_int("MOMENTUM_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} (MOMENTUM_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_momentum_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()
        n_chunks = len(chunks)
        # 진행 바 + 경과 + ETA 한 줄 렌더 (청크 완료 시·대기 중 하트비트 시 공용).
        # last_eta_str: 청크 사이 대기 중엔 ETA 재계산이 안 되므로 직전 값을 유지해 표시.
        _state = {"last_eta": "계산 중..."}

        def _render(done: int, elapsed_so_far: float, eta_str: str, running: int = 0) -> None:
            progress = (done / max(1, n_chunks)) * 100
            bar = ParamSearchProgressETA.render_bar(done / max(1, n_chunks))
            elapsed_str = ParamSearchProgressETA.format_elapsed(elapsed_so_far)
            run_tail = f" | 진행 {running}개" if (running and done > 0) else ""
            line = (f"⏳ {bar} {progress:.1f}% ({done:,}/{n_chunks:,}) | "
                    f"경과 {elapsed_str} | 남은 {eta_str}{run_tail}")
            if use_carriage_return:
                print(f"\r{line}   ", end="", flush=True)
            else:
                print(line, flush=True)

        def _heartbeat(done: int, total: int, elapsed_so_far: float, running: int) -> None:
            # 청크 사이 텀·워밍업 동안 같은 줄을 경과 시간만 갱신해 다시 그림 (멈춘 듯 안 보이게).
            _render(done, elapsed_so_far, _state["last_eta"], running)

        for local_results in iter_pool_chunk_results(
            executor, chunks, _submit, max_workers=max_workers,
            on_heartbeat=_heartbeat,
        ):
            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)

            elapsed_so_far = time.time() - start_time
            eta_str = ParamSearchProgressETA.format_sec(
                progress_eta.remaining_sec(processed, elapsed_so_far),
            )
            _state["last_eta"] = eta_str
            _render(processed, elapsed_so_far, eta_str)
        if use_carriage_return:
            print(flush=True)

    # 워커 종료(풀 close) 후 공유메모리 즉시 해제 (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

    # 힙 → 결과 취합 (힙 프리필터는 총손익 최대 기준 — 가장 수익난 후보 top_n 보관)
    results = [heapq.heappop(global_heap)[3] for _ in range(len(global_heap))]

    # ── 위험조정 점수(순수익/MDD, Calmar 유사) ─────────────────────────────
    # PF 는 "손실 대비 이익" 비율일 뿐 낙폭(자금 최대 손실)을 못 본다.
    #   → 순익 +10만/MDD 8만 (거칠게 벌었다) 보다 순익 +8만/MDD 2만 (안정적)이
    #     실전에서 더 좋은 파라미터다. score = 순수익 ÷ max(MDD, 하한) 으로
    #     '낙폭 대비 얼마나 벌었나'(자본 효율)를 랭킹한다.
    #   MDD 하한(SCORE_MDD_FLOOR)은 0낙폭·초소액 조합이 무한대 점수로 튀는 것을
    #     막는 방어값(하드코딩 금지 규칙 준수 — DB/Env 로 조정).
    mdd_floor = get_env_float("MOMENTUM_SCORE_MDD_FLOOR", 10000.0)
    for r in results:
        _pnl = float(r.get("total_pnl", 0) or 0)
        _mdd = float(r.get("mdd", 0) or 0)
        r["score"] = round(_pnl / max(_mdd, mdd_floor), 4)

    # 정렬: 위험조정 점수(순수익/MDD) 최대 → 동점 시 순익 → 승률.
    #   음수(적자) 조합은 score 도 음수라 자연히 하위로 밀린다.
    results.sort(key=lambda r: (-r.get("score", 0.0), -r["total_pnl"], -r["win_rate"]))

    # ★ 양수 조합 별도 카운트 (양수가 없어도 아래 JSON 저장은 그대로 진행된다)
    pos_results = [r for r in results if r.get("total_pnl", 0) > 0]
    print(f"\n💰 양수 조합: {len(pos_results)}건 / 전체 top {len(results)}건")

    # ── 콘솔 출력 ──
    print(f"\n완료: {elapsed:.1f}초  |  유효 결과: {len(results):,}건")
    print(f"\n{'='*100}")
    print(f" 🏆  MOMENTUM TOP {min(top_n, len(results))} (순수익/MDD 점수 기준 — 낙폭 대비 자본효율)")
    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}  {'MDD':>10}  {'점수':>8}")
    print("-" * (len(hdr) + 72))
    for r in results[:top_n]:
        p = r["params"]
        row = "  ".join(
            (f"{str(p[k]):>{col_w}}" if isinstance(p[k], bool) 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}  {r.get('mdd', 0):>10,.0f}  "
              f"{r.get('score', 0.0):>8.3f}")

    # 거래수 다양성 통계 (사용자 의문 해소용)
    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 stage2_core_lock and stage1_baselines:
        core_keys = [k for k in keys if k not in _MOMENTUM_OB_AXES]
        print("\n📊 코어별 최고 호가필터 (stage1 OFF vs stage2 ON):")
        for i, bl in enumerate(stage1_baselines, 1):
            base_core = bl.get("params") or {}
            matching = [
                r for r in results
                if all(r["params"].get(k) == base_core.get(k) for k in core_keys if k in base_core)
            ]
            if not matching:
                print(f"   코어#{i}: stage2 유효 결과 없음")
                continue
            best = max(matching, key=lambda r: (r["total_pnl"], r["win_rate"]))
            bp = best["params"]
            wr_delta = float(best["win_rate"]) - float(bl.get("win_rate") or 0)
            print(
                f"   코어#{i}: stage1 {bl.get('total_pnl', 0):+,.0f}원/{bl.get('win_rate', 0):.1f}% "
                f"→ stage2최고 {best['total_pnl']:+,.0f}원/{best['win_rate']:.1f}% "
                f"(승률 {wr_delta:+.1f}%p) | "
                f"spread={bp.get('max_spread_pct')} ratio={bp.get('min_bid_ask_ratio')} "
                f"ask×{bp.get('ask_max_mult')}"
            )

    # ── JSON 저장 ──
    out_dir = _results_dir_for_write()
    ts      = datetime.now().strftime("%Y%m%d_%H%M%S")
    _fname_mode = f"{mode}_stage2" if stage2_core_lock else mode
    out_path = os.path.join(out_dir, f"search_momentum_{_fname_mode}_{ts}.json")
    payload = {
        "strategy": "MOMENTUM",
        "mode": mode,
        "orderbook_filter": _ob_mode,
        "start": start,
        "end": end,
        "min_win_rate": min_win_rate,
        "min_pf": min_pf,
        "slot_money": slot_money_v,
        "max_stocks": max_stocks_v,
        "total_budget_krw": total_budget_v,
        "time_start_hm": FIXED_DEFAULTS.get("time_start_hm"),
        "time_end_hm": FIXED_DEFAULTS.get("time_end_hm"),
        "grid_keys": keys,
        "grid_axis_hints": {k: MOMENTUM_GRID_AXIS_HINTS_KO[k] for k in keys if k in MOMENTUM_GRID_AXIS_HINTS_KO},
        "cartesian_product": total_combos,
        "tested_combos": sampled,
        "max_combos_cap": max_combos_cap,
        "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"],
                "score": r.get("score", 0.0),
                "bot_pct": r.get("bot_pct"),
                "daily_avg_pct": r.get("daily_avg_pct"),
                "skipped_micro_buys": r.get("skipped_micro_buys", 0),
            }
            for i, r in enumerate(results)
        ],
    }
    if stage2_core_lock_meta:
        payload.update(stage2_core_lock_meta)
    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_momentum_{_fname_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}")

    # ── DB 적용 ──
    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 (momentum_engine · MomentumStrategy 실매 동일 경로)",
    )
    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", "rr", "coarse", "fine", "full"],
                        help="탐색 모드: fast(V2·200균등샘플·~15분) / rr / coarse / fine / full")
    parser.add_argument(
        "--max-combos", type=int, default=None, dest="max_combos",
        help="백테 조합 상한 (fast 기본 env MOMENTUM_FAST_MAX_COMBOS=200, coarse 등 5000, 0=무제한)",
    )
    parser.add_argument("--top",         default=1000, type=int, help="상위 N개 출력·JSON 저장 (기본 1000)")
    parser.add_argument("--min_trades",  default=1,    type=int, help="최소 거래 건수 (기본 1 — 0건 조합만 제외)")
    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_momentum_*.json 에서만 DB 적용")
    parser.add_argument("--fallback-universe", action="store_true", dest="fallback_universe",
                        help="저장 이력 무시, 시뮬 유니버스만 사용 (기본: 저장 이력 우선)")
    parser.add_argument(
        "--orderbook-filter", default="off", choices=["off", "on", "auto"],
        dest="orderbook_filter",
        help="호가필터: off=1단계(코어만·기본) / on=2단계(kiwoom 본체 스프레드·잔량비·매도벽) / auto=DB값",
    )
    parser.add_argument(
        "--stage2-core-lock", action="store_true", dest="stage2_core_lock",
        help="2단계 정밀: stage1 Top-N 코어 고정 × 호가 3축 전수 "
             "(MOMENTUM_STAGE2_CORE_TOPN 기본 7, --stage1-json 미지정 시 최신 JSON)",
    )
    parser.add_argument(
        "--stage1-json", default=None, dest="stage1_json",
        help="2단계용 stage1 결과 JSON 경로 (미지정 시 최신 search_momentum_*.json)",
    )
    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_momentum")
        if run_lock is None:
            print(
                "⛔ 이미 실행 중인 param_search_momentum 이 있습니다.\n"
                "   ps -ef | grep param_search_momentum\n"
                "   pkill -f 'param_search_momentum.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,
            stage2_core_lock = args.stage2_core_lock,
            stage1_json     = args.stage1_json,
        )
    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()
