feat(risk): resolve production_drift_status from store for NEW_RISK

docs/account_new_risk_gate.zh-CN.md

@@ -59,7 +59,7 @@
 | `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`。
+本轴**只**禁止新增风险；不启动 reopt、不写研究 ticket、不授 live。研究侧有界 reopt 仍须人工/独立 ticket 触发。助手：`production_drift_new_risk_reasons` / `production_drift_status_from_result`。平台可调用 `resolve_production_drift_status_from_store` 从 PerformanceStore 只读 probe 注入状态；无 store / parked / 探针失败时省略本轴（不发明 CRITICAL）。
 
 ### W2 只读 probe 用法
 

src/quant_platform_kit/risk/__init__.py

@@ -54,7 +54,9 @@
 from quant_platform_kit.risk.production_drift_new_risk import (
     normalize_production_drift_status,
     production_drift_new_risk_reasons,
+    production_drift_status_from_probe_summary,
     production_drift_status_from_result,
+    resolve_production_drift_status_from_store,
 )
 from quant_platform_kit.risk.capital_envelope_w2_probe import (
     format_probe_report,
@@ -122,7 +124,9 @@
     "validate_injected_snapshot",
     "normalize_production_drift_status",
     "production_drift_new_risk_reasons",
+    "production_drift_status_from_probe_summary",
     "production_drift_status_from_result",
+    "resolve_production_drift_status_from_store",
     "CycleNewRiskHealthEvidence",
     "apply_cycle_new_risk_health_axes",
     "project_cycle_new_risk_health_axes",

src/quant_platform_kit/risk/production_drift_new_risk.py

@@ -12,6 +12,8 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable, Mapping
+from datetime import date
 from typing import Any
 
 _ALLOWED_STATUSES = frozenset({"healthy", "watch", "review", "critical"})
@@ -62,8 +64,65 @@ def production_drift_status_from_result(drift: Any) -> str | None:
     return normalize_production_drift_status(status)
 
 
+def production_drift_status_from_probe_summary(
+    summary: Mapping[str, Any] | None,
+) -> str | None:
+    """Map probe summary → inject status; parked/unavailable/missing → None."""
+    if summary is None:
+        return None
+    raw = summary.get("status")
+    try:
+        normalized = normalize_production_drift_status(raw)
+    except TypeError:
+        return None
+    if normalized is None:
+        return None
+    if normalized in {"parked", "unavailable"}:
+        return None
+    if normalized in _ALLOWED_STATUSES:
+        return normalized
+    return None
+
+
+def resolve_production_drift_status_from_store(
+    *,
+    strategy_profile: str,
+    domain: str,
+    as_of: date | str | None = None,
+    store: Any | None = None,
+    probe: Callable[..., Mapping[str, Any]] | None = None,
+) -> str | None:
+    """Read-only store probe → status | None. Any exception → None (fail-soft).
+
+    Does not optimize, grant live, or reset breakers.
+    """
+    profile = (strategy_profile or "").strip()
+    domain_key = (domain or "").strip()
+    if not profile or not domain_key:
+        return None
+    active_probe = probe
+    if active_probe is None:
+        from quant_platform_kit.strategy_lifecycle.production_drift_health_probe import (
+            probe_production_drift_health_from_store,
+        )
+
+        active_probe = probe_production_drift_health_from_store
+    try:
+        summary = active_probe(
+            strategy_profile=profile,
+            domain=domain_key,
+            as_of=as_of,
+            store=store,
+        )
+    except Exception:
+        return None
+    return production_drift_status_from_probe_summary(summary)
+
+
 __all__ = [
     "normalize_production_drift_status",
     "production_drift_new_risk_reasons",
+    "production_drift_status_from_probe_summary",
     "production_drift_status_from_result",
+    "resolve_production_drift_status_from_store",
 ]

tests/test_production_drift_new_risk.py

@@ -7,7 +7,9 @@
 from quant_platform_kit.risk.production_drift_new_risk import (
     normalize_production_drift_status,
     production_drift_new_risk_reasons,
+    production_drift_status_from_probe_summary,
     production_drift_status_from_result,
+    resolve_production_drift_status_from_store,
 )
 from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus
 
@@ -52,3 +54,84 @@ def test_status_from_drift_result() -> None:
     assert production_drift_status_from_result(drift) == "critical"
     assert normalize_production_drift_status(" Review ") == "review"
     assert production_drift_status_from_result(None) is None
+
+
+
+def test_probe_summary_parked_omits_status() -> None:
+    assert production_drift_status_from_probe_summary(None) is None
+    assert (
+        production_drift_status_from_probe_summary(
+            {"status": "parked", "actionable": False}
+        )
+        is None
+    )
+    assert (
+        production_drift_status_from_probe_summary(
+            {"status": "unavailable", "actionable": False}
+        )
+        is None
+    )
+
+
+def test_probe_summary_maps_allowed_statuses() -> None:
+    assert production_drift_status_from_probe_summary({"status": "review"}) == "review"
+    assert production_drift_status_from_probe_summary({"status": "critical"}) == "critical"
+    assert production_drift_status_from_probe_summary({"status": "healthy"}) == "healthy"
+    assert production_drift_status_from_probe_summary({"status": "WATCH"}) == "watch"
+
+
+def test_resolve_from_store_uses_probe() -> None:
+    calls: list[dict[str, object]] = []
+
+    def fake_probe(**kwargs: object) -> dict[str, object]:
+        calls.append(kwargs)
+        return {"status": "review", "actionable": True}
+
+    assert (
+        resolve_production_drift_status_from_store(
+            strategy_profile="demo",
+            domain="us_equity",
+            as_of=date(2026, 9, 7),
+            probe=fake_probe,
+        )
+        == "review"
+    )
+    assert len(calls) == 1
+    assert calls[0]["strategy_profile"] == "demo"
+    assert calls[0]["domain"] == "us_equity"
+
+
+def test_resolve_from_store_probe_error_fail_soft() -> None:
+    def boom(**_kwargs: object) -> dict[str, object]:
+        raise RuntimeError("store down")
+
+    assert (
+        resolve_production_drift_status_from_store(
+            strategy_profile="demo",
+            domain="us_equity",
+            probe=boom,
+        )
+        is None
+    )
+
+
+def test_resolve_from_store_empty_profile_skips_probe() -> None:
+    def should_not_run(**_kwargs: object) -> dict[str, object]:
+        raise AssertionError("probe must not be called")
+
+    assert (
+        resolve_production_drift_status_from_store(
+            strategy_profile="",
+            domain="us_equity",
+            probe=should_not_run,
+        )
+        is None
+    )
+    assert (
+        resolve_production_drift_status_from_store(
+            strategy_profile="demo",
+            domain="  ",
+            probe=should_not_run,
+        )
+        is None
+    )