kis_bot/kis_trader/backtest/backtest_portfolio_common.py

#!/usr/bin/env python3
"""
백테·파라서치 공통 — 시각순 포트폴리오(1회투자·동시보유·총한도·소액매수 스킵) 해석·요약.
꼬리·스캘핑·모멘텀·돌파·Updow 파라서치/웹이 동일 분모·지표를 쓰도록 단일 진입.
"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

# 전략별 env 키 (첫 매칭 우선)
STRATEGY_PORTFOLIO_KEYS: Dict[str, Dict[str, Tuple[str, ...]]] = {
    "TAIL": {
        "max_stocks": ("TAIL_MAX_STOCKS",),
        "total_budget": ("TAIL_TOTAL_BUDGET_KRW",),
        "slot": ("TAIL_SLOT_MONEY",),
        "per_stock_cap": ("TAIL_MAX_BUY_AMOUNT",),
        "min_invest_env": ("TAIL_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
    "SCALP": {
        "max_stocks": ("SCALP_MAX_STOCKS", "MAX_STOCKS"),
        "total_budget": ("SCALP_TOTAL_BUDGET_KRW", "SHORT_TOTAL_BUDGET_KRW"),
        "slot": ("SLOT_MONEY_DEFAULT",),
        "per_stock_cap": ("SCALP_MAX_BUY_AMOUNT", "MAX_BUY_AMOUNT_PER_STOCK"),
        "min_invest_env": ("SCALP_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
    "MOMENTUM": {
        "max_stocks": ("MOMENTUM_MAX_STOCKS", "SCALP_MAX_STOCKS", "MAX_STOCKS"),
        "total_budget": ("MOMENTUM_TOTAL_BUDGET_KRW", "SCALP_TOTAL_BUDGET_KRW", "SHORT_TOTAL_BUDGET_KRW"),
        "slot": ("MOMENTUM_SLOT_MONEY", "SLOT_MONEY_DEFAULT"),
        "per_stock_cap": ("MOMENTUM_MAX_BUY_AMOUNT", "MAX_BUY_AMOUNT_PER_STOCK"),
        "min_invest_env": ("MOMENTUM_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
    "BREAKOUT": {
        "max_stocks": ("BREAKOUT_MAX_STOCKS", "MAX_STOCKS"),
        "total_budget": ("BREAKOUT_TOTAL_BUDGET_KRW", "SHORT_TOTAL_BUDGET_KRW"),
        "slot": ("BREAKOUT_SLOT_MONEY", "SLOT_MONEY_DEFAULT"),
        "per_stock_cap": ("BREAKOUT_MAX_BUY_AMOUNT", "MAX_BUY_AMOUNT_PER_STOCK"),
        "min_invest_env": ("BREAKOUT_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
    "RANGE_BREAK": {
        "max_stocks": ("RANGE_BREAK_MAX_STOCKS", "MAX_STOCKS"),
        "total_budget": ("RANGE_BREAK_TOTAL_BUDGET_KRW", "SHORT_TOTAL_BUDGET_KRW"),
        "slot": ("RANGE_BREAK_SLOT_MONEY", "SLOT_MONEY_DEFAULT"),
        "per_stock_cap": ("RANGE_BREAK_MAX_BUY_AMOUNT", "MAX_BUY_AMOUNT_PER_STOCK"),
        "min_invest_env": ("RANGE_BREAK_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
    "UPDOW": {
        "max_stocks": ("UPDOW_MAX_STOCKS", "MAX_STOCKS"),
        "total_budget": ("UPDOW_TOTAL_BUDGET_KRW", "SHORT_TOTAL_BUDGET_KRW"),
        "slot": ("UPDOW_SLOT_MONEY", "SLOT_MONEY_DEFAULT"),
        "per_stock_cap": ("UPDOW_MAX_BUY_AMOUNT",),
        "min_invest_env": ("UPDOW_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
    "DBBAND": {
        "max_stocks": ("DBBAND_MAX_STOCKS", "MAX_STOCKS"),
        "total_budget": ("DBBAND_TOTAL_BUDGET_KRW", "SHORT_TOTAL_BUDGET_KRW"),
        "slot": ("DBBAND_SLOT_MONEY", "SLOT_MONEY_DEFAULT"),
        "per_stock_cap": ("DBBAND_MAX_BUY_AMOUNT",),
        "min_invest_env": ("DBBAND_MIN_INVEST_RATIO_OF_SLOT", "MIN_INVEST_RATIO_OF_SLOT"),
    },
}


def _pick_env(r: Dict[str, Any], keys: Tuple[str, ...], default: Any = None) -> Any:
    for k in keys:
        v = r.get(k)
        if v not in (None, "", "None"):
            return v
    return default


def fee_and_slot_from_env_row(
    row: Optional[Dict[str, Any]],
    *,
    strategy: str = "SCALP",
) -> Tuple[float, float, float]:
    """env_config 1행 → (fee_rate, sell_tax, slot_money)."""
    if not row:
        return 0.015 / 100, 0.18 / 100, 3_000_000.0
    r = dict(row)
    fee_rate = float(r.get("FEE_RATE_PCT") or 0.015) / 100
    sell_tax = float(r.get("SELL_TAX_RATE_PCT") or 0.18) / 100
    keys = STRATEGY_PORTFOLIO_KEYS.get(strategy, STRATEGY_PORTFOLIO_KEYS["SCALP"])
    slot_money = float(
        _pick_env(r, keys["slot"], 3_000_000)
    )
    return fee_rate, sell_tax, slot_money


def resolve_portfolio_params(
    env_row: Optional[Dict[str, Any]],
    base_defaults: Optional[Dict[str, Any]] = None,
    *,
    strategy: str = "SCALP",
    slot_money: Optional[float] = None,
    max_stocks: Optional[int] = None,
    total_budget_krw: Optional[float] = None,
) -> Dict[str, Any]:
    """웹·파라서치 공통 — 1회투자·동시보유·총한도. total_budget ≤0 → max_stocks×slot."""
    r = dict(env_row) if env_row else {}
    d = dict(base_defaults) if base_defaults else {}
    keys = STRATEGY_PORTFOLIO_KEYS.get(strategy, STRATEGY_PORTFOLIO_KEYS["SCALP"])

    slot = float(slot_money) if slot_money is not None else float(
        _pick_env(r, keys["slot"], d.get("slot_money") or 3_000_000)
    )
    mxs = int(max_stocks) if max_stocks is not None else int(
        _pick_env(r, keys["max_stocks"], d.get("max_stocks") or 3)
    )
    tb_raw = total_budget_krw
    if tb_raw is None:
        tb_raw = float(_pick_env(r, keys["total_budget"], d.get("total_budget_krw") or 0) or 0)
    total_budget = float(tb_raw)
    if total_budget <= 0:
        total_budget = float(mxs * slot)

    budget_warning = None
    if total_budget < mxs * slot * 0.95:
        budget_warning = (
            f"총한도 {total_budget:,.0f}원 < 동시{mxs}×1회투자 "
            f"{mxs * slot:,.0f}원 — 잔여금 소액매수·과다 회전 위험. "
            "실매 정렬: 총한도↑ 또는 동시보유↓"
        )

    per_cap = int(float(_pick_env(r, keys["per_stock_cap"], d.get("short_max_buy_amount") or 0) or 0))

    return {
        "slot_money": slot,
        "max_stocks": max(1, mxs),
        "total_budget_krw": total_budget,
        "short_max_buy_amount": per_cap,
        "portfolio_mode": True,
        "budget_warning": budget_warning,
        "strategy": strategy,
    }


def merge_portfolio_into_params(
    params: Dict[str, Any],
    portfolio: Dict[str, Any],
) -> Dict[str, Any]:
    params["slot_money"] = float(portfolio["slot_money"])
    params["max_stocks"] = int(portfolio["max_stocks"])
    params["total_budget_krw"] = float(portfolio["total_budget_krw"])
    params.setdefault("portfolio_mode", True)
    sb = int(portfolio.get("short_max_buy_amount") or 0)
    if sb > 0:
        params["short_max_buy_amount"] = sb
    return params


def merge_param_search_apply_source(
    item: Dict[str, Any],
    search_meta: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """
    파라서치 JSON rank 항목 + 파일 meta(포트폴리오·CLI) 병합.
    apply_cfg → merged_params → params 순, params(그리드)가 최우선.
    """
    src: Dict[str, Any] = {}
    for key in ("apply_cfg", "merged_params"):
        part = item.get(key)
        if isinstance(part, dict):
            src.update(part)
    params = item.get("params")
    if isinstance(params, dict):
        src.update(params)
    meta = dict(search_meta or {})
    for k in ("slot_money", "max_stocks", "total_budget_krw", "time_start_hm", "time_end_hm"):
        if meta.get(k) is not None:
            src[k] = meta[k]
    pf = meta.get("portfolio")
    if isinstance(pf, dict):
        for k in ("slot_money", "max_stocks", "total_budget_krw", "time_start_hm", "time_end_hm"):
            if pf.get(k) is not None:
                src[k] = pf[k]
    return src


def session_env_patch(strategy: str, p: Dict[str, Any]) -> Dict[str, str]:
    """매매시간 HHMM → env (전략별 TIME_*). 꼬리(SHORT/TAIL)는 TAIL_TIME_* 만."""
    patch: Dict[str, str] = {}
    ts = p.get("time_start_hm")
    te = p.get("time_end_hm")
    if ts in (None, ""):
        ts = None
    if te in (None, ""):
        te = None
    strat = (strategy or "").upper()
    if ts is not None:
        v = str(int(float(ts)))
        if strat in ("TAIL", "SHORT"):
            patch["TAIL_TIME_START"] = v
        elif strat == "SCALP":
            patch["SCALP_TIME_START"] = v
            patch["TIME_START"] = v
        elif strat == "MOMENTUM":
            patch["MOMENTUM_TIME_START"] = v
        elif strat == "BREAKOUT":
            patch["BREAKOUT_TIME_START"] = v
        elif strat == "RANGE_BREAK":
            patch["RANGE_BREAK_TIME_START"] = v
        else:
            patch["TIME_START"] = v
    if te is not None:
        v = str(int(float(te)))
        if strat in ("TAIL", "SHORT"):
            patch["TAIL_TIME_END"] = v
        elif strat == "SCALP":
            patch["SCALP_TIME_END"] = v
            patch["TIME_END"] = v
        elif strat == "MOMENTUM":
            patch["MOMENTUM_TIME_END"] = v
        elif strat == "BREAKOUT":
            patch["BREAKOUT_TIME_END"] = v
            patch["BREAKOUT_GOLDEN_END_HM"] = v
        elif strat == "RANGE_BREAK":
            patch["RANGE_BREAK_TIME_END_HM"] = v
        else:
            patch["TIME_END"] = v
    return patch


# 전략 전용 키 prefix — 쓰기 시 '남의 전략' 키 오염 방지용.
# (읽기 fallback 튜플엔 타 전략 키가 폴백으로 들어있어, 그대로 쓰면 SCALP·SHORT 등이 덮인다)
_KNOWN_STRATEGY_PREFIXES: Tuple[str, ...] = (
    "TAIL_", "SHORT_", "SCALP_", "MOMENTUM_",
    "BREAKOUT_", "RANGE_BREAK_", "UPDOW_", "DBBAND_",
)


def _is_foreign_strategy_key(key: str, own_prefix: str) -> bool:
    """key 가 '자기 전략 prefix 가 아닌' 다른 전략 전용 키인지. (무prefix 공용키는 False)."""
    for pfx in _KNOWN_STRATEGY_PREFIXES:
        if key.startswith(pfx):
            return pfx != own_prefix
    return False


def portfolio_env_patch(strategy: str, p: Dict[str, Any]) -> Dict[str, str]:
    """1회투자·동시보유·총한도 → env_config 문자열 patch (웹 save·파라서치 apply 공통).

    ※ 쓰기 시 '자기 전략 키 + 공용 무prefix 키' 만 기록한다. 읽기용 fallback 튜플에
       섞인 다른 전략 전용 키(SCALP_*, SHORT_* 등)는 제외 → 남의 전략 설정 오염 금지.
    """
    patch: Dict[str, str] = {}
    strat = (strategy or "").upper()
    keys = STRATEGY_PORTFOLIO_KEYS.get(strat)
    if not keys:
        return patch
    own_prefix = strat + "_"

    def _set_many(env_keys: Tuple[str, ...], value: str) -> None:
        for k in env_keys:
            if _is_foreign_strategy_key(k, own_prefix):
                continue  # 남의 전략 전용 키 → 건드리지 않음
            patch[k] = value

    sm = p.get("slot_money")
    if sm not in (None, ""):
        v = str(int(float(sm)))
        _set_many(keys["slot"], v)
        _set_many(keys.get("per_stock_cap", ()), v)
    mxs = p.get("max_stocks")
    if mxs not in (None, ""):
        v = str(int(float(mxs)))
        _set_many(keys["max_stocks"], v)
    tb = p.get("total_budget_krw")
    if tb not in (None, ""):
        v = str(int(float(tb)))
        _set_many(keys["total_budget"], v)
    return patch


def min_invest_ratio_of_slot(params: Dict[str, Any], strategy: str = "SCALP") -> float:
    v = params.get("min_invest_ratio_of_slot")
    if v not in (None, ""):
        return max(0.01, min(1.0, float(v)))
    keys = STRATEGY_PORTFOLIO_KEYS.get(strategy, STRATEGY_PORTFOLIO_KEYS["SCALP"])
    try:
        from kis_trader.utils.env import get_env_float
        for k in keys["min_invest_env"]:
            raw = get_env_float(k, None)
            if raw is not None:
                return max(0.01, min(1.0, float(raw)))
        return max(0.01, min(1.0, get_env_float("MIN_INVEST_RATIO_OF_SLOT", 0.90)))
    except Exception:
        return 0.90


def target_qty_and_cost(entry_price: float, slot_money: float) -> Tuple[int, float]:
    from kis_trader.utils.position_sizing import invest_qty_for_price

    qty = invest_qty_for_price(entry_price, slot_money)
    if qty < 1:
        return 0, 0.0
    return qty, qty * entry_price


def portfolio_exposure_krw(portfolio: Dict[str, Dict[str, Any]]) -> float:
    total = 0.0
    for p in portfolio.values():
        ep = float(p.get("entry_price") or p.get("buy_price") or 0)
        total += ep * int(p.get("qty") or 1)
    return total


def build_budget_warning(
    portfolio: Dict[str, Any],
    skip_stats: Optional[Dict[str, Any]] = None,
    *,
    min_invest_ratio: float = 0.9,
) -> Optional[str]:
    msg = portfolio.get("budget_warning")
    skip_stats = skip_stats or {}
    skipped_micro = int(skip_stats.get("skipped_micro_buys") or 0)
    if skipped_micro > 0:
        micro_note = f"소액매수 스킵 {skipped_micro}건 (slot {min_invest_ratio * 100:.0f}% 미만)"
        msg = f"{msg} | {micro_note}" if msg else micro_note
    return msg


def _t2dt(candle_time: str) -> datetime:
    s = str(candle_time).strip().replace("-", "").replace(":", "").replace(" ", "")
    if len(s) >= 14:
        return datetime.strptime(s[:14], "%Y%m%d%H%M%S")
    return datetime.strptime(s[:12], "%Y%m%d%H%M")


def backtest_slip_pct(params: Optional[Dict[str, Any]]) -> float:
    """백테 체결 슬리피지(편도 %) — 엄격 체결 검증(STRICT_FILL_VERIFY)이 켜졌을 때만.

    실매 정합: 실전은 항상 fill 확인 / 모의는 STRICT_FILL_VERIFY 로 확인. 백테도
    같은 토글로 묶어 OFF=순수 이론 체결(슬립0), ON=실매 호가밀림 근사.
    BACKTEST_SLIP_PCT(전 전략 공통, 편도 %) 단일 소스. params 우선 → env(DB) 폴백.
    """
    p = params or {}
    strict = p.get("strict_fill_verify")
    if strict is None:
        from kis_trader.utils.env import get_env_bool
        strict = get_env_bool("STRICT_FILL_VERIFY", False)
    if not bool(strict):
        return 0.0
    v = p.get("backtest_slip_pct")
    if v is None:
        from kis_trader.utils.env import get_env_float
        v = get_env_float("BACKTEST_SLIP_PCT", 0.0)
    return max(0.0, float(v or 0.0))


def vol_capped_qty(qty: int, candle: Optional[Dict[str, Any]], cap_pct: float) -> int:
    """진입봉 거래량×cap_pct% 로 체결 가능 주수 제한 (실매 IOC 유동성 제약 근사).

    cap_pct<=0 → 제한 없음. 거래량 정보 없으면(0) 기존 동작 유지(보수적). 반환 0=미체결.
    실매 시장가 IOC 가 호가에 있는 만큼만 체결되는 것을 봉단위로 근사 (소형주 미체결↑).
    """
    if cap_pct <= 0:
        return qty
    try:
        vol = int(float((candle or {}).get("volume") or 0))
    except (TypeError, ValueError):
        vol = 0
    if vol <= 0:
        return qty
    fillable = int(vol * cap_pct / 100.0)
    return min(qty, fillable)


def _daily_halt_cfg(p: Dict[str, Any]) -> Dict[str, Any]:
    """백테 daily_profit 시뮬용 cfg — params 우선 → env(DB) 폴백. (실매 load_* 와 동일 키)."""
    from kis_trader.utils.env import get_env_from_db, get_env_float

    def _pf(param_keys: Tuple[str, ...], key_env: str, default: float = 0.0) -> float:
        # daily 전용 키(daily_trail_*) 우선 → 파람서치 호환 키 → env 폴백.
        # 주의: 꼬리 개별포지션 'trail_arm_pct' 와 충돌 방지 위해 arm_pct 는
        #       'daily_trail_arm_pct' 전용 키만 본다(generic 키 미참조).
        for k in param_keys:
            v = p.get(k)
            if v is not None:
                return float(v or 0.0)
        return float(get_env_float(key_env, default) or 0.0)

    mode = str(
        p.get("daily_profit_mode") or get_env_from_db("DAILY_PROFIT_MODE", "fixed") or "fixed"
    ).lower()
    # 다단계 tier(우선) — 전용키 daily_trail_tiers → env. 비면 단일 drop 폴백.
    tiers = p.get("daily_trail_tiers")
    if tiers is None:
        tiers = get_env_from_db("DAILY_PROFIT_TRAIL_TIERS", "")
    return {
        "enabled": True,
        "mode": mode,
        "trail_tiers": str(tiers or "").strip(),
        "krw": _pf(("daily_profit_target_krw",), "DAILY_PROFIT_TARGET_KRW"),
        "pct": _pf(("daily_profit_target_pct",), "DAILY_PROFIT_TARGET_PCT"),
        "trail_drop_pct": _pf(
            ("daily_trail_drop_pct", "trail_drop_pct"), "DAILY_PROFIT_TRAIL_DROP_PCT",
        ),
        "trail_arm_krw": _pf(
            ("daily_trail_arm_krw", "trail_arm_krw"), "DAILY_PROFIT_TRAIL_ARM_KRW",
        ),
        "trail_arm_pct": _pf(("daily_trail_arm_pct",), "DAILY_PROFIT_TRAIL_ARM_PCT"),
    }


def apply_daily_profit_halt_sim(
    trades: List[Dict], params: Optional[Dict[str, Any]], *, budget_krw: float,
) -> List[Dict]:
    """백테: 당일 실현손익 고정/트레일 익절 → 트리거 시각 이후 '신규 진입' 거래 제거.

    실매 daily_profit_halt 의 _target_reached/_trail_reached 를 그대로 재사용(봉단위 근사).
    당일 누적 실현손익을 청산 시각순으로 그려 고점(peak)을 추적, 고정목표 도달 또는
    고점대비 trail_drop_pct% 되돌림이 처음 발생한 시각 이후의 신규 진입을 차단한다.
    (보유분 청산은 그대로 — 실매 buy_allowed 와 동일: 신규매수만 중단.)

    명시 게이트 BACKTEST_DAILY_PROFIT_TRAIL(또는 params['_backtest_daily_profit_trail'])
    OFF 면 그대로 반환 → 기존 백테 동작 100% 불변.
    """
    p = params or {}
    gate = p.get("_backtest_daily_profit_trail")
    if gate is None:
        from kis_trader.utils.env import get_env_bool
        gate = get_env_bool("BACKTEST_DAILY_PROFIT_TRAIL", False)
    if not bool(gate):
        return trades

    from kis_trader.engine.daily_profit_halt import (
        _guard_active, _target_reached, _trail_reached,
    )
    cfg = _daily_halt_cfg(p)
    if not _guard_active(cfg):
        return trades
    mode = str(cfg.get("mode") or "fixed").lower()

    by_day: Dict[str, List[Dict]] = {}
    for t in trades:
        et = str(t.get("entry_time") or t.get("buy_time") or "")
        if len(et) < 8:
            return trades  # 진입 시각 없으면 시뮬 불가 → 불변
        by_day.setdefault(et[:8], []).append(t)

    kept: List[Dict] = []
    for _day, ts in by_day.items():
        closes = sorted(ts, key=lambda x: str(x.get("exit_time") or x.get("sell_time") or ""))
        cum = 0.0
        peak = 0.0
        trigger_time: Optional[str] = None
        for t in closes:
            cum += float(t.get("pnl") or 0)
            peak = max(peak, cum)
            # 실매 _check_hit 와 동일: both 는 고정(하드캡) 먼저 → 트레일. 먼저 닿는 쪽.
            fixed_hit = mode in ("fixed", "both") and _target_reached(cum, cfg, budget_krw)
            trail_hit = mode in ("trailing", "both") and _trail_reached(cum, peak, cfg, budget_krw)
            if fixed_hit or trail_hit:
                trigger_time = str(t.get("exit_time") or t.get("sell_time") or "")
                break
        for t in ts:
            et = str(t.get("entry_time") or t.get("buy_time") or "")
            if trigger_time and et >= trigger_time:
                continue  # 트리거 후 신규매수 차단
            kept.append(t)
    return kept


def attach_scalp_trade_pnl(
    trades: List[Dict],
    *,
    fee_rate: float,
    sell_tax: float,
    slip_pct: float = 0.0,
) -> None:
    """스캘핑·모멘텀·돌파 공통 손익 (buy_time/sell_time 필드).

    slip_pct: 백테 체결 슬리피지(편도 %). 시장가 즉시 체결 가정이라 매수는 불리(+)·
        매도는 불리(-) 로 진입·청산 양측 반영. 표시용 가격은 그대로 두고 pnl 에만 반영.
        0=OFF(동작 불변).
    """
    slip = max(0.0, float(slip_pct or 0.0)) / 100.0
    for t in trades:
        if t.get("pnl") is not None and t.get("pnl") != 0:
            continue
        ep = float(t.get("buy_price") or t.get("entry") or 0)
        xp = float(t.get("sell_price") or t.get("exit") or 0)
        qty = int(t.get("qty") or 1)
        if slip > 0:
            ep = ep * (1.0 + slip)   # 매수 체결 불리 (실매 호가 밀림)
            xp = xp * (1.0 - slip)   # 매도 체결 불리
        buy_amt = ep * qty
        sell_amt = xp * qty
        t["pnl"] = round(
            sell_amt - buy_amt - buy_amt * fee_rate - sell_amt * fee_rate - sell_amt * sell_tax
        )
        bt = t.get("buy_time") or t.get("entry_time")
        st = t.get("sell_time") or t.get("exit_time")
        if bt and st:
            t["hold_min"] = round((_t2dt(st) - _t2dt(bt)).total_seconds() / 60, 1)


def summarize_trades(
    trades: List[Dict],
    *,
    total_budget_krw: float,
    period_days: int = 1,
) -> Dict[str, Any]:
    total = len(trades)
    wins = [t for t in trades if t.get("pnl", 0) > 0]
    losses = [t for t in trades if t.get("pnl", 0) <= 0]
    total_pnl = sum(t.get("pnl", 0) for t in trades)
    win_pnl = sum(t["pnl"] for t in wins)
    loss_pnl = sum(t["pnl"] for t in losses)
    win_rate = round(len(wins) / total * 100, 2) if total else 0.0
    pf = round(abs(win_pnl / loss_pnl), 2) if loss_pnl != 0 else 9999.0
    bot_pct = round(total_pnl / total_budget_krw * 100, 2) if total_budget_krw > 0 else 0.0
    days = max(1, int(period_days))
    daily_avg_pct = round(bot_pct / days, 3) if days > 0 else 0.0
    hold_vals = [t.get("hold_min") for t in trades if t.get("hold_min") is not None]
    avg_hold = round(sum(hold_vals) / len(hold_vals), 1) if hold_vals else 0.0
    return {
        "total_trades": total,
        "wins": len(wins),
        "losses": len(losses),
        "win_rate": win_rate,
        "total_pnl": int(round(total_pnl)),
        "pf": pf,
        "bot_pct": bot_pct,
        "daily_avg_pct": daily_avg_pct,
        "avg_hold_min": avg_hold,
    }


def resolve_trigger_snapshots_for_backtest(
    candles_by_code: Dict[str, Any],
    engine_params: Dict[str, Any],
    *,
    strategy: str,
    meta_out: Optional[Dict[str, Any]] = None,
    orderbook_by_code: Optional[Dict[str, Any]] = None,
    program_by_code: Optional[Dict[str, Any]] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]], Dict[str, Any]]:
    """백테·파람서치 — ws_orderbook/ws_program 로드 (미제공 시 meta_out+DB)."""
    from kis_trader.backtest.trigger_snapshot_loader import (
        backtest_use_trigger_snapshot_db,
        load_trigger_snapshots_by_code,
    )

    snap_meta: Dict[str, Any] = {}
    if not backtest_use_trigger_snapshot_db(engine_params, strategy=strategy):
        snap_meta["trigger_snapshot_skipped"] = True
        snap_meta["skip_reason"] = "filters_off"
        return orderbook_by_code, program_by_code, snap_meta

    ob = dict(orderbook_by_code or {})
    pg = dict(program_by_code or {})
    if (not ob or not pg) and meta_out is not None:
        start_key = str(meta_out.get("start_key") or "")
        end_key = str(meta_out.get("end_key") or "")
        db = meta_out.get("db")
        if db and start_key and end_key:
            ob_loaded, pg_loaded, snap_meta = load_trigger_snapshots_by_code(
                db, start_key, end_key, set(candles_by_code.keys()),
                engine_params=engine_params, strategy=strategy,
            )
            if not ob:
                ob = ob_loaded
            if not pg:
                pg = pg_loaded
    if snap_meta.get("log_verdict_by_code") and engine_params is not None:
        engine_params["_backtest_log_verdict_by_code"] = snap_meta["log_verdict_by_code"]
    return (ob or None), (pg or None), snap_meta