diff --git a/src/quant_platform_kit/risk/engine.py b/src/quant_platform_kit/risk/engine.py index b47359c3..f8285025 100644 --- a/src/quant_platform_kit/risk/engine.py +++ b/src/quant_platform_kit/risk/engine.py @@ -120,7 +120,13 @@ def assess( *, market_data: Mapping[str, Any] | None = None, ) -> RiskAction: - """Assess a decision, rejecting missing or invalid account state.""" + """Run the legacy runtime sanity check for account state. + + This method is not evidence-grade and its ``approve`` result must not + be used as promotion authority. Promotion/evidence callers must use + :func:`quant_platform_kit.risk.gate.assess_with_evidence`, which adds + the mandate, candidate, snapshot, and exposure contract checks. + """ if portfolio_snapshot is None: return RiskAction( action="reject", diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index f599416f..d57f88e3 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -568,7 +568,7 @@ def _mandate_fields( "loss_budget": 0.0, "product_leverage_factors": {}, "allowed_nonzero_assets": None, - }, set() + }, {"missing_mandate"} if not isinstance(mandate_provenance, Mapping): return {}, {"invalid_mandate"} if mandate_provenance.get("mandate_id") == _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE: @@ -1316,7 +1316,13 @@ def assess_with_evidence( capital_base: CapitalBaseSnapshot | Mapping[str, Any] | None = None, capital_base_binding: CapitalBaseBinding | Mapping[str, Any] | None = None, ) -> RiskGateResult: - """Assess exactly once and fail closed with a redacted canonical receipt.""" + """Run the sole promotion/evidence-grade risk assessment API. + + ``mandate_provenance`` and a matching ``candidate_identity`` are required + for an evidence-grade approval. Missing or invalid authority is rejected; + the legacy :meth:`RiskEngine.assess` approval is never sufficient on its + own. The returned receipt is redacted and canonical. + """ try: risk_action = build_risk_engine().assess( decision, diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index b65f5a59..c691a6c1 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -1077,6 +1077,95 @@ def test_unmandated_fallback_rejects_stale_snapshot(self) -> None: self.assertIn("stale_portfolio_snapshot", result.assessment.reason_codes) self.assertEqual(result.decision.positions, ()) + def test_evidence_assessment_requires_mandate(self) -> None: + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + result = assess_with_evidence( + _decision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),)), + self._snapshot(), + scope="MEMBER", + mandate_provenance=None, + market_data={}, + candidate_identity=None, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("missing_mandate", result.assessment.reason_codes) + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + + def test_evidence_assessment_rejects_unbounded_target_and_unknown_symbol(self) -> None: + cases = ( + ( + "10x_target", + _decision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=10.0),)), + "effective_exposure_cap", + ), + ( + "unknown_symbol", + _decision(positions=(PositionTarget(symbol="UNKNOWN", target_weight=0.10),)), + "asset_not_authorized", + ), + ) + for name, decision, reason_code in cases: + with ( + self.subTest(name=name), + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + ): + result = assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + candidate_identity=self._candidate(), + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn(reason_code, result.assessment.reason_codes) + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + + def test_evidence_assessment_rejects_expired_mandate(self) -> None: + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + result = assess_with_evidence( + _decision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),)), + self._snapshot(), + scope="MEMBER", + mandate_provenance={ + **self._mandate(), + "expires_at": "2026-08-04T04:27:59Z", + }, + market_data={}, + candidate_identity=self._candidate(), + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("expired_mandate", result.assessment.reason_codes) + self.assertFalse(result.assessment.execution_authorized) + + def test_evidence_assessment_rejects_stale_capital_snapshot(self) -> None: + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + result = assess_with_evidence( + _decision(positions=(PositionTarget(symbol="BTCUSDT", target_value=10_000.0),)), + self._snapshot(), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + candidate_identity=self._candidate(), + capital_base=_capital_base( + as_of=self._NOW - timedelta(seconds=301), + strategy_scope="crypto_live_pool_rotation", + ), + capital_base_binding=_capital_base_binding( + strategy_scope="crypto_live_pool_rotation", + ), + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("stale_capital_base", result.assessment.reason_codes) + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + def test_risk_plugin_exception_rejects_without_exposing_exception(self) -> None: class CrashingPlugin: plugin_name = "crashing_plugin" @@ -1377,8 +1466,16 @@ def test_optional_string_nulls_and_nfc_text_remain_canonical(self) -> None: ), ), ) - first, first_engine = self._assess(decision) - second, second_engine = self._assess(decision) + first, first_engine = self._assess( + decision, + mandate=self._mandate(), + candidate=self._candidate(), + ) + second, second_engine = self._assess( + decision, + mandate=self._mandate(), + candidate=self._candidate(), + ) self.assertEqual(first.assessment.outcome, "APPROVE") self.assertEqual(first.assessment.assessment_sha256, second.assessment.assessment_sha256) @@ -1568,14 +1665,20 @@ def test_timezone_and_whole_second_timestamp_compatibility(self) -> None: first, _engine = self._assess( StrategyDecision(), snapshot={**self._SNAPSHOT, "as_of": utc.replace(microsecond=1)}, + mandate=self._mandate(), + candidate=self._candidate(), ) second, _engine = self._assess( StrategyDecision(), snapshot={**self._SNAPSHOT, "as_of": offset.replace(microsecond=999_999)}, + mandate=self._mandate(), + candidate=self._candidate(), ) rejected, _engine = self._assess( StrategyDecision(), snapshot={**self._SNAPSHOT, "as_of": "2026-08-04T12:27:55+08:00"}, + mandate=self._mandate(), + candidate=self._candidate(), ) self.assertEqual(first.assessment.outcome, "APPROVE")