fix(risk): Policy A production-drift inject for NEW_RISK gate

application/account_new_risk_gate_support.py

@@ -19,9 +19,15 @@
     CycleNewRiskHealthEvidence,
     apply_cycle_new_risk_health_axes,
 )
+from quant_platform_kit.risk.production_drift_new_risk import (
+    resolve_production_drift_status_from_store,
+)
 
 ACCOUNT_NEW_RISK_GATE_ENV = "ACCOUNT_NEW_RISK_GATE"
 
+_DEFAULT_STRATEGY_PROFILE = "soxl_soxx_trend_income"
+_DEFAULT_DOMAIN = "us_equity"
+
 _cycle_snapshot: InjectedReconciliationSnapshot | None = None
 
 
@@ -157,12 +163,41 @@ def build_account_new_risk_snapshot(
 
 
 def _resolve_production_drift_status(portfolio: Mapping[str, Any], projection: Mapping[str, Any]) -> str | None:
-    raw = projection.get("production_drift_status")
-    if raw is None or raw == "":
-        raw = portfolio.get("production_drift_status")
-    if raw is None or raw == "":
+    """Prefer explicit inject; else read-only PerformanceStore (Policy A). Fail-soft."""
+    for source in (projection, portfolio, _mapping_or_empty(portfolio.get("metadata"))):
+        raw = source.get("production_drift_status")
+        if raw is not None and raw != "":
+            return str(raw).strip()
+    profile = ""
+    domain = ""
+    for source in (projection, portfolio, _mapping_or_empty(portfolio.get("metadata"))):
+        if not profile:
+            value = source.get("strategy_profile")
+            if isinstance(value, str) and value.strip():
+                profile = value.strip()
+        if not domain:
+            value = source.get("strategy_domain") or source.get("domain")
+            if isinstance(value, str) and value.strip():
+                domain = value.strip()
+    profile = profile or str(os.environ.get("STRATEGY_PROFILE") or "").strip() or _DEFAULT_STRATEGY_PROFILE
+    domain = domain or str(os.environ.get("STRATEGY_DOMAIN") or "").strip() or _DEFAULT_DOMAIN
+    return resolve_production_drift_status_from_store(
+        strategy_profile=profile,
+        domain=domain,
+    )
+
+
+def _resolve_drawdown_from_peak(
+    *,
+    equity_usd: float | None,
+    peak_equity_usd: float | None,
+    explicit: float | None,
+) -> float | None:
+    if explicit is not None:
+        return explicit
+    if equity_usd is None or peak_equity_usd is None or peak_equity_usd <= 0.0:
         return None
-    return str(raw).strip()
+    return max(0.0, 1.0 - (equity_usd / peak_equity_usd))
 
 
 def build_snapshot_from_portfolio(
@@ -175,17 +210,27 @@ def build_snapshot_from_portfolio(
     equity_usd = _coerce_optional_float(projection.get("equity_usd"))
     if equity_usd is None:
         equity_usd = _resolve_equity_usd(portfolio, execution)
+    peak_equity_usd = (
+        _coerce_optional_float(projection.get("peak_equity_usd"))
+        if "peak_equity_usd" in projection
+        else _coerce_optional_float(portfolio.get("peak_equity_usd"))
+    )
+    explicit_dd = (
+        _coerce_optional_float(projection.get("drawdown_from_peak"))
+        if "drawdown_from_peak" in projection
+        else _coerce_optional_float(portfolio.get("drawdown_from_peak"))
+    )
     return InjectedReconciliationSnapshot(
         observation_status=str(projection.get("observation_status") or "UNAVAILABLE"),
         reconciliation_status=str(projection.get("reconciliation_status") or "UNVERIFIED"),
         circuit_breaker_state=str(projection.get("circuit_breaker_state") or "OPEN"),
         equity_usd=equity_usd,
-        peak_equity_usd=_coerce_optional_float(projection.get("peak_equity_usd"))
-        if "peak_equity_usd" in projection
-        else _coerce_optional_float(portfolio.get("peak_equity_usd")),
-        drawdown_from_peak=_coerce_optional_float(projection.get("drawdown_from_peak"))
-        if "drawdown_from_peak" in projection
-        else _coerce_optional_float(portfolio.get("drawdown_from_peak")),
+        peak_equity_usd=peak_equity_usd,
+        drawdown_from_peak=_resolve_drawdown_from_peak(
+            equity_usd=equity_usd,
+            peak_equity_usd=peak_equity_usd,
+            explicit=explicit_dd,
+        ),
         realized_vol=_coerce_optional_float(projection.get("realized_vol"))
         if "realized_vol" in projection
         else _coerce_optional_float(portfolio.get("realized_vol")),

tests/test_account_new_risk_gate.py

@@ -93,11 +93,15 @@ def test_healthy_equity_only_portfolio_allows_new_risk(self) -> None:
             "total_equity": 50_000.0,
             "metadata": {"total_equity_source": "broker_liquidation_value"},
         }
-        snapshot = build_account_new_risk_snapshot(portfolio)
-        self.assertEqual(snapshot["observation_status"], "COMPLETE")
-        self.assertEqual(snapshot["reconciliation_status"], "VERIFIED")
-        self.assertEqual(snapshot["circuit_breaker_state"], "CLOSED")
-        result = evaluate_portfolio_new_risk_admission(portfolio)
+        with patch(
+            "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+            return_value=None,
+        ):
+            snapshot = build_account_new_risk_snapshot(portfolio)
+            self.assertEqual(snapshot["observation_status"], "COMPLETE")
+            self.assertEqual(snapshot["reconciliation_status"], "VERIFIED")
+            self.assertEqual(snapshot["circuit_breaker_state"], "CLOSED")
+            result = evaluate_portfolio_new_risk_admission(portfolio)
         self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK)
 
     def test_explicit_healthy_snapshot_allows_new_risk(self) -> None:
@@ -183,12 +187,44 @@ def test_absent_production_drift_status_still_allows_when_healthy(self) -> None:
             "total_equity": 50_000.0,
             "metadata": {"total_equity_source": "broker_liquidation_value"},
         }
-        result = evaluate_portfolio_new_risk_admission(portfolio)
+        with patch(
+            "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+            return_value=None,
+        ):
+            result = evaluate_portfolio_new_risk_admission(portfolio)
         self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK)
 
-    def test_rebalance_does_not_use_unbound_research_store_for_production_drift(self) -> None:
-        import quant_platform_kit.risk.production_drift_new_risk as drift_mod
+    def test_store_critical_production_drift_prohibits_when_status_absent(self) -> None:
+        portfolio = {
+            "total_equity": 50_000.0,
+            "metadata": {"total_equity_source": "broker_liquidation_value"},
+        }
+        with patch(
+            "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+            return_value="critical",
+        ) as store_resolver:
+            result = evaluate_portfolio_new_risk_admission(portfolio)
+        store_resolver.assert_called_once()
+        self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED)
+        self.assertIn("PRODUCTION_DRIFT_CRITICAL", result.reason_codes)
 
+    def test_explicit_production_drift_status_skips_store_lookup(self) -> None:
+        portfolio = {
+            "total_equity": 50_000.0,
+            "metadata": {"total_equity_source": "broker_liquidation_value"},
+            "account_new_risk_snapshot": {"production_drift_status": "critical"},
+        }
+        with patch(
+            "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+            return_value="review",
+        ) as store_resolver:
+            result = evaluate_portfolio_new_risk_admission(portfolio)
+        store_resolver.assert_not_called()
+        self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED)
+        self.assertIn("PRODUCTION_DRIFT_CRITICAL", result.reason_codes)
+
+    def test_rebalance_policy_a_store_read_is_fail_soft_when_unbound(self) -> None:
+        """Gate may read PerformanceStore; unbound/empty probe must not invent bans."""
         plan = {
             "account_hash": "demo",
             "allocation": {
@@ -228,10 +264,9 @@ def test_rebalance_does_not_use_unbound_research_store_for_production_drift(self
             positions=(),
             metadata={},
         )
-        resolver = patch.object(
-            drift_mod,
-            "resolve_production_drift_status_from_store",
-            return_value="review",
+        resolver = patch(
+            "application.account_new_risk_gate_support.resolve_production_drift_status_from_store",
+            return_value=None,
         )
         with resolver as store_resolver:
             result = run_strategy_core(
@@ -254,7 +289,7 @@ def test_rebalance_does_not_use_unbound_research_store_for_production_drift(self
                 ),
             )
 
-        store_resolver.assert_not_called()
+        store_resolver.assert_called()
         self.assertNotIn(
             "production_drift_status",
             result.portfolio["account_new_risk_snapshot"],