From 7b7dd61966c530ef9bf9b17d9e0ea0f8f1c25eb3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 18 Sep 2026 04:14:14 +0800 Subject: [PATCH] fix(risk): add daily loss new-risk gate Co-authored-by: Cursor --- .../risk/account_new_risk_gate.py | 32 ++++++- src/quant_platform_kit/risk/contracts.py | 7 ++ tests/test_account_new_risk_gate.py | 86 +++++++++++++++++++ 3 files changed, 124 insertions(+), 1 deletion(-) diff --git a/src/quant_platform_kit/risk/account_new_risk_gate.py b/src/quant_platform_kit/risk/account_new_risk_gate.py index 29b09b8..a9ebbda 100644 --- a/src/quant_platform_kit/risk/account_new_risk_gate.py +++ b/src/quant_platform_kit/risk/account_new_risk_gate.py @@ -33,6 +33,7 @@ from typing import Protocol from quant_platform_kit.risk.capital_risk_envelope import evaluate_capital_risk_envelope +from quant_platform_kit.risk.contracts import RuntimeRiskLimits from quant_platform_kit.risk.production_drift_new_risk import ( production_drift_new_risk_reasons, ) @@ -73,6 +74,7 @@ class InjectedReconciliationSnapshot: drawdown_from_peak: float | None = None realized_vol: float | None = None production_drift_status: str | None = None + daily_loss_usd: float | None = None class ReconciliationSnapshotReader(Protocol): @@ -174,8 +176,34 @@ def _evaluate_capital_axis( return reasons, envelope.combined_scale +def _evaluate_daily_loss_axis( + snapshot: InjectedReconciliationSnapshot, + runtime_risk_limits: RuntimeRiskLimits | None, +) -> list[str]: + """Omit an unconfigured axis; otherwise require a finite nonnegative fact.""" + if runtime_risk_limits is None: + return [] + if not isinstance(runtime_risk_limits, RuntimeRiskLimits): + raise AccountNewRiskGateError("runtime_risk_limits must be RuntimeRiskLimits") + limit = runtime_risk_limits.max_daily_loss_usd + if limit is None: + return [] + + daily_loss = snapshot.daily_loss_usd + if ( + type(daily_loss) not in (int, float) + or not math.isfinite(float(daily_loss)) + or float(daily_loss) < 0.0 + ): + return ["DAILY_LOSS_UNKNOWN_FAIL_CLOSED"] + if float(daily_loss) >= float(limit): + return ["DAILY_LOSS_LIMIT_EXCEEDED"] + return [] + + def evaluate_new_risk_admission( snapshot: InjectedReconciliationSnapshot, + runtime_risk_limits: RuntimeRiskLimits | None = None, ) -> NewRiskAdmissionResult: """Map unhealthy injected snapshots / capital envelope / actionable production drift to ``NEW_RISK_PROHIBITED``. @@ -196,6 +224,7 @@ def evaluate_new_risk_admission( reasons.append("CIRCUIT_BREAKER_OPEN") capital_reasons, combined_scale = _evaluate_capital_axis(validated) reasons.extend(capital_reasons) + reasons.extend(_evaluate_daily_loss_axis(validated, runtime_risk_limits)) reasons.extend(production_drift_new_risk_reasons(validated.production_drift_status)) if reasons: return NewRiskAdmissionResult( @@ -212,6 +241,7 @@ def evaluate_new_risk_admission( def evaluate_new_risk_from_reader( reader: ReconciliationSnapshotReader, + runtime_risk_limits: RuntimeRiskLimits | None = None, ) -> NewRiskAdmissionResult: """Read via injected adapter; any reader/validation failure ⇒ prohibited.""" try: @@ -222,7 +252,7 @@ def evaluate_new_risk_from_reader( raise AccountNewRiskGateError( f"reconciliation snapshot reader failed: {type(exc).__name__}" ) from exc - return evaluate_new_risk_admission(snapshot) + return evaluate_new_risk_admission(snapshot, runtime_risk_limits) __all__ = [ diff --git a/src/quant_platform_kit/risk/contracts.py b/src/quant_platform_kit/risk/contracts.py index b23833d..509c46f 100644 --- a/src/quant_platform_kit/risk/contracts.py +++ b/src/quant_platform_kit/risk/contracts.py @@ -203,6 +203,7 @@ class RuntimeRiskLimits: total_nominal_exposure_cap: float total_effective_exposure_cap: float max_positions: int + max_daily_loss_usd: float | None = None def __post_init__(self) -> None: if isinstance(self.allowed_symbols, (str, bytes)): @@ -256,6 +257,12 @@ def _mapping(value: Mapping[str, Any], field_name: str) -> dict[str, Any]: raise ValueError(f"{field_name} must be finite and nonnegative") if type(self.max_positions) is not int or isinstance(self.max_positions, bool) or self.max_positions < 0: raise ValueError("max_positions must be a nonnegative integer") + if self.max_daily_loss_usd is not None and ( + type(self.max_daily_loss_usd) not in (int, float) + or not math.isfinite(float(self.max_daily_loss_usd)) + or float(self.max_daily_loss_usd) <= 0.0 + ): + raise ValueError("max_daily_loss_usd must be finite and positive") object.__setattr__(self, "allowed_symbols", symbols) object.__setattr__(self, "product_leverage_factors", MappingProxyType(factors)) diff --git a/tests/test_account_new_risk_gate.py b/tests/test_account_new_risk_gate.py index cc16969..c878c06 100644 --- a/tests/test_account_new_risk_gate.py +++ b/tests/test_account_new_risk_gate.py @@ -11,6 +11,7 @@ evaluate_new_risk_admission, evaluate_new_risk_from_reader, ) +from quant_platform_kit.risk.contracts import RuntimeRiskLimits def _healthy(*, equity_usd: float | None = 40_000.0, **kwargs) -> InjectedReconciliationSnapshot: @@ -23,6 +24,18 @@ def _healthy(*, equity_usd: float | None = 40_000.0, **kwargs) -> InjectedReconc ) +def _runtime_limits(*, max_daily_loss_usd: float | None) -> RuntimeRiskLimits: + return RuntimeRiskLimits( + allowed_symbols=("SPY",), + product_leverage_factors={"SPY": 1}, + nominal_caps={"SPY": 1.0}, + total_nominal_exposure_cap=1.0, + total_effective_exposure_cap=1.0, + max_positions=1, + max_daily_loss_usd=max_daily_loss_usd, + ) + + class EvaluateNewRiskAdmissionTests(unittest.TestCase): def test_healthy_with_equity_allows_new_risk_without_side_effects(self) -> None: result = evaluate_new_risk_admission(_healthy()) @@ -167,6 +180,79 @@ def test_invalid_production_drift_status_fails_closed(self) -> None: self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED) self.assertIn("PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED", result.reason_codes) + def test_daily_loss_below_limit_allows_new_risk(self) -> None: + result = evaluate_new_risk_admission( + _healthy(daily_loss_usd=99.99), + runtime_risk_limits=_runtime_limits(max_daily_loss_usd=100.0), + ) + self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK) + self.assertEqual(result.reason_codes, ()) + + def test_daily_loss_at_or_above_limit_prohibits(self) -> None: + limits = _runtime_limits(max_daily_loss_usd=100.0) + for daily_loss_usd in (100.0, 100.01): + with self.subTest(daily_loss_usd=daily_loss_usd): + result = evaluate_new_risk_admission( + _healthy(daily_loss_usd=daily_loss_usd), + runtime_risk_limits=limits, + ) + self.assertEqual( + result.disposition, + NewRiskDisposition.NEW_RISK_PROHIBITED, + ) + self.assertIn("DAILY_LOSS_LIMIT_EXCEEDED", result.reason_codes) + + def test_missing_or_invalid_daily_loss_fails_closed_when_configured(self) -> None: + limits = _runtime_limits(max_daily_loss_usd=100.0) + for daily_loss_usd in (None, -1.0, float("nan"), float("inf"), True): + with self.subTest(daily_loss_usd=daily_loss_usd): + result = evaluate_new_risk_admission( + _healthy(daily_loss_usd=daily_loss_usd), + runtime_risk_limits=limits, + ) + self.assertEqual( + result.disposition, + NewRiskDisposition.NEW_RISK_PROHIBITED, + ) + self.assertIn( + "DAILY_LOSS_UNKNOWN_FAIL_CLOSED", + result.reason_codes, + ) + + def test_unconfigured_daily_loss_axis_is_omitted(self) -> None: + for limits in (None, _runtime_limits(max_daily_loss_usd=None)): + with self.subTest(limits=limits): + result = evaluate_new_risk_admission( + _healthy(daily_loss_usd=float("nan")), + runtime_risk_limits=limits, + ) + self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK) + self.assertNotIn( + "DAILY_LOSS_UNKNOWN_FAIL_CLOSED", + result.reason_codes, + ) + self.assertNotIn("DAILY_LOSS_LIMIT_EXCEEDED", result.reason_codes) + + def test_daily_loss_reason_coexists_with_other_axes(self) -> None: + result = evaluate_new_risk_admission( + _healthy( + daily_loss_usd=100.0, + production_drift_status="critical", + ), + runtime_risk_limits=_runtime_limits(max_daily_loss_usd=100.0), + ) + self.assertEqual( + result.reason_codes, + ("DAILY_LOSS_LIMIT_EXCEEDED", "PRODUCTION_DRIFT_CRITICAL"), + ) + + +class RuntimeRiskLimitsDailyLossTests(unittest.TestCase): + def test_max_daily_loss_must_be_finite_positive_when_configured(self) -> None: + for value in (0.0, -1.0, float("nan"), float("inf"), True): + with self.subTest(value=value), self.assertRaises(ValueError): + _runtime_limits(max_daily_loss_usd=value) + class ReaderInjectionTests(unittest.TestCase): def test_reader_unhealthy_snapshot_prohibits(self) -> None: