From 04023359004719374be6ebe833904c66b260c7d6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 9 Aug 2026 02:31:50 +0800 Subject: [PATCH] feat(risk): add global ETF research mandate contract Co-Authored-By: Codex --- src/quant_platform_kit/risk/gate.py | 334 ++++++++++++++++++++- tests/test_risk_gate.py | 443 ++++++++++++++++++++++++++++ 2 files changed, 768 insertions(+), 9 deletions(-) diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index d60c0ba..08901ed 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -34,6 +34,39 @@ _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", +) +_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" @@ -297,6 +330,97 @@ 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, *, @@ -392,11 +516,18 @@ 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 = _exact_tqqq_mandate_errors( - mandate_provenance, - effective_at=effective_at, - expires_at=expires_at, - ) + 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, + ) + ) if exact_mandate_errors: return {}, exact_mandate_errors factors = mandate_provenance.get("product_leverage_factors", {}) @@ -553,6 +684,13 @@ 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): @@ -676,6 +814,165 @@ 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, @@ -733,8 +1030,13 @@ 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 ( - mandate.get("mandate_id") == _TQQQ_ETF_ONLY_RESEARCH_MANDATE + exact_research_mandate and len(active_positions) > mandate["max_nonzero_assets"] ): reason_codes.add("single_strategy_position_count") @@ -777,6 +1079,22 @@ 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 @@ -788,9 +1106,7 @@ def assess_with_evidence( product_leverage_factors=factors, effective_exposure_cap=cap, observed_effective_exposure=observed, - cash_only=( - mandate.get("mandate_id") == _TQQQ_ETF_ONLY_RESEARCH_MANDATE - ), + cash_only=exact_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 de6cecf..4584d7e 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -1261,6 +1261,449 @@ def test_over_cap_normalization_must_reduce_to_cash_and_binds_origin(self) -> No partial_engine.assess.assert_called_once() +class GlobalEtfRotationResearchMandateTests(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_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() + caps = {symbol: 0.50 for symbol in cls._ALLOWED_ASSETS} + mandate: dict[str, object] = { + "mandate_id": cls._MANDATE_ID, + "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-09T01:59:55Z", + "expires_at": "2026-09-08T01:59:55Z", + "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, + "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", + 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 + ) + 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() + 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, + actual_snapshot, + scope="MEMBER", + mandate_provenance=( + mandate if mandate is not None else self._mandate() + ), + market_data={}, + candidate_identity=self._candidate(), + risk_control_state=( + risk_state + if risk_state is not None + else self._risk_state(*active_symbols) + ), + ) + engine.assess.assert_called_once_with( + decision, + actual_snapshot, + market_data={}, + ) + return result, engine + + @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 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, ()) + + 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), + ) + 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, + ) + 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, + 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) + + class BootstrapSmallAccountV2RiskGateTests(unittest.TestCase): _MANDATE = "bootstrap_small_account_v2"