feat(risk): ban new risk on actionable production drift

docs/account_new_risk_gate.zh-CN.md

@@ -48,6 +48,19 @@
 可选：`peak_equity_usd` / `drawdown_from_peak` / `realized_vol`。缺权益 →
 `EQUITY_UNKNOWN_FAIL_CLOSED` 禁止。`ALLOW_NEW_RISK` **不是**下单许可，也不是实盘授权。
 
+### 生产 drift 轴（Policy A：禁新风险，零优化）
+
+可选注入 `production_drift_status`（`healthy` / `watch` / `review` / `critical`）：
+
+| 注入 | 门控效果 |
+| --- | --- |
+| 缺省 / 空白 | 不发明状态；本轴不加禁止理由 |
+| `healthy` / `watch` | 本轴允许 |
+| `review` / `critical` | `NEW_RISK_PROHIBITED`（`PRODUCTION_DRIFT_REVIEW` / `PRODUCTION_DRIFT_CRITICAL`） |
+| 非法值 | `PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED` |
+
+本轴**只**禁止新增风险；不启动 reopt、不写研究 ticket、不授 live。研究侧有界 reopt 仍须人工/独立 ticket 触发。助手：`production_drift_new_risk_reasons` / `production_drift_status_from_result`。
+
 ### W2 只读 probe 用法
 
 ```bash

src/quant_platform_kit/risk/__init__.py

@@ -51,6 +51,11 @@
     evaluate_new_risk_from_reader,
     validate_injected_snapshot,
 )
+from quant_platform_kit.risk.production_drift_new_risk import (
+    normalize_production_drift_status,
+    production_drift_new_risk_reasons,
+    production_drift_status_from_result,
+)
 from quant_platform_kit.risk.capital_envelope_w2_probe import (
     format_probe_report,
     probe_capital_envelope_w2,
@@ -115,6 +120,9 @@
     "evaluate_new_risk_admission",
     "evaluate_new_risk_from_reader",
     "validate_injected_snapshot",
+    "normalize_production_drift_status",
+    "production_drift_new_risk_reasons",
+    "production_drift_status_from_result",
     "CycleNewRiskHealthEvidence",
     "apply_cycle_new_risk_health_axes",
     "project_cycle_new_risk_health_axes",

src/quant_platform_kit/risk/account_new_risk_gate.py

@@ -33,6 +33,9 @@
 from typing import Protocol
 
 from quant_platform_kit.risk.capital_risk_envelope import evaluate_capital_risk_envelope
+from quant_platform_kit.risk.production_drift_new_risk import (
+    production_drift_new_risk_reasons,
+)
 
 
 class NewRiskDisposition(str, Enum):
@@ -54,6 +57,12 @@ class InjectedReconciliationSnapshot:
     inject-only numbers (no account ids, credentials, broker endpoints, or
     order payloads). Missing equity is treated as unknown → prohibit at the
     gate, not as a silent healthy default.
+
+    Optional ``production_drift_status`` is an inject-only closed enum
+    (``healthy`` / ``watch`` / ``review`` / ``critical``). Absent means the
+    drift axis was not injected (do not invent a status). ``review`` /
+    ``critical`` map to ``NEW_RISK_PROHIBITED`` only — never to optimize or
+    live enablement.
     """
 
     observation_status: str
@@ -63,6 +72,7 @@ class InjectedReconciliationSnapshot:
     peak_equity_usd: float | None = None
     drawdown_from_peak: float | None = None
     realized_vol: float | None = None
+    production_drift_status: str | None = None
 
 
 class ReconciliationSnapshotReader(Protocol):
@@ -167,11 +177,14 @@ def _evaluate_capital_axis(
 def evaluate_new_risk_admission(
     snapshot: InjectedReconciliationSnapshot,
 ) -> NewRiskAdmissionResult:
-    """Map unhealthy injected snapshots / capital envelope to ``NEW_RISK_PROHIBITED``.
+    """Map unhealthy injected snapshots / capital envelope / actionable
+    production drift to ``NEW_RISK_PROHIBITED``.
 
-    Healthy reconciliation axes **and** an allowing capital envelope may return
+    Healthy reconciliation axes, an allowing capital envelope, and a
+    non-actionable (or absent) production-drift axis may return
     ``ALLOW_NEW_RISK``. That is **not** an order permission, never grants live,
-    never flattens, and never resets breakers. Still not wired to real accounts.
+    never flattens, never resets breakers, and never starts optimization.
+    Still not wired to real accounts.
     """
     validated = validate_injected_snapshot(snapshot)
     reasons: list[str] = []
@@ -183,6 +196,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(production_drift_new_risk_reasons(validated.production_drift_status))
     if reasons:
         return NewRiskAdmissionResult(
             disposition=NewRiskDisposition.NEW_RISK_PROHIBITED,

src/quant_platform_kit/risk/production_drift_new_risk.py

@@ -0,0 +1,69 @@
+"""Map injected production-drift status → NEW_RISK prohibit reasons.
+
+Policy A (monitoring → ban new risk; zero optimize)
+---------------------------------------------------
+When a caller injects an actionable production drift status (``review`` /
+``critical``), the account NEW_RISK gate must prohibit new risk. This module
+does **not** start optimization, research tickets, or live enablement.
+
+Absent status is not invented as healthy or critical: no inject ⇒ no drift
+axis reasons. Invalid status fails closed.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+_ALLOWED_STATUSES = frozenset({"healthy", "watch", "review", "critical"})
+_ACTIONABLE_STATUSES = frozenset({"review", "critical"})
+
+
+def normalize_production_drift_status(value: object) -> str | None:
+    """Return lowercase closed-enum status, or ``None`` when not injected."""
+    if value is None:
+        return None
+    if hasattr(value, "value") and not isinstance(value, (str, bytes)):
+        # DriftStatus / similar enums
+        value = getattr(value, "value")
+    if not isinstance(value, str):
+        raise TypeError("production_drift_status must be str, enum, or None")
+    normalized = value.strip().lower()
+    if not normalized:
+        return None
+    return normalized
+
+
+def production_drift_new_risk_reasons(status: object) -> tuple[str, ...]:
+    """Return prohibit reason codes for an injected production-drift status.
+
+    ``None`` / blank ⇒ no reasons (axis not injected).
+    ``healthy`` / ``watch`` ⇒ no reasons.
+    ``review`` / ``critical`` ⇒ ``PRODUCTION_DRIFT_*`` (ban new risk only).
+    Anything else ⇒ ``PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED``.
+    """
+    try:
+        normalized = normalize_production_drift_status(status)
+    except TypeError:
+        return ("PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED",)
+    if normalized is None:
+        return ()
+    if normalized not in _ALLOWED_STATUSES:
+        return ("PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED",)
+    if normalized in _ACTIONABLE_STATUSES:
+        return (f"PRODUCTION_DRIFT_{normalized.upper()}",)
+    return ()
+
+
+def production_drift_status_from_result(drift: Any) -> str | None:
+    """Extract status string from a ``DriftResult``-like object (inject helper)."""
+    if drift is None:
+        return None
+    status = getattr(drift, "status", drift)
+    return normalize_production_drift_status(status)
+
+
+__all__ = [
+    "normalize_production_drift_status",
+    "production_drift_new_risk_reasons",
+    "production_drift_status_from_result",
+]

tests/test_account_new_risk_gate.py

@@ -126,6 +126,47 @@ def test_invalid_enum_fails_closed(self) -> None:
         with self.assertRaises(AccountNewRiskGateError):
             evaluate_new_risk_admission(snap)
 
+    def test_absent_production_drift_does_not_invent_prohibit(self) -> None:
+        result = evaluate_new_risk_admission(_healthy())
+        self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK)
+        self.assertEqual(result.reason_codes, ())
+
+    def test_healthy_or_watch_production_drift_allows_when_other_axes_ok(self) -> None:
+        for status in ("healthy", "watch", "HEALTHY", "WATCH"):
+            with self.subTest(status=status):
+                result = evaluate_new_risk_admission(
+                    _healthy(production_drift_status=status)
+                )
+                self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK)
+                self.assertEqual(result.reason_codes, ())
+
+    def test_review_production_drift_prohibits_without_side_effects(self) -> None:
+        result = evaluate_new_risk_admission(
+            _healthy(production_drift_status="review")
+        )
+        self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED)
+        self.assertIn("PRODUCTION_DRIFT_REVIEW", result.reason_codes)
+        self.assertFalse(result.live_authority_granted)
+        self.assertFalse(result.circuit_breaker_reset)
+        self.assertFalse(result.account_enablement_changed)
+
+    def test_critical_production_drift_prohibits_without_side_effects(self) -> None:
+        result = evaluate_new_risk_admission(
+            _healthy(production_drift_status="critical")
+        )
+        self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED)
+        self.assertIn("PRODUCTION_DRIFT_CRITICAL", result.reason_codes)
+        self.assertFalse(result.live_authority_granted)
+        self.assertFalse(result.circuit_breaker_reset)
+        self.assertFalse(result.account_enablement_changed)
+
+    def test_invalid_production_drift_status_fails_closed(self) -> None:
+        result = evaluate_new_risk_admission(
+            _healthy(production_drift_status="maybe")
+        )
+        self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED)
+        self.assertIn("PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED", result.reason_codes)
+
 
 class ReaderInjectionTests(unittest.TestCase):
     def test_reader_unhealthy_snapshot_prohibits(self) -> None:

tests/test_production_drift_new_risk.py

@@ -0,0 +1,54 @@
+"""Unit tests for production_drift → NEW_RISK reason mapping (Policy A)."""
+
+from __future__ import annotations
+
+from datetime import date
+
+from quant_platform_kit.risk.production_drift_new_risk import (
+    normalize_production_drift_status,
+    production_drift_new_risk_reasons,
+    production_drift_status_from_result,
+)
+from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus
+
+
+def test_absent_and_non_actionable_yield_no_reasons() -> None:
+    assert production_drift_new_risk_reasons(None) == ()
+    assert production_drift_new_risk_reasons("") == ()
+    assert production_drift_new_risk_reasons("healthy") == ()
+    assert production_drift_new_risk_reasons("watch") == ()
+    assert production_drift_new_risk_reasons(DriftStatus.HEALTHY) == ()
+    assert production_drift_new_risk_reasons(DriftStatus.WATCH) == ()
+
+
+def test_actionable_statuses_ban_new_risk_only() -> None:
+    assert production_drift_new_risk_reasons("review") == ("PRODUCTION_DRIFT_REVIEW",)
+    assert production_drift_new_risk_reasons("CRITICAL") == ("PRODUCTION_DRIFT_CRITICAL",)
+    assert production_drift_new_risk_reasons(DriftStatus.REVIEW) == (
+        "PRODUCTION_DRIFT_REVIEW",
+    )
+    assert production_drift_new_risk_reasons(DriftStatus.CRITICAL) == (
+        "PRODUCTION_DRIFT_CRITICAL",
+    )
+
+
+def test_invalid_status_fails_closed() -> None:
+    assert production_drift_new_risk_reasons("maybe") == (
+        "PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED",
+    )
+    assert production_drift_new_risk_reasons(1.5) == (
+        "PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED",
+    )
+
+
+def test_status_from_drift_result() -> None:
+    drift = DriftResult(
+        strategy_profile="demo",
+        domain="us_equity",
+        as_of=date(2026, 9, 7),
+        drift_score=0.8,
+        status=DriftStatus.CRITICAL,
+    )
+    assert production_drift_status_from_result(drift) == "critical"
+    assert normalize_production_drift_status(" Review ") == "review"
+    assert production_drift_status_from_result(None) is None