From d04c28a148db6cc7375cd8b5c8c3f2b81fcbdb72 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 19 Jun 2026 23:16:17 +0200 Subject: [PATCH] feat(surfer): regime drawdown kill-switch for the book apply_drawdown_killswitch: flatten the book once running drawdown exceeds kill_dd (default 15%), re-enter only after recovery inside reenter_dd (7%) hysteresis. The drawdown signal tracks un-throttled equity so recovery is observable while flat; output is the throttled series; no lookahead (each day uses prior-day kill-state). The small- capital tail control - survive a crypto-deleveraging by standing down, not by paying for a hedge sleeve. Pure + composable overlay on combine_book output. Test + mypy green. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fxhnt/application/book_allocator.py | 27 ++++++++++++++++++++++++ tests/integration/test_book_allocator.py | 21 ++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/src/fxhnt/application/book_allocator.py b/src/fxhnt/application/book_allocator.py index b4bb850..b19e16f 100644 --- a/src/fxhnt/application/book_allocator.py +++ b/src/fxhnt/application/book_allocator.py @@ -165,6 +165,33 @@ def combine_book(series_by_edge: dict[str, EdgeReturns], *, vol_lookback: int = return book +def apply_drawdown_killswitch(book: dict[int, float], *, kill_dd: float = 0.15, + reenter_dd: float = 0.07) -> dict[int, float]: + """Regime kill-switch: flatten the book (return 0) once its running drawdown exceeds `kill_dd`, + re-entering only after the drawdown recovers back inside `reenter_dd` (hysteresis avoids whipsaw). + + The drawdown SIGNAL tracks the UN-throttled book equity (so recovery is observable while flat); + the OUTPUT is the throttled series. No lookahead: each day's output uses the kill-state set by + strictly prior days. This is the small-capital tail control — survive a crypto-deleveraging by + standing down, not by buying an expensive hedge. Returns {epoch_day: throttled_return}. + """ + killed = False + shadow_eq = 1.0 + peak = 1.0 + out: dict[int, float] = {} + for d in sorted(book): + r = book[d] + out[d] = 0.0 if killed else r # decision uses kill-state from prior days only + shadow_eq *= (1.0 + r) # un-throttled equity = regime signal + peak = max(peak, shadow_eq) + dd = (peak - shadow_eq) / peak if peak > 0 else 0.0 + if not killed and dd > kill_dd: + killed = True + elif killed and dd < reenter_dd: + killed = False + return out + + def _json_floats(d: dict[str, Any]) -> dict[str, Any]: """Coerce model_dump scalars to JSON-native: numpy → python float, NaN/inf → 0.0.""" out: dict[str, Any] = {} diff --git a/tests/integration/test_book_allocator.py b/tests/integration/test_book_allocator.py index d05971d..a5aea00 100644 --- a/tests/integration/test_book_allocator.py +++ b/tests/integration/test_book_allocator.py @@ -5,11 +5,32 @@ import numpy as np from fxhnt.application.book_allocator import ( align_edges, + apply_drawdown_killswitch, combine_book, evaluate_book, ) +def test_drawdown_killswitch_flattens_and_reenters(): + # a sharp drawdown then recovery: killswitch flattens during the crash, re-enters after recovery + book = {} + d = 0 + for r in [0.01] * 20: # ramp up (peak) + book[d] = r; d += 1 + for r in [-0.05] * 5: # -25% crash -> breach kill_dd=0.15 + book[d] = r; d += 1 + for r in [0.02] * 30: # recovery + book[d] = r; d += 1 + out = apply_drawdown_killswitch(book, kill_dd=0.15, reenter_dd=0.07) + days = sorted(book) + # no lookahead: identical until the kill triggers + assert out[days[0]] == book[days[0]] + # at least one day is flattened to 0 during/after the crash + assert any(out[k] == 0.0 and book[k] != 0.0 for k in days) + # after full recovery the switch re-enters (last day passes through) + assert out[days[-1]] == book[days[-1]] + + def _synth_edge(seed: int, n: int = 2000, drift: float = 0.0008, vol: float = 0.008, start: int = 1000) -> dict[int, float]: """A synthetic daily net-return series (positive drift + iid noise) keyed by epoch-day.