diff --git a/docs/account_new_risk_gate.zh-CN.md b/docs/account_new_risk_gate.zh-CN.md index 401a6a6..84c86ce 100644 --- a/docs/account_new_risk_gate.zh-CN.md +++ b/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 用法 diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index 41bfa05..ecffdf9 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/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", diff --git a/src/quant_platform_kit/risk/production_drift_new_risk.py b/src/quant_platform_kit/risk/production_drift_new_risk.py index 7ccfed7..1ef341d 100644 --- a/src/quant_platform_kit/risk/production_drift_new_risk.py +++ b/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", ] diff --git a/tests/test_production_drift_new_risk.py b/tests/test_production_drift_new_risk.py index 9f6f706..1d8037d 100644 --- a/tests/test_production_drift_new_risk.py +++ b/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 + )