diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 5dddfb1..e3fe021 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -33,6 +33,9 @@ ("win_rate_drift", "win_rate", "win_rate", "win_rate", "win_rate_deviation_pct"), ] +# Reject apples-to-oranges baselines (e.g. low-vol proxy vs leveraged live path). +_BASELINE_VOL_SCALE_LIMIT = 5.0 + def _baseline_artifact_id(backtest: BacktestResult | None) -> str | None: if backtest is None: @@ -40,6 +43,24 @@ def _baseline_artifact_id(backtest: BacktestResult | None) -> str | None: return backtest.run_id or backtest.computed_at or None +def _baseline_vol_scale_incompatible( + actual_volatility: object, + expected_volatility: object, + *, + limit: float = _BASELINE_VOL_SCALE_LIMIT, +) -> bool: + """True when live vs baseline volatility differ by more than ``limit``×.""" + try: + actual = abs(float(actual_volatility)) # type: ignore[arg-type] + expected = abs(float(expected_volatility)) # type: ignore[arg-type] + except (TypeError, ValueError): + return False + if actual < 1e-6 or expected < 1e-6: + return False + ratio = max(actual / expected, expected / actual) + return ratio >= float(limit) + + def _compute_dimension( key: str, metric: str, actual_val: float, expected_val: float, threshold: float, @@ -112,6 +133,47 @@ def detect_drift( source_revision=snapshot.source_revision, drift_score=0.0, status=DriftStatus.HEALTHY) + if backtest is not None and _baseline_vol_scale_incompatible( + getattr(ref_window, "volatility", None), + getattr(backtest, "volatility", None), + ): + # Do not score leveraged live paths against low-vol proxy baselines. + # Preserve any prior restrictive status so Policy A cannot clear a ban + # via a bogus healthy comparison. + actual_vol = float(ref_window.volatility) + expected_vol = float(backtest.volatility or 0.0) + ratio = max(actual_vol, 1e-6) / max(abs(expected_vol), 1e-6) + if ratio < 1.0: + ratio = 1.0 / ratio + status = DriftStatus.REVIEW + if previous_status is not None and previous_status.severity_order > status.severity_order: + status = previous_status + score = 1.0 if status is DriftStatus.CRITICAL else 0.5 + return DriftResult( + strategy_profile=snapshot.strategy_profile, + domain=snapshot.domain, + as_of=snapshot.as_of, + drift_score=score, + source_revision=snapshot.source_revision, + status=status, + dimensions={ + "baseline_scale_incompatible": DriftDimension( + metric_name="volatility_scale", + actual=actual_vol, + expected=expected_vol, + deviation=abs(actual_vol - expected_vol), + deviation_pct=ratio - 1.0, + threshold=float(_BASELINE_VOL_SCALE_LIMIT), + breached=True, + ) + }, + previous_status=previous_status, + baseline_param_set_id=backtest.param_set_id, + baseline_param_version=backtest.param_version, + baseline_artifact_id=_baseline_artifact_id(backtest), + escalated=False, + ) + # Compute each dimension via registry dimensions: dict[str, DriftDimension] = {} for key, metric, actual_attr, expected_attr, threshold_attr in _DIMENSION_SPECS: diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 13bf2d2..cec1aae 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -79,6 +79,22 @@ def test_drawdown_breach_detected(self) -> None: self.assertIn("max_drawdown_breach", result.dimensions) self.assertTrue(result.dimensions["max_drawdown_breach"].breached) + def test_baseline_vol_scale_incompatible_preserves_critical(self) -> None: + # Leveraged live (~96% vol) vs low-vol proxy baseline (~7% vol). + snap = _make_snapshot(sharpe=1.7, cagr=2.2, dd=-0.48, vol=0.96, wr=0.63) + bt = _make_backtest(sharpe=0.5, cagr=0.035, dd=-0.05, vol=0.07, wr=0.52) + result = detect_drift(snap, backtest=bt, previous_status=DriftStatus.CRITICAL) + self.assertEqual(result.status, DriftStatus.CRITICAL) + self.assertIn("baseline_scale_incompatible", result.dimensions) + self.assertNotIn("cagr_drift", result.dimensions) + + def test_baseline_vol_scale_incompatible_defaults_to_review(self) -> None: + snap = _make_snapshot(sharpe=1.7, cagr=2.2, dd=-0.48, vol=0.96, wr=0.63) + bt = _make_backtest(sharpe=0.5, cagr=0.035, dd=-0.05, vol=0.07, wr=0.52) + result = detect_drift(snap, backtest=bt) + self.assertEqual(result.status, DriftStatus.REVIEW) + self.assertIn("baseline_scale_incompatible", result.dimensions) + def test_no_backtest_returns_healthy(self) -> None: snap = _make_snapshot(sharpe=0.6, cagr=0.01) result = detect_drift(snap, backtest=None)