diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index ed164aa..0539b13 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/src/quant_platform_kit/risk/__init__.py @@ -11,6 +11,8 @@ RegimeRoute, RiskAction, RiskAssessment, + RiskGateAssessment, + RiskGateResult, RiskSignal, ) from quant_platform_kit.risk.engine import ( @@ -18,7 +20,11 @@ aggregate_risk_signals, build_risk_engine, ) -from quant_platform_kit.risk.gate import apply_risk_gate, enrich_decision_risk_diagnostics +from quant_platform_kit.risk.gate import ( + assess_with_evidence, + apply_risk_gate, + enrich_decision_risk_diagnostics, +) from quant_platform_kit.risk.portfolio_diagnostics import ( compute_unrealized_pnl_pct, extract_portfolio_risk_diagnostics, @@ -29,9 +35,12 @@ "RegimeRoute", "RiskAction", "RiskAssessment", + "RiskGateAssessment", + "RiskGateResult", "RiskEngine", "RiskSignal", "aggregate_risk_signals", + "assess_with_evidence", "apply_risk_gate", "compute_unrealized_pnl_pct", "enrich_decision_risk_diagnostics", diff --git a/src/quant_platform_kit/risk/contracts.py b/src/quant_platform_kit/risk/contracts.py index cdc58d4..fbb2b35 100644 --- a/src/quant_platform_kit/risk/contracts.py +++ b/src/quant_platform_kit/risk/contracts.py @@ -9,6 +9,8 @@ from dataclasses import dataclass, field from datetime import datetime, timezone +import hashlib +import json from typing import Any, Mapping @@ -141,3 +143,63 @@ class RiskAction: risk_asset_scalar: float = 1.0 target_destination: str | None = None notify: bool = True + + +@dataclass(frozen=True) +class RiskGateAssessment: + """Immutable redacted evidence from a scoped risk-gate evaluation.""" + + contract_version: str + scope: str + evaluated_at: str + policy_id: str + policy_version: str + qpk_source_revision: str | None + mandate_id: str | None + mandate_version: str | None + mandate_authority_receipt_sha256: str | None + mandate_scope: str | None + decision_digest_sha256: str + portfolio_snapshot_digest_sha256: str + effective_exposure_cap: float | None + observed_effective_exposure: float | None + proposed_effective_exposure: float | None + outcome: str + reason_codes: tuple[str, ...] + assessment_sha256: str = field(init=False) + + def __post_init__(self) -> None: + payload = { + "contract_version": self.contract_version, + "scope": self.scope, + "evaluated_at": self.evaluated_at, + "policy_id": self.policy_id, + "policy_version": self.policy_version, + "qpk_source_revision": self.qpk_source_revision, + "mandate_id": self.mandate_id, + "mandate_version": self.mandate_version, + "mandate_authority_receipt_sha256": self.mandate_authority_receipt_sha256, + "mandate_scope": self.mandate_scope, + "decision_digest_sha256": self.decision_digest_sha256, + "portfolio_snapshot_digest_sha256": self.portfolio_snapshot_digest_sha256, + "effective_exposure_cap": self.effective_exposure_cap, + "observed_effective_exposure": self.observed_effective_exposure, + "proposed_effective_exposure": self.proposed_effective_exposure, + "outcome": self.outcome, + "reason_codes": self.reason_codes, + } + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + object.__setattr__(self, "assessment_sha256", hashlib.sha256(encoded).hexdigest()) + + +@dataclass(frozen=True) +class RiskGateResult: + """Risk-gated decision paired with its immutable evidence receipt.""" + + decision: Any + assessment: RiskGateAssessment diff --git a/src/quant_platform_kit/risk/engine.py b/src/quant_platform_kit/risk/engine.py index 59cf378..3f890cf 100644 --- a/src/quant_platform_kit/risk/engine.py +++ b/src/quant_platform_kit/risk/engine.py @@ -83,7 +83,16 @@ def evaluate( try: signals.append(plugin.evaluate(market_data)) except Exception: - pass + signals.append( + RiskSignal( + plugin="risk_engine", + schema_version="qpk.risk_plugin_error.v1", + route=ROUTE_BLOCKED, + confidence=1.0, + suggested_action="blocked", + reason_codes=("plugin_evaluation_error",), + ) + ) # Merge external signals if plugin_signals: diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index 54446cd..7aa0db6 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -6,10 +6,15 @@ from __future__ import annotations +from datetime import datetime, timezone +import hashlib +import json import logging import math from typing import Any, Mapping +from quant_platform_kit.common.models import PortfolioSnapshot +from quant_platform_kit.risk.contracts import RiskGateAssessment, RiskGateResult from quant_platform_kit.risk.engine import build_risk_engine from quant_platform_kit.strategy_contracts import StrategyDecision @@ -21,6 +26,415 @@ _APPROVED_BOOTSTRAP_MANDATE = "bootstrap_small_account_v2" _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP = 0.50 _BOOTSTRAP_NOMINAL_CAPS = {1: 0.50, 2: 0.25, 3: 0.15} +_ASSESSMENT_CONTRACT_VERSION = "qsl.risk_gate_assessment.v1" +_ASSESSMENT_POLICY_ID = "qpk.risk_gate" +_ASSESSMENT_POLICY_VERSION = "v1" +_FALLBACK_MAX_SNAPSHOT_AGE_SECONDS_V1 = 300.0 +_ALLOWED_SCOPES = frozenset({"MEMBER", "ACCOUNT"}) +_ALLOWED_MANDATE_SCOPES = frozenset({"RESEARCH_ONLY", "PAPER", "LIVE"}) + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _utc_timestamp(value: datetime) -> str: + return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _canonical_digest(value: Mapping[str, Any]) -> str: + encoded = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _finite_number(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + number = float(value) + return number if math.isfinite(number) else None + + +def _parse_utc_timestamp(value: Any) -> datetime | None: + if isinstance(value, datetime): + parsed = value + elif isinstance(value, str) and value.endswith("Z"): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + else: + return None + return parsed.astimezone(timezone.utc) if parsed.tzinfo is not None else None + + +def _valid_cap_value(value: Any) -> bool: + if isinstance(value, Mapping): + return bool(value) and all( + isinstance(key, str) + and bool(key) + and _valid_cap_value(candidate) + for key, candidate in value.items() + ) + cap = _finite_number(value) + return cap is not None and 0.0 <= cap <= 1.0 + + +def _sha256(value: Any) -> str | None: + if not isinstance(value, str) or len(value) != 64: + return None + return value if all(character in "0123456789abcdef" for character in value) else None + + +def _decision_metrics( + decision: StrategyDecision, + *, + total_equity: float | None, +) -> tuple[dict[str, Any], list[tuple[str, float]], set[str]]: + active: list[tuple[str, float]] = [] + reason_codes: set[str] = set() + position_payloads: list[dict[str, Any]] = [] + for position in decision.positions or (): + symbol = getattr(position, "symbol", None) + weight = _finite_number(getattr(position, "target_weight", None)) + target_value = _finite_number(getattr(position, "target_value", None)) + position_payloads.append( + { + "symbol": symbol if isinstance(symbol, str) else None, + "target_weight": weight, + "target_value": target_value, + "role": getattr(position, "role", None), + "order_preference": getattr(position, "order_preference", None), + } + ) + if ( + not isinstance(symbol, str) + or not symbol + or (weight is None) == (target_value is None) + ): + reason_codes.add("invalid_decision_exposure") + continue + normalized_weight = weight + if target_value is not None: + if total_equity is None or total_equity <= 0.0: + reason_codes.add("invalid_decision_exposure") + continue + normalized_weight = target_value / total_equity + if normalized_weight is None or normalized_weight < 0.0: + reason_codes.add("invalid_decision_exposure") + continue + if normalized_weight > 0.0: + active.append((symbol, normalized_weight)) + budget_payloads: list[dict[str, Any]] = [] + for budget in decision.budgets or (): + budget_payloads.append( + { + "name": getattr(budget, "name", None), + "symbol": getattr(budget, "symbol", None), + "amount": _finite_number(getattr(budget, "amount", None)), + "unit": getattr(budget, "unit", None), + "purpose": getattr(budget, "purpose", None), + } + ) + return { + "positions": position_payloads, + "budgets": budget_payloads, + }, active, reason_codes + + +def _snapshot_metrics( + portfolio_snapshot: Any, + *, + now: datetime, + max_snapshot_age_seconds: float | None, +) -> tuple[dict[str, Any], float | None, float | None, set[str]]: + if isinstance(portfolio_snapshot, Mapping): + as_of_value = portfolio_snapshot.get("as_of") + observed_value = portfolio_snapshot.get("observed_effective_exposure") + total_equity_value = portfolio_snapshot.get("total_equity") + elif isinstance(portfolio_snapshot, PortfolioSnapshot): + metadata = portfolio_snapshot.metadata + as_of_value = portfolio_snapshot.as_of + observed_value = metadata.get("observed_effective_exposure") + total_equity_value = portfolio_snapshot.total_equity + else: + return {}, None, None, {"invalid_portfolio_snapshot"} + as_of = _parse_utc_timestamp(as_of_value) + observed = _finite_number(observed_value) + total_equity = _finite_number(total_equity_value) + if ( + as_of is None + or observed is None + or observed < 0.0 + or total_equity is None + or total_equity <= 0.0 + ): + return {}, observed, total_equity, {"invalid_portfolio_snapshot"} + age_seconds = (now - as_of).total_seconds() + if age_seconds < 0.0 or ( + max_snapshot_age_seconds is not None and age_seconds > max_snapshot_age_seconds + ): + return {}, observed, total_equity, {"stale_portfolio_snapshot"} + return { + "as_of": _utc_timestamp(as_of), + "observed_effective_exposure": observed, + "total_equity": total_equity, + }, observed, total_equity, set() + + +def _mandate_fields( + mandate_provenance: Mapping[str, Any] | None, + *, + now: datetime, +) -> tuple[dict[str, Any], set[str]]: + if mandate_provenance is None: + return { + "mandate_id": None, + "mandate_version": None, + "authority_receipt_sha256": None, + "authority_scope": None, + "source_revision": None, + "effective_exposure_cap": _DEFAULT_MAX_SINGLE_WEIGHT, + "max_snapshot_age_seconds": _FALLBACK_MAX_SNAPSHOT_AGE_SECONDS_V1, + "loss_budget": 0.0, + "product_leverage_factors": {}, + "allowed_nonzero_assets": None, + }, set() + if not isinstance(mandate_provenance, Mapping): + return {}, {"invalid_mandate"} + required = ( + "mandate_id", + "mandate_version", + "authority_receipt_sha256", + "authority_scope", + "strategy_profile", + "account_mode", + "effective_at", + "expires_at", + "max_snapshot_age_seconds", + "effective_exposure_cap", + "loss_budget", + "product_caps", + "nominal_caps", + "product_leverage_factors", + "allowed_nonzero_assets", + "source_revision", + ) + if any( + field not in mandate_provenance + or mandate_provenance[field] is None + or mandate_provenance[field] == "" + for field in required + ): + return {}, {"invalid_mandate"} + authority_scope = mandate_provenance["authority_scope"] + receipt_sha256 = _sha256(mandate_provenance["authority_receipt_sha256"]) + effective_at = _parse_utc_timestamp(mandate_provenance["effective_at"]) + expires_at = _parse_utc_timestamp(mandate_provenance["expires_at"]) + max_snapshot_age_seconds = _finite_number(mandate_provenance["max_snapshot_age_seconds"]) + cap = _finite_number(mandate_provenance["effective_exposure_cap"]) + loss_budget = _finite_number(mandate_provenance["loss_budget"]) + if ( + not isinstance(mandate_provenance["mandate_id"], str) + or not isinstance(mandate_provenance["mandate_version"], str) + or not isinstance(mandate_provenance["source_revision"], str) + or not isinstance(mandate_provenance["strategy_profile"], str) + or not isinstance(mandate_provenance["account_mode"], str) + or authority_scope not in _ALLOWED_MANDATE_SCOPES + or receipt_sha256 is None + or effective_at is None + or expires_at is None + or max_snapshot_age_seconds is None + or max_snapshot_age_seconds <= 0.0 + or cap is None + or not 0.0 <= cap <= 1.0 + or loss_budget is None + or loss_budget < 0.0 + ): + return {}, {"invalid_mandate"} + if effective_at > now or expires_at < now or expires_at <= effective_at: + return {}, {"expired_mandate"} + factors = mandate_provenance.get("product_leverage_factors", {}) + allowed_assets = mandate_provenance.get("allowed_nonzero_assets") + if ( + not isinstance(factors, Mapping) + or not _valid_cap_value(mandate_provenance["product_caps"]) + or not _valid_cap_value(mandate_provenance["nominal_caps"]) + or ( + allowed_assets is not None + and ( + not isinstance(allowed_assets, (list, tuple)) + or not all(isinstance(asset, str) and asset for asset in allowed_assets) + ) + ) + ): + return {}, {"invalid_mandate"} + return { + "mandate_id": mandate_provenance["mandate_id"], + "mandate_version": mandate_provenance["mandate_version"], + "authority_receipt_sha256": receipt_sha256, + "authority_scope": authority_scope, + "source_revision": mandate_provenance["source_revision"], + "effective_exposure_cap": cap, + "max_snapshot_age_seconds": max_snapshot_age_seconds, + "loss_budget": loss_budget, + "product_leverage_factors": factors, + "product_caps": mandate_provenance["product_caps"], + "nominal_caps": mandate_provenance["nominal_caps"], + "allowed_nonzero_assets": set(allowed_assets) if allowed_assets is not None else None, + }, set() + + +def _position_cap(value: Any, symbol: str, leverage_factor: float) -> float | None: + if isinstance(value, Mapping): + leverage_class = str(int(leverage_factor)) + if symbol in value: + value = value[symbol] + elif leverage_class in value: + value = value[leverage_class] + else: + return None + cap = _finite_number(value) + return cap if cap is not None and 0.0 <= cap <= 1.0 else None + + +def _budget_authority_errors( + decision: StrategyDecision, + mandate: Mapping[str, Any], +) -> set[str]: + requested_budget = 0.0 + for budget in decision.budgets or (): + amount = _finite_number(getattr(budget, "amount", None)) + if amount is None or amount < 0.0: + return {"invalid_decision_budget"} + requested_budget += amount + authorized_budget = _finite_number(mandate.get("loss_budget")) + if requested_budget > 0.0 and ( + mandate.get("effective_exposure_cap") == 0.0 + or authorized_budget is None + or requested_budget > authorized_budget + 1e-9 + ): + return {"budget_authority_exceeded"} + return set() + + +def assess_with_evidence( + decision: StrategyDecision, + portfolio_snapshot: Any, + *, + scope: str, + mandate_provenance: Mapping[str, Any] | None, + market_data: Mapping[str, Any], +) -> RiskGateResult: + """Fail closed with a redacted, canonical risk-gate assessment receipt.""" + now = _utc_now() + evaluated_at = _utc_timestamp(now) + assessment_scope = scope if scope in _ALLOWED_SCOPES else "MEMBER" + mandate, mandate_errors = _mandate_fields(mandate_provenance, now=now) + reason_codes = set(mandate_errors) + cap = mandate.get("effective_exposure_cap") + snapshot_payload, observed, total_equity, snapshot_errors = _snapshot_metrics( + portfolio_snapshot, + now=now, + max_snapshot_age_seconds=mandate.get("max_snapshot_age_seconds"), + ) + reason_codes.update(snapshot_errors) + decision_payload, active_positions, decision_errors = _decision_metrics( + decision, + total_equity=total_equity, + ) + reason_codes.update(decision_errors) + if scope not in _ALLOWED_SCOPES: + reason_codes.add("invalid_scope") + can_assess_with_engine = not reason_codes + if mandate: + reason_codes.update(_budget_authority_errors(decision, mandate)) + + proposed: float | None = None + if can_assess_with_engine: + factors = mandate["product_leverage_factors"] + allowed_assets = mandate["allowed_nonzero_assets"] + weighted_exposure = 0.0 + if mandate_provenance is None and len(active_positions) > 1: + reason_codes.add("fallback_position_count") + for symbol, weight in active_positions: + if allowed_assets is not None and symbol not in allowed_assets: + reason_codes.add("asset_not_authorized") + continue + factor = 1.0 if mandate_provenance is None else _finite_number(factors.get(symbol)) + if factor is None or not factor.is_integer() or factor < 1.0: + reason_codes.add("invalid_leverage_classification") + continue + product_cap = _position_cap(mandate.get("product_caps", 1.0), symbol, factor) + nominal_cap = _position_cap(mandate.get("nominal_caps", 1.0), symbol, factor) + if product_cap is None or nominal_cap is None: + reason_codes.add("invalid_product_cap") + continue + if weight > min(product_cap, nominal_cap): + reason_codes.add("product_exposure_cap") + weighted_exposure += weight * factor + proposed = max(observed or 0.0, weighted_exposure) + if cap is None or observed is None or observed > cap + 1e-9: + reason_codes.add("observed_effective_exposure") + if cap is None or proposed > cap + 1e-9: + reason_codes.add("effective_exposure_cap") + + if can_assess_with_engine: + try: + risk_action = build_risk_engine().assess( + decision, + portfolio_snapshot, + market_data=market_data, + ) + except Exception: + reason_codes.add("risk_engine_error") + else: + if risk_action.action != "approve": + reason_codes.add("risk_engine_non_approve") + + outcome = "REJECT" if reason_codes else "APPROVE" + assessment = RiskGateAssessment( + contract_version=_ASSESSMENT_CONTRACT_VERSION, + scope=assessment_scope, + evaluated_at=evaluated_at, + policy_id=_ASSESSMENT_POLICY_ID, + policy_version=_ASSESSMENT_POLICY_VERSION, + qpk_source_revision=mandate.get("source_revision"), + mandate_id=mandate.get("mandate_id"), + mandate_version=mandate.get("mandate_version"), + mandate_authority_receipt_sha256=mandate.get("authority_receipt_sha256"), + mandate_scope=mandate.get("authority_scope"), + decision_digest_sha256=_canonical_digest(decision_payload), + portfolio_snapshot_digest_sha256=_canonical_digest(snapshot_payload), + effective_exposure_cap=cap, + observed_effective_exposure=observed, + proposed_effective_exposure=proposed, + outcome=outcome, + reason_codes=tuple(sorted(reason_codes)), + ) + if outcome == "REJECT": + return RiskGateResult( + decision=_reject( + decision, + flag="rejected:risk_gate_assessment", + reason=",".join(assessment.reason_codes), + ), + assessment=assessment, + ) + risk_flags = tuple(decision.risk_flags or ()) + ("risk_gate:passed",) + return RiskGateResult( + decision=StrategyDecision( + positions=decision.positions, + budgets=decision.budgets, + risk_flags=risk_flags, + diagnostics={**(decision.diagnostics or {}), "risk_gate": "APPROVE"}, + ), + assessment=assessment, + ) def enrich_decision_risk_diagnostics( @@ -102,6 +516,13 @@ def apply_risk_gate( reason=f"连续亏损 {int(consecutive_losses)} 笔 > {_MAX_CONSECUTIVE_LOSSES} 熔断", ) + if risk_mandate_id not in {None, _APPROVED_BOOTSTRAP_MANDATE}: + return _reject( + decision, + flag="rejected:unknown_risk_mandate", + reason="风险授权未获批准", + ) + positions = decision.positions or () if not positions: return decision @@ -192,20 +613,19 @@ def apply_risk_gate( reason=f"名义仓位 {weight:.1%} > 可用账户容量", ) else: - if risk_mandate_id is not None: + requested_single_weight = _finite_number(max_single_weight) + effective_single_weight = min( + requested_single_weight + if requested_single_weight is not None and requested_single_weight >= 0.0 + else _DEFAULT_MAX_SINGLE_WEIGHT, + _DEFAULT_MAX_SINGLE_WEIGHT, + ) + if len(weights) > 1: return _reject( decision, - flag="rejected:unknown_risk_mandate", - reason="风险授权未获批准", + flag="rejected:too_many_positions", + reason="未获授权的风险配置仅允许一个非零持仓", ) - effective_single_weight = ( - float(max_single_weight) - if isinstance(max_single_weight, (int, float)) - and not isinstance(max_single_weight, bool) - and math.isfinite(float(max_single_weight)) - and 0.0 <= float(max_single_weight) <= 1.0 - else _DEFAULT_MAX_SINGLE_WEIGHT - ) for position, weight in weights: if weight > effective_single_weight: logger.warning( @@ -252,7 +672,7 @@ def apply_risk_gate( portfolio_snapshot, market_data=market_data, ) - if assessment.action == "reject": + if assessment.action != "approve": logger.warning("risk_gate REJECT risk_engine: %s", assessment.reason) return _reject( decision, @@ -278,7 +698,7 @@ def _reject( ) -> StrategyDecision: return StrategyDecision( positions=(), - budgets=decision.budgets, + budgets=(), risk_flags=(flag,), diagnostics={ **(decision.diagnostics or {}), diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 37080e5..cd6f19a 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -1,15 +1,18 @@ from __future__ import annotations import unittest -from unittest.mock import patch +from datetime import datetime, timezone +from unittest.mock import Mock, patch -from quant_platform_kit.risk.contracts import ROUTE_BLOCKED, RiskSignal +from quant_platform_kit.common.models import PortfolioSnapshot +from quant_platform_kit.risk.contracts import ROUTE_BLOCKED, RiskAction, RiskSignal from quant_platform_kit.risk.engine import RiskEngine from quant_platform_kit.risk.gate import ( + assess_with_evidence, apply_risk_gate, enrich_decision_risk_diagnostics, ) -from quant_platform_kit.strategy_contracts import PositionTarget, StrategyDecision +from quant_platform_kit.strategy_contracts import BudgetIntent, PositionTarget, StrategyDecision def _decision( @@ -24,7 +27,7 @@ def _decision( class ApplyRiskGateTests(unittest.TestCase): - def test_no_mandate_honors_explicit_legacy_single_weight_limit(self) -> None: + def test_no_mandate_does_not_allow_caller_to_expand_default_cap(self) -> None: decision = _decision( positions=(PositionTarget(symbol="SPY", target_weight=0.11),), ) @@ -36,8 +39,9 @@ def test_no_mandate_honors_explicit_legacy_single_weight_limit(self) -> None: max_total_exposure=1.0, ) - self.assertEqual(len(result.positions), 1) - self.assertIn("risk_gate:passed", result.risk_flags) + self.assertEqual(result.positions, ()) + self.assertEqual(result.budgets, ()) + self.assertEqual(result.risk_flags, ("rejected:concentration",)) def test_no_mandate_allows_exactly_ten_percent(self) -> None: result = apply_risk_gate( @@ -47,6 +51,20 @@ def test_no_mandate_allows_exactly_ten_percent(self) -> None: self.assertEqual(len(result.positions), 1) self.assertIn("risk_gate:passed", result.risk_flags) + def test_no_mandate_rejects_multiple_nonzero_positions(self) -> None: + result = apply_risk_gate( + _decision( + positions=( + PositionTarget(symbol="SPY", target_weight=0.05), + PositionTarget(symbol="BOXX", target_weight=0.05), + ), + ), + ) + + self.assertEqual(result.positions, ()) + self.assertEqual(result.budgets, ()) + self.assertEqual(result.risk_flags, ("rejected:too_many_positions",)) + def test_unknown_mandate_remains_fail_closed(self) -> None: result = apply_risk_gate( _decision(positions=(PositionTarget(symbol="SPY", target_weight=0.10),)), @@ -126,6 +144,338 @@ def evaluate(self, market_data): self.assertEqual(gated.positions, ()) self.assertEqual(gated.risk_flags, ("rejected:risk_engine",)) + def test_rejects_non_approve_risk_engine_action_and_clears_budgets(self) -> None: + decision = StrategyDecision( + positions=(PositionTarget(symbol="SPY", target_weight=0.10),), + budgets=(BudgetIntent(name="risk_budget", amount=1.0),), + ) + engine = Mock() + engine.assess.return_value = RiskAction(action="watch", reason="not approved") + + with patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine): + gated = apply_risk_gate( + decision, + portfolio_snapshot={"total_equity": 100_000.0}, + market_data={}, + ) + + self.assertEqual(gated.positions, ()) + self.assertEqual(gated.budgets, ()) + self.assertEqual(gated.risk_flags, ("rejected:risk_engine",)) + + +class AssessWithEvidenceTests(unittest.TestCase): + _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) + + @staticmethod + def _mandate(**overrides: object) -> dict[str, object]: + mandate: dict[str, object] = { + "mandate_id": "binance_crypto_research_only_v1", + "mandate_version": "2026-08-04.1", + "authority_receipt_sha256": "a" * 64, + "authority_scope": "RESEARCH_ONLY", + "strategy_profile": "crypto_live_pool_rotation", + "account_mode": "single_strategy_account_v1", + "effective_at": "2026-08-04T04:27:55Z", + "expires_at": "2026-09-03T15:59:59Z", + "max_snapshot_age_seconds": 300, + "effective_exposure_cap": 0.50, + "loss_budget": 0.0, + "product_caps": 1.0, + "nominal_caps": 1.0, + "product_leverage_factors": {"BTCUSDT": 1}, + "allowed_nonzero_assets": ["BTCUSDT"], + "source_revision": "14b27d98bda18455439cbd6470c52a069befd002", + } + mandate.update(overrides) + return mandate + + @staticmethod + def _snapshot(**overrides: object) -> dict[str, object]: + snapshot: dict[str, object] = { + "as_of": "2026-08-04T04:27:55Z", + "observed_effective_exposure": 0.10, + "total_equity": 100_000.0, + "account_id": "private-account-id", + "positions": [{"symbol": "BTCUSDT", "quantity": 123.0}], + } + snapshot.update(overrides) + return snapshot + + def test_approved_receipt_is_immutable_and_redacts_digest_inputs(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.20),), + ) + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + first = assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + redacted_equivalent = assess_with_evidence( + decision, + self._snapshot( + account_id="another-private-account", + positions=[{"symbol": "BTCUSDT", "quantity": 999.0}], + ), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + + self.assertEqual(first.assessment.outcome, "APPROVE") + self.assertEqual(first.assessment.effective_exposure_cap, 0.50) + self.assertEqual(first.assessment.observed_effective_exposure, 0.10) + self.assertEqual(first.assessment.proposed_effective_exposure, 0.20) + self.assertEqual(first.assessment.assessment_sha256, redacted_equivalent.assessment.assessment_sha256) + self.assertEqual(len(first.decision.positions), 1) + + def test_zero_cap_research_mandate_never_produces_order_authority(self) -> None: + decision = StrategyDecision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), + budgets=(BudgetIntent(name="risk_budget", amount=1.0),), + ) + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + self._snapshot(), + scope="ACCOUNT", + mandate_provenance=self._mandate( + effective_exposure_cap=0.0, + allowed_nonzero_assets=[], + ), + market_data={}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + self.assertIn("effective_exposure_cap", result.assessment.reason_codes) + engine.assess.assert_called_once() + + def test_invalid_scope_rejects_fail_closed(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="STRATEGY", + mandate_provenance=self._mandate(), + market_data={}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.assessment.scope, "MEMBER") + self.assertEqual(result.decision.positions, ()) + + def test_unmapped_or_empty_product_caps_reject_fail_closed(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.20),), + ) + for cap_overrides in ( + {"product_caps": {"ETHUSDT": 0.10}}, + {"product_caps": {}}, + {"nominal_caps": {"ETHUSDT": 0.10}}, + {"nominal_caps": {}}, + ): + with ( + self.subTest(cap_overrides=cap_overrides), + 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(**cap_overrides), + market_data={}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + + def test_decision_digest_binds_position_and_budget_execution_fields(self) -> None: + mandate = self._mandate( + loss_budget=10.0, + product_leverage_factors={"BTCUSDT": 1, "ETHUSDT": 1}, + allowed_nonzero_assets=["BTCUSDT", "ETHUSDT"], + ) + decisions = ( + StrategyDecision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), + budgets=(BudgetIntent(name="risk_budget", amount=1.0),), + ), + StrategyDecision( + positions=(PositionTarget(symbol="ETHUSDT", target_weight=0.10),), + budgets=(BudgetIntent(name="risk_budget", amount=1.0),), + ), + StrategyDecision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), + budgets=(BudgetIntent(name="loss_budget", amount=2.0),), + ), + ) + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + results = tuple( + assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=mandate, + market_data={}, + ) + for decision in decisions + ) + + digests = {result.assessment.decision_digest_sha256 for result in results} + self.assertEqual(len(digests), len(decisions)) + self.assertTrue(all(result.assessment.outcome == "APPROVE" for result in results)) + self.assertNotIn("BTCUSDT", repr(results[0].assessment)) + self.assertNotIn("risk_budget", repr(results[0].assessment)) + + def test_unmandated_fallback_rejects_stale_snapshot(self) -> None: + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + result = assess_with_evidence( + _decision(positions=(PositionTarget(symbol="SPY", target_weight=0.10),)), + self._snapshot(as_of="2026-08-04T04:17:55Z"), + scope="MEMBER", + mandate_provenance=None, + market_data={}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("stale_portfolio_snapshot", result.assessment.reason_codes) + self.assertEqual(result.decision.positions, ()) + + def test_risk_plugin_exception_rejects_without_exposing_exception(self) -> None: + class CrashingPlugin: + plugin_name = "crashing_plugin" + schema_version = "test.v1" + + def evaluate(self, market_data): + raise RuntimeError("private plugin exception detail") + + engine = RiskEngine(plugins=(CrashingPlugin(),)) + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), + ) + self.assertEqual(engine.assess(decision, self._snapshot()).action, "reject") + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("risk_engine_non_approve", result.assessment.reason_codes) + self.assertNotIn("private plugin exception detail", repr(result)) + + def test_canonical_portfolio_snapshot_matches_mapping_normalization(self) -> None: + canonical = PortfolioSnapshot( + as_of=self._NOW.replace(minute=27, second=55), + total_equity=100_000.0, + metadata={"observed_effective_exposure": 0.10}, + ) + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.20),), + ) + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + canonical_result = assess_with_evidence( + decision, + canonical, + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + mapping_result = assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + + self.assertEqual(canonical_result.assessment.outcome, "APPROVE") + self.assertEqual( + canonical_result.assessment.portfolio_snapshot_digest_sha256, + mapping_result.assessment.portfolio_snapshot_digest_sha256, + ) + + def test_value_target_uses_positive_finite_snapshot_equity(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_value=20_000.0),), + ) + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + approved = assess_with_evidence( + decision, + self._snapshot(total_equity=100_000.0), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + + self.assertEqual(approved.assessment.outcome, "APPROVE") + self.assertEqual(approved.assessment.proposed_effective_exposure, 0.20) + for invalid_equity in (None, 0.0, float("inf")): + snapshot = self._snapshot() + if invalid_equity is None: + snapshot.pop("total_equity") + else: + snapshot["total_equity"] = invalid_equity + with ( + self.subTest(invalid_equity=invalid_equity), + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + ): + rejected = assess_with_evidence( + decision, + snapshot, + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + ) + self.assertEqual(rejected.assessment.outcome, "REJECT") + self.assertEqual(rejected.decision.positions, ()) + + def test_mandate_rejects_budget_only_decision_above_authority(self) -> None: + decision = StrategyDecision( + budgets=(BudgetIntent(name="risk_budget", amount=1.0),), + ) + for authority in ( + {"effective_exposure_cap": 0.0, "loss_budget": 0.0}, + {"effective_exposure_cap": 0.50, "loss_budget": 0.50}, + ): + with ( + self.subTest(authority=authority), + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + ): + result = assess_with_evidence( + decision, + self._snapshot(observed_effective_exposure=0.0), + scope="MEMBER", + mandate_provenance=self._mandate( + **authority, + allowed_nonzero_assets=[], + ), + market_data={}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + self.assertIn("budget_authority_exceeded", result.assessment.reason_codes) + class BootstrapSmallAccountV2RiskGateTests(unittest.TestCase): _MANDATE = "bootstrap_small_account_v2"