kis_bot/kis_trader/backtest/optuna_postprocess_topn.py

#!/usr/bin/env python3
"""
Optuna 완료 후 — gated TopN + mode_combo + 실매(참고) 호가/휩쏘/트레일 후처리.

실매 엔진·봉 정합은 건드리지 않는다. JSON·웹 표시 + apply 시 합의 트레일만.
실매 행은 과적합%에 넣지 않는다.
"""
from __future__ import annotations

import logging
import statistics
from typing import Any, Callable, Dict, List, Optional, Tuple

from kis_trader.utils.env import get_env_bool, get_env_int

logger = logging.getLogger("optuna_postprocess_topn")

EvalFn = Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]


def _krw_int(v: Any) -> int:
    try:
        x = float(v)
    except (TypeError, ValueError):
        return 0
    if x != x or abs(x) >= 1e15:
        return 0
    return int(x)


def resolve_post_top_n(default: int = 10) -> int:
    """후처리·웹 Top 표 행 수. 기본 10 (사후/학습/안정 공통)."""
    return max(1, int(get_env_int("OPTUNA_POST_TOP_N", int(default))))


def _include_mode() -> bool:
    return bool(get_env_bool("OPTUNA_POST_INCLUDE_MODE", True))


def _include_live() -> bool:
    return bool(get_env_bool("OPTUNA_POST_INCLUDE_LIVE", True))


def _include_stable() -> bool:
    return bool(get_env_bool("OPTUNA_POST_INCLUDE_STABLE", True))


def _run_ob_whipsaw_full() -> bool:
    """사후 호가8방·휩쏘 TPE.

    본 TPE에 호가축(OPTUNA_TPE_INCLUDE_ORDERBOOK)이 있으면 기본 OFF —
    성적 사기(사후필터) 방지. 강제 재실행만 OPTUNA_POST_FORCE_OB_WHIPSAW=true.
    """
    from kis_trader.backtest.optuna_tpe_common import optuna_tpe_include_orderbook
    from kis_trader.utils.env import get_env_bool

    if optuna_tpe_include_orderbook():
        return bool(get_env_bool("OPTUNA_POST_FORCE_OB_WHIPSAW", False))
    return bool(get_env_bool("OPTUNA_POST_RUN_OB_WHIPSAW", True))


def _pop_trades(result: Optional[Dict[str, Any]]) -> Tuple[Optional[Dict[str, Any]], List[Dict[str, Any]]]:
    if not isinstance(result, dict):
        return result, []
    fills = result.pop("_trades", None)
    if not isinstance(fills, list):
        fills = []
    return result, fills


def _slim_stats(st: Any) -> Dict[str, Any]:
    if not isinstance(st, dict):
        return {}
    out = dict(st)
    if "pnl" in out:
        out["pnl"] = _krw_int(out.get("pnl"))
    if "pnl_diff" in out:
        out["pnl_diff"] = _krw_int(out.get("pnl_diff"))
    return out


def _slim_axis(ax: Any) -> Dict[str, Any]:
    if not isinstance(ax, dict):
        return {"ok": False, "reason": "none", "params": {}, "recommended_stats": {}}
    rs = _slim_stats(ax.get("recommended_stats"))
    os_ = _slim_stats(ax.get("orig_stats"))
    ws_raw = ax.get("whipsaw")
    ws = None
    if isinstance(ws_raw, dict):
        ws = {
            "ok": bool(ws_raw.get("ok")),
            "reason": ws_raw.get("reason") or "",
            "n_trials": ws_raw.get("n_trials"),
            "trade_count": int(ws_raw.get("trade_count") or 0),
            "params": dict(ws_raw.get("params") or {}),
            "orig_stats": _slim_stats(ws_raw.get("orig_stats")),
            "recommended_stats": _slim_stats(ws_raw.get("recommended_stats")),
        }
    out = {
        "ok": bool(ax.get("ok")),
        "reason": ax.get("reason") or "",
        "params": dict(ax.get("params") or {}),
        "recommended_stats": rs,
        "orig_stats": os_,
        "n_trials": ax.get("n_trials"),
        "combo_id": ax.get("combo_id") or "",
        "mask": dict(ax.get("mask") or {}),
    }
    if ws is not None:
        out["whipsaw"] = ws
    return out


def _slim_ob(rec: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    if not isinstance(rec, dict):
        return {"ok": False, "reason": "none"}
    combos_in = rec.get("combos") if isinstance(rec.get("combos"), dict) else {}
    combos_out = {str(k): _slim_axis(v) for k, v in combos_in.items()}
    out = {
        "ok": bool(rec.get("ok")),
        "reason": rec.get("reason") or "",
        "trade_count": int(rec.get("trade_count") or 0),
        "fill_count": int(rec.get("fill_count") or 0),
        "params": dict(rec.get("params") or {}),
        "orig_stats": _slim_stats(rec.get("orig_stats")),
        "recommended_stats": _slim_stats(rec.get("recommended_stats")),
        "entry": _slim_axis(rec.get("entry")),
        "exit": _slim_axis(rec.get("exit")),
        "stop": _slim_axis(rec.get("stop")),
        "combos": combos_out,
    }
    # base 방 휩쏘를 앵커 top-level 에도 (구 UI 호환)
    base_ws = (combos_out.get("base") or {}).get("whipsaw")
    if isinstance(base_ws, dict):
        out["whipsaw_base"] = base_ws
    return out


def _slim_ws(rec: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    if not isinstance(rec, dict):
        return {"ok": False, "reason": "none"}
    # 호가 rec 가 아니라 휩쏘 단독 rec
    if "combos" in rec or "entry" in rec:
        return _slim_ob(rec)
    return {
        "ok": bool(rec.get("ok")),
        "reason": rec.get("reason") or "",
        "n_trials": rec.get("n_trials"),
        "trade_count": int(rec.get("trade_count") or 0),
        "params": dict(rec.get("params") or {}),
        "orig_stats": _slim_stats(rec.get("orig_stats")),
        "recommended_stats": _slim_stats(rec.get("recommended_stats")),
    }


def _slim_trail(rec: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    if not isinstance(rec, dict):
        return {"ok": False, "reason": "none", "arm_krw": 0, "anchor_krw": 0, "tiers": ""}
    return {
        "ok": bool(rec.get("ok")),
        "reason": rec.get("reason") or "",
        "prefix": rec.get("prefix"),
        "arm_krw": _krw_int(rec.get("arm_krw")),
        "anchor_krw": _krw_int(rec.get("anchor_krw")),
        "tiers": rec.get("tiers") or "",
        "mode": rec.get("mode") or "trailing",
        "enabled": bool(rec.get("enabled")),
    }


def _combo_from_row(row: Dict[str, Any]) -> Dict[str, Any]:
    p = row.get("params") or row.get("merged_params") or {}
    return dict(p) if isinstance(p, dict) else {}


def _replay_fills(evaluate_fn: Optional[EvalFn], combo: Dict[str, Any], lg: logging.Logger) -> List[Dict[str, Any]]:
    if evaluate_fn is None or not combo:
        return []
    try:
        raw = evaluate_fn(dict(combo))
        _res, fills = _pop_trades(raw)
        return fills
    except Exception as exc:
        lg.warning("⚠️ TopN 백테 재실행 실패: %s", exc)
        return []


def _trail_for_pnl(strategy: str, pnl: Any) -> Dict[str, Any]:
    from kis_trader.backtest.optuna_daily_trail_recommend import recommend_daily_trail_tiers

    v = float(pnl or 0)
    return recommend_daily_trail_tiers(
        top_pnls=[v] if v > 0 else [],
        mode_pnl=v if v > 0 else None,
        best_pnl=v if v > 0 else None,
        strategy=strategy,
    )


def _ob_for_anchor(
    *,
    strategy: str,
    out_data: Dict[str, Any],
    fills: Optional[List[Dict[str, Any]]],
    live: bool,
    lg: logging.Logger,
    n_trials: int = 0,
) -> Dict[str, Any]:
    from kis_trader.backtest.optuna_orderbook_recommend import recommend_orderbook_parameters

    hist = (
        out_data.get("universe_history_source")
        or out_data.get("_universe_history_source")
        or out_data.get("history_source")
    )
    ob_src = out_data.get("ob_source") or out_data.get("orderbook_source")
    try:
        rec = recommend_orderbook_parameters(
            strategy=strategy,
            n_trials=int(n_trials or 0),
            history_source=hist,
            ob_source=ob_src,
            log=lg,
            raw_fills=None if live else (fills or []),
            date_from=str(out_data.get("start") or "") or None,
            date_to=str(out_data.get("end") or "") or None,
        )
    except Exception as exc:
        return {"ok": False, "reason": str(exc)}
    return _slim_ob(rec)


def _ws_for_anchor(
    *,
    strategy: str,
    out_data: Dict[str, Any],
    fills: Optional[List[Dict[str, Any]]],
    live: bool,
    lg: logging.Logger,
) -> Dict[str, Any]:
    from kis_trader.backtest.optuna_whipsaw_recommend import recommend_whipsaw_parameters

    strat_u = str(strategy or "").strip().upper()
    try:
        rec = recommend_whipsaw_parameters(
            strategy=strat_u,
            n_trials=0,
            log=lg,
            raw_fills=None if live else (fills or []),
            date_from=str(out_data.get("start") or "") or None,
            date_to=str(out_data.get("end") or "") or None,
            market="US" if "US" in strat_u else "KR",
        )
    except Exception as exc:
        return {"ok": False, "reason": str(exc)}
    return _slim_ws(rec)


def _ws_from_ob_or_anchor(
    ob: Dict[str, Any],
    *,
    strategy: str,
    out_data: Dict[str, Any],
    fills: Optional[List[Dict[str, Any]]],
    live: bool,
    lg: logging.Logger,
) -> Dict[str, Any]:
    """호가 방별 휩쏘(base) 우선. 없거나 n/a 면 레거시 단독 휩쏘."""
    ws = dict((ob.get("combos") or {}).get("base", {}).get("whipsaw") or {})
    if not ws:
        ws = dict(ob.get("whipsaw_base") or {})
    if ws.get("ok") or ws.get("reason") in ("n/a_strategy", "per_combo_off", "not_enough_trades"):
        return ws if ws else {"ok": False, "reason": "none"}
    return _ws_for_anchor(
        strategy=strategy, out_data=out_data, fills=fills, live=live, lg=lg,
    )


def _median_num(vals: List[Any]) -> Optional[float]:
    nums: List[float] = []
    for v in vals:
        try:
            nums.append(float(v))
        except (TypeError, ValueError):
            continue
    if not nums:
        return None
    return float(statistics.median(nums))


def _mode_val(vals: List[Any]) -> Any:
    clean = [v for v in vals if v is not None]
    if not clean:
        return None
    try:
        return statistics.mode(clean)
    except statistics.StatisticsError:
        return clean[0]


def _consensus_axis(pool: List[Dict[str, Any]], axis: str, keys: Tuple[str, ...], bool_keys: Tuple[str, ...], int_keys: Tuple[str, ...]) -> Dict[str, Any]:
    params_list: List[Dict[str, Any]] = []
    stats_pnls: List[int] = []
    stats_cnt: List[int] = []
    stats_wrs: List[float] = []
    for a in pool:
        ob = a.get("orderbook") or {}
        nested = ob.get(axis) if isinstance(ob.get(axis), dict) else {}
        if nested.get("ok"):
            params_list.append(dict(nested.get("params") or {}))
            rs = nested.get("recommended_stats") or {}
            stats_pnls.append(_krw_int(rs.get("pnl")))
            stats_cnt.append(int(rs.get("count") or 0))
            try:
                stats_wrs.append(float(rs.get("win_rate")))
            except (TypeError, ValueError):
                pass
        elif axis == "entry" and ob.get("ok") and not nested:
            # 구 JSON: 합쳐진 params
            p = dict(ob.get("params") or {})
            if p.get("orderbook_max_spread_pct") is not None:
                params_list.append(p)
    cons: Dict[str, Any] = {}
    if params_list:
        for k in keys:
            vs = [p.get(k) for p in params_list if k in p]
            if not vs:
                continue
            if k in bool_keys:
                cons[k] = bool(_mode_val([bool(x) for x in vs]))
            elif k in int_keys:
                m = _median_num(vs)
                cons[k] = int(m) if m is not None else vs[0]
            else:
                m = _median_num(vs)
                cons[k] = round(m, 4) if m is not None else vs[0]
    rec_st: Dict[str, Any] = {}
    if stats_pnls:
        rec_st = {
            "count": int(statistics.median(stats_cnt)) if stats_cnt else 0,
            "pnl": _krw_int(statistics.median(stats_pnls)),
        }
        if stats_wrs:
            rec_st["win_rate"] = round(float(statistics.median(stats_wrs)), 1)
    return {"ok": bool(cons), "params": cons, "n": len(params_list), "recommended_stats": rec_st}


_COMBO_LABELS: Dict[str, str] = {
    # 「진입」=매수호가축. TRIGGER 진입방어(USE_DEFENSE)와 헷갈리지 않게 「매수호가」로 표기.
    "base": "000 타점만",
    "e": "100 매수호가만",
    "x": "010 익절호가만",
    "s": "001 손절호가만",
    "ex": "110 매수+익절호가",
    "es": "101 매수+손절호가",
    "xs": "011 익절+손절호가",
    "exs": "111 호가전부",
}
# 호가 8방 = 매수×익절×손절 (2³). 휩쏘는 8방 밖.
_COMBO_IDS = ("base", "e", "x", "s", "ex", "es", "xs", "exs")
_COMBO_MASK = {
    "base": (False, False, False),
    "e": (True, False, False),
    "x": (False, True, False),
    "s": (False, False, True),
    "ex": (True, True, False),
    "es": (True, False, True),
    "xs": (False, True, True),
    "exs": (True, True, True),
}


def _median_params(params_list: List[Dict[str, Any]]) -> Dict[str, Any]:
    if not params_list:
        return {}
    all_keys: set = set()
    for p in params_list:
        all_keys.update(p.keys())
    bool_keys = {
        "orderbook_filter_enabled", "exit_ob_enabled", "stop_ob_enabled",
        "whipsaw_filter_enabled", "entry_on", "exit_on", "stop_on",
    }
    int_keys = {
        "exit_ob_min_hold_bars", "exit_ob_ma_window",
        "stop_ob_min_hold_bars", "stop_ob_ma_window",
        "whipsaw_subbar_sec", "whipsaw_lookback_sec",
    }
    cons: Dict[str, Any] = {}
    for k in all_keys:
        vs = [p[k] for p in params_list if k in p and p[k] is not None]
        if not vs:
            continue
        if k in bool_keys or all(isinstance(v, bool) for v in vs):
            cons[k] = sum(1 for v in vs if v) >= (len(vs) / 2.0)
        elif k in int_keys or all(isinstance(v, int) and not isinstance(v, bool) for v in vs):
            m = _median_num(vs)
            cons[k] = int(m) if m is not None else vs[0]
        else:
            m = _median_num(vs)
            cons[k] = round(m, 4) if m is not None else vs[0]
    return cons


def _ob_8way_attempt_note(pool: List[Dict[str, Any]]) -> Tuple[bool, str]:
    """8방 TPE를 돌렸는지(구JSON 아님)와 실패 사유.

    Returns:
        (attempted, why) — attempted=True 이면 재실행만으로 8방이 안 채워질 수 있음.
    """
    bits: List[str] = []
    saw_combos = False
    for a in pool or []:
        ob = a.get("orderbook") or {}
        combos = ob.get("combos")
        if isinstance(combos, dict) and combos:
            saw_combos = True
        r = str(ob.get("reason") or "").strip()
        if r and r not in ("light_skip", "none"):
            tc = ob.get("trade_count")
            if r == "not_enough_trades" and tc is not None:
                bit = "호가스냅%s건(<3)" % int(tc)
            else:
                bit = r
            if bit not in bits:
                bits.append(bit)
    return (saw_combos or bool(bits), " · ".join(bits))


def ob_8way_web_hint(topn: Optional[Dict[str, Any]]) -> str:
    """웹 상태 한 줄. 8방이 채워졌으면 빈 문자열."""
    anchors = list((topn or {}).get("postprocess_by_anchor") or [])
    pool = [a for a in anchors if str(a.get("role") or "") in ("gated", "mode")]
    if not pool:
        return "8방 없음"
    if any(
        isinstance((a.get("orderbook") or {}).get("combos"), dict)
        and (a.get("orderbook") or {}).get("combos")
        for a in pool
    ):
        return ""
    attempted, why = _ob_8way_attempt_note(pool)
    if attempted:
        return why or "8방 미산출"
    return "구JSON"


def _anchor_combos_map(ob: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
    """앵커 orderbook → combo_id dict. 구 JSON은 e/x/s 단독만 복원."""
    if not isinstance(ob, dict):
        return {}
    raw = ob.get("combos")
    if isinstance(raw, dict) and raw:
        return {str(k): dict(v) for k, v in raw.items() if isinstance(v, dict)}
    out: Dict[str, Dict[str, Any]] = {}
    for cid, nest_key in (("e", "entry"), ("x", "exit"), ("s", "stop")):
        nested = ob.get(nest_key)
        if isinstance(nested, dict) and nested.get("ok"):
            out[cid] = nested
    if not out and ob.get("ok") and ob.get("params"):
        out["e"] = {"ok": True, "params": dict(ob.get("params") or {}), "recommended_stats": ob.get("recommended_stats") or {}}
    return out


def _combo_median_stats(recs: List[Dict[str, Any]]) -> Dict[str, Any]:
    stats_pnls: List[int] = []
    stats_cnt: List[int] = []
    stats_wrs: List[float] = []
    for r in recs:
        rs = r.get("recommended_stats") or {}
        stats_pnls.append(_krw_int(rs.get("pnl")))
        stats_cnt.append(int(rs.get("count") or 0))
        try:
            stats_wrs.append(float(rs.get("win_rate")))
        except (TypeError, ValueError):
            pass
    if not stats_pnls:
        return {}
    out: Dict[str, Any] = {
        "count": int(statistics.median(stats_cnt)) if stats_cnt else 0,
        "pnl": _krw_int(statistics.median(stats_pnls)),
    }
    if stats_wrs:
        out["win_rate"] = round(float(statistics.median(stats_wrs)), 1)
    return out


def _axis_slice_from_combo(
    combo_params: Dict[str, Any],
    *,
    use: bool,
    axis: str,
    n: int,
    rec_st: Dict[str, Any],
) -> Dict[str, Any]:
    if not use:
        return {"ok": False, "params": {}, "recommended_stats": {}, "n": 0}
    p = dict(combo_params or {})
    if axis == "entry":
        keys = ("orderbook_filter_enabled", "orderbook_max_spread_pct", "orderbook_min_bid_ask_ratio", "orderbook_entry_ask_max_mult")
    elif axis == "exit":
        keys = ("exit_ob_enabled", "exit_ob_min_hold_bars", "exit_ob_min_profit_pct", "exit_ob_ratio_min", "exit_ob_ma_window")
    else:
        keys = ("stop_ob_enabled", "stop_ob_min_hold_bars", "stop_ob_min_loss_pct", "stop_ob_ratio_min", "stop_ob_ma_window")
    sub = {k: p[k] for k in keys if k in p}
    ok = bool(sub) and (axis != "entry" or sub.get("orderbook_max_spread_pct") is not None)
    return {"ok": ok, "params": sub, "recommended_stats": dict(rec_st), "n": n}


def _consensus_from_anchors(anchors: List[Dict[str, Any]], strategy: str) -> Dict[str, Any]:
    """gated(+learn 폴백)+mode. live·stable 제외. 호가=8방 median PnL 최고 방."""
    # 사후합격 없으면 학습 Top 후처리(learn)를 합의에 넣음 — WR/PF 때문에 호가방 학습이 비지 않게
    roles = ("gated", "mode")
    if not any(str(a.get("role") or "") == "gated" for a in (anchors or [])):
        roles = ("gated", "learn", "mode")
    pool = [a for a in anchors if str(a.get("role") or "") in roles]
    by_combo: Dict[str, List[Dict[str, Any]]] = {cid: [] for cid in _COMBO_IDS}
    for a in pool:
        combos = _anchor_combos_map(a.get("orderbook") or {})
        for cid in _COMBO_IDS:
            c = combos.get(cid)
            if isinstance(c, dict) and c.get("ok"):
                by_combo[cid].append(c)

    best_cid: Optional[str] = None
    best_med_pnl = -10**15
    for cid in _COMBO_IDS:
        if cid == "base":
            continue
        recs = by_combo.get(cid) or []
        if not recs:
            continue
        pnls = [_krw_int((r.get("recommended_stats") or {}).get("pnl")) for r in recs]
        med = float(statistics.median(pnls)) if pnls else -10**15
        if med > best_med_pnl:
            best_med_pnl = med
            best_cid = cid

    combo_cons: Dict[str, Any] = {"ok": False, "combo_id": "", "label": "", "params": {}, "n": 0}
    entry: Dict[str, Any] = {"ok": False, "params": {}, "n": 0}
    exit_c: Dict[str, Any] = {"ok": False, "params": {}, "n": 0}
    stop_c: Dict[str, Any] = {"ok": False, "params": {}, "n": 0}
    ob_merged: Dict[str, Any] = {}
    note = (
        "합의=Top5(+mode) 8방 중 median PnL 최고 방 + median 파라미터. 실매 참고행 제외."
        if any(str(a.get("role") or "") == "gated" for a in (anchors or []))
        else "합의=학습Top(+mode) — 사후합격(gated) 0건이라 WR/PF 미달 후보로 호가방 학습. 실매 참고행 제외."
    )

    if best_cid:
        recs = by_combo[best_cid]
        params = _median_params([dict(r.get("params") or {}) for r in recs])
        use_e, use_x, use_s = _COMBO_MASK[best_cid]
        params["orderbook_filter_enabled"] = bool(use_e)
        params["exit_ob_enabled"] = bool(use_x)
        params["stop_ob_enabled"] = bool(use_s)
        rec_st = _combo_median_stats(recs)
        combo_cons = {
            "ok": True,
            "combo_id": best_cid,
            "label": _COMBO_LABELS.get(best_cid, best_cid),
            "mask": {"entry": bool(use_e), "exit": bool(use_x), "stop": bool(use_s)},
            "params": params,
            "recommended_stats": rec_st,
            "n": len(recs),
        }
        ob_merged = dict(params)
        entry = _axis_slice_from_combo(params, use=use_e, axis="entry", n=len(recs), rec_st=rec_st)
        exit_c = _axis_slice_from_combo(params, use=use_x, axis="exit", n=len(recs), rec_st=rec_st)
        stop_c = _axis_slice_from_combo(params, use=use_s, axis="stop", n=len(recs), rec_st=rec_st)
    elif pool:
        # 구 JSON(8방 없음): 축별 median 폴백 — apply 구스크립트 호환
        entry = _consensus_axis(
            pool, "entry",
            ("orderbook_filter_enabled", "orderbook_max_spread_pct", "orderbook_min_bid_ask_ratio", "orderbook_entry_ask_max_mult"),
            ("orderbook_filter_enabled",),
            (),
        )
        exit_c = _consensus_axis(
            pool, "exit",
            ("exit_ob_enabled", "exit_ob_min_hold_bars", "exit_ob_min_profit_pct", "exit_ob_ratio_min", "exit_ob_ma_window"),
            ("exit_ob_enabled",),
            ("exit_ob_min_hold_bars", "exit_ob_ma_window"),
        )
        stop_c = _consensus_axis(
            pool, "stop",
            ("stop_ob_enabled", "stop_ob_min_hold_bars", "stop_ob_min_loss_pct", "stop_ob_ratio_min", "stop_ob_ma_window"),
            ("stop_ob_enabled",),
            ("stop_ob_min_hold_bars", "stop_ob_ma_window"),
        )
        ob_merged = dict(entry.get("params") or {})
        ob_merged.update(exit_c.get("params") or {})
        ob_merged.update(stop_c.get("params") or {})
        attempted, why = _ob_8way_attempt_note(pool)
        if attempted:
            note = (
                "합의=8방 유효 방 없음(%s). 축분리 median 폴백. "
                "호가스냅이 늘지 않으면 재실행해도 동일."
                % (why or "TPE 실패")
            )
        else:
            note = "합의=구JSON 축분리 median(8방 없음). 후처리 재실행 권장."

    ws_params_list = [
        dict((a.get("whipsaw") or {}).get("params") or {})
        for a in pool if (a.get("whipsaw") or {}).get("ok")
    ]
    trail_ok = [a.get("trail") or {} for a in pool if (a.get("trail") or {}).get("ok")]

    ws_cons: Dict[str, Any] = {}
    if ws_params_list:
        ws_cons = _median_params(ws_params_list)
        if ws_cons:
            ws_cons["whipsaw_filter_enabled"] = True

    trail_cons: Dict[str, Any] = {"ok": False}
    if trail_ok:
        arms = [_krw_int(t.get("arm_krw")) for t in trail_ok]
        med_arm = int(statistics.median(arms)) if arms else 0
        tiers_vals = [str(t.get("tiers") or "") for t in trail_ok if t.get("tiers")]
        tiers = _mode_val(tiers_vals) if tiers_vals else ""
        prefix = str(trail_ok[0].get("prefix") or "")
        trail_cons = {
            "ok": med_arm > 0 and bool(tiers),
            "prefix": prefix,
            "arm_krw": med_arm,
            "anchor_krw": _krw_int(statistics.median([_krw_int(t.get("anchor_krw")) for t in trail_ok])),
            "tiers": tiers or "",
            "mode": "trailing",
            "enabled": True,
            "note": "gated+mode 합의(median/최빈). 실매 행 제외. 타점 적용과 별도 버튼.",
        }

    return {
        "combo": combo_cons,
        "entry": entry,
        "exit": exit_c,
        "stop": stop_c,
        "orderbook": {
            "ok": bool(ob_merged),
            "params": ob_merged,
            "combo_id": combo_cons.get("combo_id") or "",
            "n": combo_cons.get("n") or entry.get("n") or 0,
        },
        "whipsaw": {"ok": bool(ws_cons), "params": ws_cons, "n": len(ws_params_list)},
        "trail": trail_cons,
        "strategy": strategy,
        "note": note,
    }


def _dispersion_points(vals: List[float]) -> Tuple[float, str]:
    if len(vals) < 2:
        return 0.0, "표본 1 이하"
    med = abs(statistics.median(vals)) or 1.0
    try:
        iqr = statistics.quantiles(vals, n=4)[2] - statistics.quantiles(vals, n=4)[0]
    except Exception:
        iqr = max(vals) - min(vals)
    rel = abs(iqr) / med
    if rel >= 0.5:
        return 8.0, f"상대IQR {rel:.2f} (제각각)"
    if rel >= 0.25:
        return 4.0, f"상대IQR {rel:.2f}"
    return 0.0, f"상대IQR {rel:.2f} (비슷)"


def _postprocess_overfit_extra(anchors: List[Dict[str, Any]]) -> Tuple[float, List[Dict[str, Any]]]:
    pool = [a for a in anchors if str(a.get("role") or "") in ("gated", "mode")]
    factors: List[Dict[str, Any]] = []
    extra = 0.0

    spreads = []
    ratios = []
    exit_ratios: List[float] = []
    stop_ratios: List[float] = []
    dips = []
    arms = []
    for a in pool:
        ob = a.get("orderbook") or {}
        combos = _anchor_combos_map(ob)
        entry_p: Dict[str, Any] = {}
        exit_p: Dict[str, Any] = {}
        stop_p: Dict[str, Any] = {}
        if combos:
            for cid, keys, dest in (
                ("e", ("orderbook_max_spread_pct", "orderbook_min_bid_ask_ratio"), entry_p),
                ("x", ("exit_ob_ratio_min",), exit_p),
                ("s", ("stop_ob_ratio_min",), stop_p),
            ):
                c = combos.get(cid) or {}
                if c.get("ok"):
                    p = dict(c.get("params") or {})
                    for k in keys:
                        if p.get(k) is not None:
                            dest[k] = p[k]
        if not entry_p:
            op = ob.get("params") or {}
            entry_p = dict((ob.get("entry") or {}).get("params") or op)
        if not exit_p:
            exit_p = dict((ob.get("exit") or {}).get("params") or {})
        if not stop_p:
            stop_p = dict((ob.get("stop") or {}).get("params") or {})
        if ob.get("ok") or combos or ((ob.get("entry") or {}).get("ok")):
            if entry_p.get("orderbook_max_spread_pct") is not None:
                spreads.append(float(entry_p["orderbook_max_spread_pct"]))
            if entry_p.get("orderbook_min_bid_ask_ratio") is not None:
                ratios.append(float(entry_p["orderbook_min_bid_ask_ratio"]))
        if exit_p.get("exit_ob_ratio_min") is not None:
            exit_ratios.append(float(exit_p["exit_ob_ratio_min"]))
        if stop_p.get("stop_ob_ratio_min") is not None:
            stop_ratios.append(float(stop_p["stop_ob_ratio_min"]))
        wp = (a.get("whipsaw") or {}).get("params") or {}
        if (a.get("whipsaw") or {}).get("ok") and wp.get("whipsaw_dip_pct") is not None:
            dips.append(float(wp["whipsaw_dip_pct"]))
        if (a.get("trail") or {}).get("ok"):
            arms.append(float((a.get("trail") or {}).get("arm_krw") or 0))

    for fid, label, seq in (
        ("ob_spread_disp", "후처리 진입 스프레드 분산", spreads),
        ("ob_ratio_disp", "후처리 진입 잔량비 분산", ratios),
        ("ob_exit_ratio_disp", "후처리 익절호가 OR 분산", exit_ratios),
        ("ob_stop_ratio_disp", "후처리 손절호가 OR 분산", stop_ratios),
        ("ws_dip_disp", "후처리 휩쏘 dip 분산", dips),
        ("trail_arm_disp", "후처리 트레일 ARM 분산", arms),
    ):
        pts, detail = _dispersion_points(seq) if seq else (0.0, "해당 후처리 없음")
        extra += pts
        factors.append({"id": fid, "label": label, "points": pts, "detail": detail})
    extra = max(0.0, min(25.0, extra))
    return extra, factors


def _verdict(risk: float) -> Tuple[str, str]:
    if risk >= 70.0:
        return "비권장", "위험 · 비권장"
    if risk >= 40.0:
        return "주의", "주의"
    return "상대적으로낮음", "상대적으로 낮음"


def _reuse_ob_ws(anchors: List[Dict[str, Any]], trial: Any) -> Optional[Dict[str, Any]]:
    if trial is None:
        return None
    try:
        tn = int(trial)
    except (TypeError, ValueError):
        return None
    for a in anchors:
        try:
            at = int(a.get("optuna_trial_number"))
        except (TypeError, ValueError):
            continue
        if at != tn:
            continue
        ob = a.get("orderbook") or {}
        if ob.get("ok") or (ob.get("entry") or {}).get("ok"):
            return a
    return None


def append_learn_postprocess_anchors(
    data: Dict[str, Any],
    anchors: List[Dict[str, Any]],
    *,
    strat: str,
    strat_u: str,
    top_n: int,
    do_ob: bool,
    evaluate_fn: Optional[EvalFn],
    lg: logging.Logger,
    ob_n_trials: int = 0,
) -> None:
    """
    사후합격(gated)이 비었을 때 학습 TopN(results)에도 호가/휩쏘 후처리.

    탐색은 WR/PF=0 이라 −PnL도 TPE에 들어가는데, 후처리만 gated(WR40/PF1)에
    묶이면 gated=0 일 때 호가 8방이 mode/live만 남고 학습 후보가 통째 제외됨.
    → 후처리도 ‘끝난 뒤 후보 고르기’이지 ‘처음부터 제외’가 아니게 learn 폴백.
    """
    if any(str(a.get("role") or "") == "gated" for a in anchors):
        return
    if any(str(a.get("role") or "") == "learn" for a in anchors):
        return
    learn = list(
        (data or {}).get("results")
        or (data or {}).get("results_all")
        or []
    )[: max(1, int(top_n or 10))]
    if not learn:
        return
    lg.info(
        "📌 [후처리] 사후합격 0건 → 학습 Top%d 에 호가/휩쏘 후처리 (WR/PF 사후게이트와 분리)",
        len(learn),
    )
    for i, row in enumerate(learn, start=1):
        if not isinstance(row, dict):
            continue
        combo = _combo_from_row(row)
        fills: List[Dict[str, Any]] = []
        if do_ob:
            from kis_trader.backtest import optuna_post_progress as opp
            opp.next_unit(f"learn#{i}", "학습Top 호가")
        if do_ob and evaluate_fn is not None:
            lg.info("📌 [후처리] learn#%d 백테 재실행 (호가/휩쏘 체결)", i)
            fills = _replay_fills(evaluate_fn, combo, lg)
        pnl = row.get("total_pnl")
        trail = _slim_trail(_trail_for_pnl(strat, pnl))
        ob = {"ok": False, "reason": "light_skip"}
        ws = {"ok": False, "reason": "light_skip"}
        if do_ob:
            if fills:
                ob = _ob_for_anchor(
                    strategy=strat_u, out_data=data, fills=fills, live=False,
                    lg=lg, n_trials=ob_n_trials,
                )
                ws = _ws_from_ob_or_anchor(
                    ob, strategy=strat_u, out_data=data, fills=fills, live=False, lg=lg,
                )
            elif evaluate_fn is None:
                ob = {"ok": False, "reason": "no_replay_fills"}
                ws = {"ok": False, "reason": "no_replay_fills"}
            else:
                ob = {"ok": False, "reason": "not_enough_trades", "trade_count": 0}
                ws = {"ok": False, "reason": "not_enough_trades", "trade_count": 0}
        anchors.append({
            "id": f"learn#{i}",
            "role": "learn",
            "rank": i,
            "optuna_trial_number": row.get("optuna_trial_number") or row.get("_trial_number"),
            "total_pnl": _krw_int(pnl),
            "total_trades": int(row.get("total_trades") or 0),
            "win_rate": row.get("win_rate"),
            "pf": row.get("pf"),
            "orderbook": ob,
            "whipsaw": ws,
            "trail": trail,
            "note": "학습 Top(사후게이트 미달 폴백) — 호가방 학습용 · DB적용은 gated 우선",
        })


def append_stable_postprocess_anchors(
    data: Dict[str, Any],
    anchors: List[Dict[str, Any]],
    *,
    strat: str,
    strat_u: str,
    top_n: int,
    do_ob: bool,
    evaluate_fn: Optional[EvalFn],
    lg: logging.Logger,
    ob_n_trials: int = 0,
) -> None:
    """results_stable TopN 을 후처리 표 앵커로 붙인다. 같은 trial 은 gated 호가 재사용."""
    if not _include_stable():
        return
    if any(str(a.get("role") or "") == "stable" for a in anchors):
        return
    from kis_trader.backtest.optuna_common import resolve_results_stable
    stable, _meta = resolve_results_stable(data, top_n=max(1, int(top_n or 10)))
    # 후처리 중 JSON에 비어 있으면 재구성분 반영 (다음 요약·앵커 일치)
    if not list((data or {}).get("results_stable") or []) and stable:
        data["results_stable"] = list(stable)
        if _meta:
            data["stable_gates"] = dict(_meta)
    for i, row in enumerate(stable, start=1):
        if not isinstance(row, dict):
            continue
        combo = _combo_from_row(row)
        trial = row.get("optuna_trial_number") or row.get("_trial_number")
        reused = _reuse_ob_ws(anchors, trial)
        fills: List[Dict[str, Any]] = []
        pnl = row.get("total_pnl")
        trail = _slim_trail(_trail_for_pnl(strat, pnl))
        if do_ob:
            from kis_trader.backtest import optuna_post_progress as opp
            opp.next_unit(
                f"stable#{i}",
                "호가재사용" if reused is not None else "백테재실행·호가",
            )
        if reused is not None:
            ob = reused.get("orderbook") or {"ok": False, "reason": "light_skip"}
            ws = reused.get("whipsaw") or {"ok": False, "reason": "light_skip"}
            if do_ob:
                lg.info("📌 [후처리] stable#%d 호가 재사용 (gated와 동일 trial)", i)
        else:
            ob = {"ok": False, "reason": "light_skip"}
            ws = {"ok": False, "reason": "light_skip"}
            if do_ob and evaluate_fn is not None:
                lg.info("📌 [후처리] stable#%d 백테 재실행 (호가/휩쏘 체결)", i)
                fills = _replay_fills(evaluate_fn, combo, lg)
            if do_ob:
                if fills:
                    ob = _ob_for_anchor(
                        strategy=strat_u, out_data=data, fills=fills, live=False, lg=lg, n_trials=ob_n_trials,
                    )
                    ws = _ws_from_ob_or_anchor(
                        ob, strategy=strat_u, out_data=data, fills=fills, live=False, lg=lg,
                    )
                elif evaluate_fn is None:
                    ob = {"ok": False, "reason": "no_replay_fills"}
                    ws = {"ok": False, "reason": "no_replay_fills"}
                else:
                    ob = {"ok": False, "reason": "not_enough_trades", "trade_count": 0}
                    ws = {"ok": False, "reason": "not_enough_trades", "trade_count": 0}
        anchors.append({
            "id": f"stable#{i}",
            "role": "stable",
            "rank": i,
            "optuna_trial_number": trial,
            "total_pnl": _krw_int(pnl),
            "total_trades": int(row.get("total_trades") or 0),
            "win_rate": row.get("win_rate"),
            "pf": row.get("pf"),
            "orderbook": ob,
            "whipsaw": ws,
            "trail": trail,
            "note": "안정 후보" + (" · gated와 동일 trial 호가 재사용" if reused is not None else ""),
        })


def ensure_stable_postprocess_on_payload(data: Dict[str, Any]) -> None:
    """구 JSON(gated만 있는 후처리)에도 안정 앵커를 붙여 웹 표가 나오게."""
    topn = (data or {}).get("postprocess_topn")
    if not isinstance(topn, dict):
        return
    anchors = list(topn.get("postprocess_by_anchor") or [])
    if not anchors:
        return
    strat = str(data.get("strategy") or "momentum").strip().lower()
    strat_u = strat.upper()
    if strat_u in ("TAIL", "SHORT"):
        strat_u = "TAIL"
    elif strat_u in ("SCALPING", "SCALP"):
        strat_u = "SCALP"
    before = len(anchors)
    append_stable_postprocess_anchors(
        data, anchors,
        strat=strat, strat_u=strat_u, top_n=resolve_post_top_n(10),
        do_ob=False, evaluate_fn=None, lg=logger, ob_n_trials=0,
    )
    if len(anchors) != before:
        topn["postprocess_by_anchor"] = anchors


def attach_topn_postprocess(
    out_data: Dict[str, Any],
    *,
    evaluate_fn: Optional[EvalFn] = None,
    mode_fills: Optional[List[Dict[str, Any]]] = None,
    log: Optional[logging.Logger] = None,
    run_ob_whipsaw: Optional[bool] = None,
    ob_n_trials: int = 0,
) -> Dict[str, Any]:
    """
    out_data 에 postprocess_by_anchor / consensus / apply_overfit_pct 기록.

    evaluate_fn 있으면 gated(+mode 미캐시) 백테 1회씩 재실행해 체결→호가/휩쏘.
    없으면 트레일만(구 JSON 웹 요약). 실매 엔진 호출 없음.
    """
    lg = log or logger
    data = out_data or {}
    strat = str(data.get("strategy") or "momentum").strip().lower()
    strat_u = strat.upper()
    if strat_u in ("TAIL", "SHORT"):
        strat_u = "TAIL"
    elif strat_u in ("SCALPING", "SCALP"):
        strat_u = "SCALP"
    top_n = resolve_post_top_n(10)
    do_ob = _run_ob_whipsaw_full() if run_ob_whipsaw is None else bool(run_ob_whipsaw)

    gated = list(data.get("results_gated") or [])[:top_n]
    anchors: List[Dict[str, Any]] = []

    from kis_trader.backtest import optuna_post_progress as opp

    stable_preview: List[Any] = []
    if _include_stable():
        from kis_trader.backtest.optuna_common import resolve_results_stable
        stable_preview, _sg = resolve_results_stable(data, top_n=max(1, int(top_n or 10)))
        if not list((data or {}).get("results_stable") or []) and stable_preview:
            data["results_stable"] = list(stable_preview)
            if _sg:
                data["stable_gates"] = dict(_sg)
    learn_preview: List[Any] = []
    if not gated:
        learn_preview = list(
            (data or {}).get("results") or (data or {}).get("results_all") or []
        )[: max(1, int(top_n or 10))]
    n_units = len(gated) + len(learn_preview) + (len(stable_preview) if _include_stable() else 0)
    if _include_mode():
        n_units += 1
    if _include_live():
        n_units += 1
    if do_ob:
        opp.begin_job(lg, max(1, n_units))
        try:
            from kis_trader.backtest.optuna_feed_trace import (
                log_bt_feed_chain_banner,
                reset_bt_feed_sample_counter,
            )

            reset_bt_feed_sample_counter(postprocess=True)
            log_bt_feed_chain_banner(context="Optuna후처리")
            lg.info(
                "🔎 [Optuna후처리] 호가 추천은 코어 TPE(필터OFF)와 별도 — "
                "아래 [호가후처리]/[호가후처리샘플] 줄을 보면 됨"
            )
        except Exception:
            pass

    for i, row in enumerate(gated, start=1):
        combo = _combo_from_row(row)
        fills: List[Dict[str, Any]] = []
        if do_ob:
            opp.next_unit(f"gated#{i}", "백테재실행·호가")
        if do_ob and evaluate_fn is not None:
            lg.info("📌 [후처리] gated#%d 백테 재실행 (호가/휩쏘 체결)", i)
            fills = _replay_fills(evaluate_fn, combo, lg)
        pnl = row.get("total_pnl")
        trail = _slim_trail(_trail_for_pnl(strat, pnl))
        ob = {"ok": False, "reason": "light_skip"}
        ws = {"ok": False, "reason": "light_skip"}
        if do_ob:
            if fills:
                ob = _ob_for_anchor(strategy=strat_u, out_data=data, fills=fills, live=False, lg=lg, n_trials=ob_n_trials)
                ws = _ws_from_ob_or_anchor(
                    ob, strategy=strat_u, out_data=data, fills=fills, live=False, lg=lg,
                )
            elif evaluate_fn is None:
                ob = {"ok": False, "reason": "no_replay_fills"}
                ws = {"ok": False, "reason": "no_replay_fills"}
            else:
                ob = {"ok": False, "reason": "not_enough_trades", "trade_count": 0}
                ws = {"ok": False, "reason": "not_enough_trades", "trade_count": 0}
        anchors.append({
            "id": f"gated#{i}",
            "role": "gated",
            "rank": i,
            "optuna_trial_number": row.get("optuna_trial_number") or row.get("_trial_number"),
            "total_pnl": _krw_int(pnl),
            "total_trades": int(row.get("total_trades") or 0),
            "win_rate": row.get("win_rate"),
            "pf": row.get("pf"),
            "orderbook": ob,
            "whipsaw": ws,
            "trail": trail,
            "note": "사후합격 후보",
        })

    # gated 비면 학습 Top 에도 호가 후처리 (탐색 WR/PF=0 과 같은 정신)
    append_learn_postprocess_anchors(
        data, anchors,
        strat=strat, strat_u=strat_u, top_n=top_n,
        do_ob=do_ob, evaluate_fn=evaluate_fn, lg=lg, ob_n_trials=ob_n_trials,
    )

    append_stable_postprocess_anchors(
        data, anchors,
        strat=strat, strat_u=strat_u, top_n=top_n,
        do_ob=do_ob, evaluate_fn=evaluate_fn, lg=lg, ob_n_trials=ob_n_trials,
    )

    if _include_mode():
        mc = data.get("mode_combo") or {}
        bt = mc.get("backtest") or {}
        mode_pnl = bt.get("total_pnl")
        fills_m = list(mode_fills or [])
        if do_ob:
            from kis_trader.backtest import optuna_post_progress as opp
            opp.next_unit("mode", "mode_combo 호가")
        if do_ob and not fills_m and evaluate_fn is not None:
            mode_params = dict(mc.get("params") or {})
            if mode_params:
                lg.info("📌 [후처리] mode_combo 백테 재실행")
                fills_m = _replay_fills(evaluate_fn, mode_params, lg)
        trail = _slim_trail(_trail_for_pnl(strat, mode_pnl))
        ob = {"ok": False, "reason": "light_skip"}
        ws = {"ok": False, "reason": "light_skip"}
        if do_ob:
            if fills_m:
                ob = _ob_for_anchor(strategy=strat_u, out_data=data, fills=fills_m, live=False, lg=lg, n_trials=ob_n_trials)
                ws = _ws_from_ob_or_anchor(
                    ob, strategy=strat_u, out_data=data, fills=fills_m, live=False, lg=lg,
                )
            else:
                ob = {"ok": False, "reason": "no_replay_fills"}
                ws = {"ok": False, "reason": "no_replay_fills"}
        anchors.append({
            "id": "mode",
            "role": "mode",
            "rank": None,
            "optuna_trial_number": None,
            "total_pnl": _krw_int(mode_pnl),
            "total_trades": int(bt.get("total_trades") or 0),
            "win_rate": bt.get("win_rate"),
            "pf": bt.get("pf"),
            "orderbook": ob,
            "whipsaw": ws,
            "trail": trail,
            "note": "축별 최빈 조각 모음(trial 없음)",
        })

    if _include_live():
        live_ob = {"ok": False, "reason": "light_skip"}
        live_ws = {"ok": False, "reason": "light_skip"}
        if do_ob:
            from kis_trader.backtest import optuna_post_progress as opp
            opp.next_unit("live", "실매참고 호가")
            live_ob = _ob_for_anchor(strategy=strat_u, out_data=data, fills=None, live=True, lg=lg, n_trials=ob_n_trials)
            live_ws = _ws_from_ob_or_anchor(
                live_ob, strategy=strat_u, out_data=data, fills=None, live=True, lg=lg,
            )
        live_pnl = None
        if (live_ob.get("orig_stats") or {}).get("pnl") is not None:
            live_pnl = live_ob["orig_stats"]["pnl"]
        elif (live_ws.get("orig_stats") or {}).get("pnl") is not None:
            live_pnl = live_ws["orig_stats"]["pnl"]
        trail = _slim_trail(_trail_for_pnl(strat, live_pnl))
        anchors.append({
            "id": "live",
            "role": "live",
            "rank": None,
            "optuna_trial_number": None,
            "total_pnl": _krw_int(live_pnl),
            "total_trades": int((live_ob.get("orig_stats") or live_ws.get("orig_stats") or {}).get("count") or 0),
            "win_rate": (live_ob.get("orig_stats") or live_ws.get("orig_stats") or {}).get("win_rate"),
            "orderbook": live_ob,
            "whipsaw": live_ws,
            "trail": trail,
            "note": "실매 trade_history 참고 — Optuna 칸과 섞지 않음 · 과적합% 제외",
        })

    consensus = _consensus_from_anchors(anchors, strat)
    extra, extra_factors = _postprocess_overfit_extra(anchors)

    base_risk = 0.0
    try:
        from kis_trader.backtest.optuna_common import build_optuna_overfit_diagnostics
        diag = build_optuna_overfit_diagnostics(data)
        base_risk = float(diag.get("overfit_risk_pct") or 0)
        data["overfit_diagnostics"] = diag
    except Exception as exc:
        lg.warning("⚠️ overfit_diagnostics 재계산 실패: %s", exc)
        diag = data.get("overfit_diagnostics") or {}
        try:
            base_risk = float(diag.get("overfit_risk_pct") or 0)
        except (TypeError, ValueError):
            base_risk = 0.0

    apply_pct = max(0.0, min(100.0, round(base_risk + extra, 1)))
    verd, verd_ui = _verdict(apply_pct)

    payload = {
        "postprocess_by_anchor": anchors,
        "postprocess_consensus": consensus,
        "apply_overfit_pct": apply_pct,
        "apply_overfit_verdict": verd,
        "apply_overfit_verdict_ui": verd_ui,
        "apply_overfit_base_pct": round(base_risk, 1),
        "apply_overfit_post_extra": round(extra, 1),
        "apply_overfit_factors": extra_factors,
        "apply_overfit_note": (
            "과적합%=이 숫자만 믿으면 내일 틀릴 수 있는 정도(추정). AI 아님. "
            "실매 참고행은 점수에 넣지 않음. 적용=호가 8방 중 하나(또는 휩쏘/트레일 별도)."
        ),
        "run_ob_whipsaw": bool(do_ob),
    }
    data["postprocess_topn"] = payload
    data["apply_overfit_pct"] = apply_pct
    data["apply_overfit_verdict"] = verd

    # 하위호환: 기존 단일 키 = 합의 (apply 스크립트·웹 구표)
    if consensus.get("orderbook", {}).get("ok"):
        cnote = consensus.get("note") or "TopN 합의(gated+mode). 실매 단독 아님."
        cid = (consensus.get("combo") or {}).get("combo_id") or consensus.get("orderbook", {}).get("combo_id")
        if cid:
            cnote = f"{_COMBO_LABELS.get(str(cid), cid)} · {cnote}"
        data["orderbook_recommend"] = {
            "ok": True,
            "strategy": strat_u,
            "params": consensus["orderbook"]["params"],
            "combo_id": cid or "",
            "note": cnote,
        }
    if consensus.get("whipsaw", {}).get("ok"):
        data["whipsaw_recommend"] = {
            "ok": True,
            "strategy": strat_u,
            "params": consensus["whipsaw"]["params"],
            "note": "TopN 합의(gated+mode). 실매 단독 아님.",
        }
    trc = consensus.get("trail") or {}
    if trc.get("ok"):
        data["daily_trail_recommend"] = {
            "ok": True,
            "strategy": strat,
            "prefix": trc.get("prefix"),
            "arm_krw": trc.get("arm_krw"),
            "anchor_krw": trc.get("anchor_krw"),
            "tiers": trc.get("tiers"),
            "mode": "trailing",
            "enabled": True,
            "note": trc.get("note"),
        }
        mc = data.get("mode_combo")
        if isinstance(mc, dict):
            mc["daily_trail_recommend"] = data["daily_trail_recommend"]
            if data.get("orderbook_recommend"):
                mc["orderbook_recommend"] = data["orderbook_recommend"]
            if data.get("whipsaw_recommend"):
                mc["whipsaw_recommend"] = data["whipsaw_recommend"]

    lg.info(
        "📌 [후처리 TopN] anchors=%d · 과적합%%=%.1f(%s) · ob_whipsaw=%s",
        len(anchors), apply_pct, verd, do_ob,
    )
    if do_ob:
        from kis_trader.backtest import optuna_post_progress as opp
        opp.finish_job(lg)
    return data


# 구 UI 누적 upto → 방 id (하위호환). whipsaw=111방 + 휩쏘 ON.
_UPTO_TO_COMBO = {
    "base": "base",
    "entry": "e",
    "exit": "ex",
    "stop": "exs",
    "whipsaw": "exs",
}
_UPTO_ORDER = ("base", "entry", "exit", "stop", "whipsaw")  # 레거시 별칭만


def pick_postprocess_anchor(
    data: Dict[str, Any],
    source: str,
    rank: int,
) -> Optional[Dict[str, Any]]:
    topn = data.get("postprocess_topn") if isinstance(data, dict) else None
    anchors = list((topn or {}).get("postprocess_by_anchor") or [])
    src = str(source or "gated").strip().lower()
    rk = max(1, int(rank or 1))
    if src == "mode":
        for a in anchors:
            if str(a.get("role") or "") == "mode":
                return a
        return None
    if src == "live":
        return None
    role = "stable" if src == "stable" else "gated"
    for a in anchors:
        if str(a.get("role") or "") == role and int(a.get("rank") or 0) == rk:
            return a
    return None


def _normalize_strat_u(strategy: str, data: Optional[Dict[str, Any]] = None) -> str:
    strat_u = str(strategy or (data or {}).get("strategy") or "").strip().upper()
    if strat_u in ("TAIL", "SHORT"):
        return "TAIL"
    if strat_u in ("SCALPING", "SCALP"):
        return "SCALP"
    if strat_u in ("US_MOMENTUM",):
        return "US_MOMENTUM"
    if strat_u in ("MOMENTUM",):
        return "MOMENTUM"
    return strat_u


def _resolve_combo_id(raw: str) -> Tuple[str, bool]:
    """반환: (combo_id, include_whipsaw).
    `e+whip` / `100+whipsaw` → 해당 방 + 그 방 휩쏘.
    """
    u = str(raw or "base").strip().lower()
    if u == "trail":
        return "base", False
    whip = False
    for suf in ("+whipsaw", "+whip", "|whip"):
        if u.endswith(suf):
            whip = True
            u = u[: -len(suf)]
            break
    if u == "whipsaw":
        return "exs", True
    if u in _COMBO_IDS:
        return u, whip
    if u in _UPTO_TO_COMBO:
        return _UPTO_TO_COMBO[u], whip
    # 비트 표기 허용
    bit_map = {
        "000": "base", "100": "e", "010": "x", "001": "s",
        "110": "ex", "101": "es", "011": "xs", "111": "exs",
    }
    if u in bit_map:
        return bit_map[u], whip
    raise ValueError(
        f"combo/upto 는 {_COMBO_IDS}+whipsaw|trail|000~111|+whip 만 (got {raw})"
    )


def build_combo_env_patch(
    *,
    data: Dict[str, Any],
    source: str,
    rank: int,
    combo: str,
    strategy: str,
    include_whipsaw: Optional[bool] = None,
) -> Tuple[Dict[str, str], List[str]]:
    """
    8방 중 하나 적용. 켠 축만 ON·숫자 반영, 끈 축은 ENABLED=false.
    휩쏘는 8방 밖 — include_whipsaw=True 일 때만 붙임.
    """
    from kis_trader.backtest.optuna_orderbook_recommend import (
        _env_pfx,
        build_entry_ob_env_patch,
        build_exit_ob_env_patch,
        build_stop_ob_env_patch,
    )
    from kis_trader.backtest.optuna_whipsaw_recommend import (
        build_whipsaw_clear_patch,
        build_whipsaw_env_patch,
    )

    cid, whip_default = _resolve_combo_id(combo)
    do_whip = bool(whip_default if include_whipsaw is None else include_whipsaw)
    notes: List[str] = []
    strat_u = _normalize_strat_u(strategy, data)
    # 스캘핑 등 익절/손절호가 없음 → 레거시 「whipsaw」= base+휩쏘
    if str(combo or "").strip().lower() == "whipsaw" and strat_u in ("SCALP", "SCALPING", "TAIL", "SHORT"):
        cid, do_whip = "base", True
    epfx = _env_pfx(strat_u)
    use_e, use_x, use_s = _COMBO_MASK[cid]

    anchor = pick_postprocess_anchor(data, source, rank)
    if not anchor:
        notes.append("후처리 앵커 없음(구 JSON이면 재실행 필요)")
        return {}, notes

    ob = dict(anchor.get("orderbook") or {})
    combos = ob.get("combos") if isinstance(ob.get("combos"), dict) else {}
    c = combos.get(cid) if isinstance(combos.get(cid), dict) else None
    # 구 JSON(combos 없음): entry/exit/stop 중첩으로 폴백
    if not (c and c.get("ok")):
        legacy_map = {
            "e": ("entry",),
            "x": ("exit",),
            "s": ("stop",),
            "ex": ("exit", "entry"),
            "es": ("stop", "entry"),
            "xs": ("stop", "exit"),
            "exs": ("stop", "exit", "entry"),
        }
        if cid == "base":
            c = {"ok": True, "params": {}}
        else:
            merged_p: Dict[str, Any] = {}
            ok_any = False
            for ax in legacy_map.get(cid, ()):
                nested = ob.get(ax) if isinstance(ob.get(ax), dict) else {}
                if nested.get("ok"):
                    ok_any = True
                    merged_p.update(dict(nested.get("params") or {}))
            c = {"ok": ok_any, "params": merged_p} if ok_any else None
    if cid != "base" and not (c and c.get("ok")):
        notes.append(f"방 {cid} 추천 없음")
        return {}, notes

    params = dict((c or {}).get("params") or {})
    # 마스크로 enabled 강제 (방 정의가 진실)
    params["orderbook_filter_enabled"] = bool(use_e)
    params["exit_ob_enabled"] = bool(use_x)
    params["stop_ob_enabled"] = bool(use_s)
    view = {"strategy": strat_u, "params": params, "entry": {"ok": True, "params": params},
            "exit": {"ok": True, "params": params}, "stop": {"ok": True, "params": params}}

    patch: Dict[str, str] = {}
    if not epfx:
        notes.append("전략 prefix 없음")
        return {}, notes

    if use_e:
        ep = build_entry_ob_env_patch(view, strat_u)
        if ep:
            patch.update(ep)
        else:
            notes.append("진입 숫자 없음 → 진입 OFF")
            patch[f"{epfx}_ORDERBOOK_FILTER_ENABLED"] = "false"
    else:
        patch[f"{epfx}_ORDERBOOK_FILTER_ENABLED"] = "false"

    if epfx in ("MOMENTUM", "BREAKOUT"):
        if use_x:
            xp = build_exit_ob_env_patch(view, strat_u)
            if xp:
                patch.update(xp)
            else:
                notes.append("익절 숫자 없음 → 익절 OFF")
                patch[f"{epfx}_EXIT_OB_ENABLED"] = "false"
        else:
            patch[f"{epfx}_EXIT_OB_ENABLED"] = "false"
        if use_s:
            sp = build_stop_ob_env_patch(view, strat_u)
            if sp:
                patch.update(sp)
            else:
                notes.append("손절 숫자 없음 → 손절 OFF")
                patch[f"{epfx}_STOP_OB_ENABLED"] = "false"
        else:
            patch[f"{epfx}_STOP_OB_ENABLED"] = "false"
    elif use_x or use_s:
        notes.append("익절/손절호가 해당없음(전략)")

    if do_whip:
        # 방별 휩쏘 우선 (같은 호가방 통과 체결로 뽑은 값)
        ws = {}
        if isinstance(c, dict) and isinstance(c.get("whipsaw"), dict):
            ws = dict(c.get("whipsaw") or {})
        if not ws.get("ok"):
            ws = dict(anchor.get("whipsaw") or {})
        ws["strategy"] = strat_u
        wp = build_whipsaw_env_patch(ws)
        if wp:
            patch.update(wp)
        elif strat_u in ("TAIL", "SHORT", "BREAKOUT"):
            notes.append("휩쏘 DB 스킵(전략 특성)")
        else:
            notes.append("휩쏘 추천 없음(해당 방)")
    else:
        # 「호가만」/타점만 — 휩쏘 OFF + 수치 비움 (전략 공통). 예전엔 MOMENTUM만 false.
        clr = build_whipsaw_clear_patch(strat_u if epfx else strat_u)
        if not clr and epfx:
            clr = build_whipsaw_clear_patch(epfx)
        if clr:
            patch.update(clr)
            notes.append("휩쏘 OFF·수치 비움(호가만/휩쏘미포함)")

    notes.append(f"방 {cid} 적용 (진입={int(use_e)} 익절={int(use_x)} 손절={int(use_s)} 휩쏘={int(do_whip)})")
    return patch, notes


def build_upto_env_patch(
    *,
    data: Dict[str, Any],
    source: str,
    rank: int,
    upto: str,
    strategy: str,
) -> Tuple[Dict[str, str], List[str]]:
    """하위호환: 구 누적 upto → 8방 combo 패치."""
    u = str(upto or "base").strip().lower()
    if u == "trail":
        return {}, ["trail은 다단트레일 전용 경로"]
    return build_combo_env_patch(
        data=data, source=source, rank=rank, combo=u, strategy=strategy,
    )


def _later_axes_off_patch(strat_u: str, upto: str) -> Dict[str, str]:
    """레거시 누적 off. 신규는 build_combo_env_patch 마스크 사용."""
    from kis_trader.backtest.optuna_orderbook_recommend import _env_pfx

    pfx = _env_pfx(strat_u)
    if not pfx:
        return {}
    try:
        cid, do_whip = _resolve_combo_id(upto)
    except ValueError:
        return {}
    use_e, use_x, use_s = _COMBO_MASK[cid]
    patch: Dict[str, str] = {}
    if not use_e:
        patch[f"{pfx}_ORDERBOOK_FILTER_ENABLED"] = "false"
    if not use_x and pfx in ("MOMENTUM", "BREAKOUT"):
        patch[f"{pfx}_EXIT_OB_ENABLED"] = "false"
    if not use_s and pfx in ("MOMENTUM", "BREAKOUT"):
        patch[f"{pfx}_STOP_OB_ENABLED"] = "false"
    if not do_whip:
        from kis_trader.backtest.optuna_whipsaw_recommend import build_whipsaw_clear_patch
        patch.update(build_whipsaw_clear_patch(pfx))
    return patch