fix(risk): IBKR daily_loss NEW_RISK inject seam

application/account_new_risk_gate_support.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import math
 import os
 from collections.abc import Mapping
@@ -15,6 +16,7 @@
     NewRiskDisposition,
     evaluate_new_risk_admission,
 )
+from quant_platform_kit.risk.contracts import RuntimeRiskLimits
 from quant_platform_kit.risk.cycle_new_risk_health import (
     CycleNewRiskHealthEvidence,
     apply_cycle_new_risk_health_axes,
@@ -24,10 +26,14 @@
 )
 
 ACCOUNT_NEW_RISK_GATE_ENV = "ACCOUNT_NEW_RISK_GATE"
+_MAX_DAILY_LOSS_ENV_KEYS = ("IBKR_MAX_DAILY_LOSS_USD", "MAX_DAILY_LOSS_USD")
 
 _DEFAULT_STRATEGY_PROFILE = "soxl_soxx_trend_income"
 _DEFAULT_DOMAIN = "us_equity"
 
+# Minimal carrier for gate-only daily-loss axis; not a production RRL binding.
+_DAILY_LOSS_LIMIT_CARRIER_SYMBOL = "SPY"
+
 _cycle_snapshot: InjectedReconciliationSnapshot | None = None
 
 
@@ -48,6 +54,18 @@ def _coerce_optional_float(value: object) -> float | None:
     return number
 
 
+def _positive_limit_or_none(value: object) -> float | None:
+    """Accept only finite positive limits; never invent a production default."""
+    number = _coerce_optional_float(value)
+    if number is None or number <= 0.0:
+        return None
+    return number
+
+
+def _mapping_or_empty(value: object) -> Mapping[str, Any]:
+    return value if isinstance(value, Mapping) else {}
+
+
 def _resolve_equity_usd(portfolio: Mapping[str, Any], execution: Mapping[str, Any] | None) -> float | None:
     for key in ("total_equity", "total_strategy_equity", "equity"):
         equity = _coerce_optional_float(portfolio.get(key))
@@ -184,6 +202,83 @@ def _resolve_drawdown_from_peak(
     return max(0.0, 1.0 - (equity_usd / peak_equity_usd))
 
 
+def _resolve_explicit_daily_loss_usd(
+    projection: Mapping[str, Any],
+    portfolio: Mapping[str, Any],
+    execution: Mapping[str, Any] | None,
+) -> float | None:
+    """Pass through an explicit daily_loss_usd fact only; never invent one."""
+    for source in (projection, portfolio, _mapping_or_empty(execution)):
+        if "daily_loss_usd" in source:
+            return _coerce_optional_float(source.get("daily_loss_usd"))
+    return None
+
+
+def _max_daily_loss_from_runtime_target_json() -> float | None:
+    """Read max_daily_loss_usd from RUNTIME_TARGET_JSON when present; soft-omit on errors."""
+    raw_target = os.environ.get("RUNTIME_TARGET_JSON")
+    if raw_target is None or not str(raw_target).strip():
+        return None
+    try:
+        payload = json.loads(raw_target)
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+    policy = payload.get("runtime_risk_limits")
+    if not isinstance(policy, dict) or "max_daily_loss_usd" not in policy:
+        return None
+    return _positive_limit_or_none(policy.get("max_daily_loss_usd"))
+
+
+def resolve_max_daily_loss_usd(
+    portfolio: Mapping[str, Any] | None = None,
+) -> float | None:
+    """Resolve an explicit max_daily_loss_usd; omit the axis when unset.
+
+    Priority: account_new_risk_snapshot / portfolio key → RUNTIME_TARGET_JSON →
+    IBKR_MAX_DAILY_LOSS_USD / MAX_DAILY_LOSS_USD. No approved production default.
+    """
+    if portfolio is not None:
+        projection = _mapping_or_empty(portfolio.get("account_new_risk_snapshot"))
+        for source in (projection, portfolio):
+            if "max_daily_loss_usd" in source:
+                return _positive_limit_or_none(source.get("max_daily_loss_usd"))
+    policy_limit = _max_daily_loss_from_runtime_target_json()
+    if policy_limit is not None:
+        return policy_limit
+    for key in _MAX_DAILY_LOSS_ENV_KEYS:
+        raw = os.environ.get(key)
+        if raw is None or not str(raw).strip():
+            continue
+        limit = _positive_limit_or_none(raw)
+        if limit is not None:
+            return limit
+    return None
+
+
+def runtime_risk_limits_for_daily_loss_axis(
+    max_daily_loss_usd: float | None,
+) -> RuntimeRiskLimits | None:
+    """Build admission-only limits carrying ``max_daily_loss_usd``, or omit.
+
+    SPY/1.0 caps are a minimal legal RuntimeRiskLimits carrier for the gate only —
+    not a production RRL binding and not an exposure raise.
+    """
+    if max_daily_loss_usd is None:
+        return None
+    symbol = _DAILY_LOSS_LIMIT_CARRIER_SYMBOL
+    return RuntimeRiskLimits(
+        allowed_symbols=(symbol,),
+        product_leverage_factors={symbol: 1},
+        nominal_caps={symbol: 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,
+    )
+
+
 def build_snapshot_from_portfolio(
     portfolio: Mapping[str, Any],
     *,
@@ -219,6 +314,7 @@ def build_snapshot_from_portfolio(
         if "realized_vol" in projection
         else _coerce_optional_float(portfolio.get("realized_vol")),
         production_drift_status=_resolve_production_drift_status(portfolio, projection),
+        daily_loss_usd=_resolve_explicit_daily_loss_usd(projection, portfolio, execution),
     )
 
 
@@ -229,7 +325,8 @@ def evaluate_portfolio_new_risk_admission(
 ) -> NewRiskAdmissionResult:
     try:
         snapshot = build_snapshot_from_portfolio(portfolio, execution=execution)
-        return evaluate_new_risk_admission(snapshot)
+        limits = runtime_risk_limits_for_daily_loss_axis(resolve_max_daily_loss_usd(portfolio))
+        return evaluate_new_risk_admission(snapshot, limits)
     except AccountNewRiskGateError:
         return NewRiskAdmissionResult(
             disposition=NewRiskDisposition.NEW_RISK_PROHIBITED,
@@ -289,7 +386,8 @@ def evaluate_cycle_new_risk_admission() -> NewRiskAdmissionResult:
             reason_codes=("EQUITY_UNKNOWN_FAIL_CLOSED",),
         )
     try:
-        return evaluate_new_risk_admission(_cycle_snapshot)
+        limits = runtime_risk_limits_for_daily_loss_axis(resolve_max_daily_loss_usd())
+        return evaluate_new_risk_admission(_cycle_snapshot, limits)
     except AccountNewRiskGateError:
         return NewRiskAdmissionResult(
             disposition=NewRiskDisposition.NEW_RISK_PROHIBITED,

strategy_runtime.py

@@ -489,7 +489,7 @@ def _build_runtime_risk_capabilities(
             "max_positions",
             "exit_parameters",
         }
-        optional_policy_keys = {"small_account_hold"}
+        optional_policy_keys = {"small_account_hold", "max_daily_loss_usd"}
         policy_keys = set(policy)
         if (
             not expected_policy_keys.issubset(policy_keys)
@@ -545,13 +545,17 @@ def _build_runtime_risk_capabilities(
             )
             return {**capabilities, "runtime_risk_limits": object()}, "unavailable:runtime_binding_mismatch"
         try:
+            daily_loss_kwargs: dict[str, Any] = {}
+            if "max_daily_loss_usd" in policy:
+                daily_loss_kwargs["max_daily_loss_usd"] = policy.get("max_daily_loss_usd")
             limits = RuntimeRiskLimits(
                 allowed_symbols=tuple(policy["allowed_symbols"]),
                 product_leverage_factors=policy["product_leverage_factors"],
                 nominal_caps=policy["nominal_caps"],
                 total_nominal_exposure_cap=policy["total_nominal_exposure_cap"],
                 total_effective_exposure_cap=policy["total_effective_exposure_cap"],
                 max_positions=policy["max_positions"],
+                **daily_loss_kwargs,
             )
         except (TypeError, ValueError):
             return {**capabilities, "runtime_risk_limits": object()}, "unavailable:invalid_runtime_risk_limits"

tests/test_account_new_risk_gate.py

@@ -18,10 +18,12 @@
     apply_combined_scale,
     build_account_new_risk_snapshot,
     build_portfolio_from_account_values,
+    build_snapshot_from_portfolio,
     evaluate_account_values_new_risk_admission,
     evaluate_cycle_new_risk_admission,
     evaluate_portfolio_new_risk_admission,
     is_account_new_risk_gate_enabled,
+    new_risk_buy_prohibited,
     set_cycle_snapshot,
 )
 from application.ibkr_order_execution import submit_order_intent
@@ -30,8 +32,20 @@
 @pytest.fixture(autouse=True)
 def _clear_cycle_snapshot():
     set_cycle_snapshot(None)
+    for key in (
+        "IBKR_MAX_DAILY_LOSS_USD",
+        "MAX_DAILY_LOSS_USD",
+        "RUNTIME_TARGET_JSON",
+    ):
+        os.environ.pop(key, None)
     yield
     set_cycle_snapshot(None)
+    for key in (
+        "IBKR_MAX_DAILY_LOSS_USD",
+        "MAX_DAILY_LOSS_USD",
+        "RUNTIME_TARGET_JSON",
+    ):
+        os.environ.pop(key, None)
 
 
 def test_gate_enabled_by_default():
@@ -141,6 +155,60 @@ def test_explicit_healthy_snapshot_allows_new_risk():
     assert result.live_authority_granted is False
 
 
+def test_explicit_daily_loss_at_limit_prohibits_buy():
+    portfolio = {
+        "total_equity": 50_000.0,
+        "peak_equity_usd": 50_000.0,
+        "account_new_risk_snapshot": {
+            "daily_loss_usd": 100.0,
+            "max_daily_loss_usd": 100.0,
+        },
+    }
+    with mock.patch(
+        "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+        return_value=None,
+    ):
+        snapshot = build_snapshot_from_portfolio(portfolio)
+        result = evaluate_portfolio_new_risk_admission(portfolio)
+    assert snapshot.daily_loss_usd == 100.0
+    assert result.disposition == NewRiskDisposition.NEW_RISK_PROHIBITED
+    assert "DAILY_LOSS_LIMIT_EXCEEDED" in result.reason_codes
+    assert new_risk_buy_prohibited(result)
+
+
+def test_unconfigured_daily_loss_limit_omits_axis():
+    portfolio = {
+        "total_equity": 50_000.0,
+        "peak_equity_usd": 50_000.0,
+        # daily_loss fact absent / invalid must not invent a prohibit when
+        # no max_daily_loss_usd is configured.
+        "account_new_risk_snapshot": {"daily_loss_usd": float("nan")},
+    }
+    with mock.patch(
+        "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+        return_value=None,
+    ):
+        result = evaluate_portfolio_new_risk_admission(portfolio)
+    assert result.disposition == NewRiskDisposition.ALLOW_NEW_RISK
+    assert "DAILY_LOSS_UNKNOWN_FAIL_CLOSED" not in result.reason_codes
+    assert "DAILY_LOSS_LIMIT_EXCEEDED" not in result.reason_codes
+
+
+def test_configured_limit_without_daily_loss_fact_fails_closed():
+    portfolio = {
+        "total_equity": 50_000.0,
+        "peak_equity_usd": 50_000.0,
+        "account_new_risk_snapshot": {"max_daily_loss_usd": 100.0},
+    }
+    with mock.patch(
+        "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+        return_value=None,
+    ):
+        result = evaluate_portfolio_new_risk_admission(portfolio)
+    assert result.disposition == NewRiskDisposition.NEW_RISK_PROHIBITED
+    assert "DAILY_LOSS_UNKNOWN_FAIL_CLOSED" in result.reason_codes
+
+
 def test_unknown_pending_orders_prohibits_and_opens_breaker():
     portfolio = {
         "total_equity": 50_000.0,