feat: gate SOXL pre-inception asset eligibility

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,33 @@ 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")
+        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")
@@ -970,6 +999,8 @@ 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"]
         decision = StrategyDecision(
             positions=tuple(
                 PositionTarget(
@@ -985,6 +1016,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:

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,136 @@ 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.15, "QQQ": 0.021428571428571432}),
+        (None, {"SOXL": 0.15}),
+    ),
+)
+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", "SOXX", "SCHD", "DGRO", "QQQ"}
+            },
+            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}
+    assert result.decision.diagnostics["promotion_research_ineligible_assets_to_cash"] == (
+        ("BOXX",) if fallback_symbol == "QQQ" else ("BOXX", "QQQI")
+    )
+    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),

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)