From 6a3a15baea6b7ef91ceea5a45f0dc41fccd1ac6b Mon Sep 17 00:00:00 2001 From: Brian McMahon Date: Mon, 18 May 2026 18:48:51 -0700 Subject: [PATCH] feat(regime): drawdown-leg predictor-veto clamp (PR 3, gated default-off) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR 3 of the drawdown-regime arc — first consumer. Mirrors the Stage-F forced-bear veto clamp. Gated `drawdown_regime_enabled` (S3 predictor_params, default False) → zero behavior change on merge. Plan: regime-drawdown-hysteresis-260518.md; ROADMAP config #230. - get_veto_threshold: new drawdown_effective_regime param. When drawdown_regime_enabled AND the composed daily drawdown effective regime (ctx.drawdown_effective_regime, stamped by PR 2's daily stage) is protective, clamp the veto threshold — bear → the same bear floor as forced-bear (base-cap); caution → a milder floor (base-cap/2), mirroring the discrete -0.20/-0.10 relationship. - Refactored the clamp tail from the forced-bear early-return into a single most-protective accumulator (min over {Wire4/discrete, forced-bear, drawdown}). Behavior-preserving for every currently-reachable state: with the drawdown flag off (always, today) the return is identical to before; verified by the existing forced-bear/Wire-4 tests + a new all-off regression test. - Flag off ⇒ parallel-observe: the would-be clamp is logged (for the promotion gate) without changing the threshold. - call site passes ctx.drawdown_effective_regime (getattr default None). - tests: +7 (TestDrawdownVetoClamp) — flag-off observe-only, bear/ caution floors, benign no-clamp, most-protective vs existing threshold, composition with forced-bear, all-off regression. 361 veto/write_output/regime/daily_predict tests pass, zero failures. Co-Authored-By: Claude Opus 4.7 (1M context) --- inference/stages/write_output.py | 71 +++++++++++++++++++++++++------- tests/test_veto_threshold.py | 71 ++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 16 deletions(-) diff --git a/inference/stages/write_output.py b/inference/stages/write_output.py index 5f7c06b..7304002 100644 --- a/inference/stages/write_output.py +++ b/inference/stages/write_output.py @@ -116,6 +116,7 @@ def get_veto_threshold( market_regime: str = "", regime_intensity_z: float | None = None, forced_bear: bool = False, + drawdown_effective_regime: str | None = None, ) -> float: """ Return the active veto confidence threshold, adjusted by market regime. @@ -159,6 +160,20 @@ def get_veto_threshold( Default ``regime_forced_bear_enabled=False``: the clamp is dormant but, when ``forced_bear=True``, the counterfactual is logged (parallel-observe) so the F2 gate has data during the gated window. + + Drawdown-regime clamp (regime-drawdown-hysteresis-260518.md, PR 3): + When ``drawdown_regime_enabled=True`` in S3 predictor_params AND + the composed daily drawdown effective regime + (``ctx.drawdown_effective_regime``, stamped by the daily stage) + is protective, the threshold is clamped — ``bear`` → the same + bear floor as forced-bear (``base - cap``), ``caution`` → a milder + floor (``base - cap/2``), mirroring the discrete −0.20/−0.10 + relationship. Composed most-protective with the forced-bear clamp + and the Wire-4/discrete result: all clamps accumulate via ``min`` + (lower threshold = more aggressive veto); none is half-honored. + + Default ``drawdown_regime_enabled=False``: dormant; the + counterfactual is logged (parallel-observe) for the promotion gate. """ params = _load_predictor_params_from_s3(s3_bucket) if params and "veto_confidence" in params: @@ -204,30 +219,51 @@ def get_veto_threshold( base, adjustment, regime, threshold, ) - # ── F2 forced-bear max-protection clamp ────────────────────────────── + # ── Max-protection clamps (forced-bear + drawdown) ─────────────────── + # Both accumulate via min() so the most-protective wins and neither + # is half-honored. Each is independently gated default-OFF; flag-off + # logs the counterfactual (parallel-observe) without changing + # behavior. With both flags off (the only live state today) the + # return value is exactly ``threshold`` — unchanged. + effective = threshold + if forced_bear: bear_floor = max(0.0, min(0.80, base - cap)) clamped = min(threshold, bear_floor) # lower = more aggressive veto - clamp_enabled = bool( - params and params.get("regime_forced_bear_enabled", False) - ) - if clamp_enabled: - if clamped < threshold: + if bool(params and params.get("regime_forced_bear_enabled", False)): + if clamped < effective: log.warning( "Veto threshold FORCED-BEAR clamp: %.3f → %.3f " "(bear_floor=base-cap=%.2f-%.2f)", - threshold, clamped, base, cap, + effective, clamped, base, cap, ) - return clamped - # Parallel-observe: flag off, behavior unchanged, log the - # counterfactual so the F2 gate can measure the would-be effect. - log.info( - "regime_forced_bear OBSERVE (flag off): forced_bear=True, " - "veto would clamp %.3f → %.3f (bear_floor=%.3f)", - threshold, clamped, bear_floor, - ) + effective = min(effective, clamped) + else: + log.info( + "regime_forced_bear OBSERVE (flag off): forced_bear=True, " + "veto would clamp %.3f → %.3f (bear_floor=%.3f)", + threshold, clamped, bear_floor, + ) + + dd = (drawdown_effective_regime or "").lower().strip() + if dd in ("bear", "caution"): + dd_floor = max(0.0, min(0.80, base - (cap if dd == "bear" else cap / 2.0))) + dd_clamped = min(threshold, dd_floor) + if bool(params and params.get("drawdown_regime_enabled", False)): + if dd_clamped < effective: + log.warning( + "Veto threshold DRAWDOWN clamp (%s): %.3f → %.3f " + "(dd_floor=%.3f)", dd, effective, dd_clamped, dd_floor, + ) + effective = min(effective, dd_clamped) + else: + log.info( + "drawdown_regime OBSERVE (flag off): effective_regime=%s, " + "veto would clamp %.3f → %.3f (dd_floor=%.3f)", + dd, threshold, dd_clamped, dd_floor, + ) - return threshold + return effective # ── Output writing (migrated from daily_predict.py) ────────────────────────── @@ -1037,6 +1073,9 @@ def run(ctx: PipelineContext) -> None: market_regime=market_regime, regime_intensity_z=getattr(ctx, "regime_intensity_z", None), forced_bear=getattr(ctx, "regime_forced_bear", False), + drawdown_effective_regime=getattr( + ctx, "drawdown_effective_regime", None + ), ) # Veto fires only when the model is BOTH bearish AND confident. diff --git a/tests/test_veto_threshold.py b/tests/test_veto_threshold.py index c62dc2e..fe80e1b 100644 --- a/tests/test_veto_threshold.py +++ b/tests/test_veto_threshold.py @@ -84,3 +84,74 @@ def test_caches_after_first_call(self): wo._predictor_params_cache = {"veto_confidence": 0.70} result = _load_predictor_params_from_s3("bucket") assert result == {"veto_confidence": 0.70} + + +class TestDrawdownVetoClamp: + """PR 3: drawdown effective-regime veto clamp (gated default-off, + most-protective composition with the forced-bear clamp).""" + + # base=0.30, cap=0.20 ⇒ bear_floor=0.10, caution_floor=0.20. + _OFF = {"veto_confidence": 0.30} + _ON = {"veto_confidence": 0.30, "drawdown_regime_enabled": True} + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_flag_off_is_observe_only_no_behavior_change(self, m): + m.return_value = self._OFF + # dd=bear but flag off ⇒ threshold unchanged (0.30); counterfactual + # only logged. + assert get_veto_threshold( + "b", drawdown_effective_regime="bear" + ) == pytest.approx(0.30) + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_flag_on_bear_clamps_to_bear_floor(self, m): + m.return_value = self._ON + assert get_veto_threshold( + "b", drawdown_effective_regime="bear" + ) == pytest.approx(0.10) + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_flag_on_caution_clamps_to_milder_floor(self, m): + m.return_value = self._ON + assert get_veto_threshold( + "b", drawdown_effective_regime="caution" + ) == pytest.approx(0.20) + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_flag_on_benign_regimes_no_clamp(self, m): + m.return_value = self._ON + for r in ("bull", "neutral", None, ""): + assert get_veto_threshold( + "b", drawdown_effective_regime=r + ) == pytest.approx(0.30) + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_most_protective_min_with_existing_threshold(self, m): + # bull market_regime would raise threshold to 0.40, but dd=bear + # clamp wins (min ⇒ 0.10). + m.return_value = self._ON + assert get_veto_threshold( + "b", market_regime="bull", drawdown_effective_regime="bear" + ) == pytest.approx(0.10) + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_composition_with_forced_bear_is_most_protective(self, m): + # forced_bear (enabled) → bear_floor 0.10; dd=caution → 0.20. + # Most-protective = min = 0.10. + m.return_value = { + "veto_confidence": 0.30, + "drawdown_regime_enabled": True, + "regime_forced_bear_enabled": True, + } + assert get_veto_threshold( + "b", forced_bear=True, drawdown_effective_regime="caution" + ) == pytest.approx(0.10) + + @patch.object(wo, "_load_predictor_params_from_s3") + def test_both_flags_off_default_path_unchanged(self, m): + # Regression: the accumulation refactor must not change the + # all-off path (forced_bear False, no dd) — plain base. + m.return_value = {"veto_confidence": 0.30} + assert get_veto_threshold("b", market_regime="neutral") == pytest.approx( + 0.30 + )