diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index 08901ed..d5ab882 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -34,39 +34,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" +_RETIRED_GLOBAL_ETF_RESEARCH_MANDATE = ( + "global_etf_rotation_etf_only_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", -) -_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" @@ -98,7 +68,10 @@ 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)): return None - number = float(value) + try: + number = float(value) + except (OverflowError, TypeError, ValueError): + return None return number if math.isfinite(number) else None @@ -330,97 +303,6 @@ def _exact_tqqq_mandate_errors( 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 +323,8 @@ 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", @@ -516,18 +400,11 @@ def _mandate_fields( 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", {}) @@ -684,13 +561,6 @@ 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): @@ -814,165 +684,6 @@ def _risk_control_fields( }, 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, - "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 assess_with_evidence( decision: StrategyDecision, portfolio_snapshot: Any, @@ -1030,13 +741,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") @@ -1079,22 +785,6 @@ def assess_with_evidence( ): 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") target_weights: dict[str, float] = {} for symbol, weight in active_positions: target_weights[symbol] = target_weights.get(symbol, 0.0) + weight @@ -1106,7 +796,9 @@ def assess_with_evidence( product_leverage_factors=factors, effective_exposure_cap=cap, observed_effective_exposure=observed, - cash_only=exact_research_mandate, + cash_only=( + mandate.get("mandate_id") == _TQQQ_ETF_ONLY_RESEARCH_MANDATE + ), ) if not valid_normalization: reason_codes.add("invalid_reduce_only_normalization") diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 4584d7e..96aa306 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -1261,42 +1261,15 @@ def test_over_cap_normalization_must_reduce_to_cash_and_binds_origin(self) -> No partial_engine.assess.assert_called_once() -class GlobalEtfRotationResearchMandateTests(unittest.TestCase): +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 +1278,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 +1297,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 +1336,80 @@ 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) - - 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"}, - ) - 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: - cases = ( - ( - _decision( - 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, ()) + 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_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, - }, - } - 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), + 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 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") + self._assert_terminal_rejection(self._assess(decision)) - 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", - ), + 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),), ) - for state, reason in breaker_cases: - with self.subTest(reason=reason): - result, _engine = self._assess( - self._two_position_decision(), - risk_state=state, + cases = ( + {"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),), + ) + }, + {"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"), - engine_error=RuntimeError("redacted"), - ) - engine_error, _error_engine = self._assess( - decision, + 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"), ) - 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):