feat(옵투나·웹): 후처리 재탐색·ob_modes·적용감사·수집통계

- Optuna web jobs/TPE/apply snapshot·틱로더 정합, jobs limit·감사로그
- 백테 UI 호가모드·후보 적용 흐름, feed_collect_stats API/탭
- 가설검증·교차검증 룰, 4전략 스모크·OB slot41 진단 스크립트

Co-authored-by: Cursor <cursoragent@cursor.com>

kis_trader/backtest/optuna_web_jobs.py

@@ -25,11 +25,91 @@ from typing import Any, Dict, List, Optional
 ROOT = Path(__file__).resolve().parents[2]
 JOBS_DIR = ROOT / "logs" / "optuna_web_jobs"
 RESULTS_DIR = ROOT / "kis_trader" / "backtest" / "results"
+# Optuna 웹/API 적용 감사 — 한 줄 JSON (역추적: source=mode|gated · tp/sl)
+APPLY_AUDIT_PATH = ROOT / "logs" / "optuna_apply_audit.jsonl"
 PY = ROOT / ".venv" / "bin" / "python"
 
 _STRATS = ("momentum", "us_momentum", "tail", "breakout", "scalp")
 
 
+def _pct_keys_from_params(params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    """적용 로그용 — 익절/손절 등 UI% 축만 짧게."""
+    if not isinstance(params, dict):
+        return {}
+    out: Dict[str, Any] = {}
+    for k in (
+        "tp_pct", "sl_pct", "tp_max_pct", "drop_rate",
+        "trail_pct", "trail_arm_pct", "trail_trigger", "trail_stop",
+        "shoulder_min_high", "shoulder_min_high_pct",
+        "shoulder_cut_pct",
+    ):
+        if k in params and params[k] is not None:
+            out[k] = params[k]
+    return out
+
+
+def record_optuna_apply_audit(
+    *,
+    ok: bool,
+    strategy: str = "",
+    source: str = "",
+    rank: int = 0,
+    upto: str = "",
+    trial: Any = None,
+    job_id: Optional[str] = None,
+    study_name: str = "",
+    result_json: str = "",
+    params: Optional[Dict[str, Any]] = None,
+    metrics: Optional[Dict[str, Any]] = None,
+    error: str = "",
+    note: str = "",
+) -> None:
+    """적용 성공/실패를 JSONL + 표준 로그에 남김 (HTTP access body 없음 보완)."""
+    import logging
+
+    lg = logging.getLogger("optuna_apply")
+    row = {
+        "ts": _now_iso(),
+        "ok": bool(ok),
+        "strategy": str(strategy or ""),
+        "source": str(source or ""),
+        "rank": int(rank or 0),
+        "upto": str(upto or ""),
+        "trial": trial,
+        "job_id": job_id or None,
+        "study_name": str(study_name or "") or None,
+        "result_json": str(result_json or "") or None,
+        "params": _pct_keys_from_params(params),
+        "metrics": {
+            k: (metrics or {}).get(k)
+            for k in (
+                "total_pnl", "total_trades", "win_rate", "pf",
+                "optuna_trial_number",
+            )
+            if metrics and k in metrics
+        } or None,
+        "error": (error or "")[:500] or None,
+        "note": (note or "")[:300] or None,
+    }
+    try:
+        APPLY_AUDIT_PATH.parent.mkdir(parents=True, exist_ok=True)
+        with APPLY_AUDIT_PATH.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(row, ensure_ascii=False) + "\n")
+    except Exception as exc:
+        lg.warning("optuna apply audit file write failed: %s", exc)
+    if ok:
+        lg.info(
+            "OPTUNA_APPLY ok strat=%s source=%s rank=%s upto=%s trial=%s params=%s job=%s",
+            row["strategy"], row["source"], row["rank"], row["upto"],
+            row["trial"], row["params"], row["job_id"],
+        )
+    else:
+        lg.warning(
+            "OPTUNA_APPLY fail strat=%s source=%s rank=%s err=%s job=%s",
+            row["strategy"], row["source"], row["rank"], row["error"], row["job_id"],
+        )
+
+
 def _result_study_name(meta: Optional[Dict[str, Any]]) -> str:
     """테이블 PK. 순차 잡의 seq_* 이름은 쓰지 않고 현재/활성 study."""
     m = meta or {}
@@ -135,6 +215,42 @@ def _normalize_breakout_sl_modes(raw: Any) -> List[str]:
     return items or ["fixed"]
 
 
+def _normalize_breakout_ob_modes(raw: Any) -> List[str]:
+    """웹/CLI: off, on. 빈값이면 off 1개(호가OFF 스터디)."""
+    from kis_trader.backtest.optuna_breakout_tpe_space import (
+        normalize_tpe_breakout_ob_mode,
+    )
+
+    items: List[str] = []
+    if raw is None or raw is False:
+        seq: List[Any] = []
+    elif isinstance(raw, str):
+        seq = [x for x in raw.replace(",", " ").split() if x.strip()]
+    elif isinstance(raw, (list, tuple)):
+        seq = list(raw)
+    else:
+        seq = [raw]
+    for x in seq:
+        om = normalize_tpe_breakout_ob_mode(x)
+        if om not in items:
+            items.append(om)
+    return items or ["off"]
+
+
+def _parse_breakout_seq_extra(extra: str) -> Tuple[str, str]:
+    """순차 extra → (sl_mode, ob_mode). 예: fixed_ob_off / atr_ob_on / fixed(구형)."""
+    from kis_trader.backtest.optuna_breakout_tpe_space import (
+        normalize_tpe_breakout_ob_mode,
+        normalize_tpe_breakout_sl_mode,
+    )
+
+    e = str(extra or "").strip().lower()
+    if "_ob_" in e:
+        sm, _, om = e.partition("_ob_")
+        return normalize_tpe_breakout_sl_mode(sm), normalize_tpe_breakout_ob_mode(om)
+    return normalize_tpe_breakout_sl_mode(e or "fixed"), "off"
+
+
 def _normalize_tail_entry_modes(raw: Any) -> List[str]:
     """웹/CLI: align, limit_atr. 빈값이면 align 1개(기존 TPE와 동일)."""
     from kis_trader.backtest.optuna_tail_tpe_space import normalize_tpe_tail_entry_mode
@@ -507,13 +623,16 @@ _SEQ_START_RE = re.compile(
 
 
 def _seq_step_catalog(meta: Dict[str, Any]) -> List[Dict[str, str]]:
-    """순차 한 칸 = 전략(+꼬리 진입/+돌파 손절). 스크립트 run_one 과 동일 순서."""
+    """순차 한 칸 = 전략(+꼬리 진입/+돌파 손절×호가). 스크립트 run_one 과 동일 순서."""
+    from kis_trader.backtest.optuna_breakout_tpe_space import breakout_tpe_study_extra
+
     picked = list(meta.get("strategies") or [])
     if not picked:
         raw = str(meta.get("strategy") or "")
         picked = [s.strip() for s in raw.split(",") if s.strip()]
     tail_ems = list(meta.get("tail_entry_modes") or ["align"])
     bo_sms = list(meta.get("breakout_sl_modes") or ["fixed"])
+    bo_oms = list(meta.get("breakout_ob_modes") or ["off"])
     out: List[Dict[str, str]] = []
     for s in picked:
         s = str(s or "").strip().lower()
@@ -522,7 +641,11 @@ def _seq_step_catalog(meta: Dict[str, Any]) -> List[Dict[str, str]]:
                 out.append({"strategy": "tail", "extra": str(em or "align")})
         elif s == "breakout":
             for sm in bo_sms:
-                out.append({"strategy": "breakout", "extra": str(sm or "fixed")})
+                for om in bo_oms:
+                    out.append({
+                        "strategy": "breakout",
+                        "extra": breakout_tpe_study_extra(sm, om),
+                    })
         elif s:
             out.append({"strategy": s, "extra": ""})
     return out
@@ -659,8 +782,23 @@ def _join_argv_for_study(
         ems = list(meta.get("tail_entry_modes") or ["align"])
         argv.extend(["--entry-mode", extra or str(ems[0] if ems else "align")])
     if strat == "breakout":
-        sms = list(meta.get("breakout_sl_modes") or ["fixed"])
-        argv.extend(["--sl-mode", extra or str(sms[0] if sms else "fixed")])
+        if extra:
+            sm, om = _parse_breakout_seq_extra(extra)
+        else:
+            sms = list(meta.get("breakout_sl_modes") or ["fixed"])
+            oms = list(meta.get("breakout_ob_modes") or ["off"])
+            sm = str(sms[0] if sms else "fixed")
+            om = str(oms[0] if oms else "off")
+        argv.extend(["--sl-mode", sm])
+        # 베이스 argv 의 --orderbook-filter off 를 스터디 스위치로 덮어씀
+        if "--orderbook-filter" in argv:
+            _i = argv.index("--orderbook-filter")
+            if _i + 1 < len(argv):
+                argv[_i + 1] = om
+            else:
+                argv.extend(["--orderbook-filter", om])
+        else:
+            argv.extend(["--orderbook-filter", om])
     for flag, key in (
         ("--candle-source", "candle_source"),
         ("--tick-source", "tick_source"),
@@ -864,6 +1002,133 @@ def _study_progress(study_name: str, trials_total: int) -> Dict[str, Any]:
     return out
 
 
+def _as_bool_opt(v: Any) -> Optional[bool]:
+    if v is None or v == "":
+        return None
+    if isinstance(v, bool):
+        return v
+    s = str(v).strip().lower()
+    if s in ("1", "true", "yes", "on"):
+        return True
+    if s in ("0", "false", "no", "off"):
+        return False
+    return None
+
+
+def _ob_whip_ui_from_params(params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    """TopN 표용 — 본 TPE trial 호가·익절·손절·휩쏘 요약 (사후 8방과 무관).
+
+    상세 보기 없이 적용값을 고를 수 있게 ON/OFF + 핵심 수치를 전부 내려준다.
+    """
+    p = params if isinstance(params, dict) else {}
+    ob_on = _as_bool_opt(p.get("_orderbook_filter_enabled"))
+    if ob_on is None:
+        ob_on = _as_bool_opt(p.get("ob_filter_enabled"))
+    whip_on = _as_bool_opt(p.get("whipsaw_enabled"))
+    if whip_on is None:
+        whip_on = _as_bool_opt(p.get("whipsaw_filter_enabled"))
+
+    def _f(key: str) -> Optional[float]:
+        v = p.get(key)
+        if v is None or v == "":
+            return None
+        try:
+            return float(v)
+        except (TypeError, ValueError):
+            return None
+
+    def _i(key: str) -> Optional[int]:
+        v = _f(key)
+        if v is None:
+            return None
+        try:
+            return int(round(v))
+        except (TypeError, ValueError):
+            return None
+
+    spread = _f("max_spread_pct")
+    if spread is None:
+        spread = _f("orderbook_max_spread_pct")
+    ratio = _f("min_bid_ask_ratio")
+    if ratio is None:
+        ratio = _f("orderbook_min_bid_ask_ratio")
+    ask = _f("ask_max_mult")
+    if ask is None:
+        ask = _f("orderbook_entry_ask_max_mult")
+
+    tp = _f("tp_pct")
+    if tp is None:
+        tp = _f("take_profit_pct")
+    tp_max = _f("tp_max_pct")
+    if tp_max is None:
+        tp_max = _f("take_profit_max_pct")
+    sl = _f("sl_pct")
+    if sl is None:
+        sl = _f("stop_loss_pct")
+
+    whip_sub = _i("whipsaw_subbar_sec")
+    whip_lb = _i("whipsaw_lookback_sec")
+    whip_dip = _f("whipsaw_dip_pct")
+    whip_tol = _f("whipsaw_recovery_tol_pct")
+
+    lines: List[str] = []
+    if ob_on is True:
+        ob_bits = ["호가ON"]
+        if spread is not None:
+            ob_bits.append(f"spr{spread:.1f}")
+        if ratio is not None:
+            ob_bits.append(f"r{ratio:.2f}")
+        if ask is not None:
+            ob_bits.append(f"ask×{ask:.0f}")
+        lines.append(" ".join(ob_bits))
+    elif ob_on is False:
+        lines.append("호가OFF")
+
+    exit_bits: List[str] = []
+    if tp is not None:
+        exit_bits.append(f"익절{tp:.1f}%")
+    if tp_max is not None:
+        exit_bits.append(f"상한{tp_max:.1f}%")
+    if sl is not None:
+        exit_bits.append(f"손절{sl:.1f}%")
+    if exit_bits:
+        lines.append(" ".join(exit_bits))
+
+    if whip_on is True:
+        w_bits = ["휩쏘ON"]
+        if whip_sub is not None:
+            w_bits.append(f"sub{whip_sub}s")
+        if whip_lb is not None:
+            w_bits.append(f"lb{whip_lb}s")
+        if whip_dip is not None:
+            # 저장값 0.007 → 화면 0.70% (비율→퍼센트)
+            dip_pct = whip_dip * 100.0 if whip_dip < 0.5 else whip_dip
+            w_bits.append(f"dip{dip_pct:.2f}%")
+        if whip_tol is not None:
+            tol_pct = whip_tol * 100.0 if whip_tol < 0.5 else whip_tol
+            w_bits.append(f"tol{tol_pct:.2f}%")
+        lines.append(" ".join(w_bits))
+    elif whip_on is False and (ob_on is not None or exit_bits):
+        lines.append("휩쏘OFF")
+
+    return {
+        "ob_on": ob_on,
+        "whip_on": whip_on,
+        "ob_spread": spread,
+        "ob_ratio": ratio,
+        "ob_ask": ask,
+        "tp_pct": tp,
+        "tp_max_pct": tp_max,
+        "sl_pct": sl,
+        "whip_subbar_sec": whip_sub,
+        "whip_lookback_sec": whip_lb,
+        "whip_dip_pct": whip_dip,
+        "whip_recovery_tol_pct": whip_tol,
+        "ob_summary": " · ".join(lines) if lines else None,
+        "ob_summary_lines": lines,
+    }
+
+
 def _row_metrics(
     row: Optional[Dict[str, Any]],
     *,
@@ -900,6 +1165,11 @@ def _row_metrics(
             out["overfit_verdict_ui"] = of.get("verdict_ui")
         except Exception:
             pass
+    try:
+        prm = row.get("merged_params") or row.get("params") or {}
+        out.update(_ob_whip_ui_from_params(prm if isinstance(prm, dict) else {}))
+    except Exception:
+        pass
     return out
 
 
@@ -984,6 +1254,10 @@ def _summarize_result_data(
             "daily_pnl": daily,
             "rank": 1,
         }
+        try:
+            mode_row.update(_ob_whip_ui_from_params(mc.get("params") or {}))
+        except Exception:
+            pass
         try:
             from kis_trader.backtest.optuna_common import overfit_risk_pct_for_row
             of = overfit_risk_pct_for_row(data, mode_row)
@@ -1104,6 +1378,15 @@ def _summarize_result_data(
         "top5_learn": top5_learn,
         "top5_stable": top5_stable,
         "top5_mode": top5_mode,
+        # 본 TPE 호가축 여부 · 사후8방 생략 안내
+        "tpe_includes_orderbook": (
+            "_orderbook_filter_enabled" in list(data.get("grid_keys") or [])
+            or "max_spread_pct" in list(data.get("grid_keys") or [])
+        ),
+        "post_run_ob_whipsaw": (
+            bool(post_topn.get("run_ob_whipsaw"))
+            if isinstance(post_topn, dict) else None
+        ),
         "mode_combo_note": mc.get("note"),
         "daily_trail_recommend": (
             data.get("daily_trail_recommend")
@@ -1292,10 +1575,14 @@ def apply_optuna_result(
     src = str(source or "gated").strip().lower()
     rank = max(1, int(rank or 1))
     upto_s = str(upto or "base").strip().lower() or "base"
+    # full/as_scored/trial = Top10「적용」: 그 trial 점수에 쓰인 타점·익절·손절·호가·휩쏘 전부
+    # (구 base=타점만+호가OFF 후처리 덮어쓰기 — TPE 호가축 ON일 때 호가가 꺼지는 원인)
+    _FULL_UPTO = frozenset({"full", "as_scored", "trial", "all"})
     _allowed = (
         "base", "entry", "exit", "stop", "whipsaw", "trail",
         "e", "x", "s", "ex", "es", "xs", "exs",
         "000", "100", "010", "001", "110", "101", "011", "111",
+        "full", "as_scored", "trial", "all",
     )
     # 방별 휩쏘: e+whip / base+whipsaw / 100+whip …
     _chk = upto_s
@@ -1305,9 +1592,10 @@ def apply_optuna_result(
             break
     if _chk not in _allowed:
         raise ValueError(
-            "upto/combo 는 base|e|x|s|ex|es|xs|exs|whipsaw|trail|000~111 "
-            "(구 entry/exit/stop · +whip/+whipsaw 방별휩쏘 포함) 만"
+            "upto/combo 는 full|base|e|x|s|ex|es|xs|exs|whipsaw|trail|000~111 "
+            "(full=줄전체 · 구 entry/exit/stop · +whip 방별휩쏘 포함) 만"
         )
+    _apply_full_trial = upto_s in _FULL_UPTO
     item: Optional[Dict[str, Any]] = None
     merged: Dict[str, Any] = {}
 
@@ -1403,7 +1691,12 @@ def apply_optuna_result(
         else:
             raise ValueError(strat)
 
-        if not sym:
+        if _apply_full_trial:
+            # trial params 그대로 — 8방 후처리로 호가OFF 덮어쓰지 않음
+            axis_notes.append(
+                "upto=full · 본 trial 호가/휩쏘 ON·OFF·수치 유지(사후8방 미적용)"
+            )
+        elif not sym:
             from kis_trader.backtest.optuna_postprocess_topn import build_upto_env_patch
             axis_patch, axis_notes = build_upto_env_patch(
                 data=data, source=src, rank=rank, upto=upto_s, strategy=strat,
@@ -1449,6 +1742,41 @@ def apply_optuna_result(
         pass
 
     apply_target = f"stock_config:{sym}" if (strat == "us_momentum" and sym) else "global"
+    note = (
+        "TIME_* 는 session_env_patch 기본 OFF — 운영 시간창 유지"
+        + (
+            ""
+            if upto_s == "trail"
+            else (
+                " · 줄전체 적용(타점·익절·손절·호가·휩쏘)"
+                if _apply_full_trial
+                else " · 타점/후처리 적용(upto=" + upto_s + ")"
+            )
+        )
+        + trail_note
+        + axis_note
+        + " · 웹 폼은 DB로 다시 채움(새로고침 불필요)"
+    )
+    study_nm = str(
+        (meta or {}).get("study_name")
+        or data.get("optuna_study_name")
+        or data.get("study_name")
+        or "",
+    )
+    record_optuna_apply_audit(
+        ok=True,
+        strategy=strat,
+        source=src,
+        rank=rank,
+        upto=upto_s,
+        trial=metrics.get("optuna_trial_number"),
+        job_id=(meta or {}).get("job_id") if meta else job_id,
+        study_name=study_nm,
+        result_json=str(path or ""),
+        params=merged if upto_s != "trail" else None,
+        metrics=metrics,
+        note=note,
+    )
     return {
         "ok": True,
         "strategy": strat,
@@ -1464,13 +1792,7 @@ def apply_optuna_result(
         "axis_patch": axis_patch,
         "axis_notes": axis_notes,
         "reload_forms": True,
-        "note": (
-            "TIME_* 는 session_env_patch 기본 OFF — 운영 시간창 유지"
-            + ("" if upto_s == "trail" else " · 타점 적용(upto=" + upto_s + ")")
-            + trail_note
-            + axis_note
-            + " · 웹 폼은 DB로 다시 채움(새로고침 불필요)"
-        ),
+        "note": note,
     }
 
 
@@ -1791,17 +2113,33 @@ def refresh_job_status(meta: Dict[str, Any]) -> Dict[str, Any]:
     pp_log = str((rerun or {}).get("log_path") or "") if str((rerun or {}).get("status") or "") == "running" else ""
     post = _parse_postprocess_progress(pp_log or active_log)
     topn = ((m.get("result_summary") or {}).get("postprocess_topn") or {}) if isinstance(m.get("result_summary"), dict) else {}
-    if isinstance(topn, dict) and topn.get("postprocess_by_anchor") and topn.get("run_ob_whipsaw"):
-        if not alive and str((rerun or {}).get("status") or "") != "running":
-            post["ready"] = True
-            post["pct"] = 100.0
-            from kis_trader.backtest.optuna_postprocess_topn import ob_8way_web_hint
-            miss8 = ob_8way_web_hint(topn)
-            post["hint"] = (
-                ("후처리 끝 · 8방 미산출(%s) · 「상세」 가능" % miss8)
-                if miss8 else "후처리 끝 · 「상세」 가능"
-            )
-            post["stage"] = post.get("stage") or "done"
+    # 본 TPE 호가축 ON → 사후 8방 기본 OFF. 웹「후처리」바는 8방 TPE 전용으로만 길게 보이게.
+    try:
+        from kis_trader.backtest.optuna_postprocess_topn import _run_ob_whipsaw_full
+        expect_ob_post = bool(_run_ob_whipsaw_full())
+    except Exception:
+        expect_ob_post = True
+    post["expect_ob"] = bool(expect_ob_post)
+    if isinstance(topn, dict) and topn.get("postprocess_by_anchor") is not None:
+        ran_ob = bool(topn.get("run_ob_whipsaw"))
+        if ran_ob:
+            if not alive and str((rerun or {}).get("status") or "") != "running":
+                post["ready"] = True
+                post["pct"] = 100.0
+                from kis_trader.backtest.optuna_postprocess_topn import ob_8way_web_hint
+                miss8 = ob_8way_web_hint(topn)
+                post["hint"] = (
+                    ("후처리 끝 · 8방 미산출(%s) · 「상세」 가능" % miss8)
+                    if miss8 else "후처리 끝 · 「상세」 가능"
+                )
+                post["stage"] = post.get("stage") or "done"
+        else:
+            # A: 사후호가방 생략 — 바를「후처리」로 붙잡지 않음
+            if not alive:
+                post["ready"] = True
+                post["pct"] = 100.0
+                post["stage"] = "skipped_ob"
+                post["hint"] = "본TPE 호가포함 · 사후호가방 생략 · 「상세」가능"
     trials_tot = int(prog.get("trials_total") or 0)
     trials_done = int(prog.get("trials_done") or 0)
     is_seq = m.get("kind") in ("seq4", "seq")
@@ -1816,14 +2154,21 @@ def refresh_job_status(meta: Dict[str, Any]) -> Dict[str, Any]:
         if not post.get("hint"):
             post["hint"] = "후처리 재실행 중 · 「상세」는 끝난 뒤"
     elif alive and trial_finished and not post.get("ready"):
-        m["phase"] = "postprocess"
-        if not post.get("stage"):
-            post["stage"] = "wait"
-            post["hint"] = "학습 끝 · 후처리 시작 대기 · 「상세」는 아직"
+        if expect_ob_post:
+            m["phase"] = "postprocess"
+            if not post.get("stage"):
+                post["stage"] = "wait"
+                post["hint"] = "학습 끝 · 후처리 시작 대기 · 「상세」는 아직"
+        else:
+            # mode_combo·JSON 저장만 — 호가8방 TPE 아님
+            m["phase"] = "finalize"
+            post["stage"] = post.get("stage") or "finalize"
+            if not post.get("hint"):
+                post["hint"] = "학습 끝 · JSON 저장 중 (사후호가방 생략)"
     elif alive:
         m["phase"] = "trials"
         if not post.get("hint"):
-            post["hint"] = "학습 trial 중 · 후처리는 그 다음"
+            post["hint"] = "학습 trial 중 · 후처리는 그 다음" if expect_ob_post else "학습 trial 중 (호가=본TPE)"
     else:
         m["phase"] = str(m.get("status") or "idle")
         if post.get("ready") and not post.get("hint"):
@@ -1926,6 +2271,7 @@ def start_optuna_job(
     ob_source: Optional[str] = None,
     entry_modes: Optional[Any] = None,
     sl_modes: Optional[Any] = None,
+    ob_modes: Optional[Any] = None,
     study_trials: Optional[int] = None,
     study_name_override: Optional[str] = None,
 ) -> Dict[str, Any]:
@@ -1937,7 +2283,8 @@ def start_optuna_job(
     universe_history_source: kiwoom|ls (저장 후보 이력 테이블).
     candle_source: ''|kis|kiwoom — CANDLE_SOURCE / --candle-source (실매 읽기쌍과 동일).
     entry_modes: 꼬리 TPE 고정 진입 align|limit_atr. 둘 다=순차 2스터디(한 스터디에 섞지 않음).
-    sl_modes: 돌파 TPE 고정 손절 fixed|atr. 둘 다=순차 2스터디.
+    sl_modes: 돌파 TPE 고정 손절 fixed|atr. 둘 다=순차(호가와 곱).
+    ob_modes: 돌파 TPE 호가 스터디 스위치 off|on. 손절×호가 최대 4순차(한 스터디에 안 섞음).
     """
     _ensure_dirs()
     running = find_running_jobs()
@@ -2006,8 +2353,12 @@ def start_optuna_job(
     bo_sms = _normalize_breakout_sl_modes(sl_modes)
     if "breakout" not in picked:
         bo_sms = ["fixed"]
-    bo_dual = "breakout" in picked and len(bo_sms) >= 2
-    use_seq = len(picked) >= 2 or tail_dual or bo_dual
+    bo_oms = _normalize_breakout_ob_modes(ob_modes)
+    if "breakout" not in picked:
+        bo_oms = ["off"]
+    bo_steps = (len(bo_sms) * len(bo_oms)) if "breakout" in picked else 0
+    bo_multi = bo_steps >= 2
+    use_seq = len(picked) >= 2 or tail_dual or bo_multi
     seq_active = None
 
     if use_seq:
@@ -2035,6 +2386,7 @@ def start_optuna_job(
         env["UNIVERSE_HISTORY_SOURCE"] = hist_src
         env["TAIL_OPTUNA_ENTRY_MODES"] = " ".join(tail_ems if "tail" in picked else ["align"])
         env["BREAKOUT_OPTUNA_SL_MODES"] = " ".join(bo_sms if "breakout" in picked else ["fixed"])
+        env["BREAKOUT_OPTUNA_OB_MODES"] = " ".join(bo_oms if "breakout" in picked else ["off"])
         if candle_source:
             env["CANDLE_SOURCE"] = candle_source
         if tick_source:
@@ -2046,7 +2398,8 @@ def start_optuna_job(
         if "tail" in picked and tail_ems:
             _lab = _lab.replace("꼬리", "꼬리(" + "+".join(tail_ems) + ")")
         if "breakout" in picked and bo_sms:
-            _lab = _lab.replace("돌파", "돌파(" + "+".join(bo_sms) + ")")
+            _bo_bits = [f"{sm}×{om}" for sm in bo_sms for om in bo_oms]
+            _lab = _lab.replace("돌파", "돌파(" + "+".join(_bo_bits) + ")")
         label = "순차(" + _lab + ")"
         strat_field = ",".join(picked)
     else:
@@ -2066,12 +2419,13 @@ def start_optuna_job(
             log_path = ROOT / "logs" / f"optuna_web_{strat}_{_em}_{ts}.log"
             label = f"꼬리({_em})"
         elif strat == "breakout":
-            _sm = bo_sms[0]
+            from kis_trader.backtest.optuna_breakout_tpe_space import breakout_tpe_study_extra
+            _extra = breakout_tpe_study_extra(bo_sms[0], bo_oms[0])
             study_name = (
-                f"{strat}_{_sm}_{mode}_{start.replace('-', '')}_{end.replace('-', '')}_{ts}"
+                f"{strat}_{_extra}_{mode}_{start.replace('-', '')}_{end.replace('-', '')}_{ts}"
             )
-            log_path = ROOT / "logs" / f"optuna_web_{strat}_{_sm}_{ts}.log"
-            label = f"돌파({_sm})"
+            log_path = ROOT / "logs" / f"optuna_web_{strat}_{_extra}_{ts}.log"
+            label = f"돌파({_extra})"
         else:
             study_name = f"{strat}_{mode}_{start.replace('-', '')}_{end.replace('-', '')}_{ts}"
             log_path = ROOT / "logs" / f"optuna_web_{strat}_{ts}.log"
@@ -2089,7 +2443,7 @@ def start_optuna_job(
             "--min_trades", "1",
             "--min_win_rate", "0",
             "--min_pf", "0",
-            "--orderbook-filter", "off",
+            "--orderbook-filter", (bo_oms[0] if strat == "breakout" else "off"),
             "--no-progress",
             "--study-name", study_name,
             "--sort-by", sort_by,
@@ -2157,6 +2511,7 @@ def start_optuna_job(
         "ob_source": ob_source,
         "tail_entry_modes": tail_ems if "tail" in picked else None,
         "breakout_sl_modes": bo_sms if "breakout" in picked else None,
+        "breakout_ob_modes": bo_oms if "breakout" in picked else None,
         "log_path": str(log_path),
         "pid": int(proc.pid),
         "status": "running",
@@ -2213,6 +2568,7 @@ def continue_optuna_job(job_id: str) -> Dict[str, Any]:
         ob_source=meta.get("ob_source"),
         entry_modes=meta.get("tail_entry_modes"),
         sl_modes=meta.get("breakout_sl_modes"),
+        ob_modes=meta.get("breakout_ob_modes"),
         study_trials=goal,
         study_name_override=name,
     )