From abade7639c96823a68a724622f23bd1767c7a8c5 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 5 Aug 2026 22:37:27 +0800 Subject: [PATCH 1/2] feat: gate SOXL pre-inception asset eligibility Co-Authored-By: Codex --- .../entrypoints/__init__.py | 30 ++++ tests/test_entrypoint_risk_gate.py | 141 +++++++++++++++++- tests/test_entrypoints.py | 1 + 3 files changed, 171 insertions(+), 1 deletion(-) diff --git a/src/us_equity_strategies/entrypoints/__init__.py b/src/us_equity_strategies/entrypoints/__init__.py index 6304b30..40f12e8 100644 --- a/src/us_equity_strategies/entrypoints/__init__.py +++ b/src/us_equity_strategies/entrypoints/__init__.py @@ -933,6 +933,8 @@ def evaluate_soxl_soxx_trend_income_promotion_research( stop_loss_distances: Mapping[str, float], drawdown_scalar: float, inputs_fresh: bool, + point_in_time_eligible_assets: frozenset[str] | None = None, + qqqi_preinception_fallback_symbol: str | None = None, normalization_origin_weights: Mapping[str, float] | None = None, ) -> RiskGateResult: """Build, size and assess one offline SOXL promotion-research decision.""" @@ -950,6 +952,23 @@ def evaluate_soxl_soxx_trend_income_promotion_research( if position.target_weight is not None } mandate = mandate_provenance if isinstance(mandate_provenance, Mapping) else {} + registered_assets = tuple(soxl_soxx_trend_income_manifest.default_config["managed_symbols"]) + promotion_assets = (*registered_assets, "QQQ") + eligible_assets = point_in_time_eligible_assets + if ( + not isinstance(eligible_assets, frozenset) + or not eligible_assets + or not eligible_assets.issubset(promotion_assets) + or qqqi_preinception_fallback_symbol not in {None, "QQQ"} + or set(raw_weights) - set(registered_assets) + or mandate.get("allowed_nonzero_assets") != list(promotion_assets) + ): + raise ValueError("invalid promotion eligibility contract") + if "QQQI" not in eligible_assets and qqqi_preinception_fallback_symbol == "QQQ": + if "QQQ" not in eligible_assets: + raise ValueError("QQQ fallback is not point-in-time eligible") + if "QQQI" in raw_weights: + raw_weights["QQQ"] = raw_weights.pop("QQQI") mandate_factors = mandate.get("product_leverage_factors") if not isinstance(mandate_factors, Mapping): raise ValueError("promotion mandate is missing product leverage factors") @@ -970,6 +989,14 @@ def evaluate_soxl_soxx_trend_income_promotion_research( positions_by_symbol = { position.symbol: position for position in weighted_decision.positions } + if "QQQ" in raw_weights: + positions_by_symbol["QQQ"] = positions_by_symbol["QQQI"] + unavailable_targets = tuple( + sorted(symbol for symbol, weight in sized_weights.items() if weight > 0.0 and symbol not in eligible_assets) + ) + sized_weights = { + symbol: weight for symbol, weight in sized_weights.items() if symbol in eligible_assets + } decision = StrategyDecision( positions=tuple( PositionTarget( @@ -985,6 +1012,9 @@ def evaluate_soxl_soxx_trend_income_promotion_research( diagnostics={ **weighted_decision.diagnostics, "promotion_research_sized": True, + "promotion_research_eligible_assets": tuple(sorted(eligible_assets)), + "promotion_research_qqqi_fallback_symbol": qqqi_preinception_fallback_symbol, + "promotion_research_ineligible_assets_to_cash": unavailable_targets, }, ) except Exception: diff --git a/tests/test_entrypoint_risk_gate.py b/tests/test_entrypoint_risk_gate.py index 9b9e526..80f27bf 100644 --- a/tests/test_entrypoint_risk_gate.py +++ b/tests/test_entrypoint_risk_gate.py @@ -14,7 +14,7 @@ from us_equity_strategies.entrypoints._common import apply_risk_gate -_SOXL_ASSETS = ("SOXL", "SOXX", "BOXX", "SCHD", "DGRO", "SGOV", "SPYI", "QQQI") +_SOXL_ASSETS = ("SOXL", "SOXX", "BOXX", "SCHD", "DGRO", "SGOV", "SPYI", "QQQI", "QQQ") _SOXL_NOW = datetime(2026, 8, 5, 12, 0, tzinfo=timezone.utc) @@ -85,6 +85,17 @@ def _soxl_raw_decision() -> StrategyDecision: ) +def _soxl_preinception_decision() -> StrategyDecision: + return StrategyDecision( + positions=( + PositionTarget(symbol="SOXL", target_value=70_000.0), + PositionTarget(symbol="BOXX", target_value=20_000.0, role="safe_haven"), + PositionTarget(symbol="QQQI", target_value=10_000.0, role="income"), + ), + diagnostics={"source": "shared_builder"}, + ) + + def test_soxl_promotion_research_sizes_and_assesses_exactly_once(monkeypatch) -> None: candidate = _soxl_candidate() mandate = _soxl_mandate(candidate) @@ -108,6 +119,8 @@ def test_soxl_promotion_research_sizes_and_assesses_exactly_once(monkeypatch) -> stop_loss_distances={"SOXL": 0.05, "BOXX": 0.05}, drawdown_scalar=1.0, inputs_fresh=True, + point_in_time_eligible_assets=frozenset(_SOXL_ASSETS), + qqqi_preinception_fallback_symbol="QQQ", ) assert result.assessment.outcome == "APPROVE" @@ -153,6 +166,8 @@ def test_soxl_promotion_research_identity_failures_clear_decision_and_assess_onc stop_loss_distances={"SOXL": 0.05, "BOXX": 0.05}, drawdown_scalar=1.0, inputs_fresh=True, + point_in_time_eligible_assets=frozenset(_SOXL_ASSETS), + qqqi_preinception_fallback_symbol="QQQ", ) assert result.assessment.outcome == "REJECT" @@ -193,6 +208,8 @@ def test_soxl_promotion_research_builder_exception_fails_closed_and_assesses_onc stop_loss_distances={"SOXL": 0.05, "BOXX": 0.05}, drawdown_scalar=1.0, inputs_fresh=True, + point_in_time_eligible_assets=frozenset(_SOXL_ASSETS), + qqqi_preinception_fallback_symbol="QQQ", ) assert result.assessment.outcome == "REJECT" @@ -205,6 +222,128 @@ def test_soxl_promotion_research_builder_exception_fails_closed_and_assesses_onc monitor.assert_not_called() +@pytest.mark.parametrize( + ("fallback_symbol", "expected"), + (("QQQ", {"SOXL": 0.14, "QQQ": 0.02}), (None, {"SOXL": 0.14})), +) +def test_soxl_promotion_research_keeps_unavailable_targets_as_cash( + monkeypatch, + fallback_symbol, + expected, +) -> None: + candidate = _soxl_candidate() + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + monkeypatch.setattr( + entrypoints, + "_build_soxl_soxx_trend_income_decision", + Mock(return_value=_soxl_preinception_decision()), + ) + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=_SOXL_NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = entrypoints.evaluate_soxl_soxx_trend_income_promotion_research( + _soxl_context(), + candidate_identity=candidate, + mandate_provenance=_soxl_mandate(candidate), + stop_loss_distances={symbol: 0.05 for symbol in _SOXL_ASSETS}, + drawdown_scalar=1.0, + inputs_fresh=True, + point_in_time_eligible_assets=frozenset({"SOXL", "SOXX", "SCHD", "DGRO", "QQQ"}), + qqqi_preinception_fallback_symbol=fallback_symbol, + ) + + assert result.assessment.outcome == "APPROVE" + assert {position.symbol: position.target_weight for position in result.decision.positions} == pytest.approx( + expected + ) + assert not {"BOXX", "QQQI"} & {position.symbol for position in result.decision.positions} + engine.assess.assert_called_once() + + +@pytest.mark.parametrize( + ("eligible_assets", "fallback_symbol", "raw_decision"), + ( + (None, "QQQ", _soxl_preinception_decision()), + (frozenset({"SOXL", "QQQ", "SPY"}), "QQQ", _soxl_preinception_decision()), + (frozenset({"SOXL", "QQQ"}), "SPY", _soxl_preinception_decision()), + (frozenset({"SOXL"}), "QQQ", _soxl_preinception_decision()), + ( + frozenset(_SOXL_ASSETS), + "QQQ", + StrategyDecision(positions=(PositionTarget(symbol="QQQ", target_value=10_000.0),)), + ), + ), +) +def test_soxl_promotion_research_invalid_eligibility_fails_closed_and_assesses_once( + monkeypatch, + eligible_assets, + fallback_symbol, + raw_decision, +) -> None: + candidate = _soxl_candidate() + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + monkeypatch.setattr( + entrypoints, + "_build_soxl_soxx_trend_income_decision", + Mock(return_value=raw_decision), + ) + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=_SOXL_NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = entrypoints.evaluate_soxl_soxx_trend_income_promotion_research( + _soxl_context(), + candidate_identity=candidate, + mandate_provenance=_soxl_mandate(candidate), + stop_loss_distances={symbol: 0.05 for symbol in _SOXL_ASSETS}, + drawdown_scalar=1.0, + inputs_fresh=True, + point_in_time_eligible_assets=eligible_assets, + qqqi_preinception_fallback_symbol=fallback_symbol, + ) + + assert result.assessment.outcome == "REJECT" + assert result.decision.positions == () + assert result.decision.budgets == () + engine.assess.assert_called_once() + + +def test_soxl_promotion_research_missing_eligibility_fails_closed_and_assesses_once( + monkeypatch, +) -> None: + candidate = _soxl_candidate() + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + monkeypatch.setattr( + entrypoints, + "_build_soxl_soxx_trend_income_decision", + Mock(return_value=_soxl_preinception_decision()), + ) + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=_SOXL_NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = entrypoints.evaluate_soxl_soxx_trend_income_promotion_research( + _soxl_context(), + candidate_identity=candidate, + mandate_provenance=_soxl_mandate(candidate), + stop_loss_distances={symbol: 0.05 for symbol in _SOXL_ASSETS}, + drawdown_scalar=1.0, + inputs_fresh=True, + ) + + assert result.assessment.outcome == "REJECT" + assert result.decision.positions == () + assert result.decision.budgets == () + engine.assess.assert_called_once() + + def test_apply_risk_gate_enriches_stop_loss_diagnostics_from_portfolio() -> None: snapshot = PortfolioSnapshot( as_of=datetime(2026, 7, 9, tzinfo=timezone.utc), diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index 9e6069e..fe625f1 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -851,6 +851,7 @@ def test_soxl_soxx_trend_income_entrypoint_maps_target_values_without_execution_ entrypoint.manifest.default_config["managed_symbols"], ("SOXL", "SOXX", "BOXX", "SCHD", "DGRO", "SGOV", "SPYI", "QQQI"), ) + self.assertNotIn("QQQ", entrypoint.manifest.default_config["managed_symbols"]) self.assertIs(entrypoint.manifest.default_config["income_layer_enabled"], True) self.assertEqual(entrypoint.manifest.default_config["income_layer_ratio_mode"], "log_total_drawdown_budget") self.assertEqual(entrypoint.manifest.default_config["income_layer_start_usd"], 150000.0) From 24fe759cca1140095ccb36badd872b36bd8ab837 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 5 Aug 2026 22:56:37 +0800 Subject: [PATCH 2/2] fix: filter ineligible targets before sizing Co-Authored-By: Codex --- src/us_equity_strategies/entrypoints/__init__.py | 16 ++++++++++------ tests/test_entrypoint_risk_gate.py | 12 ++++++++++-- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/src/us_equity_strategies/entrypoints/__init__.py b/src/us_equity_strategies/entrypoints/__init__.py index 40f12e8..63bad29 100644 --- a/src/us_equity_strategies/entrypoints/__init__.py +++ b/src/us_equity_strategies/entrypoints/__init__.py @@ -969,6 +969,16 @@ def evaluate_soxl_soxx_trend_income_promotion_research( raise ValueError("QQQ fallback is not point-in-time eligible") if "QQQI" in raw_weights: raw_weights["QQQ"] = raw_weights.pop("QQQI") + unavailable_targets = tuple( + sorted( + symbol + for symbol, weight in raw_weights.items() + if weight > 0.0 and symbol not in eligible_assets + ) + ) + raw_weights = { + symbol: weight for symbol, weight in raw_weights.items() if symbol in eligible_assets + } mandate_factors = mandate.get("product_leverage_factors") if not isinstance(mandate_factors, Mapping): raise ValueError("promotion mandate is missing product leverage factors") @@ -991,12 +1001,6 @@ def evaluate_soxl_soxx_trend_income_promotion_research( } if "QQQ" in raw_weights: positions_by_symbol["QQQ"] = positions_by_symbol["QQQI"] - unavailable_targets = tuple( - sorted(symbol for symbol, weight in sized_weights.items() if weight > 0.0 and symbol not in eligible_assets) - ) - sized_weights = { - symbol: weight for symbol, weight in sized_weights.items() if symbol in eligible_assets - } decision = StrategyDecision( positions=tuple( PositionTarget( diff --git a/tests/test_entrypoint_risk_gate.py b/tests/test_entrypoint_risk_gate.py index 80f27bf..e4ffdca 100644 --- a/tests/test_entrypoint_risk_gate.py +++ b/tests/test_entrypoint_risk_gate.py @@ -224,7 +224,10 @@ def test_soxl_promotion_research_builder_exception_fails_closed_and_assesses_onc @pytest.mark.parametrize( ("fallback_symbol", "expected"), - (("QQQ", {"SOXL": 0.14, "QQQ": 0.02}), (None, {"SOXL": 0.14})), + ( + ("QQQ", {"SOXL": 0.15, "QQQ": 0.021428571428571432}), + (None, {"SOXL": 0.15}), + ), ) def test_soxl_promotion_research_keeps_unavailable_targets_as_cash( monkeypatch, @@ -248,7 +251,9 @@ def test_soxl_promotion_research_keeps_unavailable_targets_as_cash( _soxl_context(), candidate_identity=candidate, mandate_provenance=_soxl_mandate(candidate), - stop_loss_distances={symbol: 0.05 for symbol in _SOXL_ASSETS}, + stop_loss_distances={ + symbol: 0.05 for symbol in {"SOXL", "SOXX", "SCHD", "DGRO", "QQQ"} + }, drawdown_scalar=1.0, inputs_fresh=True, point_in_time_eligible_assets=frozenset({"SOXL", "SOXX", "SCHD", "DGRO", "QQQ"}), @@ -260,6 +265,9 @@ def test_soxl_promotion_research_keeps_unavailable_targets_as_cash( expected ) assert not {"BOXX", "QQQI"} & {position.symbol for position in result.decision.positions} + assert result.decision.diagnostics["promotion_research_ineligible_assets_to_cash"] == ( + ("BOXX",) if fallback_symbol == "QQQ" else ("BOXX", "QQQI") + ) engine.assess.assert_called_once()