#!/usr/bin/env python3
"""
kis_trader/backtest/tail_param_search.py — 꼬리잡기 백테스트 파라미터 자동 탐색 (Grid Search)
==============================================================================================
tail_engine 을 직접 임포트해 run_tail_backtest 호출. 기본값은 DB(env_config) 단일 소스.
실매매·백테스트·파라서치가 동일한 env 값을 사용해 결과 예측 가능.
[V3 통합]: MA20, 피뢰침, ATR 동적 배수, 리스크 비율 등 고급 방어 로직 변수 탐색 추가.
[성능 최적화]: 멀티프로세싱(ProcessPool) 및 Heapq 기반 메모리 최적화 적용.

실행:
    cd /home/hoon/kis_bot
    python3 kis_trader/backtest/tail_param_search.py
    # 또는
    python3 -m kis_trader.backtest.tail_param_search --mode full --apply

옵션:
    --start       시작일 (기본: 오늘-7일)
    --end         종료일 (기본: 오늘)
    --mode        탐색 모드: fast / coarse / fine / full / massive (기본: fast ≈768조합·~10–20분·포트폴리오)
    --timeframe   ws_candles 분봉 (3·5·15·60, 기본 3 — backtest_web 꼬리 탭과 동일)
    --slot-money  1회투자금(원). 미지정 시 DB (웹 tl_slot)
    --max-stocks  동시보유 종목. 미지정 시 DB SHORT_MAX_STOCKS
    --total-budget 총 운용한도(원). 0/미지정 시 동시×1회투자 (웹 tl_total_budget)
    --top         상위 N개 출력·JSON 저장 (기본: 5000)
    --min_trades  최소 거래 건수 필터 (기본: 1)
    --min_win_rate 승률 하한 (기본: 40.0).
    --min_pf Profit Factor 하한 (기본: 1.0).
    --max-combos  백테 조합 상한 (기본 5000, 0=무제한·fine/full 비추)
    --apply [N]   1위(또는 N위) 결과를 DB에 적용. N 생략 시 1.
    --from-file   --apply N 과 함께 사용 시, 최근 결과 JSON에서 N번째 적용 (탐색 생략).

⚠️ 그리드 축 env 오버라이드 (콤마 구분, DB/env_config 또는 셸 export):
    TAIL_GRID_{FAST|COARSE|FINE|FULL|MASSIVE}_{축이름}
    예) TAIL_GRID_COARSE_MAX_DAILY_CHG=15,20,25,30
    예) TAIL_GRID_FAST_LIMIT_ATR_MULT=1.5,2.0

⚠️ 파라서치 진입 모드 (체결 수 확보): TAIL_PARAM_SEARCH_ENTRY_MODE=align|limit_atr
    미설정 시 DB TAIL_ENTRY_MODE 사용. limit_atr 은 거래 수가 적을 수 있음.

UPDOW(직전봉 하락·다음봉 시가 매수, holding_min_candles):
    kis_trader/backtest/param_search_updow.py

위치 이관 (2026-04 기준):
    backtest_scalping/tail_param_search.py → kis_trader/backtest/tail_param_search.py
    - ROOT = kis_bot 프로젝트 루트 (__file__ 기준 3단계 위)
    - 결과 저장: kis_trader/backtest/results/search_tail_{mode}_YYYYMMDD_HHMMSS.json
      (``search_momentum_*`` / ``search_breakout_*`` 와 동일 패턴 · 매 실행 신규 파일)
    - 쓰기 실패 시: ~/.kis_bot_search_results/ 에 동일 파일명 폴백
    - --from-file /--apply-ai: 최신 search_tail_*.json 우선 → 구 tail_search_*.json →
      레거시 tail_param_result.json (kis_trader/backtest·backtest_scalping)
"""

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

# 프로젝트 루트 경로 추가 (database, tail_engine 등 임포트용)
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)

from database import TradeDB
from kis_trader.engine import tail_engine as te
from kis_trader.backtest import tail_backtest_common as tbc
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.engine.indicator_cache import attach_indicator_caches_to_params
from kis_trader.utils.env import get_env_bool, get_env_float, get_env_from_db, get_env_int

logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger("tail_param_search")

DEFAULT_MAX_COMBOS = 5000  # coarse 그리드 전체(~3500) + fine/full 상한
FAST_MAX_COMBOS = 256      # fast 16축 → 균등샘플 (각 축 값별 marginal 스캔 · ~15–30초)
TAIL_BACKTEST_STRATEGY_ID = tbc.TAIL_STRATEGY_ID  # backtest_web 과 동일

# ─────────────────────────────────────────────────────────────────────────────
# 결과 디렉터리 (신규 위치 우선, 구 경로 fallback)
# ─────────────────────────────────────────────────────────────────────────────

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]:
    """타임스탬프 JSON 탐색 — 신규 results · 홈 폴백 · 구 backtest_scalping/results."""
    seen: Set[str] = set()
    out: List[str] = []
    for d in (
        os.path.join(HERE, "results"),
        os.path.join(os.path.expanduser("~"), ".kis_bot_search_results"),
        os.path.join(ROOT, "backtest_scalping", "results"),
    ):
        if d and d not in seen:
            seen.add(d)
            out.append(d)
    return out


def _legacy_tail_param_fixed_paths() -> List[str]:
    """구버전 고정 파일명 (--from-file 레거시 호환)."""
    return [
        os.path.join(HERE, "tail_param_result.json"),
        os.path.join(ROOT, "backtest_scalping", "tail_param_result.json"),
    ]


def _latest_tail_json_path() -> Optional[str]:
    """가장 최근 수정 시각인 search_tail_*.json 또는 tail_search_*.json 경로."""
    prefixes = ("search_tail_", "tail_search_")
    best_path: Optional[str] = None
    best_mtime = -1.0
    for d in _results_dirs_for_read():
        if not os.path.isdir(d):
            continue
        try:
            names = os.listdir(d)
        except OSError:
            continue
        for f in names:
            if not f.endswith(".json"):
                continue
            if not any(f.startswith(p) for p in prefixes):
                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 _find_tail_result_json() -> Optional[str]:
    """--from-file / --apply / apply_ai 가 읽을 JSON (최신 타임스탬프 파일 우선)."""
    ts_path = _latest_tail_json_path()
    if ts_path:
        return ts_path
    for p in _legacy_tail_param_fixed_paths():
        if os.path.isfile(p):
            return p
    return None

def _parse_csv_floats(env_key: str, fallback: List[float]) -> List[float]:
    """env/DB 콤마 구분 float 리스트 → 그리드 축."""
    raw = get_env_from_db(env_key, "")
    if not raw or str(raw).strip() in ("", "None"):
        return list(fallback)
    out: List[float] = []
    for part in str(raw).split(","):
        part = part.strip().split("#", 1)[0].strip()
        if not part:
            continue
        try:
            out.append(float(part))
        except ValueError:
            continue
    return out if out else list(fallback)


def _parse_csv_strings(env_key: str, fallback: List[str]) -> List[str]:
    """env 콤마 문자열 → 그리드 축 (entry_mode 등)."""
    raw = get_env_from_db(env_key, "")
    if raw in (None, "", "None"):
        return list(fallback)
    out: List[str] = []
    for part in str(raw).split(","):
        part = part.strip().split("#", 1)[0].strip().lower()
        if part:
            out.append(part)
    return out if out else list(fallback)


def _parse_semi_strings(env_key: str, fallback: List[str]) -> List[str]:
    """env 세미콜론(;) 구분 문자열 → 그리드 축.

    tier 프리셋처럼 값 자체에 콤마가 들어가는 경우용(예: '30000:50,100000:35').
    예) TAIL_GRID_FAST_TRAIL_TIERS='off;20000:40,50000:30;30000:50,100000:35'
    """
    raw = get_env_from_db(env_key, "")
    if raw in (None, "", "None"):
        return list(fallback)
    out: List[str] = []
    for part in str(raw).split(";"):
        part = part.strip()
        if part:
            out.append(part)
    return out if out else list(fallback)


def _parse_csv_bools(env_key: str, fallback: List[bool]) -> List[bool]:
    """env 콤마 bool 리스트 → 그리드 축 (패턴 ON/OFF 등). 'true/1/on' = True."""
    raw = get_env_from_db(env_key, "")
    if raw in (None, "", "None"):
        return list(fallback)
    out: List[bool] = []
    for part in str(raw).split(","):
        part = part.strip().split("#", 1)[0].strip().lower()
        if not part:
            continue
        out.append(part in ("1", "true", "t", "y", "yes", "on"))
    return out if out else list(fallback)


def _format_param_cell(value: Any, col_w: int) -> str:
    """그리드 표시: 문자열·bool은 그대로, 숫자는 .4g."""
    if isinstance(value, bool):
        return f"{str(value):>{col_w}}"
    if isinstance(value, str):
        return f"{value:>{col_w}}"
    return f"{value:>{col_w}.4g}"


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


def _snap_float(snap: Optional[Dict[str, Any]], key: str, default: float) -> float:
    """env_config 스냅샷 → float (그리드 끝값)."""
    if snap and key in snap and snap[key] is not None:
        raw = str(snap[key]).split("#", 1)[0].strip()
        if raw:
            try:
                return float(raw)
            except (TypeError, ValueError):
                pass
    try:
        return float(get_env_from_db(key, default) or default)
    except (TypeError, ValueError):
        return float(default)


def _tail_grids(mode: str) -> Dict[str, List[Any]]:
    """
    꼬리 파라서치 그리드 — TAIL_GRID_{MODE}_* env 로 축 오버라이드 (돌파 BREAKOUT_GRID_* 와 동일 패턴).
    fast: entry_mode(align/limit_atr) + limit_atr_mult — 틱 OFF·3분봉 OHLC (모멘텀과 동일).
    """
    m = (mode or "fast").strip().lower()
    p = f"TAIL_GRID_{m.upper()}_"

    if m == "fast":
        # 16축 — tail_vol_mult×4, atr_tp_max×4 + 청산방식(ratchet)·RSI·반전패턴 (균등샘플 cap)
        return {
            "entry_mode": _parse_csv_strings(f"{p}ENTRY_MODE", ["align", "limit_atr"]),
            "max_daily_change": _parse_csv_floats(f"{p}MAX_DAILY_CHG", [15.0, 20.0, 25.0, 30.0]),
            "min_drop_rate": _parse_csv_floats(f"{p}MIN_DROP_RATE", [0.02, 0.03]),
            "min_recovery_ratio": _parse_csv_floats(f"{p}MIN_RECOVERY_RATIO", [0.25, 0.4, 0.5]),
            "tail_ratio_min": _parse_csv_floats(f"{p}TAIL_RATIO_MIN", [1.0, 1.5]),
            "max_rec_3m": _parse_csv_floats(f"{p}MAX_REC_3M", [0.70, 0.80, 0.90]),
            "shoulder_min_high": _parse_csv_floats(f"{p}SHOULDER_MIN_HIGH", [0.003, 0.005]),
            "shoulder_cut_pct": _parse_csv_floats(f"{p}SHOULDER_CUT_PCT", [0.002, 0.003]),
            "stop_atr_mult": _parse_csv_floats(f"{p}STOP_ATR_MULT", [1.5, 2.0]),
            "target_atr_mult": _parse_csv_floats(f"{p}TARGET_ATR_MULT", [1.5, 2.0]),
            "atr_tp_max_pct": _parse_csv_floats(f"{p}ATR_TP_MAX_PCT", [1.0, 1.5, 2.0, 3.0]),
            "tail_vol_mult": _parse_csv_floats(f"{p}VOL_MULT", [0.0, 1.5, 2.0, 2.5]),
            "tail_vol_win": _parse_csv_floats(f"{p}VOL_WIN", [5.0]),
            "limit_atr_mult": _parse_csv_floats(f"{p}LIMIT_ATR_MULT", [1.0, 1.5, 2.0]),
            # ── 청산방식 탐색: ratchet_tiers ──────────────────────────────
            #   off = 단일 어깨컷(shoulder_*) 사용 / 나머지 = 다단 래칫 트레일.
            #   "off" 포함으로 shoulder_min_high·shoulder_cut_pct 축이 비로소
            #   의미를 갖는다(래칫 ON 조합에선 어깨컷은 엔진상 무시됨).
            "ratchet_tiers": _parse_csv_strings(
                f"{p}RATCHET_TIERS",
                ["off", "0.5:0.3,1.0:0.25,2.0:0.2", "1.0:0.4,2.0:0.3"],
            ),
            # ── 과매매 제어: 종목당 일일 매수횟수 · 재진입 콜다운(분) ───────
            #   실매 정합 핵심 — 현재 DB(20회·1분)는 같은 종목 연타로 PF 왜곡.
            #   파람서치가 "적게 사는 게 PF↑"를 찾으면 DB에 반영된다.
            "max_daily": _parse_csv_ints(f"{p}MAX_DAILY", [2, 3, 5]),
            "cooldown_min": _parse_csv_floats(f"{p}COOLDOWN_MIN", [3.0, 5.0, 15.0]),
            # ── 진입 캔들 등락 필터(%) — 신호봉이 이 구간 안일 때만 매수 ─────
            #   bar_chg_min: 하한(과대 급락 제외) / bar_chg_max: 상한(최소 하락폭)
            "bar_chg_min_pct": _parse_csv_floats(f"{p}BAR_CHG_MIN", [-10.0, -7.0]),
            "bar_chg_max_pct": _parse_csv_floats(f"{p}BAR_CHG_MAX", [-1.0, -1.5, -2.0]),
            # ── RSI 과열 진입 거부 기준(%) ────────────────────────────────
            "rsi_threshold": _parse_csv_floats(f"{p}RSI_THRESHOLD", [78.0, 82.0, 85.0]),
            # ── 반전 캔들 패턴 ON/OFF (hammer 는 base 고정, 추가 패턴만 탐색) ─
            #   pin: 핀바(긴 아래꼬리 단일봉, 꼬리잡기와 직결)
            #   engulfing/piercing: 2봉 반전 — 신호 다양성·표본 확대용
            "pattern_pin": _parse_csv_bools(f"{p}PATTERN_PIN", [False, True]),
            "pattern_engulfing": _parse_csv_bools(f"{p}PATTERN_ENGULFING", [False, True]),
            "pattern_piercing": _parse_csv_bools(f"{p}PATTERN_PIERCING", [False, True]),
            # ── 호가필터(2단계 조인트) 축 — 기본 단일값=동작 불변 ──────────
            #   max_spread_pct    : 스프레드 상한(%)               ↑완화
            #   min_bid_ask_ratio : 총매수/총매도 잔량비 하한(꼬리지지) ↓완화
            "max_spread_pct": _parse_csv_floats(f"{p}MAX_SPREAD_PCT", [0.45]),
            "min_bid_ask_ratio": _parse_csv_floats(f"{p}MIN_BID_ASK_RATIO", [0.85]),
            # ── 당일 누적손익 트레일 익절(%) — 0=OFF(기본·동작 불변) ─────────
            #   고점 대비 N% 되돌리면 그날 신규매수 중단(실매 daily_profit_halt 동일).
            #   일(日) 단위 이벤트라 표본=거래일 수 → 긴 기간에서만 유의미(강건 구간 검증).
            #   스윕하려면 env TAIL_GRID_FAST_TRAIL_DROP_PCT=0,30,40 + DB
            #   DAILY_PROFIT_TRAIL_ARM_KRW(발동 최소수익) 설정 필요.
            # ── 당일 누적손익 다단계 트레일(ratchet 철학) — 벌수록 타이트 ──────
            #   'off'=트레일 미사용(기준군) / '수익원:컷%,…'=다단 trail.
            #   peak 가 높은 구간에 들수록 되돌림 컷%를 좁혀 이익 보존(워런 버핏식).
            #   tier 가 켜지면 단일 trail_drop_pct·arm 은 무시됨. 일(日)단위 표본.
            #   env TAIL_GRID_FAST_TRAIL_TIERS 로 세미콜론(;) 구분 override.
            "trail_tiers": _parse_semi_strings(f"{p}TRAIL_TIERS", [
                "off",
                "5000:40,10000:30,20000:20",    # 보수: 일찍·타이트하게 사수
                "30000:50,100000:35,300000:25",   # 중도
                "50000:60,200000:45,500000:30",   # 공격: 많이 벌 때만 강하게 사수
            ]),
            # 단일 트레일(tier='off' 일 때만 의미) — 하위호환·기준 비교용 단일값.
            "trail_drop_pct": _parse_csv_floats(f"{p}TRAIL_DROP_PCT", [0.0]),
            "trail_arm_krw": _parse_csv_floats(f"{p}TRAIL_ARM_KRW", [10000.0]),
        }

    if m == "coarse":
        return {
            "max_daily_change": _parse_csv_floats(f"{p}MAX_DAILY_CHG", [15.0, 20.0, 25.0, 30.0]),
            "min_drop_rate": _parse_csv_floats(f"{p}MIN_DROP_RATE", [0.02, 0.03]),
            "min_recovery_ratio": _parse_csv_floats(f"{p}MIN_RECOVERY_RATIO", [0.4, 0.45, 0.5]),
            "tail_ratio_min": _parse_csv_floats(f"{p}TAIL_RATIO_MIN", [1.0, 1.5]),
            "max_rec_3m": _parse_csv_floats(f"{p}MAX_REC_3M", [0.85, 0.9]),
            "shoulder_min_high": _parse_csv_floats(f"{p}SHOULDER_MIN_HIGH", [0.002, 0.003, 0.005]),
            "shoulder_cut_pct": _parse_csv_floats(f"{p}SHOULDER_CUT_PCT", [0.002, 0.003]),
            "stop_atr_mult": _parse_csv_floats(f"{p}STOP_ATR_MULT", [1.5, 2.0]),
            "target_atr_mult": _parse_csv_floats(f"{p}TARGET_ATR_MULT", [1.5, 2.0]),
            "atr_tp_max_pct": _parse_csv_floats(f"{p}ATR_TP_MAX_PCT", [0.8, 1.0, 1.2]),
            "max_loss_krw": _parse_csv_floats(f"{p}MAX_LOSS_KRW", [100000.0, 200000.0]),
            "limit_atr_mult": _parse_csv_floats(f"{p}LIMIT_ATR_MULT", [1.0, 1.5, 2.0]),
        }

    # fine / full / massive — 기존 PARAM_GRIDS 폴백 + env 축만 덮어쓰기
    base = dict(PARAM_GRIDS.get(m) or {})
    if not base:
        return _tail_grids("fast")
    overrides = {
        "max_daily_change": _parse_csv_floats(f"{p}MAX_DAILY_CHG", base.get("max_daily_change", [20.0, 25.0])),
        "limit_atr_mult": _parse_csv_floats(f"{p}LIMIT_ATR_MULT", [1.0, 1.5, 2.0]),
    }
    for axis, env_suffix, fb_key in (
        ("min_drop_rate", "MIN_DROP_RATE", "min_drop_rate"),
        ("min_recovery_ratio", "MIN_RECOVERY_RATIO", "min_recovery_ratio"),
        ("tail_ratio_min", "TAIL_RATIO_MIN", "tail_ratio_min"),
        ("max_rec_3m", "MAX_REC_3M", "max_rec_3m"),
        ("shoulder_min_high", "SHOULDER_MIN_HIGH", "shoulder_min_high"),
        ("shoulder_cut_pct", "SHOULDER_CUT_PCT", "shoulder_cut_pct"),
        ("stop_atr_mult", "STOP_ATR_MULT", "stop_atr_mult"),
        ("target_atr_mult", "TARGET_ATR_MULT", "target_atr_mult"),
    ):
        if fb_key in base or env_suffix:
            overrides[axis] = _parse_csv_floats(
                f"{p}{env_suffix}",
                base.get(fb_key, overrides.get(axis, [0.03])),
            )
    merged = dict(base)
    merged.update(overrides)
    return merged


# ─────────────────────────────────────────────────────────────────────────────
# 파라미터 그리드 정의 (fine/full/massive 폴백 — fast/coarse 는 _tail_grids)
# ─────────────────────────────────────────────────────────────────────────────
PARAM_GRIDS = {
    # ─────────────────────────────────────────────────────────────────────
    # [FAST] ~10–15분 — 포트폴리오 ON · limit_atr_mult×3 포함 시 ~768조합 (2^8×3)
    # 진입 4축 + 청산 4축, atr_tp_max·회복률 3단계는 DB base 고정 (coarse에서 탐색)
    # ─────────────────────────────────────────────────────────────────────
    "fine": {
        "min_drop_rate":      [0.015, 0.02, 0.025, 0.03, 0.04],
        "min_recovery_ratio": [0.35, 0.4, 0.45, 0.5, 0.6],
        "tail_ratio_min":     [1.0, 1.2, 1.5, 1.8, 2.0],
        "tail_pct_min":       [0.001, 0.002, 0.003, 0.005],
        "stop_atr_mult":      [1.5, 2.0, 2.5],
        "target_atr_mult":    [1.5, 2.0, 2.5, 3.0],
        "atr_sl_min_pct":     [0.5, 0.8],
        "atr_sl_max_pct":     [0.8, 1.0, 1.2],
        "atr_tp_min_pct":     [0.3, 0.5, 0.8],
        "atr_tp_max_pct":     [0.8, 1.0, 1.2],
        "shoulder_min_high":  [0.002, 0.0025, 0.003, 0.004, 0.005],
        "shoulder_cut_pct":   [0.0015, 0.002, 0.0025, 0.003, 0.004, 0.005],
        "max_rec_3m":         [0.85, 0.9, 0.95],
        "sl_pct":             [0.015, 0.02, 0.025],
        "tp_pct":             [0.03, 0.04, 0.05],
        "rsi_threshold":      [72, 75, 78, 82],
        "max_loss_krw":       [150000, 200000],
        "min_drop_pct_for_loss_cut": [0.01, 0.015, 0.02],
    },
    "full": {
        "min_drop_rate":      [0.015, 0.02, 0.025, 0.03, 0.04, 0.05],
        "min_recovery_ratio": [0.35, 0.4, 0.5, 0.6],
        "tail_ratio_min":     [1.0, 1.2, 1.5, 1.8, 2.0],
        "tail_pct_min":       [0.001, 0.002, 0.003],
        "stop_atr_mult":      [1.5, 2.0, 2.5, 3.0],
        "target_atr_mult":    [1.5, 2.0, 2.5, 3.0],
        "atr_sl_min_pct":     [0.5, 0.8],
        "atr_sl_max_pct":     [0.8, 1.0, 1.2, 1.5],
        "atr_tp_min_pct":     [0.3, 0.5, 0.8],
        "atr_tp_max_pct":     [0.8, 1.0, 1.2, 1.5],
        "shoulder_min_high":  [0.002, 0.003, 0.004, 0.005, 0.007],
        "shoulder_cut_pct":   [0.0015, 0.002, 0.003, 0.004, 0.005],
        "max_rec_3m":         [0.85, 0.9, 0.95],
        "sl_pct":             [0.015, 0.02, 0.025, 0.03],
        "tp_pct":             [0.03, 0.04, 0.05, 0.06],
        "ma20_max_above":     [3.0, 5.0],
        "max_daily_change":   [15.0, 20.0, 25.0],
        "max_loss_krw":       [100000, 200000],
        "min_drop_pct_for_loss_cut": [0.01, 0.015, 0.02, 0.025],
    },
    "massive": {
        "min_drop_rate":      [0.015, 0.02, 0.025, 0.03, 0.04, 0.05],
        "min_recovery_ratio": [0.35, 0.4, 0.45, 0.5, 0.6],
        "tail_ratio_min":     [1.0, 1.2, 1.5, 1.8, 2.0],
        "tail_pct_min":       [0.001, 0.002, 0.003, 0.005],
        "stop_atr_mult":      [1.5, 2.0, 2.5, 3.0],
        "target_atr_mult":    [1.5, 2.0, 2.5, 3.0],
        "atr_sl_min_pct":     [0.5, 0.8, 1.0],
        "atr_sl_max_pct":     [0.8, 1.0, 1.2, 1.5],
        "atr_tp_min_pct":     [0.3, 0.5, 0.8],
        "atr_tp_max_pct":     [0.8, 1.0, 1.2, 1.5],
        "shoulder_min_high":  [0.002, 0.0025, 0.003, 0.004, 0.005, 0.007],
        "shoulder_cut_pct":   [0.0015, 0.002, 0.0025, 0.003, 0.004, 0.005],
        "max_rec_3m":         [0.85, 0.9, 0.95],
        "ma20_max_above":     [2.0, 3.0, 5.0],
        "max_daily_change":   [15.0, 20.0, 25.0],
        "sl_pct":             [0.015, 0.02, 0.025, 0.03],
        "tp_pct":             [0.03, 0.04, 0.05, 0.06],
        "risk_pct":           [0.005, 0.01, 0.02],
        "kelly_mult":         [0.15, 0.25, 0.5],
        "max_loss_krw":       [100000, 200000],
        "min_drop_pct_for_loss_cut": [0.01, 0.015, 0.02],
    },
}


TAIL_GRID_AXIS_HINTS_KO: Dict[str, str] = {
    "min_drop_rate": "당일 낙폭(시가→저점) 하한",
    "min_recovery_ratio": "저점 대비 회복률 하한 (낮을수록 매수 많음)",
    "tail_ratio_min": "망치 꼬리/몸통 비율 하한",
    "tail_pct_min": "꼬리 길이 최소 비율",
    "max_rec_3m": "3분봉 회복 위치 상한 (높을수록 매수 많음)",
    "stop_atr_mult": "ATR 손절 배수 (실제 손절=배수×ATR 후 %캡)",
    "target_atr_mult": "ATR 익절 배수 (실제 익절=배수×ATR 후 %캡)",
    "atr_sl_min_pct": "ATR 손절 % 하한",
    "atr_sl_max_pct": "ATR 손절 % 상한",
    "atr_tp_min_pct": "ATR 익절 % 하한",
    "atr_tp_max_pct": "ATR 익절 % 상한 (잡주 과대 목표 방지) — fast 그리드",
    "tail_vol_mult": "TRIGGER 거래량 — 신호봉 vol ≥ N봉평균×배수 (0=OFF)",
    "tail_vol_win": "거래량 평균 창(봉 수) — tail_vol_mult>0 일 때",
    "shoulder_min_high": "어깨 발동: 진입가 대비 최소 수익(0.003=0.3%) 후 max_price 추적",
    "shoulder_cut_pct": "어깨 폭: 갱신된 max_price 대비 되돌림(0.002=0.2%, 매도선=max×(1-cut))",
    "ratchet_tiers": "청산방식: off=단일 어깨컷 / '수익%:컷%,…'=다단 래칫 트레일",
    "trail_tiers": "당일손익 다단 트레일 'off'/'수익원:컷%,…' — 벌수록 컷% 타이트(이익 보존), 일단위 표본",
    "trail_drop_pct": "당일손익 단일 트레일(고점대비 %, 0=OFF) — tier='off' 일 때만 의미",
    "trail_arm_krw": "단일 트레일 발동 최소수익(원) — tier='off' & drop>0 일 때만 의미",
    "max_daily": "종목당 일일 매수 횟수 상한 (낮을수록 과매매↓·실매정합↑)",
    "cooldown_min": "청산 후 동일종목 재진입 콜다운(분) — 연타 방지",
    "bar_chg_min_pct": "진입 신호봉 등락 하한(%) — 과대 급락 제외",
    "bar_chg_max_pct": "진입 신호봉 등락 상한(%) — 최소 하락폭 요구",
    "rsi_threshold": "RSI 과열 진입 거부 기준(%) — 낮을수록 매수 적음",
    "pattern_pin": "핀바(긴 아래꼬리 단일봉) 진입 패턴 ON/OFF",
    "pattern_engulfing": "장악형(2봉 반전) 진입 패턴 ON/OFF",
    "pattern_piercing": "관통형(2봉 반전) 진입 패턴 ON/OFF",
    "max_loss_krw": "1회 최대 금액 손실(원, 트레일 미발동 시)",
    "limit_atr_mult": "ATR 지정가 깊이(anchor−ATR×배수) — limit_atr 전용, 클수록 체결↓",
    "entry_mode": "진입: align=다음봉 시가 / limit_atr=지정가(틱 체결 재생)",
    "max_daily_change": "급등주 필터(당일 등락률 상한, %)",
    "trail_pct": "보조 트레일 폭(고점 대비, 0=OFF). 래칫 ON이면 1단계보다 arm이 커야 효과",
    "trail_arm_pct": "보조 트레일 무장: 진입×(1+arm) 도달 후 (0=즉시)",
    "max_spread_pct": "호가 스프레드 상한(%) — kiwoom_0d 본체 재계산(6/25~)",
    "min_bid_ask_ratio": "총매수/총매도 잔량비 하한(꼬리지지) — kiwoom_0d 본체 재계산(6/25~)",
}


def evaluate_tail_param_combo(
    combo: Dict[str, Any],
    *,
    base_params: Dict[str, Any],
    candles_by_code: Dict[str, List[Dict]],
    fee_rate: float,
    sell_tax: float,
    min_trades: int,
    min_win_rate: float,
    min_pf: float,
    universe_by_slot: Optional[Dict[str, List[str]]] = None,
    slot_money: float = 3_000_000.0,
    max_stocks: int = 3,
    total_budget_krw: float = 9_000_000.0,
    period_days: int = 1,
    cache_holder: Optional[Dict[str, Any]] = None,
    ticks_by_code: Any = None,
    orderbook_by_code: Any = None,
    program_by_code: Any = None,
    log_verdict_by_code: Any = None,
) -> Optional[Dict[str, Any]]:
    """
    단일 파라미터 조합 백테 — Grid 워커·Optuna objective 공통.
    min_trades·승률·PF 게이트 미통과 시 None.
    """
    test_params = dict(base_params)
    test_params.update(combo)
    if "max_spread_pct" in combo and combo.get("max_spread_pct") is not None:
        test_params["_ob_max_spread_pct"] = float(combo["max_spread_pct"])
    if "min_bid_ask_ratio" in combo and combo.get("min_bid_ask_ratio") is not None:
        test_params["_ob_min_bid_ask_ratio"] = float(combo["min_bid_ask_ratio"])
    # 당일손익 트레일 익절 — tier(우선) 또는 단일 drop 이 켜졌을 때만 시뮬 게이트 ON.
    _tier_combo = str(combo.get("trail_tiers") or "").strip().lower()
    _tier_on = _tier_combo not in ("", "off")
    if _tier_on:
        test_params["_backtest_daily_profit_trail"] = True
        test_params.setdefault("daily_profit_mode", "trailing")
        test_params["daily_trail_tiers"] = str(combo["trail_tiers"]).strip()
    elif float(combo.get("trail_drop_pct") or 0) > 0:
        test_params["_backtest_daily_profit_trail"] = True
        test_params.setdefault("daily_profit_mode", "trailing")
        test_params["daily_trail_drop_pct"] = float(combo["trail_drop_pct"])
        if combo.get("trail_arm_krw") is not None:
            test_params["daily_trail_arm_krw"] = float(combo["trail_arm_krw"])
        else:
            _arm_env = os.environ.get("TAIL_GRID_TRAIL_ARM_KRW", "")
            if str(_arm_env).strip() != "":
                test_params["daily_trail_arm_krw"] = float(_arm_env)
    if cache_holder:
        test_params.update(cache_holder)
    if log_verdict_by_code:
        test_params["_backtest_log_verdict_by_code"] = log_verdict_by_code

    meta: Dict[str, Any] = {}
    trades = tbc.run_tail_backtest_web_aligned(
        candles_by_code, test_params, universe_by_slot,
        slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax,
        max_stocks=max_stocks, total_budget_krw=total_budget_krw,
        ticks_by_code=ticks_by_code,
        orderbook_by_code=orderbook_by_code,
        program_by_code=program_by_code,
        meta_out=meta,
    )

    stats = tbc.summarize_tail_trades(
        trades,
        total_budget_krw=total_budget_krw,
        period_days=period_days,
    )
    total_trades = stats["total_trades"]
    if total_trades < min_trades:
        return None

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

    return {
        "params": combo,
        "apply_cfg": {**base_params, **combo},
        "total_trades": total_trades,
        "win_rate": win_rate,
        "total_pnl": total_pnl,
        "wins": stats["wins"],
        "losses": stats["losses"],
        "pf": stats["pf"],
        "bot_pct": stats["bot_pct"],
        "daily_avg_pct": stats["daily_avg_pct"],
        "avg_hold_min": stats["avg_hold_min"],
        "sell_reasons": _count_tail_sell_reasons(trades),
        "skipped_micro_buys": int(
            (meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0
        ),
    }


def evaluate_param_chunk(
    param_chunk: List[Dict[str, Any]],
    base_params: Dict[str, Any],
    candles_by_code: Optional[Dict[str, List[Dict]]],
    fee_rate: float,
    sell_tax: float,
    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,
    period_days: int = 1,
) -> List[Tuple[float, Dict]]:
    """
    워커 프로세스에서 실행될 백테스트 평가 함수.
    universe_by_slot이 있으면 유니버스 히스토리(5분마다 후보)만 매수 검사.
    시각순 포트폴리오·총한도·소액매수 스킵은 tail_backtest_common 과 웹과 동일.
    """
    shared = worker_shared_get()
    ticks_preloaded = None
    orderbook_preloaded = None
    program_preloaded = None
    log_verdict_preloaded = None
    if shared:
        if candles_by_code is None:
            candles_by_code = shared.get("candles_by_code") or {}
        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")
    if candles_by_code is None:
        candles_by_code = {}
    cache_holder: Dict[str, Any] = {}
    attach_indicator_caches_to_params(cache_holder, candles_by_code)
    local_heap = []

    for combo in param_chunk:
        assert_parent_alive()
        result_pkg = evaluate_tail_param_combo(
            combo,
            base_params=base_params,
            candles_by_code=candles_by_code,
            fee_rate=fee_rate,
            sell_tax=sell_tax,
            min_trades=min_trades,
            min_win_rate=min_win_rate,
            min_pf=min_pf,
            universe_by_slot=universe_by_slot,
            slot_money=slot_money,
            max_stocks=max_stocks,
            total_budget_krw=total_budget_krw,
            period_days=period_days,
            cache_holder=cache_holder,
            ticks_by_code=ticks_preloaded,
            orderbook_by_code=orderbook_preloaded,
            program_by_code=program_preloaded,
            log_verdict_by_code=log_verdict_preloaded,
        )
        if result_pkg is None:
            continue

        win_rate = result_pkg["win_rate"]
        total_pnl = result_pkg["total_pnl"]
        if len(local_heap) < top_n:
            heapq.heappush(local_heap, (win_rate, total_pnl, id(result_pkg), result_pkg))
        else:
            heapq.heappushpop(local_heap, (win_rate, total_pnl, id(result_pkg), result_pkg))

    return local_heap


def run_search(
    start: str,
    end: str,
    mode: str,
    top_n: int,
    min_trades: int,
    min_win_rate: float,
    min_pf: float,
    sort_by: str = "pnl",
    use_fallback_universe: bool = False,
    max_combos: int = 0,
    timeframe: int = 3,
    time_start_hm: Optional[int] = None,
    time_end_hm: Optional[int] = None,
    slot_money: Optional[float] = None,
    max_stocks: Optional[int] = None,
    total_budget_krw: Optional[float] = None,
    orderbook_filter: str = "off",
) -> bool:
    """탐색 실행. 결과가 있어서 JSON 저장까지 했으면 True, 조건 만족 조합 없이 조기 return 시 False."""
    db = TradeDB()
    try:
        # 1. Base Environment Parameters 로드 (backtest_web _get_tail_defaults_for_backtest 와 동일)
        base_params = te.get_tail_defaults_from_db(db)
        if time_start_hm is not None:
            base_params["time_start_hm"] = int(time_start_hm)
        if time_end_hm is not None:
            base_params["time_end_hm"] = int(time_end_hm)

        # ── 호가필터 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":
            base_params["_orderbook_filter_enabled"] = False
        elif _ob_mode == "on":
            base_params["_orderbook_filter_enabled"] = True
        _ob_filter_on = bool(base_params.get("_orderbook_filter_enabled")) or _ob_mode == "auto"
        logger.info(
            "📌 호가필터: %s (%s)",
            _ob_mode.upper(),
            "적용" if _ob_filter_on else "스킵 — 코어 파라미터 순수 탐색",
        )

        row = db.conn.execute("SELECT * FROM env_config ORDER BY id DESC LIMIT 1").fetchone()
        r = dict(row) if row else {}
        fee_rate, sell_tax, _slot_from_fee = tbc.fee_and_slot_from_env_row(r)
        portfolio = tbc.resolve_tail_portfolio_params(
            r,
            base_params,
            slot_money=slot_money if slot_money is not None else _slot_from_fee,
            max_stocks=max_stocks,
            total_budget_krw=total_budget_krw,
        )
        slot_money = float(portfolio["slot_money"])
        max_stocks = int(portfolio["max_stocks"])
        total_budget_krw = float(portfolio["total_budget_krw"])
        tbc.merge_tail_portfolio_into_params(base_params, portfolio)
        base_params["capital"] = float(
            r.get("BACKTEST_CAPITAL") or base_params.get("capital") or 100_000_000.0
        )
        period_days = max(
            1,
            (datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days + 1,
        )

        tail_tf = int(timeframe)
        if tail_tf not in tbc.VALID_TIMEFRAMES:
            logger.error("❌ timeframe 은 3·5·15·60 중 하나여야 합니다 (backtest_web 과 동일)")
            return False

        start_key, end_key, start_ymd, end_ymd = tbc.date_keys(start, end)

        use_saved_history = not use_fallback_universe
        universe_by_slot, universe_source, universe_history_slots, scan_interval_min = (
            tbc.resolve_tail_universe(start_ymd, end_ymd, use_saved_history=use_saved_history)
        )
        if use_fallback_universe:
            print("📌 [유니버스] --fallback-universe: 저장 이력 무시 → ws_candles 전 종목")
        elif universe_source == "history":
            avg = (
                sum(len(v) for v in universe_by_slot.values()) / max(1, universe_history_slots)
                if universe_by_slot else 0
            )
            print(
                f"✅ 유니버스: SHORT 저장 이력 (웹백테 동일) | "
                f"{universe_history_slots:,}슬롯 · 평균 {avg:.1f}종목"
            )
        else:
            print("📌 [유니버스] 저장 이력 없음 → ws_candles 전 종목 (웹 폴백과 동일)")

        base_params = dict(base_params)
        base_params["scan_interval_min"] = scan_interval_min
        base_params["timeframe"] = tail_tf
        base_params.setdefault("backtest_tick_fallback_ohlc", True)
        # ── 틱재생(ws_ticks) — 기본 ON (실매 체결 정합) ──────────────────────
        # 3분봉 OHLC(open→high→low→close 가정)는 손절보다 익절을 먼저 체결하는
        # 낙관 편향을 만든다. 모멘텀·돌파와 동일하게 tail_tick_replay 기본값(ON)을
        # 그대로 따르되, params/env(TAIL_BACKTEST_USE_TICK_DB)로 오버라이드 가능.
        from kis_trader.engine.tail_tick_replay import tail_backtest_use_tick_db as _tail_use_tick
        base_params.setdefault("backtest_use_tick_db", _tail_use_tick(None))
        if base_params.get("backtest_use_tick_db"):
            logger.info("📌 틱재생(ws_ticks): ON — 실매 체결 정합 모드 (OHLC 낙관편향 제거)")

        logger.info(
            f"📅 데이터 로드: {start} ~ {end} | TF={tail_tf} | "
            f"유니버스={universe_source} | 매수시간 "
            f"{base_params.get('time_start_hm', 930):04d}-{base_params.get('time_end_hm', 1500):04d}"
        )

        rsi_period = int(base_params.get("rsi_period", 14))
        candles_by_code, total_candles, has_holding_peak = tbc.load_tail_candles_by_code(
            db, start_key, end_key, tail_tf, rsi_period=rsi_period,
        )

        if not candles_by_code:
            logger.info("❌ 백테스트할 데이터가 없습니다.")
            return False

        ticks_by_code: Dict[str, Dict[str, List[Dict]]] = {}
        tick_backtest_meta: Dict[str, Any] = {}
        from kis_trader.engine.tail_tick_replay import tail_backtest_wants_tick_replay
        from kis_trader.backtest.tail_tick_loader import load_tail_ticks_by_code, tick_coverage_stats
        _tick_probe = dict(base_params)
        _tick_probe["entry_mode"] = "align"
        if tail_backtest_wants_tick_replay(_tick_probe):
            _tick_db = TradeDB()
            try:
                ticks_by_code, tick_rows = load_tail_ticks_by_code(
                    _tick_db, start_key, end_key, set(candles_by_code.keys()),
                )
            finally:
                _tick_db.close()
        else:
            tick_rows = 0
        if tick_rows > 0:
            tick_backtest_meta = tick_coverage_stats(candles_by_code, ticks_by_code)
            tick_backtest_meta["ws_tick_rows_loaded"] = tick_rows
            cov = tick_backtest_meta.get("tick_bar_coverage_pct", 0)
            logger.info(
                "✅ ws_ticks %s건 | 3분봉 커버리지 %s%% (%s/%s종목)",
                f"{tick_rows:,}",
                cov,
                tick_backtest_meta.get("tick_codes_with_data", 0),
                tick_backtest_meta.get("tick_codes_total", 0),
            )
        elif tail_backtest_wants_tick_replay(_tick_probe):
            logger.warning("⚠️ ws_ticks 없음 — OHLC 폴백 (WS_TICK_SAVE_ENABLED 후 재탐색)")

        _pre_grid = _tail_grids(mode)
        # kiwoom_0d 본체 재계산 — 필터 ON + 호가필터 축 **실제 스윕**(값 2개↑) 일 때만.
        _ob_axes = ("max_spread_pct", "min_bid_ask_ratio")
        _ob_sweeping = any(len(set(_pre_grid.get(k) or [])) > 1 for k in _ob_axes)
        if _ob_filter_on and _ob_sweeping:
            base_params["backtest_use_kiwoom_body_snapshot"] = True
            base_params["_backtest_use_kiwoom_body"] = True
            _ob_axis_vals = {k: _pre_grid.get(k) for k in _ob_axes if len(set(_pre_grid.get(k) or [])) > 1}
            logger.info(
                "📌 호가필터 스윕 활성 → kiwoom_0d 본체 재계산 "
                f"(축 {_ob_axis_vals}, 본체 없는 날은 log_backfill 폴백)"
            )
        elif _ob_filter_on:
            logger.info("📌 호가필터 ON · 스윕 없음 → 본체 재계산 생략(판정 재생 경로)")

        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] = {}
        try:
            from kis_trader.backtest.trigger_snapshot_loader import load_trigger_snapshots_by_code
            orderbook_by_code, program_by_code, trigger_snap_meta = load_trigger_snapshots_by_code(
                db, start_key, end_key, set(candles_by_code.keys()),
                engine_params=base_params, strategy="TAIL",
            )
            log_verdict_by_code = trigger_snap_meta.get("log_verdict_by_code") or {}
            ob_rows = int(trigger_snap_meta.get("ws_orderbook_rows_loaded") or 0)
            pg_rows = int(trigger_snap_meta.get("ws_program_rows_loaded") or 0)
            lv_rows = int(trigger_snap_meta.get("log_verdict_rows") or 0)
            logger.info(
                "✅ TRIGGER 스냅샷 ws_orderbook %s건 | ws_program %s건 | log_backfill 판정 %s건",
                f"{ob_rows:,}", f"{pg_rows:,}", f"{lv_rows:,}",
            )
            if ob_rows <= 0 and pg_rows <= 0:
                logger.warning(
                    "⚠️ TRIGGER 스냅샷 없음 — 호가·프로그램 필터 스킵 (실매 수집 후 재탐색)"
                )
        except Exception as _snap_ex:
            logger.debug("trigger snapshot 로드 스킵: %s", _snap_ex)

        logger.info(
            f"📦 종목: {len(candles_by_code)}개 (ws_candles 전체) | "
            f"캔들: {total_candles:,}개 | portfolio_mode=ON"
        )
        logger.info(
            f"💼 포트폴리오: 1회 {slot_money:,.0f}원 | 동시 {max_stocks}종 | "
            f"총한도 {total_budget_krw:,.0f}원 (웹백테 동일)"
        )
        if portfolio.get("budget_warning"):
            logger.warning(f"💰 {portfolio['budget_warning']}")

        # 4. Grid 조합 생성
        grid = _tail_grids(mode)
        if not grid:
            logger.error(f"❌ 알 수 없는 모드: {mode}")
            return False

        if "entry_mode" not in grid:
            _search_entry = get_env_from_db("TAIL_PARAM_SEARCH_ENTRY_MODE", "")
            if _search_entry not in (None, "", "None"):
                base_params["entry_mode"] = str(_search_entry).strip().lower()

        if mode == "fast":
            print(
                f"📌 [fast] TAIL_GRID_FAST_* · 22축(거래량·atr_tp_max·청산방식·RSI·반전패턴·"
                f"과매매제어·봉등락) · →{FAST_MAX_COMBOS}샘플 · "
                f"ws_ticks={'ON' if tick_rows > 0 else 'OFF'}"
            )
        if mode == "coarse":
            print(
                "📌 [coarse] TAIL_GRID_COARSE_* env · max_daily_chg 포함 (기본 ~4천조합, cap 5000)"
            )
        if "entry_mode" in grid:
            print(f"📌 진입모드 그리드: {grid['entry_mode']} (조합별 align/limit_atr + 틱 재생)")
        else:
            print(
                f"📌 진입모드: {base_params.get('entry_mode', 'align')} "
                f"(TAIL_PARAM_SEARCH_ENTRY_MODE)"
            )

        keys = list(grid.keys())
        axes = list(grid.values())
        # 데카르트곱 전체를 RAM 에 펼치지 않는다 (호가 스윕 시 수천만 → OOM 방지).
        dict_combos, total_grid, max_combos_cap, dropped_by_cap = cap_combos_uniform_lazy(
            keys,
            axes,
            mode,
            strategy_env_prefix="TAIL",
            default_fast=FAST_MAX_COMBOS,
            default_other=DEFAULT_MAX_COMBOS,
            max_combos_override=max_combos,
        )

        sampled = len(dict_combos)
        logger.info(f"🔍 탐색 모드: {mode.upper()} | 그리드 조합: {total_grid:,} → 백테: {sampled:,}")
        if mode == "fast":
            logger.info(f"📌 [fast] {total_grid:,}→{max_combos_cap}균등샘플")
        if dropped_by_cap:
            logger.info(f"   (max-combos={max_combos_cap} 균등 샘플, 제외 {dropped_by_cap:,}개)")
        for k in keys:
            hint = TAIL_GRID_AXIS_HINTS_KO.get(k)
            if hint:
                logger.info(f"   [{k}] {hint}")
        logger.info(f"📌 1위 정렬 기준: {'총손익 최대 (수익 나는 조합 우선)' if sort_by == 'pnl' else '승률 최대'}")

        # ── ws_ticks 공유메모리 — 워커별 사본 대신 1벌 공유 (momentum 과 동일) ────────
        # 기본 ON(검증 완료). ON 이면 컬럼(numpy)로 shared_memory 에 1벌 올리고 워커는
        # read-only attach → 메모리 N배 제거 + 워커 상한 해제. 엔진 핫루프는 TickColumnView
        # 로 배열 직접 읽어 dict 경로와 bit-identical. 끄려면 TAIL_PARAM_SEARCH_SHARED_TICKS=0.
        # numpy/shm 미지원·빌드 실패 시 자동 폴백.
        shared_tick_store = None
        if get_env_bool("TAIL_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)
                    logger.info("📦 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:
                    logger.info("⚠️  ws_ticks 공유메모리 build 실패 — 기존 경로 폴백")
            else:
                logger.info("⚠️  numpy/shared_memory 미지원 — 기존 경로 폴백")

        # 5. 멀티프로세싱을 위한 청크 분할 (메모리 및 부하 분산)
        shared = ParamSearchSharedPayload({
            "candles_by_code": candles_by_code,
            "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),
            "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,
        })
        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)
        if ticks_by_code:
            tick_cap = get_env_int("TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS", 3)
            if tick_cap > 0:
                max_workers = min(max_workers, tick_cap)
                logger.info(
                    "📌 ws_ticks payload — 워커 상한 %d (TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS)",
                    max_workers,
                )
        # 공유메모리 틱은 사본이 없어 OOM 위험이 없다 → 기본 상한 없음(CPU/메모리 계획대로).
        if shared_tick_store is not None:
            shared_cap = get_env_int("TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_SHARED_TICKS", 0)
            if shared_cap > 0 and max_workers > shared_cap:
                max_workers = shared_cap
                logger.info(
                    "📌 ws_ticks 공유메모리 — 워커 상한 %d (TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_SHARED_TICKS)",
                    max_workers,
                )
        chunks = [dict_combos[i:i + chunk_size] for i in range(0, len(dict_combos), chunk_size)]

        logger.info(param_search_worker_budget_line(payload_bytes))
        logger.info(
            f"⚙️ 멀티프로세싱 시작 (코어: {n_cpu}, 워커: {max_workers}, CPU {_cpu_frac*100:.0f}%) | "
            f"청크: {len(chunks):,}개 (청크당 ~{chunk_size}조합)"
        )

        start_time = time.time()
        global_heap = []
        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_param_chunk, chunk, base_params, None,
                    fee_rate, sell_tax, min_trades, min_win_rate, min_pf, top_n, None,
                    slot_money, max_stocks, total_budget_krw, period_days,
                )

            logger.info(f"⏳ 청크 처리 중… (청크당 최대 {chunk_size:,}개 조합, 완료되는 대로 진행률·ETA 출력)")

            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 i, item in enumerate(local_results):
                    wr, pnl, _, result_pkg = item
                    if sort_by == "pnl":
                        unique_item = (-pnl, wr, (processed, i), result_pkg)
                    else:
                        unique_item = (wr, pnl, (processed, i), result_pkg)
                    if len(global_heap) < top_n:
                        heapq.heappush(global_heap, unique_item)
                    else:
                        heapq.heappushpop(global_heap, unique_item)

                # ETA — 워밍업 1파도 제외 누적 평균 (param_search_pool.ParamSearchProgressETA)
                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)
                eta_msg = f" | 경과: {elapsed_str} | 남은시간: {eta_str}"
                line = f"⏳ 진행률: {progress:.1f}% ({processed:,}/{len(chunks):,} 청크 완료){eta_msg}"

                if use_carriage_return:
                    print(f"\r{line}", end="", flush=True)
                else:
                    logger.info(line)

            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

        # 5. 결과 정렬 및 출력
        if not global_heap:
            logger.info("⚠️ 조건을 만족하는 조합이 없습니다. (min_trades를 낮추거나 기간을 늘려보세요)")
            print("📌 DB 미적용. 기존 설정 유지.")
            return False

        # sort_by pnl → (-pnl, wr) 최소힙 → heappop 순이 이미 [best pnl, ..., worst] 이므로 reverse 금지
        # sort_by win_rate → (wr, pnl) 최소힙 → heappop 순은 [low wr, ..., high wr] 이므로 reverse 필요
        best_results = [heapq.heappop(global_heap)[3] for _ in range(len(global_heap))]
        if sort_by == "win_rate":
            best_results.reverse()

        # sl_pct(손절%)별 상위 보장 → 한 값만 상위 독점 방지, 동점이면 손절 낮은 쪽(보수적) 1위
        if "sl_pct" in keys and best_results:
            sl_vals = sorted(set(r["params"]["sl_pct"] for r in best_results))
            per_sl = max(1, top_n // len(sl_vals))
            by_sl = {}
            for r in best_results:
                v = r["params"]["sl_pct"]
                if v not in by_sl:
                    by_sl[v] = []
                if len(by_sl[v]) < per_sl:
                    by_sl[v].append(r)
            best_results = []
            for v in sl_vals:
                best_results.extend(by_sl.get(v, []))
            best_results.sort(key=lambda r: (-r["total_pnl"], r["params"]["sl_pct"], -r["win_rate"]))
            logger.info(f"✅ 손절(sl_pct)별 상위 {per_sl}개씩 보장 → {len(best_results)}건 (동점 시 손절 낮은 쪽 1위)")

        filtered = [
            r for r in best_results
            if combo_passes_search_filters(
                win_rate=float(r.get("win_rate") or 0),
                pf=float(r.get("pf") or 0),
                min_win_rate=min_win_rate,
                min_pf=min_pf,
            )
        ]
        if filtered:
            best_results = filtered
            order_msg = "수익→승률 순" if sort_by == "pnl" else "승률→수익 순"
            logger.info(
                f"✅ 승률≥{min_win_rate}% · PF≥{min_pf} {len(best_results)}건 중 {order_msg} 상위"
            )
        else:
            logger.info(
                f"⚠️ 승률≥{min_win_rate}% · PF≥{min_pf} 없음 → 차악(상위) 적용"
            )

        # 손익 마이너스인 조합 제외 (수익 나는 것만 표시·저장)
        profitable = [r for r in best_results if r["total_pnl"] > 0]
        if profitable:
            best_results = profitable
            logger.info(f"✅ 총손익 플러스만 사용: {len(best_results)}건 (손실 조합 제외)")
        else:
            logger.info(f"⚠️ 수익 나는 조합 없음 → 손실 최소 순으로 표시")

        order_label = "수익" if sort_by == "pnl" else "승률"
        hdr_keys = [k for k in keys if k in (best_results[0]["params"] if best_results else {})]
        col_w = max((len(k) for k in hdr_keys), default=6) + 2
        sep_w = len(hdr_keys) * (col_w + 2) + 50
        print(f"\n✅ 탐색 완료! 총 소요 시간: {elapsed:.1f}초")
        print(f"\n{'='*min(sep_w, 100)}")
        print(f" 🏆 꼬리잡기 {order_label} TOP {min(top_n, len(best_results))}")
        print(f"{'='*min(sep_w, 100)}")
        hdr = "  ".join(f"{k:>{col_w}}" for k in hdr_keys)
        print(f"{hdr}  | {'손익(원)':>12}  {'수익%':>7}  {'승률':>6}  {'거래':>5}  {'PF':>5}")
        print("-" * min(sep_w + 10, 110))
        for i, res in enumerate(best_results[:top_n]):
            p = res["params"]
            row = "  ".join(_format_param_cell(p.get(k, ""), col_w) for k in hdr_keys)
            pf = res.get("pf", 0) or 0
            bot_pct = res.get("bot_pct", 0) or 0
            print(
                f"{row}  | {res['total_pnl']:>+12,.0f}  {bot_pct:>+6.2f}%  "
                f"{res['win_rate']:>5.1f}%  {res['total_trades']:>5}  {pf:>5.2f}"
            )
        best = best_results[0]
        bp = best["params"]
        print(f"""
╔══════════════════════════════════════════╗
║          🏆 1위 최적 파라미터            ║
╠══════════════════════════════════════════╣""")
        for k in hdr_keys:
            v = bp.get(k, "")
            print(f"║  {k:<28s} : {str(v):>6}  ║")
        print(f"""╠══════════════════════════════════════════╣
║  총 손익       : {best['total_pnl']:>+12,.0f} 원          ║
║  수익률(한도)  : {best.get('bot_pct', 0):>+6.2f}%                ║
║  승률          : {best['win_rate']:>6.1f}%                ║
║  총 거래       : {best['total_trades']:>5} 건              ║
║  Profit Factor : {best.get('pf', 0):>5.2f}                    ║
╚══════════════════════════════════════════╝""")
        reasons = best.get("sell_reasons") or {}
        if reasons:
            parts = ", ".join(f"{k}={v}" for k, v in sorted(reasons.items(), key=lambda x: -x[1]))
            print(f"🏅 1위 청산분포: {parts}")

        # 6. JSON 파일로 결과 저장 (매 실행 새 파일 · momentum/breakout 과 동일 정책)
        out_data = {
            "mode": mode,
            "start": start,
            "end": end,
            "timeframe": tail_tf,
            "universe_source": universe_source,
            "universe_history_slots": universe_history_slots,
            "slot_money": int(slot_money),
            "max_stocks": max_stocks,
            "total_budget_krw": int(total_budget_krw),
            "portfolio_mode": True,
            "budget_warning": portfolio.get("budget_warning"),
            "backtest_days": period_days,
            "web_backtest_aligned": True,
            "min_trades": min_trades,
            "min_win_rate": min_win_rate,
            "min_pf": min_pf,
            "grid_keys": keys,
            "cartesian_product": total_grid,
            "sampled_backtests": sampled,
            "dropped_by_combo_cap": dropped_by_cap,
            "max_combos": max_combos_cap,
            "grid_axis_hints": {k: TAIL_GRID_AXIS_HINTS_KO[k] for k in keys if k in TAIL_GRID_AXIS_HINTS_KO},
            "holding_peak_in_candles": has_holding_peak,
            "tested_combos": sampled,
            "elapsed_sec": round(elapsed, 1),
            "results": best_results[:top_n],
        }
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        out_name = f"search_tail_{mode}_{ts}.json"
        out_dir = _results_dir_for_write()
        out_path = os.path.join(out_dir, out_name)
        try:
            with open(out_path, "w", encoding="utf-8") as f:
                json.dump(out_data, f, indent=2, ensure_ascii=False)
        except OSError:
            fb = os.path.join(os.path.expanduser("~"), ".kis_bot_search_results")
            os.makedirs(fb, exist_ok=True)
            out_path = os.path.join(fb, out_name)
            with open(out_path, "w", encoding="utf-8") as f:
                json.dump(out_data, f, indent=2, ensure_ascii=False)
            logger.warning("⚠️  results/ 쓰기 권한 없음 → 폴백 저장: %s", out_path)

        logger.info("💾 결과 저장: %s", out_path)
        return True

    except Exception as e:
        logger.error(f"❌ 탐색 중 오류 발생: {e}", exc_info=True)
        return False
    finally:
        db.close()


def _merge_tail_apply_source(
    target: Dict[str, Any],
    search_meta: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """tail JSON → apply 소스 (공통 merge 사용)."""
    return merge_param_search_apply_source(target, search_meta)


def _count_tail_sell_reasons(trades: List[Dict[str, Any]]) -> Dict[str, int]:
    """청산 사유별 건수 — tail trades 는 reason 또는 sell_reason."""
    out: Dict[str, int] = {}
    for t in trades or []:
        k = str(t.get("sell_reason") or t.get("reason") or "unknown")
        out[k] = out.get(k, 0) + 1
    return out


def _tail_params_to_env_map(p: Dict[str, Any]) -> Dict[str, str]:
    """꼬리 파라서치·웹 save_config 와 동일 — TAIL_* 키만."""
    from kis_trader.engine.tail_env_keys import params_to_tail_env_patch
    env_map = params_to_tail_env_patch(p)
    if "risk_pct" in p:
        env_map["RISK_PCT_PER_TRADE"] = str(p["risk_pct"])
    if "kelly_mult" in p:
        env_map["KELLY_MULTIPLIER"] = str(p["kelly_mult"])
    return env_map


def apply_from_json(apply_idx: int):
    """
    저장된 JSON 결과에서 N위 조합을 추출하여 env_config에 즉시 반영
    """
    out_file = _find_tail_result_json()
    if not out_file:
        logger.error(
            "❌ 결과를 찾을 수 없습니다. 후보:\n"
            " - kis_trader/backtest/results/search_tail_*.json (최신)\n"
            " - 구 형식 tail_search_*.json 또는 tail_param_result.json"
        )
        return

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

    results = data.get("results", [])
    if not results:
        logger.error("❌ JSON에 저장된 결과가 없습니다.")
        return

    if apply_idx < 1 or apply_idx > len(results):
        logger.error(f"❌ 유효하지 않은 순위입니다. (1~{len(results)} 사이 입력)")
        return

    target = results[apply_idx - 1]
    if target.get("total_pnl", 0) <= 0:
        logger.warning(f"⚠️ {apply_idx}번째 결과는 총손익 ≤ 0 (조건 미충족). DB 미적용. 기존 설정 유지.")
        print("⚠️ 해당 순위는 총손익 ≤ 0 (조건 미충족). DB 미적용. 기존 설정 유지.")
        return

    merged = _merge_tail_apply_source(target, data)
    env_map = _tail_params_to_env_map(merged)

    db = TradeDB()
    try:
        latest = db.get_latest_env()
        snap = dict(latest["snapshot"]) if latest else {}
        snap.update(env_map)

        env_id = db.insert_env_snapshot(snap)
        logger.info(f"\n🚀 [자동 반영 완료] {apply_idx}위 조합을 DB에 적용했습니다. (env_id: {env_id})")
        logger.info(f"📂 파일: {out_file}")
        logger.info(f"적용된 값: {json.dumps(env_map, indent=2, ensure_ascii=False)}")
        logger.info("실매매 봇(kis_trader/main.py TailCatchStrategy)이 다음 루프부터 이 설정을 사용하여 매매를 시작합니다.")
    finally:
        db.close()


def apply_params_to_db(p: dict):
    """
    병합된 꼬리 설정 dict → env_config 반영.
    (_merge_tail_apply_source 결과 또는 apply_from_json 과 동일 형태)
    """
    env_map = _tail_params_to_env_map(dict(p or {}))
    if not env_map:
        logger.warning("DB 적용할 env 키가 없습니다.")
        return
    db = TradeDB()
    try:
        latest = db.get_latest_env()
        snap = dict(latest["snapshot"]) if latest else {}
        snap.update(env_map)
        db.insert_env_snapshot(snap)
    finally:
        db.close()


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

    parser = argparse.ArgumentParser(description="꼬리잡기 V3 백테스트 파라미터 최적화 (Grid Search)")
    parser.add_argument("--start",      default=week_ago, help="시작일 (YYYY-MM-DD, 기본 7일)")
    parser.add_argument("--end",        default=today,    help="종료일 (YYYY-MM-DD)")
    parser.add_argument("--timeframe", "--tf", default=3, type=int, dest="timeframe",
                        help="ws_candles 분봉 3·5·15·60 (backtest_web 꼬리 탭과 동일, 기본 3)")
    add_portfolio_cli_args(parser)
    parser.add_argument("--mode",       default="fast", choices=["fast", "coarse", "fine", "full", "massive"],
                        help="탐색 모드 (fast≈768조합·포트폴리오·~10–20분, coarse≈10k)")
    parser.add_argument("--max-combos", type=int, default=-1,
                        help=f"백테 조합 상한 (기본 {DEFAULT_MAX_COMBOS}, 0=무제한·수억 조합 모드 비추)")
    parser.add_argument("--top",          default=5000,   type=int, help="메모리에 유지·JSON 저장할 상위 N개 (기본 5000)")
    parser.add_argument("--min_trades",   default=3,    type=int,
                        help="최소 거래 건수 (1건 플러스 과최적화 방지, 기본 3)")
    add_search_filter_cli_args(parser)
    parser.add_argument("--apply",        nargs="?", const=1, type=int, default=None, metavar="N",
                        help="N번째 결과를 DB에 적용 (기본 1위). --from-file 과 함께 쓰면 재탐색 없이 즉시 적용")
    parser.add_argument("--from-file",    action="store_true", help="탐색 생략하고 기존 저장된 JSON에서 적용")
    parser.add_argument("--sort-by",      default="pnl", choices=["pnl", "win_rate"],
                        help="1위 기준: pnl=총손익 최대(기본), win_rate=승률 최대")
    parser.add_argument("--fallback-universe", action="store_true", dest="fallback_universe",
                        help="저장 이력 무시, 전체 종목으로 매수 검사 (조합별 거래 수 확대·느림)")
    parser.add_argument("--use-universe-history", action="store_true", dest="use_universe_history",
                        help="SHORT 유니버스 저장 이력만 매수 (기본 동작과 동일, 명시용)")
    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("--apply-ai",     action="store_true", dest="apply_ai",
                        help="Gemini가 수익·승률 기준으로 하나 골라 DB 적용. --from-file 과 함께 쓰면 탐색 없이 최근 JSON만 사용; 그 외에는 탐색 완료 후 방금 생성된 JSON으로 적용")
    args = parser.parse_args()

    start = args.start
    end = args.end

    # 탐색 없이 최근 JSON으로만 AI 적용 (--from-file --apply-ai)
    if args.apply_ai and args.from_file:
        import param_apply_ai
        param_apply_ai.apply_ai_tail()
        return
    if args.from_file and args.apply is not None:
        apply_from_json(args.apply)
        return

    # SIGTERM 도 KeyboardInterrupt 로 전환 (systemd·운영자 kill 대응)
    def _sigterm_to_kbd(_sig, _frm):
        raise KeyboardInterrupt("SIGTERM 수신 → 워커 정리 후 종료")
    try:
        signal.signal(signal.SIGTERM, _sigterm_to_kbd)
    except Exception:
        pass

    run_lock = try_acquire_run_lock("tail_param_search")
    if run_lock is None:
        logger.error(
            "⛔ 이미 실행 중인 tail_param_search 가 있습니다.\n"
            "   ps -ef | grep tail_param_search\n"
            "   pkill -f 'tail_param_search.py'  후 재실행하세요."
        )
        sys.exit(2)

    try:
        if args.max_combos >= 0:
            max_combos = max(0, int(args.max_combos))
        elif args.mode == "fast":
            max_combos = FAST_MAX_COMBOS
        else:
            max_combos = DEFAULT_MAX_COMBOS

        # 기본: SHORT 유니버스 저장 이력 (backtest_web universe=history). 전종목은 --fallback-universe.
        use_fallback = bool(args.fallback_universe)
        if args.use_universe_history:
            use_fallback = False

        had_results = run_search(
            start, end, args.mode, args.top,
            args.min_trades, args.min_win_rate, args.min_pf,
            sort_by=args.sort_by, use_fallback_universe=use_fallback,
            max_combos=max_combos,
            timeframe=args.timeframe,
            time_start_hm=args.time_start,
            time_end_hm=args.time_end,
            slot_money=args.slot_money,
            max_stocks=args.max_stocks,
            total_budget_krw=args.total_budget,
            orderbook_filter=args.orderbook_filter,
        )
    except KeyboardInterrupt as e:
        print(f"\n⛔ {e} — 미완료 결과 없이 종료합니다.", flush=True)
        sys.exit(130)
    finally:
        run_lock.release()

    if args.apply is not None:
        if _find_tail_result_json():
            apply_from_json(args.apply)
        else:
            logger.warning("결과가 없어 DB 적용을 건너뜁니다.")

    # 탐색에서 조건 만족 조합이 있었을 때만 방금 저장된 JSON으로 AI 적용 (없으면 기존 설정 유지)
    if args.apply_ai and had_results:
        import param_apply_ai
        param_apply_ai.apply_ai_tail()
    elif args.apply_ai and not had_results:
        print("📌 이번 탐색에서 조건 만족 조합 없음 → apply_ai 스킵. DB 미적용. 기존 설정 유지.")


if __name__ == "__main__":
    main()
