diff --git a/docs/account_new_risk_gate.zh-CN.md b/docs/account_new_risk_gate.zh-CN.md index b972b28..401a6a6 100644 --- a/docs/account_new_risk_gate.zh-CN.md +++ b/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 diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index 4e69049..41bfa05 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/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", diff --git a/src/quant_platform_kit/risk/account_new_risk_gate.py b/src/quant_platform_kit/risk/account_new_risk_gate.py index 22031d3..29b09b8 100644 --- a/src/quant_platform_kit/risk/account_new_risk_gate.py +++ b/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, diff --git a/src/quant_platform_kit/risk/production_drift_new_risk.py b/src/quant_platform_kit/risk/production_drift_new_risk.py new file mode 100644 index 0000000..7ccfed7 --- /dev/null +++ b/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", +] diff --git a/tests/test_account_new_risk_gate.py b/tests/test_account_new_risk_gate.py index a4a232a..cc16969 100644 --- a/tests/test_account_new_risk_gate.py +++ b/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: diff --git a/tests/test_production_drift_new_risk.py b/tests/test_production_drift_new_risk.py new file mode 100644 index 0000000..9f6f706 --- /dev/null +++ b/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