feat(surfer): wire regime kill-switch into the live combined-book forward track

The ForwardTracker now optionally throttles booked exposure off the live track's own
running drawdown: flatten to 0 once drawdown exceeds kill_dd, re-enter once it recovers
inside reenter_dd. The drawdown SIGNAL tracks shadow (un-throttled) equity so recovery is
observable while flat; the booked NAV reflects the throttled book. No-lookahead (each
day's decision uses prior-day kill-state); state is persisted (shadow_nav/shadow_peak/
killed) and lazy-inits for tracks that predate the switch.

Wired ON in both live entry points via new settings combined_book_kill_dd (0.15) /
combined_book_reenter_dd (0.07): the Dagster combined_forward_nav asset and the
forward-track CLI. Default OFF at the tracker/factory level so existing trackers and
tests are unchanged. Tests 54 passed (incl. new live-killswitch test); mypy clean on
changed lines (pre-existing debt untouched).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/fxhnt/adapters/orchestration/assets.py

@@ -30,11 +30,13 @@ def _data_dir() -> str:
 
 
 def build_combined_forward_nav(
-    source: CombinedBookSource, futures: List[str], state_path: str
+    source: CombinedBookSource, futures: List[str], state_path: str,
+    *, kill_dd: float | None = None, reenter_dd: float = 0.07,
 ) -> ForwardStatus:
-    """Build and step the forward tracker from the given source. Unit-testable without Dagster."""
+    """Build and step the forward tracker from the given source. Unit-testable without Dagster.
+    `kill_dd` (when set) wires the regime drawdown kill-switch into the live track."""
     book = CombinedBook("warehouse", futures, source=source)
-    return CombinedBookForwardTracker(book, state_path).step()
+    return CombinedBookForwardTracker(book, state_path, kill_dd=kill_dd, reenter_dd=reenter_dd).step()
 
 
 @asset
@@ -107,6 +109,8 @@ def combined_forward_nav(context: AssetExecutionContext) -> dict:  # type: ignor
         make_combined_book_source(s, _data_dir()),
         _FUTURES,
         f"{_data_dir()}/fxhnt_combined_forward.json",
+        kill_dd=s.combined_book_kill_dd,
+        reenter_dd=s.combined_book_reenter_dd,
     )
     context.log.info(
         f"combined_forward_nav: {st.forward_days} forward days through {st.last_date}"

src/fxhnt/application/forward_tracker.py

@@ -47,9 +47,12 @@ class ForwardTracker:
     """Strategy-agnostic forward-validation tracker. Freezes inception on first run (books nothing — no
     backfill), then books only days strictly after last_date. Owns the days-list state file."""
 
-    def __init__(self, strategy: ForwardStrategy, state_path: str) -> None:
+    def __init__(self, strategy: ForwardStrategy, state_path: str, *,
+                 kill_dd: float | None = None, reenter_dd: float = 0.07) -> None:
         self._strategy = strategy
         self._path = state_path
+        self._kill_dd = kill_dd            # regime kill-switch: flatten booked exposure beyond this DD
+        self._reenter_dd = reenter_dd       # re-enter once the (shadow) drawdown recovers inside this
 
     def _load(self) -> dict | None:
         if os.path.exists(self._path):
@@ -75,12 +78,34 @@ class ForwardTracker:
             state: dict[str, Any] = {"inception": latest, "last_date": latest, "nav": 1.0, "days": [], "extra": new_extra}
         else:
             state = loaded
+            if self._kill_dd is not None:
+                # lazy-init kill-switch state (supports tracks that predate the switch): the existing
+                # un-throttled NAV IS the shadow equity; its running max is the shadow peak.
+                shadow = float(state.get("shadow_nav", state["nav"]))
+                speak = float(state.get("shadow_peak",
+                                        max([d["nav"] for d in state["days"]] + [state["nav"]])))
+                killed = bool(state.get("killed", False))
             for d, r in rows:
                 if d > state["last_date"]:
-                    state["nav"] *= (1.0 + r)
-                    state["days"].append({"date": d, "ret": r, "nav": state["nav"]})
+                    if self._kill_dd is not None:
+                        was_killed = killed
+                        eff = 0.0 if was_killed else r        # decision uses prior-day kill-state
+                        state["nav"] *= (1.0 + eff)
+                        shadow *= (1.0 + r)                    # shadow tracks un-throttled = regime signal
+                        speak = max(speak, shadow)
+                        dd = (speak - shadow) / speak if speak > 0 else 0.0
+                        if not killed and dd > self._kill_dd:
+                            killed = True
+                        elif killed and dd < self._reenter_dd:
+                            killed = False
+                        state["days"].append({"date": d, "ret": eff, "nav": state["nav"], "killed": was_killed})
+                    else:
+                        state["nav"] *= (1.0 + r)
+                        state["days"].append({"date": d, "ret": r, "nav": state["nav"]})
                     state["last_date"] = d
                     booked += 1
+            if self._kill_dd is not None:
+                state["shadow_nav"], state["shadow_peak"], state["killed"] = shadow, speak, killed
             state["extra"] = new_extra
         self._save(state)
 
@@ -105,6 +130,8 @@ class CombinedBookStrategy:
         return rows, extra
 
 
-def CombinedBookForwardTracker(book: CombinedBook, state_path: str) -> ForwardTracker:  # noqa: N802
-    """Back-compat factory: the combined book is just a recomputable ForwardStrategy."""
-    return ForwardTracker(CombinedBookStrategy(book), state_path)
+def CombinedBookForwardTracker(book: CombinedBook, state_path: str, *,  # noqa: N802
+                               kill_dd: float | None = None, reenter_dd: float = 0.07) -> ForwardTracker:
+    """Back-compat factory: the combined book is just a recomputable ForwardStrategy. `kill_dd` (when
+    set) wires the regime drawdown kill-switch into the live track."""
+    return ForwardTracker(CombinedBookStrategy(book), state_path, kill_dd=kill_dd, reenter_dd=reenter_dd)

src/fxhnt/cli.py

@@ -213,9 +213,11 @@ def forward_track(
     from fxhnt.application.combined_book_factory import make_combined_book_source
     from fxhnt.application.forward_tracker import CombinedBookForwardTracker
 
+    _s = get_settings()
     book = CombinedBook(data_dir, _FUTURES_UNIVERSE,
-                        source=make_combined_book_source(get_settings(), data_dir))
-    st = CombinedBookForwardTracker(book, state).step()
+                        source=make_combined_book_source(_s, data_dir))
+    st = CombinedBookForwardTracker(book, state, kill_dd=_s.combined_book_kill_dd,
+                                    reenter_dd=_s.combined_book_reenter_dd).step()
     typer.echo(f"forward track since {st.inception} (last {st.last_date}): {st.forward_days} forward days, "
                f"booked {st.booked_today} today")
     typer.echo(f"  forward return {st.forward_return_pct:+.2f}%  Sharpe {st.forward_sharpe:+.2f}  "

src/fxhnt/config.py

@@ -113,6 +113,10 @@ class Settings(BaseSettings):
     # Combined-book data path: "raw" (crypto_pit/.dbn files) or "warehouse" (SSOT). Default raw for safety;
     # the B0 Dagster deploy sets FXHNT_COMBINED_BOOK_DATA_SOURCE=warehouse.
     combined_book_data_source: str = Field(default="raw")
+    # Regime kill-switch on the live combined-book track: flatten booked exposure once the forward
+    # drawdown exceeds combined_book_kill_dd, re-enter once it recovers inside combined_book_reenter_dd.
+    combined_book_kill_dd: float = Field(default=0.15)
+    combined_book_reenter_dd: float = Field(default=0.07)
     crypto_pit_dir: str = Field(default=str(_DATA_DIR / "crypto_pit"))   # FXHNT_CRYPTO_PIT_DIR; survivorship-free npz panel
 
     cost_bps_per_turnover: float = 10.0   # round-trip cost model (bps of traded notional)

tests/integration/test_forward_tracker.py

@@ -50,6 +50,22 @@ def test_idempotent_rerun_books_nothing(tmp_path) -> None:
     assert st.forward_days == 0
 
 
+def test_killswitch_flattens_live_track_and_reenters(tmp_path) -> None:
+    p = str(tmp_path / "s.json")
+    # ramp (peak) -> crash past kill_dd -> recovery; freeze on first run, then book forward
+    series = [(f"2026-01-{i+1:02d}", 0.01) for i in range(10)]                     # ramp
+    ForwardTracker(_Recomputable(series), p, kill_dd=0.15).step()                  # freeze at last ramp day
+    series += [(f"2026-02-{i+1:02d}", -0.06) for i in range(5)]                    # ~-26% crash -> breach
+    st = ForwardTracker(_Recomputable(series), p, kill_dd=0.15).step()
+    import json
+    days = json.load(open(p))["days"]
+    crash = [d for d in days if d["date"].startswith("2026-02")]
+    assert any(d["ret"] == 0.0 and d["killed"] for d in crash)                     # flattened during crash
+    # a flattened day books 0 even though the underlying return was -0.06
+    assert any(d["killed"] and d["ret"] == 0.0 for d in crash)
+    assert st.forward_days == 5
+
+
 def test_extra_carry_state_round_trips(tmp_path) -> None:
     p = str(tmp_path / "s.json")
     ForwardTracker(_LiveBooking("2026-01-01", 0.0), p).step()         # freeze, extra={"runs":1}