From bb4ec3e2ae2eeb3afe943d0211edc10e25ac442d Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 19 Jun 2026 23:22:15 +0200 Subject: [PATCH] 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) --- src/fxhnt/adapters/orchestration/assets.py | 10 ++++-- src/fxhnt/application/forward_tracker.py | 39 ++++++++++++++++++---- src/fxhnt/cli.py | 6 ++-- src/fxhnt/config.py | 4 +++ tests/integration/test_forward_tracker.py | 16 +++++++++ 5 files changed, 64 insertions(+), 11 deletions(-) diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py index 5a4d4d7..6630b13 100644 --- a/src/fxhnt/adapters/orchestration/assets.py +++ b/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}" diff --git a/src/fxhnt/application/forward_tracker.py b/src/fxhnt/application/forward_tracker.py index 1b176fd..35c0bb6 100644 --- a/src/fxhnt/application/forward_tracker.py +++ b/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) diff --git a/src/fxhnt/cli.py b/src/fxhnt/cli.py index 31ae707..21ad95c 100644 --- a/src/fxhnt/cli.py +++ b/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} " diff --git a/src/fxhnt/config.py b/src/fxhnt/config.py index 0c9b0e1..d211857 100644 --- a/src/fxhnt/config.py +++ b/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) diff --git a/tests/integration/test_forward_tracker.py b/tests/integration/test_forward_tracker.py index b6763d9..69130ea 100644 --- a/tests/integration/test_forward_tracker.py +++ b/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}