From d6876783220565a80dd17c8a5edb42ba1bc7014d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 9 Aug 2026 00:57:39 +0800 Subject: [PATCH] feat(risk): freeze TQQQ research mandate contract Co-Authored-By: Codex --- src/quant_platform_kit/position_sizing.py | 35 ++- src/quant_platform_kit/risk/contracts.py | 16 ++ src/quant_platform_kit/risk/gate.py | 304 +++++++++++++++++++- tests/test_position_sizing.py | 71 +++++ tests/test_risk_gate.py | 331 ++++++++++++++++++++++ 5 files changed, 751 insertions(+), 6 deletions(-) diff --git a/src/quant_platform_kit/position_sizing.py b/src/quant_platform_kit/position_sizing.py index 1ac421c..c11dad5 100644 --- a/src/quant_platform_kit/position_sizing.py +++ b/src/quant_platform_kit/position_sizing.py @@ -20,9 +20,14 @@ class KellyResult: _APPROVED_BOOTSTRAP_MANDATE = "bootstrap_small_account_v2" +_TQQQ_ETF_ONLY_RESEARCH_MANDATE = "tqqq_etf_only_research_v1" _BOOTSTRAP_LOSS_BUDGET_CAP = 0.01 _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP = 0.50 _BOOTSTRAP_NOMINAL_CAPS = {1: 0.50, 2: 0.25, 3: 0.15} +_TQQQ_ETF_ONLY_PRODUCTS = { + "TQQQ": (3, 0.15, 0.45), + "BOXX": (1, 0.50, 0.50), +} def _weight_mapping(value: object, *, allow_empty: bool) -> dict[str, float] | None: @@ -140,6 +145,7 @@ def validate_reduce_only_normalization( product_leverage_factors: Mapping[str, int], effective_exposure_cap: float, observed_effective_exposure: float, + cash_only: bool = False, ) -> bool: """Validate one explicit transition from an over-cap origin toward cash.""" origin = _weight_mapping(origin_weights, allow_empty=False) @@ -153,6 +159,7 @@ def validate_reduce_only_normalization( or not isinstance(effective_exposure_cap, (int, float)) or isinstance(observed_effective_exposure, bool) or not isinstance(observed_effective_exposure, (int, float)) + or not isinstance(cash_only, bool) ): return False cap = float(effective_exposure_cap) @@ -178,6 +185,8 @@ def validate_reduce_only_normalization( target_active = {symbol for symbol, weight in target.items() if weight > 0.0} if not origin_active or not target_active.issubset(origin_active): return False + if cash_only and target_active: + return False if any(target.get(symbol, 0.0) > origin[symbol] + 1e-9 for symbol in origin): return False if any( @@ -199,6 +208,7 @@ def validate_reduce_only_normalization( def risk_budgeted_target_weight( *, risk_mandate_id: str | None = None, + product_symbol: str | None = None, account_equity: float | None = None, risk_fraction: float | None = None, stop_loss_distance: float | None = None, @@ -209,9 +219,9 @@ def risk_budgeted_target_weight( ) -> float: """Return a fail-closed single-account target weight. - ``bootstrap_small_account_v2`` permits one ETF position with its approved - product cap. Without that mandate, the legacy 10% and unlevered fallback - applies. This helper does not allocate across strategies. + Approved mandates permit one ETF position with their product caps. Without + a mandate, the legacy 10% and unlevered fallback applies. This pure helper + sizes a target; it does not authorize a product or execution. """ numeric_inputs = ( account_equity, @@ -256,6 +266,18 @@ def risk_budgeted_target_weight( nominal_cap = _BOOTSTRAP_NOMINAL_CAPS.get(product_leverage_factor, 0.0) if nominal_cap == 0.0: return 0.0 + elif risk_mandate_id == _TQQQ_ETF_ONLY_RESEARCH_MANDATE: + product = _TQQQ_ETF_ONLY_PRODUCTS.get(product_symbol or "") + if ( + product is None + or product_leverage_factor != product[0] + or risk_fraction != _BOOTSTRAP_LOSS_BUDGET_CAP + or stop_loss_distance != 0.05 + or drawdown_scalar not in {0.0, 0.5, 1.0} + ): + return 0.0 + nominal_cap = product[1] + product_effective_cap = product[2] else: return 0.0 @@ -263,11 +285,16 @@ def risk_budgeted_target_weight( if not math.isfinite(risk_weight): return 0.0 + effective_cap = ( + product_effective_cap + if risk_mandate_id == _TQQQ_ETF_ONLY_RESEARCH_MANDATE + else _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP + ) return min( risk_weight, nominal_cap, available_account_exposure, - _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP / product_leverage_factor, + effective_cap / product_leverage_factor, ) diff --git a/src/quant_platform_kit/risk/contracts.py b/src/quant_platform_kit/risk/contracts.py index 70f7641..f05481c 100644 --- a/src/quant_platform_kit/risk/contracts.py +++ b/src/quant_platform_kit/risk/contracts.py @@ -222,6 +222,14 @@ class RiskGateAssessment: proposed_effective_exposure: float | None outcome: str reason_codes: tuple[str, ...] + execution_authorized: bool = False + stop_loss_distance: float | None = None + stop_intent_ready: bool | None = None + strategy_breaker_triggered: bool | None = None + account_breaker_triggered: bool | None = None + account_drawdown_fraction: float | None = None + drawdown_scalar: float | None = None + risk_control_state_digest_sha256: str | None = None assessment_sha256: str = field(init=False) def __post_init__(self) -> None: @@ -245,6 +253,14 @@ def __post_init__(self) -> None: "proposed_effective_exposure": self.proposed_effective_exposure, "outcome": self.outcome, "reason_codes": self.reason_codes, + "execution_authorized": self.execution_authorized, + "stop_loss_distance": self.stop_loss_distance, + "stop_intent_ready": self.stop_intent_ready, + "strategy_breaker_triggered": self.strategy_breaker_triggered, + "account_breaker_triggered": self.account_breaker_triggered, + "account_drawdown_fraction": self.account_drawdown_fraction, + "drawdown_scalar": self.drawdown_scalar, + "risk_control_state_digest_sha256": self.risk_control_state_digest_sha256, } encoded = json.dumps( payload, diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index a362865..d60c0ba 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -29,11 +29,16 @@ _MAX_CONSECUTIVE_LOSSES = 5 _DEFAULT_MAX_SINGLE_WEIGHT = 0.10 _APPROVED_BOOTSTRAP_MANDATE = "bootstrap_small_account_v2" +_TQQQ_ETF_ONLY_RESEARCH_MANDATE = "tqqq_etf_only_research_v1" +_TQQQ_ETF_ONLY_STRATEGY_PROFILE = "tqqq_etf_only_single_strategy_research_v1" +_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} _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP = 0.50 _BOOTSTRAP_NOMINAL_CAPS = {1: 0.50, 2: 0.25, 3: 0.15} -_ASSESSMENT_CONTRACT_VERSION = "qsl.risk_gate_assessment.v1" +_ASSESSMENT_CONTRACT_VERSION = "qsl.risk_gate_assessment.v2" _ASSESSMENT_POLICY_ID = "qpk.risk_gate" -_ASSESSMENT_POLICY_VERSION = "v1" +_ASSESSMENT_POLICY_VERSION = "v2" _FALLBACK_MAX_SNAPSHOT_AGE_SECONDS_V1 = 300.0 _ALLOWED_SCOPES = frozenset({"MEMBER", "ACCOUNT"}) _ALLOWED_MANDATE_SCOPES = frozenset({"RESEARCH_ONLY", "PAPER", "LIVE"}) @@ -197,6 +202,101 @@ def _snapshot_metrics( }, observed, total_equity, set() +def _exact_numeric_mapping(value: Any, expected: Mapping[str, float]) -> bool: + if not isinstance(value, Mapping) or set(value) != set(expected): + return False + return all( + (number := _finite_number(value[key])) is not None + and number == expected_value + for key, expected_value in expected.items() + ) + + +def _exact_tqqq_mandate_errors( + mandate_provenance: Mapping[str, Any], + *, + effective_at: datetime, + expires_at: datetime, +) -> set[str]: + if mandate_provenance.get("mandate_id") != _TQQQ_ETF_ONLY_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", + "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() + ) + ) + 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") + != _TQQQ_ETF_ONLY_STRATEGY_PROFILE + or mandate_provenance.get("account_mode") != "single_strategy_account_v1" + 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"), + _TQQQ_ETF_ONLY_NOMINAL_CAPS, + ) + or not _exact_numeric_mapping( + mandate_provenance.get("nominal_caps"), + _TQQQ_ETF_ONLY_NOMINAL_CAPS, + ) + or not _exact_numeric_mapping( + mandate_provenance.get("product_effective_caps"), + _TQQQ_ETF_ONLY_EFFECTIVE_CAPS, + ) + or not exact_factors + or not isinstance(allowed_assets, (list, tuple)) + 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 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 _finite_number( + mandate_provenance.get("precommitted_executable_stop_distance") + ) + != 0.05 + 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_tqqq_research_mandate"} if invalid else set() + + def _mandate_fields( mandate_provenance: Mapping[str, Any] | None, *, @@ -292,6 +392,13 @@ 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, + ) + if exact_mandate_errors: + return {}, exact_mandate_errors factors = mandate_provenance.get("product_leverage_factors", {}) allowed_assets = mandate_provenance.get("allowed_nonzero_assets") if ( @@ -326,7 +433,12 @@ def _mandate_fields( "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"), }, set() @@ -425,6 +537,145 @@ def _budget_authority_errors( return set() +def _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 mandate.get("mandate_id") != _TQQQ_ETF_ONLY_RESEARCH_MANDATE: + return empty, set() + 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_intent_ready", + "tqqq_entry_fill_identity_sha256", + "stop_entry_fill_identity_sha256", + "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 not isinstance(risk_control_state.get("stop_intent_ready"), bool): + errors.add("invalid_stop_state") + 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") + + tqqq_active = any(symbol == "TQQQ" for symbol, _ in active_positions) + if tqqq_active and risk_control_state.get("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 + or entry_fill_identity != stop_entry_fill_identity + ): + errors.add("stop_entry_fill_identity_mismatch") + 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_intent_ready": ( + risk_control_state.get("stop_intent_ready") + if isinstance(risk_control_state.get("stop_intent_ready"), bool) + else None + ), + "tqqq_entry_fill_identity_sha256": entry_fill_identity, + "stop_entry_fill_identity_sha256": stop_entry_fill_identity, + "consecutive_completed_losing_exits": losses, + "account_drawdown_fraction": account_drawdown, + "drawdown_scalar": drawdown_scalar, + } + 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 assess_with_evidence( decision: StrategyDecision, portfolio_snapshot: Any, @@ -434,6 +685,7 @@ def assess_with_evidence( 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.""" now = _utc_now() @@ -465,6 +717,13 @@ def assess_with_evidence( can_evaluate_policy = not reason_codes if mandate: reason_codes.update(_budget_authority_errors(decision, mandate)) + control_fields, control_errors = _risk_control_fields( + risk_control_state, + mandate=mandate, + now=now, + active_positions=active_positions, + ) + reason_codes.update(control_errors) proposed: float | None = None normalization_origin_digest_sha256: str | None = None @@ -474,6 +733,11 @@ def assess_with_evidence( weighted_exposure = 0.0 if mandate_provenance is None and len(active_positions) > 1: reason_codes.add("fallback_position_count") + if ( + mandate.get("mandate_id") == _TQQQ_ETF_ONLY_RESEARCH_MANDATE + and len(active_positions) > mandate["max_nonzero_assets"] + ): + reason_codes.add("single_strategy_position_count") for symbol, weight in active_positions: if allowed_assets is not None and symbol not in allowed_assets: reason_codes.add("asset_not_authorized") @@ -489,6 +753,29 @@ def assess_with_evidence( continue if weight > min(product_cap, nominal_cap): reason_codes.add("product_exposure_cap") + product_effective_cap = _position_cap( + mandate.get("product_effective_caps", 1.0), + symbol, + factor, + ) + if ( + product_effective_cap is None + or weight * factor > product_effective_cap + 1e-9 + ): + reason_codes.add("product_effective_exposure_cap") + if mandate.get("mandate_id") == _TQQQ_ETF_ONLY_RESEARCH_MANDATE: + stop_distance = control_fields["stop_loss_distance"] + drawdown_scalar = control_fields["drawdown_scalar"] + loss_budget = mandate.get("loss_budget") + if ( + stop_distance is not None + 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 + ): + reason_codes.add("risk_budget_exposure_cap") weighted_exposure += weight * factor target_weights: dict[str, float] = {} for symbol, weight in active_positions: @@ -501,6 +788,9 @@ 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 + ), ) if not valid_normalization: reason_codes.add("invalid_reduce_only_normalization") @@ -563,6 +853,16 @@ def assess_with_evidence( proposed_effective_exposure=proposed, outcome=outcome, reason_codes=tuple(sorted(reason_codes)), + execution_authorized=False, + stop_loss_distance=control_fields["stop_loss_distance"], + stop_intent_ready=control_fields["stop_intent_ready"], + strategy_breaker_triggered=control_fields["strategy_breaker_triggered"], + account_breaker_triggered=control_fields["account_breaker_triggered"], + account_drawdown_fraction=control_fields["account_drawdown_fraction"], + drawdown_scalar=control_fields["drawdown_scalar"], + risk_control_state_digest_sha256=control_fields[ + "risk_control_state_digest_sha256" + ], ) if outcome == "REJECT": return RiskGateResult( diff --git a/tests/test_position_sizing.py b/tests/test_position_sizing.py index 98f6bda..c51099c 100644 --- a/tests/test_position_sizing.py +++ b/tests/test_position_sizing.py @@ -177,6 +177,59 @@ def test_invalid_stale_or_over_budget_inputs_fail_closed(self) -> None: ) +class TqqqEtfOnlyResearchSizingTests(unittest.TestCase): + _MANDATE_ID = "tqqq_etf_only_research_v1" + + def _inputs(self, **overrides: object) -> dict[str, object]: + return { + "risk_mandate_id": self._MANDATE_ID, + "product_symbol": "TQQQ", + "account_equity": 100_000.0, + "risk_fraction": 0.01, + "stop_loss_distance": 0.05, + "drawdown_scalar": 1.0, + "available_account_exposure": 0.50, + "product_leverage_factor": 3, + "inputs_fresh": True, + **overrides, + } + + def test_tqqq_uses_one_percent_budget_five_percent_stop_and_fifteen_percent_cap( + self, + ) -> None: + self.assertEqual(risk_budgeted_target_weight(**self._inputs()), 0.15) + self.assertAlmostEqual( + risk_budgeted_target_weight(**self._inputs(drawdown_scalar=0.50)), + 0.10, + ) + self.assertEqual( + risk_budgeted_target_weight(**self._inputs(drawdown_scalar=0.0)), + 0.0, + ) + + def test_only_exact_tqqq_and_boxx_product_contracts_are_sized(self) -> None: + self.assertAlmostEqual( + risk_budgeted_target_weight( + **self._inputs(product_symbol="BOXX", product_leverage_factor=1) + ), + 0.20, + ) + invalid_cases = ( + {"product_symbol": "QQQ", "product_leverage_factor": 1}, + {"product_symbol": "TQQQ", "product_leverage_factor": 1}, + {"product_symbol": "BOXX", "product_leverage_factor": 3}, + {"stop_loss_distance": 0.06}, + {"risk_fraction": 0.010001}, + {"drawdown_scalar": 0.75}, + ) + for overrides in invalid_cases: + with self.subTest(overrides=overrides): + self.assertEqual( + risk_budgeted_target_weight(**self._inputs(**overrides)), + 0.0, + ) + + class RiskBudgetedTargetWeightsTests(unittest.TestCase): def _approved_inputs(self, **overrides: object) -> dict[str, object]: return { @@ -249,6 +302,24 @@ def test_one_hundred_percent_boxx_can_normalize_to_compliant_boxx_cash(self) -> ) ) + def test_cash_only_normalization_rejects_any_residual_position(self) -> None: + inputs = { + "origin_weights": {"TQQQ": 0.20}, + "product_leverage_factors": {"TQQQ": 3}, + "effective_exposure_cap": 0.50, + "observed_effective_exposure": 0.60, + "cash_only": True, + } + self.assertTrue( + validate_reduce_only_normalization(target_weights={}, **inputs) + ) + self.assertFalse( + validate_reduce_only_normalization( + target_weights={"TQQQ": 0.10}, + **inputs, + ) + ) + def test_normalization_rejects_new_exposure_non_reduction_or_bad_origin(self) -> None: invalid_cases = ( ({"BOXX": 0.40, "SOXX": 0.10}, {"BOXX": 1, "SOXX": 1}, 1.0), diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 0434f5b..de6cecf 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -584,6 +584,10 @@ def test_reduce_only_normalization_can_exit_over_cap_origin_once(self) -> None: ) self.assertEqual(result.assessment.outcome, "APPROVE") + self.assertEqual( + result.assessment.contract_version, + "qsl.risk_gate_assessment.v2", + ) self.assertEqual(result.assessment.proposed_effective_exposure, 0.50) self.assertIsNotNone(result.assessment.normalization_origin_digest_sha256) engine.assess.assert_called_once_with( @@ -930,6 +934,333 @@ def test_mandate_rejects_budget_only_decision_above_authority(self) -> None: self.assertIn("budget_authority_exceeded", result.assessment.reason_codes) +class TqqqEtfOnlyResearchMandateTests(unittest.TestCase): + _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) + _MANDATE_ID = "tqqq_etf_only_research_v1" + _STRATEGY_PROFILE = "tqqq_etf_only_single_strategy_research_v1" + + @classmethod + def _candidate(cls) -> CandidateRiskIdentity: + return CandidateRiskIdentity( + strategy_profile=cls._STRATEGY_PROFILE, + account_mode="single_strategy_account_v1", + strategy_revision="b" * 40, + runner_revision="c" * 40, + config_sha256="d" * 64, + input_manifest_sha256="e" * 64, + authority_receipt_sha256="a" * 64, + ) + + @classmethod + def _mandate(cls, **overrides: object) -> dict[str, object]: + candidate = cls._candidate() + 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-04T04:27:55Z", + "expires_at": "2026-09-03T15:59:59Z", + "max_snapshot_age_seconds": 300, + "effective_exposure_cap": 0.50, + "loss_budget": 0.01, + "loss_budget_equity_reference": "completed_session_equity", + "product_caps": {"TQQQ": 0.15, "BOXX": 0.50}, + "nominal_caps": {"TQQQ": 0.15, "BOXX": 0.50}, + "product_effective_caps": {"TQQQ": 0.45, "BOXX": 0.50}, + "product_leverage_factors": {"TQQQ": 3, "BOXX": 1}, + "allowed_nonzero_assets": ["TQQQ", "BOXX"], + "max_nonzero_assets": 1, + "broker_margin_factor": 1, + "margin_stacking": False, + "borrowing": False, + "shorting": False, + "income_sleeve_enabled": False, + "option_overlay_enabled": False, + "precommitted_executable_stop_distance": 0.05, + "max_consecutive_completed_losing_exits": 5, + "source_revision": "f" * 40, + } + mandate.update(overrides) + return mandate + + @staticmethod + def _snapshot(**overrides: object) -> dict[str, object]: + snapshot: dict[str, object] = { + "as_of": "2026-08-04T04:27:55Z", + "observed_effective_exposure": 0.0, + "total_equity": 100_000.0, + } + snapshot.update(overrides) + return snapshot + + @classmethod + def _risk_state(cls, **overrides: object) -> dict[str, object]: + state: dict[str, object] = { + "as_of": "2026-08-04T04:27:55Z", + "mandate_id": cls._MANDATE_ID, + "candidate_identity_sha256": cls._candidate().candidate_sha256, + "stop_loss_distance": 0.05, + "stop_intent_ready": True, + "tqqq_entry_fill_identity_sha256": "1" * 64, + "stop_entry_fill_identity_sha256": "1" * 64, + "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, + origin: dict[str, float] | None = None, + ) -> tuple[object, Mock]: + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + snapshot or self._snapshot(), + scope="MEMBER", + mandate_provenance=mandate or self._mandate(), + market_data={}, + candidate_identity=self._candidate(), + normalization_origin_weights=origin, + risk_control_state=( + self._risk_state() if risk_state is None else risk_state + ), + ) + return result, engine + + def test_valid_research_mandate_approves_evidence_but_never_execution(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.15),) + ) + result, engine = self._assess(decision) + + self.assertEqual(result.assessment.outcome, "APPROVE") + self.assertEqual(result.assessment.mandate_id, self._MANDATE_ID) + self.assertAlmostEqual(result.assessment.proposed_effective_exposure, 0.45) + 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.account_drawdown_fraction, 0.05) + 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) + engine.assess.assert_called_once_with(decision, self._snapshot(), market_data={}) + + def test_assessment_identity_is_bound_to_risk_control_state(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.15),) + ) + first, first_engine = self._assess( + decision, + risk_state=self._risk_state(account_drawdown_fraction=0.04), + ) + second, second_engine = self._assess( + decision, + risk_state=self._risk_state(account_drawdown_fraction=0.05), + ) + + self.assertEqual(first.assessment.outcome, "APPROVE") + self.assertEqual(second.assessment.outcome, "APPROVE") + self.assertEqual( + first.assessment.decision_digest_sha256, + second.assessment.decision_digest_sha256, + ) + self.assertNotEqual( + first.assessment.risk_control_state_digest_sha256, + second.assessment.risk_control_state_digest_sha256, + ) + self.assertNotEqual( + first.assessment.assessment_sha256, + second.assessment.assessment_sha256, + ) + first_engine.assess.assert_called_once() + second_engine.assess.assert_called_once() + + def test_exact_mandate_values_and_exclusions_are_fail_closed(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.15),) + ) + invalid_cases = ( + {"authority_scope": "PAPER"}, + {"strategy_profile": "other"}, + {"account_mode": "smart_portfolio"}, + {"effective_exposure_cap": 0.51}, + {"loss_budget": 0.011}, + {"loss_budget_equity_reference": "current_equity"}, + {"product_caps": {"TQQQ": 0.16, "BOXX": 0.50}}, + {"product_effective_caps": {"TQQQ": 0.46, "BOXX": 0.50}}, + {"product_leverage_factors": {"TQQQ": 2, "BOXX": 1}}, + {"allowed_nonzero_assets": ["TQQQ", "BOXX", "QQQ"]}, + {"max_nonzero_assets": 2}, + {"broker_margin_factor": 2}, + {"margin_stacking": True}, + {"borrowing": True}, + {"shorting": True}, + {"income_sleeve_enabled": True}, + {"option_overlay_enabled": True}, + {"precommitted_executable_stop_distance": 0.06}, + {"max_consecutive_completed_losing_exits": 6}, + ) + 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_tqqq_research_mandate", + result.assessment.reason_codes, + ) + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_missing_stale_nonfinite_or_mismatched_control_state_rejects(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.15),) + ) + invalid_cases = ( + {}, + self._risk_state(as_of="2026-08-04T04:17:55Z"), + self._risk_state(account_drawdown_fraction=float("nan")), + self._risk_state(candidate_identity_sha256="0" * 64), + self._risk_state(mandate_id="other"), + self._risk_state(stop_loss_distance=0.06), + self._risk_state(stop_intent_ready=False), + self._risk_state(stop_entry_fill_identity_sha256="2" * 64), + self._risk_state(drawdown_scalar=0.50), + ) + for risk_state in invalid_cases: + with self.subTest(risk_state=risk_state): + result, engine = self._assess(decision, risk_state=risk_state) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_drawdown_and_strategy_breaker_boundaries(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.15),) + ) + approved_cases = ( + ( + decision, + self._risk_state(account_drawdown_fraction=0.05, drawdown_scalar=1.0), + ), + ( + _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ), + self._risk_state( + account_drawdown_fraction=0.050001, + drawdown_scalar=0.50, + ), + ), + ( + _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ), + self._risk_state( + account_drawdown_fraction=0.10, + drawdown_scalar=0.50, + ), + ), + (decision, self._risk_state(consecutive_completed_losing_exits=4)), + ) + for approved_decision, state in approved_cases: + with self.subTest(state=state): + result, engine = self._assess( + approved_decision, + risk_state=state, + ) + self.assertEqual(result.assessment.outcome, "APPROVE") + engine.assess.assert_called_once() + + breaker_cases = ( + ( + self._risk_state(consecutive_completed_losing_exits=5), + "strategy_breaker_triggered", + ), + ( + self._risk_state( + account_drawdown_fraction=0.100001, + drawdown_scalar=0.0, + ), + "account_breaker_triggered", + ), + ) + for state, reason in breaker_cases: + with self.subTest(state=state): + 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, ()) + engine.assess.assert_called_once() + + def test_single_strategy_rule_and_product_caps_reject_excess(self) -> None: + invalid_decisions = ( + _decision( + positions=( + PositionTarget(symbol="TQQQ", target_weight=0.10), + PositionTarget(symbol="BOXX", target_weight=0.10), + ) + ), + _decision(positions=(PositionTarget(symbol="TQQQ", target_weight=0.151),)), + _decision(positions=(PositionTarget(symbol="BOXX", target_weight=0.201),)), + _decision(positions=(PositionTarget(symbol="BOXX", target_weight=0.501),)), + _decision(positions=(PositionTarget(symbol="QQQ", target_weight=0.10),)), + ) + for decision in invalid_decisions: + with self.subTest(decision=decision): + result, engine = self._assess(decision) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_over_cap_normalization_must_reduce_to_cash_and_binds_origin(self) -> None: + snapshot = self._snapshot(observed_effective_exposure=0.60) + cash_result, cash_engine = self._assess( + _decision(), + snapshot=snapshot, + origin={"TQQQ": 0.20}, + ) + partial_result, partial_engine = self._assess( + _decision(positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),)), + snapshot=snapshot, + origin={"TQQQ": 0.20}, + ) + + self.assertEqual(cash_result.assessment.outcome, "APPROVE") + self.assertEqual(len(cash_result.assessment.normalization_origin_digest_sha256), 64) + self.assertEqual(partial_result.assessment.outcome, "REJECT") + self.assertIn( + "invalid_reduce_only_normalization", + partial_result.assessment.reason_codes, + ) + cash_engine.assess.assert_called_once() + partial_engine.assess.assert_called_once() + + class BootstrapSmallAccountV2RiskGateTests(unittest.TestCase): _MANDATE = "bootstrap_small_account_v2"