fix(risk): add daily loss NEW_RISK contract

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__ = [

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))

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: