From 27b786a76148568dd8fbd6b305478979906cf011 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:05:32 +0800 Subject: [PATCH 1/2] Harden canonical risk metadata ingress Co-Authored-By: Codex --- src/quant_platform_kit/risk/gate.py | 1464 +++++++++++++++------------ tests/test_risk_gate.py | 954 ++++++++++------- 2 files changed, 1420 insertions(+), 998 deletions(-) diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index 08901ed..2eba0ec 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -6,11 +6,12 @@ from __future__ import annotations -from datetime import datetime, timezone import hashlib import json import logging import math +import unicodedata +from datetime import datetime, timezone from typing import Any, Mapping from quant_platform_kit.common.models import PortfolioSnapshot @@ -21,7 +22,11 @@ RiskGateResult, ) from quant_platform_kit.risk.engine import build_risk_engine -from quant_platform_kit.strategy_contracts import StrategyDecision +from quant_platform_kit.strategy_contracts import ( + BudgetIntent, + PositionTarget, + StrategyDecision, +) logger = logging.getLogger(__name__) @@ -34,39 +39,9 @@ _TQQQ_ETF_ONLY_FACTORS = {"TQQQ": 3, "BOXX": 1} _TQQQ_ETF_ONLY_NOMINAL_CAPS = {"TQQQ": 0.15, "BOXX": 0.50} _TQQQ_ETF_ONLY_EFFECTIVE_CAPS = {"TQQQ": 0.45, "BOXX": 0.50} -_GLOBAL_ETF_RESEARCH_MANDATE = "global_etf_rotation_etf_only_research_v1" -_GLOBAL_ETF_STRATEGY_PROFILE = ( - "global_etf_rotation_etf_only_single_strategy_research_v1" -) -_GLOBAL_ETF_ACCOUNT_MODE = "single_strategy_research_v1" -_GLOBAL_ETF_ALLOWED_ASSETS = ( - "EWY", - "EWT", - "INDA", - "FXI", - "EWJ", - "VGK", - "VOO", - "XLK", - "SMH", - "GLD", - "SLV", - "USO", - "DBA", - "XLE", - "XLF", - "ITA", - "XLP", - "XLU", - "XLV", - "IHI", - "VNQ", - "KRE", - "BIL", +_RETIRED_GLOBAL_ETF_RESEARCH_MANDATE = ( + "global_etf_rotation_etf_only_research_v1" ) -_GLOBAL_ETF_FACTORS = {symbol: 1 for symbol in _GLOBAL_ETF_ALLOWED_ASSETS} -_GLOBAL_ETF_CAPS = {symbol: 0.50 for symbol in _GLOBAL_ETF_ALLOWED_ASSETS} -_GLOBAL_ETF_STOP_FILL_POLICY = "gap_aware_min_open_or_stop_v1" _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.v2" @@ -75,6 +50,8 @@ _FALLBACK_MAX_SNAPSHOT_AGE_SECONDS_V1 = 300.0 _ALLOWED_SCOPES = frozenset({"MEMBER", "ACCOUNT"}) _ALLOWED_MANDATE_SCOPES = frozenset({"RESEARCH_ONLY", "PAPER", "LIVE"}) +_MAX_JSON_SAFE_INTEGER = (1 << 53) - 1 +_MAX_MATERIAL_ITEMS = 1_000 def _utc_now() -> datetime: @@ -96,47 +73,165 @@ def _canonical_digest(value: Mapping[str, Any]) -> str: def _finite_number(value: Any) -> float | None: - if isinstance(value, bool) or not isinstance(value, (int, float)): + if type(value) not in (int, float): + return None + if type(value) is int and abs(value) > _MAX_JSON_SAFE_INTEGER: + return None + try: + number = float(value) + except (OverflowError, TypeError, ValueError): 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: +def _bounded_nonnegative_int(value: Any) -> int | None: + if type(value) is not int or not 0 <= value <= _MAX_JSON_SAFE_INTEGER: + return None + return value + + +def _canonical_string( + value: Any, + *, + optional: bool = False, +) -> tuple[str | None, bool]: + if value is None: + return None, optional + if type(value) is not str or not value or value != value.strip(): + return None, False + try: + value.encode("utf-8") + except UnicodeEncodeError: + return None, False + if unicodedata.normalize("NFC", value) != value: + return None, False + return value, True + + +def _canonical_string_list(value: Any) -> list[str] | None: + if type(value) not in (list, tuple) or len(value) > _MAX_MATERIAL_ITEMS: + return None + result: list[str] = [] + for candidate in value: + normalized, valid = _canonical_string(candidate) + if not valid or normalized is None: return None - else: + result.append(normalized) + return result + + +def _canonical_numeric_mapping( + value: Any, + *, + integer: bool = False, + minimum: float = 0.0, + maximum: float | None = None, +) -> dict[str, int | float] | None: + if not isinstance(value, Mapping) or len(value) > _MAX_MATERIAL_ITEMS: return None - return parsed.astimezone(timezone.utc) if parsed.tzinfo is not None else None + result: dict[str, int | float] = {} + for raw_key, raw_value in value.items(): + key, valid_key = _canonical_string(raw_key) + if not valid_key or key is None: + return None + if integer: + number = _bounded_nonnegative_int(raw_value) + else: + number = _finite_number(raw_value) + if ( + number is None + or number < minimum + or (maximum is not None and number > maximum) + ): + return None + result[key] = number + return result -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() +def _canonical_cap_material( + value: Any, + *, + _depth: int = 0, + _item_count: list[int] | None = None, +) -> Any: + number = _finite_number(value) + if number is not None: + return number if 0.0 <= number <= 1.0 else None + if ( + _depth >= 8 + or not isinstance(value, Mapping) + or not value + or len(value) > _MAX_MATERIAL_ITEMS + ): + return None + item_count = _item_count if _item_count is not None else [0] + result: dict[str, Any] = {} + for raw_key, raw_value in value.items(): + item_count[0] += 1 + key, valid_key = _canonical_string(raw_key) + if ( + item_count[0] > _MAX_MATERIAL_ITEMS + or not valid_key + or key is None + ): + return None + candidate = _canonical_cap_material( + raw_value, + _depth=_depth + 1, + _item_count=item_count, ) - cap = _finite_number(value) - return cap is not None and 0.0 <= cap <= 1.0 + if candidate is None: + return None + result[key] = candidate + return result + + +def _safe_diagnostics(value: Any) -> tuple[dict[str, Any], bool]: + if not isinstance(value, Mapping) or len(value) > _MAX_MATERIAL_ITEMS: + return {}, False + result: dict[str, Any] = {} + for raw_key, raw_value in value.items(): + key, valid = _canonical_string(raw_key) + if not valid or key is None: + return {}, False + result[key] = raw_value + return result, True + + +def _parse_utc_timestamp(value: Any) -> datetime | None: + if type(value) is datetime: + parsed = value + else: + timestamp, valid = _canonical_string(value) + if not valid or timestamp is None or not timestamp.endswith("Z"): + return None + try: + parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + except (OverflowError, TypeError, ValueError): + return None + return parsed.astimezone(timezone.utc) if parsed.tzinfo is not None else None def _sha256(value: Any) -> str | None: - if not isinstance(value, str) or len(value) != 64: + normalized, valid = _canonical_string(value) + if not valid or normalized is None or len(normalized) != 64: return None - return value if all(character in "0123456789abcdef" for character in value) else None + return ( + normalized + if all(character in "0123456789abcdef" for character in normalized) + else None + ) def _git_revision(value: Any) -> str | None: - if not isinstance(value, str) or len(value) != 40: + normalized, valid = _canonical_string(value) + if not valid or normalized is None or len(normalized) != 40: return None - return value if all(character in "0123456789abcdef" for character in value) else None + return ( + normalized + if all(character in "0123456789abcdef" for character in normalized) + else None + ) def _decision_metrics( @@ -147,22 +242,42 @@ def _decision_metrics( 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)) + raw_positions = decision.positions or () + if type(raw_positions) is not tuple or len(raw_positions) > _MAX_MATERIAL_ITEMS: + raw_positions = () + reason_codes.add("invalid_risk_metadata") + for position in raw_positions: + if type(position) is not PositionTarget: + reason_codes.add("invalid_risk_metadata") + continue + symbol, valid_symbol = _canonical_string(position.symbol) + role, valid_role = _canonical_string(position.role, optional=True) + order_preference, valid_order = _canonical_string( + position.order_preference, + optional=True, + ) + weight = _finite_number(position.target_weight) + target_value = _finite_number(position.target_value) + if ( + not valid_symbol + or not valid_role + or not valid_order + or (position.target_weight is not None and weight is None) + or (position.target_value is not None and target_value is None) + ): + reason_codes.add("invalid_risk_metadata") position_payloads.append( { - "symbol": symbol if isinstance(symbol, str) else None, + "symbol": symbol, "target_weight": weight, "target_value": target_value, - "role": getattr(position, "role", None), - "order_preference": getattr(position, "order_preference", None), + "role": role, + "order_preference": order_preference, } ) if ( - not isinstance(symbol, str) - or not symbol + not valid_symbol + or symbol is None or (weight is None) == (target_value is None) ): reason_codes.add("invalid_decision_exposure") @@ -173,20 +288,46 @@ def _decision_metrics( reason_codes.add("invalid_decision_exposure") continue normalized_weight = target_value / total_equity - if normalized_weight is None or normalized_weight < 0.0: + if ( + normalized_weight is None + or not math.isfinite(normalized_weight) + 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 (): + raw_budgets = decision.budgets or () + if type(raw_budgets) is not tuple or len(raw_budgets) > _MAX_MATERIAL_ITEMS: + raw_budgets = () + reason_codes.add("invalid_risk_metadata") + for budget in raw_budgets: + if type(budget) is not BudgetIntent: + reason_codes.add("invalid_risk_metadata") + continue + name, valid_name = _canonical_string(budget.name) + symbol, valid_symbol = _canonical_string(budget.symbol, optional=True) + unit, valid_unit = _canonical_string(budget.unit) + purpose, valid_purpose = _canonical_string(budget.purpose, optional=True) + amount = _finite_number(budget.amount) + if ( + not valid_name + or not valid_symbol + or not valid_unit + or not valid_purpose + or (budget.amount is not None and amount is None) + ): + reason_codes.add("invalid_risk_metadata") + if amount is None or amount < 0.0: + reason_codes.add("invalid_decision_budget") 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), + "name": name, + "symbol": symbol, + "amount": amount, + "unit": unit, + "purpose": purpose, } ) return { @@ -236,10 +377,11 @@ def _snapshot_metrics( def _exact_numeric_mapping(value: Any, expected: Mapping[str, float]) -> bool: - if not isinstance(value, Mapping) or set(value) != set(expected): + normalized = _canonical_numeric_mapping(value, maximum=1.0) + if normalized is None or set(normalized) != set(expected): return False return all( - (number := _finite_number(value[key])) is not None + (number := _finite_number(normalized[key])) is not None and number == expected_value for key, expected_value in expected.items() ) @@ -266,18 +408,15 @@ def _exact_tqqq_mandate_errors( "precommitted_executable_stop_distance", "max_consecutive_completed_losing_exits", ) - allowed_assets = mandate_provenance.get("allowed_nonzero_assets") - factors = mandate_provenance.get("product_leverage_factors") - exact_factors = ( - isinstance(factors, Mapping) - and set(factors) == set(_TQQQ_ETF_ONLY_FACTORS) - and all( - not isinstance(factors[symbol], bool) - and isinstance(factors[symbol], int) - and factors[symbol] == expected - for symbol, expected in _TQQQ_ETF_ONLY_FACTORS.items() - ) + allowed_assets = _canonical_string_list( + mandate_provenance.get("allowed_nonzero_assets") ) + factors = _canonical_numeric_mapping( + mandate_provenance.get("product_leverage_factors"), + integer=True, + minimum=1.0, + ) + exact_factors = factors == _TQQQ_ETF_ONLY_FACTORS invalid = ( any(field not in mandate_provenance for field in required) or mandate_provenance.get("mandate_version") != "v1" @@ -304,13 +443,13 @@ def _exact_tqqq_mandate_errors( _TQQQ_ETF_ONLY_EFFECTIVE_CAPS, ) or not exact_factors - or not isinstance(allowed_assets, (list, tuple)) + or allowed_assets is None or len(allowed_assets) != 2 or set(allowed_assets) != set(_TQQQ_ETF_ONLY_FACTORS) - or isinstance(mandate_provenance.get("max_nonzero_assets"), bool) - or mandate_provenance.get("max_nonzero_assets") != 1 - or isinstance(mandate_provenance.get("broker_margin_factor"), bool) - or mandate_provenance.get("broker_margin_factor") != 1 + or _bounded_nonnegative_int(mandate_provenance.get("max_nonzero_assets")) + != 1 + or _bounded_nonnegative_int(mandate_provenance.get("broker_margin_factor")) + != 1 or mandate_provenance.get("margin_stacking") is not False or mandate_provenance.get("borrowing") is not False or mandate_provenance.get("shorting") is not False @@ -320,107 +459,15 @@ def _exact_tqqq_mandate_errors( mandate_provenance.get("precommitted_executable_stop_distance") ) != 0.05 - or isinstance( - mandate_provenance.get("max_consecutive_completed_losing_exits"), - bool, + or _bounded_nonnegative_int( + mandate_provenance.get("max_consecutive_completed_losing_exits") ) - or mandate_provenance.get("max_consecutive_completed_losing_exits") != 5 + != 5 or (expires_at - effective_at).total_seconds() > 90 * 24 * 60 * 60 ) return {"invalid_tqqq_research_mandate"} if invalid else set() -def _exact_global_etf_mandate_errors( - mandate_provenance: Mapping[str, Any], - *, - effective_at: datetime, - expires_at: datetime, -) -> set[str]: - if mandate_provenance.get("mandate_id") != _GLOBAL_ETF_RESEARCH_MANDATE: - return set() - required = ( - "loss_budget_equity_reference", - "product_effective_caps", - "max_nonzero_assets", - "broker_margin_factor", - "margin_stacking", - "borrowing", - "shorting", - "income_sleeve_enabled", - "option_overlay_enabled", - "ai_overlay_enabled", - "market_regime_overlay_enabled", - "precommitted_executable_stop_distance", - "stop_fill_policy", - "max_consecutive_completed_losing_exits", - ) - allowed_assets = mandate_provenance.get("allowed_nonzero_assets") - factors = mandate_provenance.get("product_leverage_factors") - exact_factors = ( - isinstance(factors, Mapping) - and set(factors) == set(_GLOBAL_ETF_FACTORS) - and all( - not isinstance(factors[symbol], bool) - and isinstance(factors[symbol], int) - and factors[symbol] == expected - for symbol, expected in _GLOBAL_ETF_FACTORS.items() - ) - ) - invalid = ( - any(field not in mandate_provenance for field in required) - or mandate_provenance.get("mandate_version") != "v1" - or mandate_provenance.get("authority_scope") != "RESEARCH_ONLY" - or mandate_provenance.get("strategy_profile") - != _GLOBAL_ETF_STRATEGY_PROFILE - or mandate_provenance.get("account_mode") != _GLOBAL_ETF_ACCOUNT_MODE - or _finite_number(mandate_provenance.get("max_snapshot_age_seconds")) - != 300.0 - or _finite_number(mandate_provenance.get("effective_exposure_cap")) != 0.50 - or _finite_number(mandate_provenance.get("loss_budget")) != 0.01 - or mandate_provenance.get("loss_budget_equity_reference") - != "completed_session_equity" - or not _exact_numeric_mapping( - mandate_provenance.get("product_caps"), - _GLOBAL_ETF_CAPS, - ) - or not _exact_numeric_mapping( - mandate_provenance.get("nominal_caps"), - _GLOBAL_ETF_CAPS, - ) - or not _exact_numeric_mapping( - mandate_provenance.get("product_effective_caps"), - _GLOBAL_ETF_CAPS, - ) - or not exact_factors - or not isinstance(allowed_assets, (list, tuple)) - or tuple(allowed_assets) != _GLOBAL_ETF_ALLOWED_ASSETS - or isinstance(mandate_provenance.get("max_nonzero_assets"), bool) - or mandate_provenance.get("max_nonzero_assets") != 2 - or isinstance(mandate_provenance.get("broker_margin_factor"), bool) - or mandate_provenance.get("broker_margin_factor") != 1 - or mandate_provenance.get("margin_stacking") is not False - or mandate_provenance.get("borrowing") is not False - or mandate_provenance.get("shorting") is not False - or mandate_provenance.get("income_sleeve_enabled") is not False - or mandate_provenance.get("option_overlay_enabled") is not False - or mandate_provenance.get("ai_overlay_enabled") is not False - or mandate_provenance.get("market_regime_overlay_enabled") is not False - or _finite_number( - mandate_provenance.get("precommitted_executable_stop_distance") - ) - != 0.05 - or mandate_provenance.get("stop_fill_policy") - != _GLOBAL_ETF_STOP_FILL_POLICY - or isinstance( - mandate_provenance.get("max_consecutive_completed_losing_exits"), - bool, - ) - or mandate_provenance.get("max_consecutive_completed_losing_exits") != 5 - or (expires_at - effective_at).total_seconds() > 90 * 24 * 60 * 60 - ) - return {"invalid_global_etf_research_mandate"} if invalid else set() - - def _mandate_fields( mandate_provenance: Mapping[str, Any] | None, *, @@ -441,6 +488,9 @@ def _mandate_fields( }, set() if not isinstance(mandate_provenance, Mapping): return {}, {"invalid_mandate"} + if mandate_provenance.get("mandate_id") == _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE: + return {}, {"retired_global_etf_research_mandate"} + required = ( "mandate_id", "mandate_version", @@ -464,17 +514,28 @@ def _mandate_fields( "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 - ): + if any(field not in mandate_provenance for field in required): return {}, {"invalid_mandate"} - authority_scope = mandate_provenance["authority_scope"] + + mandate_id, valid_mandate_id = _canonical_string( + mandate_provenance["mandate_id"] + ) + mandate_version, valid_mandate_version = _canonical_string( + mandate_provenance["mandate_version"] + ) + authority_scope, valid_authority_scope = _canonical_string( + mandate_provenance["authority_scope"] + ) + strategy_profile, valid_strategy_profile = _canonical_string( + mandate_provenance["strategy_profile"] + ) + account_mode, valid_account_mode = _canonical_string( + mandate_provenance["account_mode"] + ) receipt_sha256 = _sha256(mandate_provenance["authority_receipt_sha256"]) strategy_revision = _git_revision(mandate_provenance["strategy_revision"]) runner_revision = _git_revision(mandate_provenance["runner_revision"]) + source_revision = _git_revision(mandate_provenance["source_revision"]) config_sha256 = _sha256(mandate_provenance["config_sha256"]) input_manifest_sha256 = _sha256(mandate_provenance["input_manifest_sha256"]) candidate_identity_sha256 = _sha256( @@ -482,28 +543,44 @@ def _mandate_fields( ) 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"]) + 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"]) + product_caps = _canonical_cap_material(mandate_provenance["product_caps"]) + nominal_caps = _canonical_cap_material(mandate_provenance["nominal_caps"]) + product_effective_caps = _canonical_cap_material( + mandate_provenance.get("product_effective_caps", 1.0) + ) + factors = _canonical_numeric_mapping( + mandate_provenance["product_leverage_factors"], + integer=True, + minimum=1.0, + ) + allowed_assets = _canonical_string_list( + mandate_provenance["allowed_nonzero_assets"] + ) + max_nonzero_assets_value = mandate_provenance.get("max_nonzero_assets") + max_nonzero_assets = ( + None + if max_nonzero_assets_value is None + else _bounded_nonnegative_int(max_nonzero_assets_value) + ) if ( - not isinstance(mandate_provenance["mandate_id"], str) - or not isinstance(mandate_provenance["mandate_version"], str) - or _git_revision(mandate_provenance["source_revision"]) is None - or not isinstance(mandate_provenance["strategy_profile"], str) - or not mandate_provenance["strategy_profile"] - or mandate_provenance["strategy_profile"] - != mandate_provenance["strategy_profile"].strip() - or not isinstance(mandate_provenance["account_mode"], str) - or not mandate_provenance["account_mode"] - or mandate_provenance["account_mode"] - != mandate_provenance["account_mode"].strip() + not valid_mandate_id + or not valid_mandate_version + or not valid_authority_scope + or not valid_strategy_profile + or not valid_account_mode + or authority_scope not in _ALLOWED_MANDATE_SCOPES + or receipt_sha256 is None or strategy_revision is None or runner_revision is None + or source_revision is None or config_sha256 is None or input_manifest_sha256 is None or candidate_identity_sha256 is None - 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 @@ -512,47 +589,33 @@ def _mandate_fields( or not 0.0 <= cap <= 1.0 or loss_budget is None or loss_budget < 0.0 + or product_caps is None + or nominal_caps is None + or product_effective_caps is None + or factors is None + or not factors + or allowed_assets is None + or (max_nonzero_assets_value is not None and max_nonzero_assets is None) ): return {}, {"invalid_mandate"} if effective_at > now or expires_at < now or expires_at <= effective_at: return {}, {"expired_mandate"} - exact_mandate_errors = set() - for validator in ( - _exact_tqqq_mandate_errors, - _exact_global_etf_mandate_errors, - ): - exact_mandate_errors.update( - validator( - mandate_provenance, - effective_at=effective_at, - expires_at=expires_at, - ) - ) + + exact_mandate_errors = _exact_tqqq_mandate_errors( + mandate_provenance, + effective_at=effective_at, + expires_at=expires_at, + ) if exact_mandate_errors: return {}, exact_mandate_errors - 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"], + "mandate_id": mandate_id, + "mandate_version": mandate_version, "authority_receipt_sha256": receipt_sha256, "authority_scope": authority_scope, - "source_revision": mandate_provenance["source_revision"], - "strategy_profile": mandate_provenance["strategy_profile"], - "account_mode": mandate_provenance["account_mode"], + "source_revision": source_revision, + "strategy_profile": strategy_profile, + "account_mode": account_mode, "strategy_revision": strategy_revision, "runner_revision": runner_revision, "config_sha256": config_sha256, @@ -562,17 +625,13 @@ def _mandate_fields( "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"], - "product_effective_caps": mandate_provenance.get( - "product_effective_caps", - 1.0, - ), - "allowed_nonzero_assets": set(allowed_assets) if allowed_assets is not None else None, - "max_nonzero_assets": mandate_provenance.get("max_nonzero_assets"), + "product_caps": product_caps, + "nominal_caps": nominal_caps, + "product_effective_caps": product_effective_caps, + "allowed_nonzero_assets": set(allowed_assets), + "max_nonzero_assets": max_nonzero_assets, }, set() - def _candidate_binding_errors( mandate_provenance: Mapping[str, Any] | None, mandate: Mapping[str, Any], @@ -582,7 +641,26 @@ def _candidate_binding_errors( return {"candidate_without_mandate"} if candidate_identity is not None else set() if candidate_identity is None: return {"missing_candidate_identity"} - if not isinstance(candidate_identity, CandidateRiskIdentity): + if type(candidate_identity) is not CandidateRiskIdentity: + return {"invalid_candidate_identity"} + strategy_profile, valid_strategy_profile = _canonical_string( + candidate_identity.strategy_profile + ) + account_mode, valid_account_mode = _canonical_string( + candidate_identity.account_mode + ) + if ( + not valid_strategy_profile + or not valid_account_mode + or strategy_profile is None + or account_mode is None + or _git_revision(candidate_identity.strategy_revision) is None + or _git_revision(candidate_identity.runner_revision) is None + or _sha256(candidate_identity.config_sha256) is None + or _sha256(candidate_identity.input_manifest_sha256) is None + or _sha256(candidate_identity.authority_receipt_sha256) is None + or _sha256(candidate_identity.candidate_sha256) is None + ): return {"invalid_candidate_identity"} if not mandate: return set() @@ -684,16 +762,12 @@ def _risk_control_fields( "drawdown_scalar": None, "risk_control_state_digest_sha256": None, } - if mandate.get("mandate_id") == _GLOBAL_ETF_RESEARCH_MANDATE: - return _global_etf_risk_control_fields( - risk_control_state, - mandate=mandate, - now=now, - active_positions=active_positions, - ) if mandate.get("mandate_id") != _TQQQ_ETF_ONLY_RESEARCH_MANDATE: return empty, set() - if not isinstance(risk_control_state, Mapping): + if ( + not isinstance(risk_control_state, Mapping) + or len(risk_control_state) > _MAX_MATERIAL_ITEMS + ): return empty, {"missing_risk_control_state"} required = ( @@ -713,33 +787,43 @@ def _risk_control_fields( errors.add("invalid_risk_control_state") as_of = _parse_utc_timestamp(risk_control_state.get("as_of")) + mandate_id, valid_mandate_id = _canonical_string( + risk_control_state.get("mandate_id") + ) + candidate_identity_sha256 = _sha256( + risk_control_state.get("candidate_identity_sha256") + ) stop_loss_distance = _finite_number(risk_control_state.get("stop_loss_distance")) account_drawdown = _finite_number( risk_control_state.get("account_drawdown_fraction") ) drawdown_scalar = _finite_number(risk_control_state.get("drawdown_scalar")) - raw_losses = risk_control_state.get("consecutive_completed_losing_exits") - losses = ( - raw_losses - if not isinstance(raw_losses, bool) - and isinstance(raw_losses, int) - and raw_losses >= 0 - else None + losses = _bounded_nonnegative_int( + risk_control_state.get("consecutive_completed_losing_exits") + ) + stop_intent_value = risk_control_state.get("stop_intent_ready") + stop_intent_ready = stop_intent_value if type(stop_intent_value) is bool else None + entry_fill_identity = _sha256( + risk_control_state.get("tqqq_entry_fill_identity_sha256") + ) + stop_entry_fill_identity = _sha256( + risk_control_state.get("stop_entry_fill_identity_sha256") ) max_age = _finite_number(mandate.get("max_snapshot_age_seconds")) + + if not valid_mandate_id: + errors.add("invalid_risk_metadata") if as_of is None or max_age is None: errors.add("invalid_risk_control_state") elif (age := (now - as_of).total_seconds()) < 0.0 or age > max_age: errors.add("stale_risk_control_state") - if risk_control_state.get("mandate_id") != mandate.get("mandate_id"): + if mandate_id != mandate.get("mandate_id"): errors.add("risk_control_mandate_mismatch") - if _sha256(risk_control_state.get("candidate_identity_sha256")) != mandate.get( - "candidate_identity_sha256" - ): + if candidate_identity_sha256 != mandate.get("candidate_identity_sha256"): errors.add("risk_control_candidate_mismatch") if stop_loss_distance != 0.05: errors.add("invalid_stop_loss_distance") - if not isinstance(risk_control_state.get("stop_intent_ready"), bool): + if stop_intent_ready is None: errors.add("invalid_stop_state") if account_drawdown is None or not 0.0 <= account_drawdown <= 1.0: errors.add("invalid_account_drawdown") @@ -760,14 +844,8 @@ def _risk_control_fields( errors.add("invalid_drawdown_scalar") tqqq_active = any(symbol == "TQQQ" for symbol, _ in active_positions) - if tqqq_active and risk_control_state.get("stop_intent_ready") is not True: + if tqqq_active and stop_intent_ready is not True: errors.add("stop_intent_not_ready") - entry_fill_identity = _sha256( - risk_control_state.get("tqqq_entry_fill_identity_sha256") - ) - stop_entry_fill_identity = _sha256( - risk_control_state.get("stop_entry_fill_identity_sha256") - ) if tqqq_active and ( entry_fill_identity is None or stop_entry_fill_identity is None @@ -783,16 +861,10 @@ def _risk_control_fields( payload = { "as_of": _utc_timestamp(as_of) if as_of is not None else None, - "mandate_id": risk_control_state.get("mandate_id"), - "candidate_identity_sha256": _sha256( - risk_control_state.get("candidate_identity_sha256") - ), + "mandate_id": mandate_id, + "candidate_identity_sha256": candidate_identity_sha256, "stop_loss_distance": stop_loss_distance, - "stop_intent_ready": ( - risk_control_state.get("stop_intent_ready") - if isinstance(risk_control_state.get("stop_intent_ready"), bool) - else None - ), + "stop_intent_ready": stop_intent_ready, "tqqq_entry_fill_identity_sha256": entry_fill_identity, "stop_entry_fill_identity_sha256": stop_entry_fill_identity, "consecutive_completed_losing_exits": losses, @@ -801,170 +873,7 @@ def _risk_control_fields( } return { "stop_loss_distance": stop_loss_distance, - "stop_intent_ready": ( - risk_control_state.get("stop_intent_ready") - if isinstance(risk_control_state.get("stop_intent_ready"), bool) - else None - ), - "strategy_breaker_triggered": strategy_breaker, - "account_breaker_triggered": account_breaker, - "account_drawdown_fraction": account_drawdown, - "drawdown_scalar": drawdown_scalar, - "risk_control_state_digest_sha256": _canonical_digest(payload), - }, errors - - -def _global_etf_risk_control_fields( - risk_control_state: Mapping[str, Any] | None, - *, - mandate: Mapping[str, Any], - now: datetime, - active_positions: list[tuple[str, float]], -) -> tuple[dict[str, Any], set[str]]: - empty = { - "stop_loss_distance": None, - "stop_intent_ready": None, - "strategy_breaker_triggered": None, - "account_breaker_triggered": None, - "account_drawdown_fraction": None, - "drawdown_scalar": None, - "risk_control_state_digest_sha256": None, - } - if not isinstance(risk_control_state, Mapping): - return empty, {"missing_risk_control_state"} - - required = ( - "as_of", - "mandate_id", - "candidate_identity_sha256", - "stop_loss_distance", - "stop_fill_policy", - "position_stop_states", - "consecutive_completed_losing_exits", - "account_drawdown_fraction", - "drawdown_scalar", - ) - errors: set[str] = set() - if any(field not in risk_control_state for field in required): - errors.add("invalid_risk_control_state") - - as_of = _parse_utc_timestamp(risk_control_state.get("as_of")) - stop_loss_distance = _finite_number(risk_control_state.get("stop_loss_distance")) - account_drawdown = _finite_number( - risk_control_state.get("account_drawdown_fraction") - ) - drawdown_scalar = _finite_number(risk_control_state.get("drawdown_scalar")) - raw_losses = risk_control_state.get("consecutive_completed_losing_exits") - losses = ( - raw_losses - if not isinstance(raw_losses, bool) - and isinstance(raw_losses, int) - and raw_losses >= 0 - else None - ) - max_age = _finite_number(mandate.get("max_snapshot_age_seconds")) - if as_of is None or max_age is None: - errors.add("invalid_risk_control_state") - elif (age := (now - as_of).total_seconds()) < 0.0 or age > max_age: - errors.add("stale_risk_control_state") - if risk_control_state.get("mandate_id") != mandate.get("mandate_id"): - errors.add("risk_control_mandate_mismatch") - if _sha256(risk_control_state.get("candidate_identity_sha256")) != mandate.get( - "candidate_identity_sha256" - ): - errors.add("risk_control_candidate_mismatch") - if stop_loss_distance != 0.05: - errors.add("invalid_stop_loss_distance") - if risk_control_state.get("stop_fill_policy") != _GLOBAL_ETF_STOP_FILL_POLICY: - errors.add("invalid_stop_fill_policy") - if account_drawdown is None or not 0.0 <= account_drawdown <= 1.0: - errors.add("invalid_account_drawdown") - if losses is None: - errors.add("invalid_strategy_breaker_state") - - expected_scalar: float | None = None - if account_drawdown is not None and 0.0 <= account_drawdown <= 1.0: - if account_drawdown <= 0.05: - expected_scalar = 1.0 - elif account_drawdown <= 0.10: - expected_scalar = 0.50 - else: - expected_scalar = 0.0 - if drawdown_scalar != expected_scalar: - errors.add("drawdown_scalar_mismatch") - elif drawdown_scalar is None: - errors.add("invalid_drawdown_scalar") - - active_symbols = [symbol for symbol, _weight in active_positions] - if len(active_symbols) != len(set(active_symbols)): - errors.add("duplicate_active_symbol") - raw_stop_states = risk_control_state.get("position_stop_states") - normalized_stop_states: dict[str, dict[str, Any]] = {} - all_stop_intents_ready = True - if not isinstance(raw_stop_states, Mapping): - errors.add("invalid_position_stop_states") - all_stop_intents_ready = False - elif set(raw_stop_states) != set(active_symbols): - errors.add("stop_state_positions_mismatch") - all_stop_intents_ready = False - else: - expected_stop_fields = { - "stop_intent_ready", - "entry_fill_identity_sha256", - "stop_entry_fill_identity_sha256", - } - for symbol in sorted(set(active_symbols)): - raw_stop = raw_stop_states.get(symbol) - if not isinstance(raw_stop, Mapping) or set(raw_stop) != expected_stop_fields: - errors.add("invalid_position_stop_state") - all_stop_intents_ready = False - continue - ready = raw_stop.get("stop_intent_ready") - entry_fill_identity = _sha256( - raw_stop.get("entry_fill_identity_sha256") - ) - stop_entry_fill_identity = _sha256( - raw_stop.get("stop_entry_fill_identity_sha256") - ) - if ready is not True: - errors.add("stop_intent_not_ready") - all_stop_intents_ready = False - if ( - entry_fill_identity is None - or stop_entry_fill_identity is None - or entry_fill_identity != stop_entry_fill_identity - ): - errors.add("stop_entry_fill_identity_mismatch") - all_stop_intents_ready = False - normalized_stop_states[symbol] = { - "stop_intent_ready": ready if isinstance(ready, bool) else None, - "entry_fill_identity_sha256": entry_fill_identity, - "stop_entry_fill_identity_sha256": stop_entry_fill_identity, - } - - strategy_breaker = losses is not None and losses >= 5 - account_breaker = account_drawdown is not None and account_drawdown > 0.10 - if strategy_breaker: - errors.add("strategy_breaker_triggered") - if account_breaker: - errors.add("account_breaker_triggered") - - payload = { - "as_of": _utc_timestamp(as_of) if as_of is not None else None, - "mandate_id": risk_control_state.get("mandate_id"), - "candidate_identity_sha256": _sha256( - risk_control_state.get("candidate_identity_sha256") - ), - "stop_loss_distance": stop_loss_distance, - "stop_fill_policy": risk_control_state.get("stop_fill_policy"), - "position_stop_states": normalized_stop_states, - "consecutive_completed_losing_exits": losses, - "account_drawdown_fraction": account_drawdown, - "drawdown_scalar": drawdown_scalar, - } - return { - "stop_loss_distance": stop_loss_distance, - "stop_intent_ready": all_stop_intents_ready, + "stop_intent_ready": stop_intent_ready, "strategy_breaker_triggered": strategy_breaker, "account_breaker_triggered": account_breaker, "account_drawdown_fraction": account_drawdown, @@ -972,22 +881,28 @@ def _global_etf_risk_control_fields( "risk_control_state_digest_sha256": _canonical_digest(payload), }, errors - -def assess_with_evidence( +def _assess_with_evidence_static( decision: StrategyDecision, portfolio_snapshot: Any, *, - scope: str, + scope: Any, mandate_provenance: Mapping[str, Any] | None, - market_data: Mapping[str, Any], - candidate_identity: CandidateRiskIdentity | None = None, - normalization_origin_weights: Mapping[str, float] | None = None, - risk_control_state: Mapping[str, Any] | None = None, + candidate_identity: CandidateRiskIdentity | None, + normalization_origin_weights: Mapping[str, float] | None, + risk_control_state: Mapping[str, Any] | None, + now: datetime, + risk_action: Any, + risk_engine_failed: bool, ) -> RiskGateResult: - """Assess exactly once and fail closed with a redacted canonical receipt.""" - now = _utc_now() + if type(decision) is not StrategyDecision: + raise TypeError("invalid decision") evaluated_at = _utc_timestamp(now) - assessment_scope = scope if scope in _ALLOWED_SCOPES else "MEMBER" + normalized_scope, valid_scope = _canonical_string(scope) + assessment_scope = ( + normalized_scope + if valid_scope and normalized_scope in _ALLOWED_SCOPES + else "MEMBER" + ) mandate, mandate_errors = _mandate_fields(mandate_provenance, now=now) reason_codes = set(mandate_errors) reason_codes.update( @@ -1009,7 +924,11 @@ def assess_with_evidence( total_equity=total_equity, ) reason_codes.update(decision_errors) - if scope not in _ALLOWED_SCOPES: + diagnostics, valid_diagnostics = _safe_diagnostics(decision.diagnostics) + risk_flags = _canonical_string_list(decision.risk_flags or ()) + if not valid_diagnostics or risk_flags is None: + reason_codes.add("invalid_risk_metadata") + if not valid_scope or normalized_scope not in _ALLOWED_SCOPES: reason_codes.add("invalid_scope") can_evaluate_policy = not reason_codes if mandate: @@ -1030,13 +949,8 @@ def assess_with_evidence( weighted_exposure = 0.0 if mandate_provenance is None and len(active_positions) > 1: reason_codes.add("fallback_position_count") - mandate_id = mandate.get("mandate_id") - exact_research_mandate = mandate_id in { - _TQQQ_ETF_ONLY_RESEARCH_MANDATE, - _GLOBAL_ETF_RESEARCH_MANDATE, - } if ( - exact_research_mandate + mandate.get("mandate_id") == _TQQQ_ETF_ONLY_RESEARCH_MANDATE and len(active_positions) > mandate["max_nonzero_assets"] ): reason_codes.add("single_strategy_position_count") @@ -1044,12 +958,24 @@ def assess_with_evidence( 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)) + 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) + 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 @@ -1074,50 +1000,57 @@ def assess_with_evidence( and stop_distance > 0.0 and drawdown_scalar is not None and loss_budget is not None - and weight - > loss_budget * drawdown_scalar / stop_distance + 1e-9 + and weight > loss_budget * drawdown_scalar / stop_distance + 1e-9 ): reason_codes.add("risk_budget_exposure_cap") - weighted_exposure += weight * factor - if mandate_id == _GLOBAL_ETF_RESEARCH_MANDATE: - stop_distance = control_fields["stop_loss_distance"] - drawdown_scalar = control_fields["drawdown_scalar"] - loss_budget = mandate.get("loss_budget") - modeled_stop_loss = ( - sum(weight for _symbol, weight in active_positions) * stop_distance - if stop_distance is not None - else None - ) - if ( - modeled_stop_loss is not None - and drawdown_scalar is not None - and loss_budget is not None - and modeled_stop_loss > loss_budget * drawdown_scalar + 1e-9 - ): - reason_codes.add("risk_budget_exposure_cap") + effective_weight = weight * factor + if not math.isfinite(effective_weight): + reason_codes.add("invalid_risk_metadata") + continue + weighted_exposure += effective_weight + if not math.isfinite(weighted_exposure): + reason_codes.add("invalid_risk_metadata") + weighted_exposure = 0.0 + target_weights: dict[str, float] = {} for symbol, weight in active_positions: - target_weights[symbol] = target_weights.get(symbol, 0.0) + weight + combined_weight = target_weights.get(symbol, 0.0) + weight + if not math.isfinite(combined_weight): + reason_codes.add("invalid_risk_metadata") + continue + target_weights[symbol] = combined_weight valid_normalization = False if normalization_origin_weights is not None: - valid_normalization = validate_reduce_only_normalization( - origin_weights=normalization_origin_weights, - target_weights=target_weights, - product_leverage_factors=factors, - effective_exposure_cap=cap, - observed_effective_exposure=observed, - cash_only=exact_research_mandate, + normalized_origin_material = _canonical_numeric_mapping( + normalization_origin_weights, + minimum=0.0, ) - if not valid_normalization: - reason_codes.add("invalid_reduce_only_normalization") + if normalized_origin_material is None: + reason_codes.update( + {"invalid_risk_metadata", "invalid_reduce_only_normalization"} + ) else: normalized_origin = { symbol: float(weight) - for symbol, weight in sorted(normalization_origin_weights.items()) + for symbol, weight in sorted(normalized_origin_material.items()) } - normalization_origin_digest_sha256 = _canonical_digest( - {"weights": normalized_origin} + valid_normalization = validate_reduce_only_normalization( + origin_weights=normalized_origin, + target_weights=target_weights, + product_leverage_factors=factors, + effective_exposure_cap=cap, + observed_effective_exposure=observed, + cash_only=( + mandate.get("mandate_id") + == _TQQQ_ETF_ONLY_RESEARCH_MANDATE + ), ) + if not valid_normalization: + reason_codes.add("invalid_reduce_only_normalization") + else: + normalization_origin_digest_sha256 = _canonical_digest( + {"weights": normalized_origin} + ) proposed = ( weighted_exposure if valid_normalization @@ -1130,18 +1063,10 @@ def assess_with_evidence( if cap is None or proposed > cap + 1e-9: reason_codes.add("effective_exposure_cap") - has_static_rejection = bool(reason_codes) - try: - risk_action = build_risk_engine().assess( - decision, - portfolio_snapshot, - market_data=market_data, - ) - except Exception: - if not has_static_rejection: + if not reason_codes: + if risk_engine_failed: reason_codes.add("risk_engine_error") - else: - if risk_action.action != "approve" and not has_static_rejection: + elif getattr(risk_action, "action", None) != "approve": reason_codes.add("risk_engine_non_approve") outcome = "REJECT" if reason_codes else "APPROVE" @@ -1158,7 +1083,8 @@ def assess_with_evidence( mandate_scope=mandate.get("authority_scope"), candidate_identity_sha256=( candidate_identity.candidate_sha256 - if isinstance(candidate_identity, CandidateRiskIdentity) + if type(candidate_identity) is CandidateRiskIdentity + and _sha256(candidate_identity.candidate_sha256) is not None else None ), decision_digest_sha256=_canonical_digest(decision_payload), @@ -1186,21 +1112,131 @@ def assess_with_evidence( decision, flag="rejected:risk_gate_assessment", reason=",".join(assessment.reason_codes), + diagnostics=diagnostics, ), 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"}, + risk_flags=tuple(risk_flags or ()) + ("risk_gate:passed",), + diagnostics={**diagnostics, "risk_gate": "APPROVE"}, ), assessment=assessment, ) +def _invalid_assessment_result( + decision: StrategyDecision, + *, + scope: Any, + mandate_provenance: Mapping[str, Any] | None, + candidate_identity: CandidateRiskIdentity | None, + now: datetime, +) -> RiskGateResult: + reason_codes = {"invalid_risk_metadata"} + try: + if ( + isinstance(mandate_provenance, Mapping) + and mandate_provenance.get("mandate_id") + == _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE + ): + reason_codes.add("retired_global_etf_research_mandate") + except Exception: + pass + normalized_scope, valid_scope = _canonical_string(scope) + assessment_scope = ( + normalized_scope + if valid_scope and normalized_scope in _ALLOWED_SCOPES + else "MEMBER" + ) + candidate_sha256 = None + try: + if type(candidate_identity) is CandidateRiskIdentity: + candidate_sha256 = _sha256(candidate_identity.candidate_sha256) + except Exception: + candidate_sha256 = None + invalid_digest = _canonical_digest({"invalid_risk_metadata": True}) + assessment = RiskGateAssessment( + contract_version=_ASSESSMENT_CONTRACT_VERSION, + scope=assessment_scope, + evaluated_at=_utc_timestamp(now), + policy_id=_ASSESSMENT_POLICY_ID, + policy_version=_ASSESSMENT_POLICY_VERSION, + qpk_source_revision=None, + mandate_id=None, + mandate_version=None, + mandate_authority_receipt_sha256=None, + mandate_scope=None, + candidate_identity_sha256=candidate_sha256, + decision_digest_sha256=invalid_digest, + portfolio_snapshot_digest_sha256=invalid_digest, + normalization_origin_digest_sha256=None, + effective_exposure_cap=None, + observed_effective_exposure=None, + proposed_effective_exposure=None, + outcome="REJECT", + reason_codes=tuple(sorted(reason_codes)), + execution_authorized=False, + ) + return RiskGateResult( + decision=_reject( + decision, + flag="rejected:risk_gate_assessment", + reason=",".join(assessment.reason_codes), + diagnostics={}, + ), + assessment=assessment, + ) + + +def assess_with_evidence( + decision: StrategyDecision, + portfolio_snapshot: Any, + *, + scope: str, + mandate_provenance: Mapping[str, Any] | None, + market_data: Mapping[str, Any], + candidate_identity: CandidateRiskIdentity | None = None, + normalization_origin_weights: Mapping[str, float] | None = None, + risk_control_state: Mapping[str, Any] | None = None, +) -> RiskGateResult: + """Assess exactly once and fail closed with a redacted canonical receipt.""" + try: + risk_action = build_risk_engine().assess( + decision, + portfolio_snapshot, + market_data=market_data, + ) + except Exception: + risk_action = None + risk_engine_failed = True + else: + risk_engine_failed = False + now = _utc_now() + try: + return _assess_with_evidence_static( + decision, + portfolio_snapshot, + scope=scope, + mandate_provenance=mandate_provenance, + candidate_identity=candidate_identity, + normalization_origin_weights=normalization_origin_weights, + risk_control_state=risk_control_state, + now=now, + risk_action=risk_action, + risk_engine_failed=risk_engine_failed, + ) + except Exception: + return _invalid_assessment_result( + decision, + scope=scope, + mandate_provenance=mandate_provenance, + candidate_identity=candidate_identity, + now=now, + ) + def enrich_decision_risk_diagnostics( decision: StrategyDecision, *, @@ -1215,9 +1251,17 @@ def enrich_decision_risk_diagnostics( """ diagnostics = dict(decision.diagnostics or {}) if unrealized_pnl_pct is not None: - diagnostics["unrealized_pnl_pct"] = float(unrealized_pnl_pct) + normalized_pnl = _finite_number(unrealized_pnl_pct) + diagnostics["unrealized_pnl_pct"] = ( + normalized_pnl if normalized_pnl is not None else unrealized_pnl_pct + ) if consecutive_losses is not None: - diagnostics["consecutive_losses"] = int(consecutive_losses) + normalized_losses = _bounded_nonnegative_int(consecutive_losses) + diagnostics["consecutive_losses"] = ( + normalized_losses + if normalized_losses is not None + else consecutive_losses + ) if diagnostics == dict(decision.diagnostics or {}): return decision return StrategyDecision( @@ -1228,111 +1272,220 @@ def enrich_decision_risk_diagnostics( ) -def apply_risk_gate( +def _apply_risk_gate_static( decision: StrategyDecision, *, - risk_mandate_id: str | None = None, - product_leverage_factors: Mapping[str, int] | None = None, - available_account_exposure: float | None = None, - max_single_weight: float = _DEFAULT_MAX_SINGLE_WEIGHT, - max_positions: int = 20, - max_total_exposure: float = 1.0, - portfolio_snapshot: Any | None = None, - market_data: Mapping[str, Any] | None = None, + risk_mandate_id: Any, + product_leverage_factors: Any, + available_account_exposure: Any, + max_single_weight: Any, + max_positions: Any, + max_total_exposure: Any, + engine_action: Any, + engine_failed: bool, ) -> StrategyDecision: - """Apply hard risk checks to a strategy decision. - - Checks (in order): - 1. Circuit breaker from diagnostics (unrealized_pnl_pct, consecutive_losses) - 2. Mandate-specific single-account and leverage classification limits - 3. Legacy caller-supplied concentration limits when no mandate is supplied - 4. Legacy position-count and total-exposure limits - 5. RiskEngine.assess() exactly once; missing/invalid snapshots reject - - Returns an empty-position StrategyDecision on REJECT. - """ - diagnostics = dict(decision.diagnostics or {}) + if type(decision) is not StrategyDecision: + raise TypeError("invalid decision") + diagnostics, valid_diagnostics = _safe_diagnostics(decision.diagnostics) static_rejection: tuple[str, str] | None = None - - pnl_pct = diagnostics.get("unrealized_pnl_pct") - if pnl_pct is not None and float(pnl_pct) < _STOP_LOSS_THRESHOLD: - logger.warning( - "risk_gate REJECT stop_loss: unrealized_pnl_pct=%.2f%%", - float(pnl_pct) * 100, - ) + if not valid_diagnostics: static_rejection = ( - "rejected:stop_loss", - f"未实现亏损 {float(pnl_pct):.1%} < {_STOP_LOSS_THRESHOLD:.0%} 止损线", + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", ) - consecutive_losses = diagnostics.get("consecutive_losses") + normalized_risk_flags = _canonical_string_list(decision.risk_flags or ()) + raw_positions = decision.positions or () + raw_budgets = decision.budgets or () if ( - static_rejection is None - and consecutive_losses is not None - and int(consecutive_losses) > _MAX_CONSECUTIVE_LOSSES + normalized_risk_flags is None + or type(raw_positions) is not tuple + or len(raw_positions) > _MAX_MATERIAL_ITEMS + or type(raw_budgets) is not tuple + or len(raw_budgets) > _MAX_MATERIAL_ITEMS ): - logger.warning( - "risk_gate REJECT circuit_breaker: consecutive_losses=%d", - int(consecutive_losses), + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", ) - static_rejection = ( - "rejected:circuit_breaker", - f"连续亏损 {int(consecutive_losses)} 笔 > {_MAX_CONSECUTIVE_LOSSES} 熔断", + + mandate_id, valid_mandate_id = _canonical_string( + risk_mandate_id, + optional=True, + ) + requested_single_weight = _finite_number(max_single_weight) + position_limit = _bounded_nonnegative_int(max_positions) + total_exposure_limit = _finite_number(max_total_exposure) + available_exposure = ( + None + if available_account_exposure is None + else _finite_number(available_account_exposure) + ) + factors = ( + None + if product_leverage_factors is None + else _canonical_numeric_mapping( + product_leverage_factors, + integer=True, + minimum=1.0, + ) + ) + if ( + not valid_mandate_id + or requested_single_weight is None + or requested_single_weight < 0.0 + or position_limit is None + or total_exposure_limit is None + or total_exposure_limit < 0.0 + or ( + available_account_exposure is not None + and available_exposure is None + ) + or (product_leverage_factors is not None and factors is None) + ): + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + + for position in raw_positions if type(raw_positions) is tuple else (): + if type(position) is not PositionTarget: + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + continue + symbol, valid_symbol = _canonical_string(position.symbol) + _role, valid_role = _canonical_string(position.role, optional=True) + _preference, valid_preference = _canonical_string( + position.order_preference, + optional=True, ) + target_value = _finite_number(position.target_value) + if ( + not valid_symbol + or symbol is None + or not valid_role + or not valid_preference + or (position.target_value is not None and target_value is None) + ): + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + + for budget in raw_budgets if type(raw_budgets) is tuple else (): + if type(budget) is not BudgetIntent: + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + continue + _name, valid_name = _canonical_string(budget.name) + _symbol, valid_symbol = _canonical_string(budget.symbol, optional=True) + _unit, valid_unit = _canonical_string(budget.unit) + _purpose, valid_purpose = _canonical_string(budget.purpose, optional=True) + amount = _finite_number(budget.amount) + if ( + not valid_name + or not valid_symbol + or not valid_unit + or not valid_purpose + or amount is None + or amount < 0.0 + ): + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + + pnl_pct_value = diagnostics.get("unrealized_pnl_pct") + if pnl_pct_value is not None: + pnl_pct = _finite_number(pnl_pct_value) + if pnl_pct is None: + diagnostics.pop("unrealized_pnl_pct", None) + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + elif static_rejection is None and pnl_pct < _STOP_LOSS_THRESHOLD: + logger.warning( + "risk_gate REJECT stop_loss: unrealized_pnl_pct=%.2f%%", + pnl_pct * 100, + ) + static_rejection = ( + "rejected:stop_loss", + f"未实现亏损 {pnl_pct:.1%} < {_STOP_LOSS_THRESHOLD:.0%} 止损线", + ) + + losses_value = diagnostics.get("consecutive_losses") + if losses_value is not None: + consecutive_losses = _bounded_nonnegative_int(losses_value) + if consecutive_losses is None: + diagnostics.pop("consecutive_losses", None) + static_rejection = static_rejection or ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) + elif ( + static_rejection is None + and consecutive_losses > _MAX_CONSECUTIVE_LOSSES + ): + logger.warning( + "risk_gate REJECT circuit_breaker: consecutive_losses=%d", + consecutive_losses, + ) + static_rejection = ( + "rejected:circuit_breaker", + f"连续亏损 {consecutive_losses} 笔 > {_MAX_CONSECUTIVE_LOSSES} 熔断", + ) if ( static_rejection is None - and risk_mandate_id not in {None, _APPROVED_BOOTSTRAP_MANDATE} + and mandate_id not in {None, _APPROVED_BOOTSTRAP_MANDATE} ): static_rejection = ( "rejected:unknown_risk_mandate", "风险授权未获批准", ) - positions = decision.positions or () - weights: list[tuple[Any, float]] = [] + positions = raw_positions if type(raw_positions) is tuple else () + weights: list[tuple[PositionTarget, float]] = [] if static_rejection is None: for position in positions: raw_weight = position.target_weight if raw_weight is None: weight = 0.0 - elif isinstance(raw_weight, bool) or not isinstance( - raw_weight, - (int, float), - ): - static_rejection = ( - "rejected:invalid_weight", - f"{position.symbol} 目标仓位无效", - ) - break else: - weight = abs(float(raw_weight)) - if not math.isfinite(weight): - static_rejection = ( - "rejected:invalid_weight", - f"{position.symbol} 目标仓位无效", - ) + normalized_weight = _finite_number(raw_weight) + if normalized_weight is None: + if type(raw_weight) is str: + static_rejection = ( + "rejected:invalid_weight", + f"{position.symbol} 目标仓位无效", + ) + else: + static_rejection = ( + "rejected:invalid_risk_metadata", + "invalid_risk_metadata", + ) break + weight = abs(normalized_weight) if weight > 0.0: weights.append((position, weight)) if ( static_rejection is None and positions - and risk_mandate_id == _APPROVED_BOOTSTRAP_MANDATE + and mandate_id == _APPROVED_BOOTSTRAP_MANDATE ): if len(weights) > 1: static_rejection = ( "rejected:too_many_positions", "bootstrap_small_account_v2 仅允许一个非零持仓", ) - elif available_account_exposure is None or ( - isinstance(available_account_exposure, bool) - or not isinstance(available_account_exposure, (int, float)) - or not math.isfinite(float(available_account_exposure)) - or not 0.0 - <= float(available_account_exposure) - <= _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP + elif available_exposure is None or not ( + 0.0 <= available_exposure <= _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP ): static_rejection = ( "rejected:overexposed", @@ -1340,22 +1493,15 @@ def apply_risk_gate( ) elif weights: active_symbols = {position.symbol for position, _ in weights} - if ( - product_leverage_factors is None - or set(product_leverage_factors) != active_symbols - ): + if factors is None or set(factors) != active_symbols: static_rejection = ( "rejected:leverage_classification", "缺少或不一致的产品杠杆分类", ) else: position, weight = weights[0] - leverage_factor = product_leverage_factors[position.symbol] - if ( - isinstance(leverage_factor, bool) - or not isinstance(leverage_factor, int) - or leverage_factor not in _BOOTSTRAP_NOMINAL_CAPS - ): + leverage_factor = factors[position.symbol] + if leverage_factor not in _BOOTSTRAP_NOMINAL_CAPS: static_rejection = ( "rejected:leverage_classification", "产品杠杆分类无效", @@ -1373,17 +1519,14 @@ def apply_risk_gate( "rejected:overexposed", f"有效敞口 {effective_exposure:.1%} > 50%", ) - elif weight > float(available_account_exposure) + 1e-9: + elif weight > available_exposure + 1e-9: static_rejection = ( "rejected:overexposed", f"名义仓位 {weight:.1%} > 可用账户容量", ) elif static_rejection is None and positions: - 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, + requested_single_weight, _DEFAULT_MAX_SINGLE_WEIGHT, ) if len(weights) > 1: @@ -1403,86 +1546,145 @@ def apply_risk_gate( "rejected:concentration", f"{position.symbol} {weight:.1%} > {effective_single_weight:.0%} 上限", ) - elif len(positions) > max_positions: + elif len(positions) > position_limit: logger.warning( "risk_gate REJECT position_count: %d > %d", len(positions), - max_positions, + position_limit, ) static_rejection = ( "rejected:too_many_positions", - f"{len(positions)} 个持仓 > {max_positions} 上限", + f"{len(positions)} 个持仓 > {position_limit} 上限", ) else: total_weight = sum(weight for _, weight in weights) - if total_weight > max_total_exposure + 1e-9: + if total_weight > total_exposure_limit + 1e-9: logger.warning( "risk_gate REJECT total_exposure: %.2f%% > %.0f%%", total_weight * 100, - max_total_exposure * 100, + total_exposure_limit * 100, ) static_rejection = ( "rejected:overexposed", - f"总仓位 {total_weight:.1%} > {max_total_exposure:.0%}", + f"总仓位 {total_weight:.1%} > {total_exposure_limit:.0%}", ) elif weights: active_symbols = {position.symbol for position, _ in weights} if ( - product_leverage_factors is None - or set(product_leverage_factors) != active_symbols - or any( - isinstance(factor, bool) - or not isinstance(factor, int) - or factor != 1 - for factor in product_leverage_factors.values() - ) + factors is None + or set(factors) != active_symbols + or any(factor != 1 for factor in factors.values()) ): static_rejection = ( "rejected:leverage_classification", "未获授权的风险配置必须明确为无杠杆产品", ) - try: - assessment = build_risk_engine().assess( - decision, - portfolio_snapshot, - market_data=market_data, - ) - except Exception: + engine_rejection: tuple[str, str] | None + if engine_failed: engine_rejection = ("rejected:risk_engine", "risk_engine_error") + elif getattr(engine_action, "action", None) != "approve": + raw_reason = getattr(engine_action, "reason", None) + reason, valid_reason = _canonical_string(raw_reason) + engine_rejection = ( + "rejected:risk_engine", + reason if valid_reason and reason is not None else "risk_engine_non_approve", + ) else: - if assessment.action != "approve": - logger.warning("risk_gate REJECT risk_engine: %s", assessment.reason) - engine_rejection = ("rejected:risk_engine", assessment.reason) - else: - engine_rejection = None + engine_rejection = None rejection = static_rejection or engine_rejection if rejection is not None: - return _reject(decision, flag=rejection[0], reason=rejection[1]) + rejection_diagnostics = ( + {} if rejection[0] == "rejected:invalid_risk_metadata" else diagnostics + ) + return _reject( + decision, + flag=rejection[0], + reason=rejection[1], + diagnostics=rejection_diagnostics, + ) - risk_flags = list(decision.risk_flags or ()) - risk_flags.append("risk_gate:passed") return StrategyDecision( positions=decision.positions, budgets=decision.budgets, - risk_flags=tuple(risk_flags), + risk_flags=tuple(normalized_risk_flags or ()) + ("risk_gate:passed",), diagnostics={**diagnostics, "risk_gate": "APPROVE"}, ) +def apply_risk_gate( + decision: StrategyDecision, + *, + risk_mandate_id: str | None = None, + product_leverage_factors: Mapping[str, int] | None = None, + available_account_exposure: float | None = None, + max_single_weight: float = _DEFAULT_MAX_SINGLE_WEIGHT, + max_positions: int = 20, + max_total_exposure: float = 1.0, + portfolio_snapshot: Any | None = None, + market_data: Mapping[str, Any] | None = None, +) -> StrategyDecision: + """Apply hard checks and call RiskEngine.assess exactly once.""" + try: + engine_action = build_risk_engine().assess( + decision, + portfolio_snapshot, + market_data=market_data, + ) + except Exception: + engine_action = None + engine_failed = True + else: + engine_failed = False + try: + return _apply_risk_gate_static( + decision, + risk_mandate_id=risk_mandate_id, + product_leverage_factors=product_leverage_factors, + available_account_exposure=available_account_exposure, + max_single_weight=max_single_weight, + max_positions=max_positions, + max_total_exposure=max_total_exposure, + engine_action=engine_action, + engine_failed=engine_failed, + ) + except Exception: + return _reject( + decision, + flag="rejected:invalid_risk_metadata", + reason="invalid_risk_metadata", + diagnostics={}, + ) + def _reject( decision: StrategyDecision, *, flag: str, reason: str, + diagnostics: Mapping[str, Any] | None = None, ) -> StrategyDecision: + safe_diagnostics: dict[str, Any] = {} + if diagnostics is not None: + try: + safe_diagnostics, valid = _safe_diagnostics(diagnostics) + except Exception: + valid = False + if not valid: + safe_diagnostics = {} + else: + try: + safe_diagnostics, valid = _safe_diagnostics(decision.diagnostics) + except Exception: + valid = False + if not valid: + safe_diagnostics = {} return StrategyDecision( positions=(), budgets=(), risk_flags=(flag,), diagnostics={ - **(decision.diagnostics or {}), + **safe_diagnostics, "risk_gate": "REJECT", "reason": reason, }, diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 4584d7e..55a70da 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -1,7 +1,11 @@ from __future__ import annotations import unittest -from datetime import datetime, timezone +from collections.abc import Iterator, Mapping +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from fractions import Fraction +import unicodedata from unittest.mock import Mock, patch from quant_platform_kit.common.models import PortfolioSnapshot @@ -20,6 +24,17 @@ from quant_platform_kit.strategy_contracts import BudgetIntent, PositionTarget, StrategyDecision +class ExplodingMapping(Mapping[str, object]): + def __getitem__(self, key: str) -> object: + raise RuntimeError("untrusted mapping") + + def __iter__(self) -> Iterator[str]: + raise RuntimeError("untrusted mapping") + + def __len__(self) -> int: + raise RuntimeError("untrusted mapping") + + def _decision( *, positions: tuple[PositionTarget, ...] = (), @@ -364,6 +379,83 @@ def test_risk_engine_exception_rejects_without_order_truth(self) -> None: engine.assess.assert_called_once() +class ApplyRiskMetadataSanitizationTests(unittest.TestCase): + def _apply(self, decision: StrategyDecision, **kwargs: object) -> StrategyDecision: + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + with patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine): + result = apply_risk_gate( + decision, + portfolio_snapshot={"total_equity": 100_000.0}, + **kwargs, + ) + engine.assess.assert_called_once_with( + decision, + {"total_equity": 100_000.0}, + market_data=None, + ) + return result + + def _assert_invalid(self, result: StrategyDecision) -> None: + self.assertEqual(result.positions, ()) + self.assertEqual(result.budgets, ()) + self.assertEqual(result.risk_flags, ("rejected:invalid_risk_metadata",)) + self.assertEqual(result.diagnostics.get("risk_gate"), "REJECT") + self.assertEqual(result.diagnostics.get("reason"), "invalid_risk_metadata") + + def test_malformed_diagnostics_still_assess_once_and_reject(self) -> None: + invalid_values = ( + True, + float("nan"), + float("inf"), + 10**10_000, + Decimal("0.1"), + Fraction(1, 10), + "0.1", + [], + {}, + object(), + ) + for field in ("unrealized_pnl_pct", "consecutive_losses"): + for value in invalid_values: + with self.subTest(field=field, value_type=type(value).__name__): + decision = StrategyDecision( + positions=(PositionTarget(symbol="SPY", target_weight=0.05),), + budgets=(BudgetIntent(name="risk", amount=0.0),), + diagnostics={field: value}, + ) + self._assert_invalid(self._apply(decision)) + + def test_malformed_limits_mandate_and_classification_reject(self) -> None: + decision = StrategyDecision( + positions=(PositionTarget(symbol="SPY", target_weight=0.05),), + budgets=(BudgetIntent(name="risk", amount=0.0),), + ) + cases = ( + {"max_single_weight": 10**10_000}, + {"max_single_weight": Decimal("0.1")}, + {"max_positions": True}, + {"max_positions": "20"}, + {"max_total_exposure": None}, + {"max_total_exposure": float("nan")}, + {"available_account_exposure": 10**10_000}, + {"risk_mandate_id": []}, + {"product_leverage_factors": ExplodingMapping()}, + ) + for kwargs in cases: + with self.subTest(kwargs=kwargs): + self._assert_invalid(self._apply(decision, **kwargs)) + + def test_malformed_position_material_rejects_after_one_assessment(self) -> None: + for value in (True, float("nan"), float("inf"), 10**10_000, Decimal("0.1"), object()): + with self.subTest(value_type=type(value).__name__): + decision = StrategyDecision( + positions=(PositionTarget(symbol="SPY", target_weight=value),), + budgets=(BudgetIntent(name="risk", amount=0.0),), + ) + self._assert_invalid(self._apply(decision)) + + class AssessWithEvidenceTests(unittest.TestCase): _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) @@ -934,6 +1026,355 @@ def test_mandate_rejects_budget_only_decision_above_authority(self) -> None: self.assertIn("budget_authority_exceeded", result.assessment.reason_codes) +class CanonicalRiskMetadataSanitizationTests(unittest.TestCase): + _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) + _SNAPSHOT = { + "as_of": "2026-08-04T04:27:55Z", + "observed_effective_exposure": 0.0, + "total_equity": 100_000.0, + } + + @staticmethod + def _candidate() -> CandidateRiskIdentity: + return CandidateRiskIdentity( + strategy_profile="canonical_risk_metadata", + account_mode="single_strategy_account_v1", + strategy_revision="1" * 40, + runner_revision="2" * 40, + config_sha256="3" * 64, + input_manifest_sha256="4" * 64, + authority_receipt_sha256="5" * 64, + ) + + @classmethod + def _mandate(cls, **overrides: object) -> dict[str, object]: + candidate = cls._candidate() + mandate: dict[str, object] = { + "mandate_id": "canonical_risk_metadata_v1", + "mandate_version": "v1", + "authority_receipt_sha256": candidate.authority_receipt_sha256, + "authority_scope": "RESEARCH_ONLY", + "strategy_profile": candidate.strategy_profile, + "account_mode": candidate.account_mode, + "strategy_revision": candidate.strategy_revision, + "runner_revision": candidate.runner_revision, + "config_sha256": candidate.config_sha256, + "input_manifest_sha256": candidate.input_manifest_sha256, + "candidate_identity_sha256": candidate.candidate_sha256, + "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.01, + "product_caps": {"XLK": 0.50}, + "nominal_caps": {"XLK": 0.50}, + "product_leverage_factors": {"XLK": 1}, + "allowed_nonzero_assets": ["XLK"], + "source_revision": "6" * 40, + } + mandate.update(overrides) + return mandate + + def _assess( + self, + decision: StrategyDecision, + *, + snapshot: object = _SNAPSHOT, + scope: object = "MEMBER", + mandate: object | None = None, + candidate: CandidateRiskIdentity | None = None, + origin: object | None = None, + risk_control_state: object | None = None, + engine_error: Exception | None = None, + ) -> tuple[object, Mock]: + engine = Mock() + if engine_error is None: + engine.assess.return_value = RiskAction(action="approve", reason="passed") + else: + engine.assess.side_effect = engine_error + 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, + snapshot, + scope=scope, + mandate_provenance=mandate, + market_data={}, + candidate_identity=candidate, + normalization_origin_weights=origin, + risk_control_state=risk_control_state, + ) + engine.assess.assert_called_once_with(decision, snapshot, market_data={}) + return result, engine + + def _assert_rejected(self, result: object) -> None: + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + + @staticmethod + def _noncanonical_values() -> tuple[object, ...]: + nfc = unicodedata.normalize("NFC", "e\u0301") + return ( + True, + float("nan"), + float("inf"), + Decimal("1"), + Fraction(1, 2), + [], + {}, + datetime(2026, 8, 4, tzinfo=timezone.utc), + object(), + "", + " drift ", + "\ud800", + unicodedata.normalize("NFD", nfc), + ) + + def test_digest_bound_strings_require_canonical_text(self) -> None: + for field in ("symbol", "role", "order_preference"): + for value in self._noncanonical_values(): + kwargs: dict[str, object] = { + "symbol": "XLK", + "target_weight": 0.05, + } + kwargs[field] = value + with self.subTest(kind="position", field=field, value_type=type(value).__name__): + result, _engine = self._assess( + StrategyDecision(positions=(PositionTarget(**kwargs),)), + ) + self._assert_rejected(result) + self.assertIn("invalid_risk_metadata", result.assessment.reason_codes) + + for field in ("name", "symbol", "unit", "purpose"): + for value in self._noncanonical_values(): + kwargs = {"name": "risk", "amount": 0.0, "unit": "quote_ccy"} + kwargs[field] = value + with self.subTest(kind="budget", field=field, value_type=type(value).__name__): + result, _engine = self._assess( + StrategyDecision(budgets=(BudgetIntent(**kwargs),)), + ) + self._assert_rejected(result) + self.assertIn("invalid_risk_metadata", result.assessment.reason_codes) + + def test_optional_string_nulls_and_nfc_text_remain_canonical(self) -> None: + nfc_role = unicodedata.normalize("NFC", "e\u0301") + decision = StrategyDecision( + positions=( + PositionTarget( + symbol="XLK", + target_weight=0.05, + role=nfc_role, + order_preference=None, + ), + ), + budgets=( + BudgetIntent( + name="risk", + symbol=None, + amount=0.0, + unit="quote_ccy", + purpose="研究预算", + ), + ), + ) + first, first_engine = self._assess(decision) + second, second_engine = self._assess(decision) + + self.assertEqual(first.assessment.outcome, "APPROVE") + self.assertEqual(first.assessment.assessment_sha256, second.assessment.assessment_sha256) + first_engine.assess.assert_called_once() + second_engine.assess.assert_called_once() + + def test_required_null_material_rejects(self) -> None: + cases = ( + StrategyDecision( + positions=(PositionTarget(symbol=None, target_weight=0.05),), + ), + StrategyDecision( + positions=(PositionTarget(symbol="XLK"),), + ), + StrategyDecision( + budgets=(BudgetIntent(name=None, amount=0.0),), + ), + StrategyDecision( + budgets=(BudgetIntent(name="risk", amount=0.0, unit=None),), + ), + StrategyDecision( + budgets=(BudgetIntent(name="risk", amount=None),), + ), + ) + for decision in cases: + with self.subTest(decision=decision): + result, _engine = self._assess(decision) + self._assert_rejected(result) + + for field in ("mandate_id", "effective_exposure_cap", "product_caps"): + with self.subTest(mandate_field=field): + result, _engine = self._assess( + StrategyDecision(), + mandate=self._mandate(**{field: None}), + candidate=self._candidate(), + ) + self._assert_rejected(result) + + def test_noncanonical_candidate_identity_is_rejected_at_gate(self) -> None: + canonical = self._candidate() + nfd_profile = unicodedata.normalize("NFD", "é") + for profile in (nfd_profile, "\ud800"): + with self.subTest(profile=ascii(profile)): + candidate = CandidateRiskIdentity( + strategy_profile=profile, + account_mode=canonical.account_mode, + strategy_revision=canonical.strategy_revision, + runner_revision=canonical.runner_revision, + config_sha256=canonical.config_sha256, + input_manifest_sha256=canonical.input_manifest_sha256, + authority_receipt_sha256=canonical.authority_receipt_sha256, + ) + result, _engine = self._assess( + StrategyDecision(), + mandate=self._mandate( + strategy_profile=profile, + candidate_identity_sha256=candidate.candidate_sha256, + ), + candidate=candidate, + ) + self._assert_rejected(result) + + def test_numeric_types_are_exact_finite_and_bounded(self) -> None: + invalid_values = ( + True, + float("nan"), + float("inf"), + 10**10_000, + Decimal("0.1"), + Fraction(1, 10), + "0.1", + [], + {}, + object(), + ) + for value in invalid_values: + cases = ( + StrategyDecision( + positions=(PositionTarget(symbol="XLK", target_weight=value),), + ), + StrategyDecision( + budgets=(BudgetIntent(name="risk", amount=value),), + ), + ) + for decision in cases: + with self.subTest(value_type=type(value).__name__, decision=decision): + result, _engine = self._assess(decision) + self._assert_rejected(result) + + for field in ("observed_effective_exposure", "total_equity"): + with self.subTest(value_type=type(value).__name__, snapshot_field=field): + result, _engine = self._assess( + StrategyDecision(), + snapshot={**self._SNAPSHOT, field: value}, + ) + self._assert_rejected(result) + + def test_malformed_mandate_material_is_redacted_fail_closed(self) -> None: + invalid_values = ( + True, + float("nan"), + float("inf"), + 10**10_000, + Decimal("0.1"), + Fraction(1, 10), + "0.1", + object(), + ) + for field in ( + "max_snapshot_age_seconds", + "effective_exposure_cap", + "loss_budget", + ): + for value in invalid_values: + with self.subTest(field=field, value_type=type(value).__name__): + result, _engine = self._assess( + StrategyDecision(), + mandate=self._mandate(**{field: value}), + candidate=self._candidate(), + ) + self._assert_rejected(result) + + deep_caps: object = 0.5 + for index in range(1_100): + deep_caps = {f"level_{index}": deep_caps} + for caps in (ExplodingMapping(), deep_caps): + with self.subTest(caps_type=type(caps).__name__): + result, _engine = self._assess( + StrategyDecision(), + mandate=self._mandate(product_caps=caps), + candidate=self._candidate(), + ) + self._assert_rejected(result) + + def test_scope_mapping_and_normalization_errors_assess_once(self) -> None: + for scope in ([], {}, object()): + with self.subTest(scope_type=type(scope).__name__): + result, _engine = self._assess(StrategyDecision(), scope=scope) + self._assert_rejected(result) + + for origin in ( + {"XLK": 10**10_000}, + {"XLK": Decimal("0.2")}, + {"XLK": True}, + {1: 0.2}, + ExplodingMapping(), + ): + with self.subTest(origin_type=type(origin).__name__): + result, _engine = self._assess( + StrategyDecision(), + snapshot={**self._SNAPSHOT, "observed_effective_exposure": 0.2}, + origin=origin, + ) + self._assert_rejected(result) + + def test_timezone_and_whole_second_timestamp_compatibility(self) -> None: + utc = datetime(2026, 8, 4, 4, 27, 55, tzinfo=timezone.utc) + offset = utc.astimezone(timezone(timedelta(hours=8))) + first, _engine = self._assess( + StrategyDecision(), + snapshot={**self._SNAPSHOT, "as_of": utc.replace(microsecond=1)}, + ) + second, _engine = self._assess( + StrategyDecision(), + snapshot={**self._SNAPSHOT, "as_of": offset.replace(microsecond=999_999)}, + ) + rejected, _engine = self._assess( + StrategyDecision(), + snapshot={**self._SNAPSHOT, "as_of": "2026-08-04T12:27:55+08:00"}, + ) + + self.assertEqual(first.assessment.outcome, "APPROVE") + self.assertEqual( + first.assessment.portfolio_snapshot_digest_sha256, + second.assessment.portfolio_snapshot_digest_sha256, + ) + self._assert_rejected(rejected) + + def test_engine_exception_and_invalid_static_material_remain_redacted(self) -> None: + result, engine = self._assess( + StrategyDecision( + positions=(PositionTarget(symbol="XLK", target_weight=10**10_000),), + ), + engine_error=RuntimeError("private engine material"), + ) + + self._assert_rejected(result) + self.assertIn("invalid_risk_metadata", result.assessment.reason_codes) + self.assertNotIn("private engine material", repr(result)) + engine.assess.assert_called_once() + + class TqqqEtfOnlyResearchMandateTests(unittest.TestCase): _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) _MANDATE_ID = "tqqq_etf_only_research_v1" @@ -1260,43 +1701,63 @@ def test_over_cap_normalization_must_reduce_to_cash_and_binds_origin(self) -> No cash_engine.assess.assert_called_once() partial_engine.assess.assert_called_once() + def test_malformed_digest_bound_control_material_never_throws(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.15),), + ) + nfc = unicodedata.normalize("NFC", "e\u0301") + invalid_mandate_ids = ( + float("nan"), + float("inf"), + Decimal("1"), + Fraction(1, 2), + True, + [], + {}, + object(), + "", + " other ", + "\ud800", + unicodedata.normalize("NFD", nfc), + ) + for mandate_id in invalid_mandate_ids: + with self.subTest(field="mandate_id", value_type=type(mandate_id).__name__): + result, engine = self._assess( + decision, + risk_state=self._risk_state(mandate_id=mandate_id), + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once() -class GlobalEtfRotationResearchMandateTests(unittest.TestCase): + malformed_states = ( + self._risk_state(consecutive_completed_losing_exits=10**10_000), + self._risk_state(consecutive_completed_losing_exits=Decimal("1")), + self._risk_state(account_drawdown_fraction=10**10_000), + self._risk_state(account_drawdown_fraction=Fraction(1, 20)), + ExplodingMapping(), + ) + for risk_state in malformed_states: + with self.subTest(risk_state_type=type(risk_state).__name__): + result, engine = self._assess(decision, risk_state=risk_state) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once() + + +class RetiredGlobalEtfRotationMandateTests(unittest.TestCase): _NOW = datetime(2026, 8, 9, 2, 0, tzinfo=timezone.utc) _MANDATE_ID = "global_etf_rotation_etf_only_research_v1" - _STRATEGY_PROFILE = "global_etf_rotation_etf_only_single_strategy_research_v1" - _ACCOUNT_MODE = "single_strategy_research_v1" - _ALLOWED_ASSETS = ( - "EWY", - "EWT", - "INDA", - "FXI", - "EWJ", - "VGK", - "VOO", - "XLK", - "SMH", - "GLD", - "SLV", - "USO", - "DBA", - "XLE", - "XLF", - "ITA", - "XLP", - "XLU", - "XLV", - "IHI", - "VNQ", - "KRE", - "BIL", - ) @classmethod def _candidate(cls) -> CandidateRiskIdentity: return CandidateRiskIdentity( - strategy_profile=cls._STRATEGY_PROFILE, - account_mode=cls._ACCOUNT_MODE, + strategy_profile="retired_global_etf_candidate", + account_mode="single_strategy_research_v1", strategy_revision="1" * 40, runner_revision="2" * 40, config_sha256="3" * 64, @@ -1305,10 +1766,9 @@ def _candidate(cls) -> CandidateRiskIdentity: ) @classmethod - def _mandate(cls, **overrides: object) -> dict[str, object]: + def _otherwise_valid_generic_mandate(cls) -> dict[str, object]: candidate = cls._candidate() - caps = {symbol: 0.50 for symbol in cls._ALLOWED_ASSETS} - mandate: dict[str, object] = { + return { "mandate_id": cls._MANDATE_ID, "mandate_version": "v1", "authority_receipt_sha256": candidate.authority_receipt_sha256, @@ -1325,96 +1785,36 @@ def _mandate(cls, **overrides: object) -> dict[str, object]: "max_snapshot_age_seconds": 300, "effective_exposure_cap": 0.50, "loss_budget": 0.01, - "loss_budget_equity_reference": "completed_session_equity", - "product_caps": caps, - "nominal_caps": caps, - "product_effective_caps": caps, - "product_leverage_factors": { - symbol: 1 for symbol in cls._ALLOWED_ASSETS - }, - "allowed_nonzero_assets": list(cls._ALLOWED_ASSETS), - "max_nonzero_assets": 2, - "broker_margin_factor": 1, - "margin_stacking": False, - "borrowing": False, - "shorting": False, - "income_sleeve_enabled": False, - "option_overlay_enabled": False, - "ai_overlay_enabled": False, - "market_regime_overlay_enabled": False, - "precommitted_executable_stop_distance": 0.05, - "stop_fill_policy": "gap_aware_min_open_or_stop_v1", - "max_consecutive_completed_losing_exits": 5, + "product_caps": {"XLK": 0.50}, + "nominal_caps": {"XLK": 0.50}, + "product_leverage_factors": {"XLK": 1}, + "allowed_nonzero_assets": ["XLK"], "source_revision": "6" * 40, } - mandate.update(overrides) - return mandate - - @staticmethod - def _snapshot(**overrides: object) -> dict[str, object]: - snapshot: dict[str, object] = { - "as_of": "2026-08-09T01:59:55Z", - "observed_effective_exposure": 0.0, - "total_equity": 100_000.0, - } - snapshot.update(overrides) - return snapshot - - @classmethod - def _risk_state( - cls, - *symbols: str, - position_stop_states: dict[str, object] | None = None, - **overrides: object, - ) -> dict[str, object]: - stops = { - symbol: { - "stop_intent_ready": True, - "entry_fill_identity_sha256": str(index + 1) * 64, - "stop_entry_fill_identity_sha256": str(index + 1) * 64, - } - for index, symbol in enumerate(symbols) - } - state: dict[str, object] = { - "as_of": "2026-08-09T01:59:55Z", - "mandate_id": cls._MANDATE_ID, - "candidate_identity_sha256": cls._candidate().candidate_sha256, - "stop_loss_distance": 0.05, - "stop_fill_policy": "gap_aware_min_open_or_stop_v1", - "position_stop_states": ( - stops if position_stop_states is None else position_stop_states - ), - "consecutive_completed_losing_exits": 0, - "account_drawdown_fraction": 0.05, - "drawdown_scalar": 1.0, - } - state.update(overrides) - return state def _assess( self, decision: StrategyDecision, *, - mandate: dict[str, object] | None = None, - risk_state: dict[str, object] | None = None, - snapshot: dict[str, object] | None = None, - engine_action: str = "approve", + mandate: object | None = None, + snapshot: object | None = None, + risk_control_state: object | None = None, engine_error: Exception | None = None, - ) -> tuple[object, Mock]: - active_symbols = tuple( - position.symbol - for position in decision.positions - if (position.target_weight or 0.0) > 0.0 - ) + ) -> object: engine = Mock() if engine_error is not None: engine.assess.side_effect = engine_error else: - engine.assess.return_value = RiskAction( - action=engine_action, - reason="test", - ) - actual_snapshot = snapshot if snapshot is not None else self._snapshot() + engine.assess.return_value = RiskAction(action="approve", reason="test") + actual_snapshot = ( + snapshot + if snapshot is not None + else { + "as_of": "2026-08-09T01:59:55Z", + "observed_effective_exposure": 0.0, + "total_equity": 100_000.0, + } + ) 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), @@ -1424,284 +1824,104 @@ def _assess( actual_snapshot, scope="MEMBER", mandate_provenance=( - mandate if mandate is not None else self._mandate() + mandate + if mandate is not None + else self._otherwise_valid_generic_mandate() ), market_data={}, candidate_identity=self._candidate(), - risk_control_state=( - risk_state - if risk_state is not None - else self._risk_state(*active_symbols) - ), + risk_control_state=risk_control_state, ) engine.assess.assert_called_once_with( decision, actual_snapshot, market_data={}, ) - return result, engine + return result - @staticmethod - def _two_position_decision( - first: float = 0.15, - second: float = 0.05, - ) -> StrategyDecision: - return _decision( - positions=( - PositionTarget(symbol="XLK", target_weight=first), - PositionTarget(symbol="BIL", target_weight=second), - ) + def _assert_terminal_rejection(self, result: object) -> None: + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn( + "retired_global_etf_research_mandate", + result.assessment.reason_codes, ) - - def test_valid_research_decision_approves_but_never_authorizes_execution( - self, - ) -> None: - decision = self._two_position_decision() - - result, _engine = self._assess(decision) - - self.assertEqual(result.assessment.outcome, "APPROVE") - self.assertEqual(result.assessment.mandate_id, self._MANDATE_ID) - self.assertEqual(result.assessment.proposed_effective_exposure, 0.20) - self.assertEqual(result.assessment.stop_loss_distance, 0.05) - self.assertTrue(result.assessment.stop_intent_ready) - self.assertFalse(result.assessment.strategy_breaker_triggered) - self.assertFalse(result.assessment.account_breaker_triggered) - self.assertEqual(result.assessment.drawdown_scalar, 1.0) - self.assertEqual(len(result.assessment.risk_control_state_digest_sha256), 64) self.assertFalse(result.assessment.execution_authorized) - self.assertEqual(result.decision.positions, decision.positions) + self.assertIsNone(result.assessment.stop_loss_distance) + self.assertIsNone(result.assessment.risk_control_state_digest_sha256) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) - def test_exact_mandate_shape_is_fail_closed(self) -> None: - decision = self._two_position_decision() - caps = {symbol: 0.50 for symbol in self._ALLOWED_ASSETS} - invalid_cases = ( - {"authority_scope": "PAPER"}, - {"strategy_profile": "global_etf_rotation"}, - {"account_mode": "single_strategy_account_v1"}, - {"effective_exposure_cap": 0.51}, - {"loss_budget": 0.011}, - {"loss_budget_equity_reference": "current_equity"}, - {"product_caps": {**caps, "XLK": 0.51}}, - {"product_leverage_factors": {"XLK": 1, "BIL": 1}}, - {"allowed_nonzero_assets": [*self._ALLOWED_ASSETS, "SPY"]}, - {"max_nonzero_assets": 3}, - {"broker_margin_factor": 2}, - {"margin_stacking": True}, - {"borrowing": True}, - {"shorting": True}, - {"income_sleeve_enabled": True}, - {"option_overlay_enabled": True}, - {"ai_overlay_enabled": True}, - {"market_regime_overlay_enabled": True}, - {"precommitted_executable_stop_distance": 0.06}, - {"stop_fill_policy": "stop_price_only"}, - {"max_consecutive_completed_losing_exits": 6}, - {"expires_at": "2027-08-09T01:59:55Z"}, + def test_otherwise_valid_generic_payload_is_explicitly_retired(self) -> None: + decision = StrategyDecision( + positions=(PositionTarget(symbol="XLK", target_weight=0.10),), + budgets=(BudgetIntent(name="risk_budget", amount=0.005),), ) - for overrides in invalid_cases: - with self.subTest(overrides=overrides): - result, _engine = self._assess( - decision, - mandate=self._mandate(**overrides), - ) - self.assertEqual(result.assessment.outcome, "REJECT") - self.assertIn( - "invalid_global_etf_research_mandate", - result.assessment.reason_codes, - ) - self.assertEqual(result.decision.positions, ()) - def test_position_count_assets_caps_and_aggregate_risk_budget_fail_closed( - self, - ) -> None: + self._assert_terminal_rejection(self._assess(decision)) + + def test_retired_id_is_fail_closed_for_malformed_material(self) -> None: + decision = StrategyDecision( + positions=(PositionTarget(symbol="XLK", target_weight=0.10),), + budgets=(BudgetIntent(name="risk_budget", amount=0.005),), + ) cases = ( - ( - _decision( + {"mandate": {"mandate_id": self._MANDATE_ID}}, + {"snapshot": {"total_equity": 10**400}}, + { + "snapshot": { + "as_of": "2026-08-09T01:59:55Z", + "observed_effective_exposure": float("nan"), + "total_equity": float("inf"), + } + }, + { + "decision": StrategyDecision( + positions=(PositionTarget(symbol="XLK", target_weight=10**400),), + budgets=(BudgetIntent(name="risk_budget", amount=10**400),), + ) + }, + { + "decision": StrategyDecision( positions=( - PositionTarget(symbol="XLK", target_weight=0.05), - PositionTarget(symbol="BIL", target_weight=0.05), - PositionTarget(symbol="GLD", target_weight=0.05), - ) - ), - None, - "single_strategy_position_count", - ), - ( - _decision( - positions=(PositionTarget(symbol="SPY", target_weight=0.10),) - ), - None, - "asset_not_authorized", - ), - ( - _decision( - positions=(PositionTarget(symbol="XLK", target_weight=0.501),) - ), - None, - "product_exposure_cap", - ), - ( - self._two_position_decision(first=0.151, second=0.05), - None, - "risk_budget_exposure_cap", - ), - ( - self._two_position_decision(first=0.06, second=0.05), - self._risk_state( - "XLK", - "BIL", - account_drawdown_fraction=0.050001, - drawdown_scalar=0.50, - ), - "risk_budget_exposure_cap", - ), - ) - for decision, state, reason in cases: - with self.subTest(reason=reason): - result, _engine = self._assess(decision, risk_state=state) - self.assertEqual(result.assessment.outcome, "REJECT") - self.assertIn(reason, result.assessment.reason_codes) - self.assertEqual(result.decision.positions, ()) - - def test_per_position_gap_aware_stop_state_is_fail_closed(self) -> None: - decision = self._two_position_decision() - valid_stops = self._risk_state("XLK", "BIL")["position_stop_states"] - assert isinstance(valid_stops, dict) - mismatched_fill = { - **valid_stops, - "XLK": { - **valid_stops["XLK"], - "stop_entry_fill_identity_sha256": "9" * 64, + PositionTarget( + symbol="XLK", + target_weight=0.10, + role=float("nan"), + ), + ), + budgets=(BudgetIntent(name="risk_budget", amount=0.005),), + ) }, - } - not_ready = { - **valid_stops, - "BIL": {**valid_stops["BIL"], "stop_intent_ready": False}, - } - invalid_cases = ( - {}, - self._risk_state("XLK", "BIL", as_of="2026-08-09T01:49:55Z"), - self._risk_state("XLK", "BIL", as_of="2026-08-09T02:00:01Z"), - self._risk_state( - "XLK", - "BIL", - account_drawdown_fraction=float("nan"), - ), - self._risk_state("XLK", "BIL", candidate_identity_sha256="0" * 64), - self._risk_state("XLK", "BIL", mandate_id="other"), - self._risk_state("XLK", "BIL", stop_loss_distance=0.06), - self._risk_state("XLK", "BIL", stop_fill_policy="stop_price_only"), - self._risk_state( - "XLK", - position_stop_states={"XLK": valid_stops["XLK"]}, - ), - self._risk_state( - "XLK", - "BIL", - position_stop_states=mismatched_fill, - ), - self._risk_state("XLK", "BIL", position_stop_states=not_ready), - self._risk_state("XLK", "BIL", drawdown_scalar=0.50), - ) - for state in invalid_cases: - with self.subTest(state=state): - result, _engine = self._assess(decision, risk_state=state) - self.assertEqual(result.assessment.outcome, "REJECT") - self.assertEqual(result.decision.positions, ()) - - def test_drawdown_and_strategy_breaker_boundaries(self) -> None: - approved_cases = ( - ( - self._two_position_decision(), - self._risk_state("XLK", "BIL"), - ), - ( - self._two_position_decision(first=0.075, second=0.025), - self._risk_state( - "XLK", - "BIL", - account_drawdown_fraction=0.050001, - drawdown_scalar=0.50, - ), - ), - ( - self._two_position_decision(first=0.075, second=0.025), - self._risk_state( - "XLK", - "BIL", - account_drawdown_fraction=0.10, - drawdown_scalar=0.50, - ), - ), - ( - self._two_position_decision(), - self._risk_state( - "XLK", - "BIL", - consecutive_completed_losing_exits=4, - ), - ), - ) - for decision, state in approved_cases: - with self.subTest(state=state): - result, _engine = self._assess(decision, risk_state=state) - self.assertEqual(result.assessment.outcome, "APPROVE") - - breaker_cases = ( - ( - self._risk_state( - "XLK", - "BIL", - consecutive_completed_losing_exits=5, - ), - "strategy_breaker_triggered", - ), - ( - self._risk_state( - "XLK", - "BIL", - account_drawdown_fraction=0.100001, - drawdown_scalar=0.0, - ), - "account_breaker_triggered", - ), - ) - for state, reason in breaker_cases: - with self.subTest(reason=reason): - result, _engine = self._assess( - self._two_position_decision(), - risk_state=state, + { + "decision": StrategyDecision( + positions=(PositionTarget(symbol="XLK", target_weight=0.10),), + budgets=( + BudgetIntent( + name="risk_budget", + amount=0.005, + unit=float("nan"), + ), + ), + ) + }, + {"risk_control_state": {"material": float("nan")}}, + ) + for case in cases: + with self.subTest(case=case): + actual_decision = case.get("decision", decision) + kwargs = {key: value for key, value in case.items() if key != "decision"} + self._assert_terminal_rejection( + self._assess(actual_decision, **kwargs), ) - self.assertEqual(result.assessment.outcome, "REJECT") - self.assertIn(reason, result.assessment.reason_codes) - self.assertEqual(result.decision.positions, ()) - def test_engine_is_exactly_once_for_static_reject_error_and_nonapprove( - self, - ) -> None: - decision = self._two_position_decision() - static_reject, _static_engine = self._assess( - decision, - mandate=self._mandate(authority_scope="PAPER"), + def test_retired_static_reject_still_assesses_risk_exactly_once(self) -> None: + result = self._assess( + _decision(positions=(PositionTarget(symbol="XLK", target_weight=0.10),)), engine_error=RuntimeError("redacted"), ) - engine_error, _error_engine = self._assess( - decision, - engine_error=RuntimeError("redacted"), - ) - nonapprove, _nonapprove_engine = self._assess( - decision, - engine_action="reject", - ) - self.assertEqual(static_reject.assessment.outcome, "REJECT") - self.assertNotIn("risk_engine_error", static_reject.assessment.reason_codes) - self.assertIn("risk_engine_error", engine_error.assessment.reason_codes) - self.assertIn("risk_engine_non_approve", nonapprove.assessment.reason_codes) - self.assertFalse(static_reject.assessment.execution_authorized) - self.assertFalse(engine_error.assessment.execution_authorized) - self.assertFalse(nonapprove.assessment.execution_authorized) + self._assert_terminal_rejection(result) + self.assertNotIn("risk_engine_error", result.assessment.reason_codes) class BootstrapSmallAccountV2RiskGateTests(unittest.TestCase): From 87b86cde2b407cf53d9dac5c1a64666ae7f571b1 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:29:49 +0800 Subject: [PATCH 2/2] Address risk metadata review findings Co-Authored-By: Codex --- src/quant_platform_kit/risk/gate.py | 30 ++++++++--- tests/test_risk_gate.py | 81 ++++++++++++++++++++++++++++- 2 files changed, 102 insertions(+), 9 deletions(-) diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index 2eba0ec..556eb9c 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -130,7 +130,9 @@ def _canonical_numeric_mapping( if not isinstance(value, Mapping) or len(value) > _MAX_MATERIAL_ITEMS: return None result: dict[str, int | float] = {} - for raw_key, raw_value in value.items(): + for item_count, (raw_key, raw_value) in enumerate(value.items(), start=1): + if item_count > _MAX_MATERIAL_ITEMS: + return None key, valid_key = _canonical_string(raw_key) if not valid_key or key is None: return None @@ -190,7 +192,9 @@ def _safe_diagnostics(value: Any) -> tuple[dict[str, Any], bool]: if not isinstance(value, Mapping) or len(value) > _MAX_MATERIAL_ITEMS: return {}, False result: dict[str, Any] = {} - for raw_key, raw_value in value.items(): + for item_count, (raw_key, raw_value) in enumerate(value.items(), start=1): + if item_count > _MAX_MATERIAL_ITEMS: + return {}, False key, valid = _canonical_string(raw_key) if not valid or key is None: return {}, False @@ -242,7 +246,7 @@ def _decision_metrics( active: list[tuple[str, float]] = [] reason_codes: set[str] = set() position_payloads: list[dict[str, Any]] = [] - raw_positions = decision.positions or () + raw_positions = decision.positions if type(raw_positions) is not tuple or len(raw_positions) > _MAX_MATERIAL_ITEMS: raw_positions = () reason_codes.add("invalid_risk_metadata") @@ -298,7 +302,7 @@ def _decision_metrics( if normalized_weight > 0.0: active.append((symbol, normalized_weight)) budget_payloads: list[dict[str, Any]] = [] - raw_budgets = decision.budgets or () + raw_budgets = decision.budgets if type(raw_budgets) is not tuple or len(raw_budgets) > _MAX_MATERIAL_ITEMS: raw_budgets = () reason_codes.add("invalid_risk_metadata") @@ -829,6 +833,8 @@ def _risk_control_fields( errors.add("invalid_account_drawdown") if losses is None: errors.add("invalid_strategy_breaker_state") + if entry_fill_identity is None or stop_entry_fill_identity is None: + errors.add("invalid_stop_identity") expected_scalar: float | None = None if account_drawdown is not None and 0.0 <= account_drawdown <= 1.0: @@ -925,7 +931,11 @@ def _assess_with_evidence_static( ) reason_codes.update(decision_errors) diagnostics, valid_diagnostics = _safe_diagnostics(decision.diagnostics) - risk_flags = _canonical_string_list(decision.risk_flags or ()) + risk_flags = ( + _canonical_string_list(decision.risk_flags) + if type(decision.risk_flags) is tuple + else None + ) if not valid_diagnostics or risk_flags is None: reason_codes.add("invalid_risk_metadata") if not valid_scope or normalized_scope not in _ALLOWED_SCOPES: @@ -1294,9 +1304,13 @@ def _apply_risk_gate_static( "invalid_risk_metadata", ) - normalized_risk_flags = _canonical_string_list(decision.risk_flags or ()) - raw_positions = decision.positions or () - raw_budgets = decision.budgets or () + normalized_risk_flags = ( + _canonical_string_list(decision.risk_flags) + if type(decision.risk_flags) is tuple + else None + ) + raw_positions = decision.positions + raw_budgets = decision.budgets if ( normalized_risk_flags is None or type(raw_positions) is not tuple diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 55a70da..c19b322 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -35,6 +35,20 @@ def __len__(self) -> int: raise RuntimeError("untrusted mapping") +class UnderreportedMapping(Mapping[str, object]): + def __init__(self, values: dict[str, object]) -> None: + self._values = values + + def __getitem__(self, key: str) -> object: + return self._values[key] + + def __iter__(self) -> Iterator[str]: + return iter(self._values) + + def __len__(self) -> int: + return 0 + + def _decision( *, positions: tuple[PositionTarget, ...] = (), @@ -455,6 +469,23 @@ def test_malformed_position_material_rejects_after_one_assessment(self) -> None: ) self._assert_invalid(self._apply(decision)) + def test_falsey_mutable_decision_containers_reject(self) -> None: + for field in ("positions", "budgets", "risk_flags"): + values: dict[str, object] = { + "positions": (), + "budgets": (), + "risk_flags": (), + } + values[field] = [] + with self.subTest(field=field): + self._assert_invalid(self._apply(StrategyDecision(**values))) + + def test_underreported_diagnostics_mapping_rejects(self) -> None: + diagnostics = UnderreportedMapping( + {f"diagnostic_{index}": 0 for index in range(1_001)} + ) + self._assert_invalid(self._apply(StrategyDecision(diagnostics=diagnostics))) + class AssessWithEvidenceTests(unittest.TestCase): _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) @@ -1212,7 +1243,12 @@ def test_required_null_material_rejects(self) -> None: result, _engine = self._assess(decision) self._assert_rejected(result) - for field in ("mandate_id", "effective_exposure_cap", "product_caps"): + for field in ( + "mandate_id", + "effective_exposure_cap", + "product_caps", + "allowed_nonzero_assets", + ): with self.subTest(mandate_field=field): result, _engine = self._assess( StrategyDecision(), @@ -1221,6 +1257,29 @@ def test_required_null_material_rejects(self) -> None: ) self._assert_rejected(result) + def test_falsey_mutable_decision_containers_reject(self) -> None: + for field in ("positions", "budgets", "risk_flags"): + values: dict[str, object] = { + "positions": (), + "budgets": (), + "risk_flags": (), + } + values[field] = [] + with self.subTest(field=field): + result, _engine = self._assess(StrategyDecision(**values)) + self._assert_rejected(result) + + def test_underreported_factor_mapping_rejects(self) -> None: + factors = UnderreportedMapping( + {"XLK": 1, **{f"ASSET_{index}": 1 for index in range(1_000)}} + ) + result, _engine = self._assess( + StrategyDecision(), + mandate=self._mandate(product_leverage_factors=factors), + candidate=self._candidate(), + ) + self._assert_rejected(result) + def test_noncanonical_candidate_identity_is_rejected_at_gate(self) -> None: canonical = self._candidate() nfd_profile = unicodedata.normalize("NFD", "é") @@ -1748,6 +1807,26 @@ def test_malformed_digest_bound_control_material_never_throws(self) -> None: self.assertEqual(result.decision.budgets, ()) engine.assess.assert_called_once() + def test_inactive_tqqq_stop_identities_are_still_required(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="BOXX", target_weight=0.10),), + ) + for field in ( + "tqqq_entry_fill_identity_sha256", + "stop_entry_fill_identity_sha256", + ): + for value in (None, object()): + with self.subTest(field=field, value_type=type(value).__name__): + result, engine = self._assess( + decision, + risk_state=self._risk_state(**{field: value}), + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once() + class RetiredGlobalEtfRotationMandateTests(unittest.TestCase): _NOW = datetime(2026, 8, 9, 2, 0, tzinfo=timezone.utc)