Enable SOXL market signal runtime (#117)

Author: Pigbibi

market_signal_runtime.py

@@ -9,11 +9,17 @@
     MARKET_SIGNAL_REFERENCE_CONSUMPTION_AUDIT,
     MARKET_SIGNAL_REFERENCE_PLATFORM_HANDOFF,
     MARKET_SIGNAL_REFERENCE_PLATFORM_HANDOFF_INDEX,
+    SOXL_SOXX_TREND_INCOME_MARKET_SIGNAL_CONSUMER,
     extract_consumer_market_signal_inputs_from_reference,
 )
 
 
 IBIT_SMART_DCA_PROFILE = "ibit_smart_dca"
+SOXL_SOXX_TREND_INCOME_PROFILE = "soxl_soxx_trend_income"
+MARKET_SIGNAL_CONSUMER_BY_PROFILE = {
+    IBIT_SMART_DCA_PROFILE: IBIT_SMART_DCA_MARKET_SIGNAL_CONSUMER,
+    SOXL_SOXX_TREND_INCOME_PROFILE: SOXL_SOXX_TREND_INCOME_MARKET_SIGNAL_CONSUMER,
+}
 DEFAULT_MARKET_SIGNAL_CACHE_DIR = "/tmp/quant-platform-market-signals"
 
 
@@ -26,21 +32,28 @@ def resolve_external_market_signal_inputs(
     logger: Callable[[str], None] = print,
     client_factory: Any = None,
 ) -> dict[str, Any]:
-    if str(strategy_profile or "").strip().lower() != IBIT_SMART_DCA_PROFILE:
+    normalized_profile = str(strategy_profile or "").strip().lower()
+    consumer = MARKET_SIGNAL_CONSUMER_BY_PROFILE.get(normalized_profile)
+    if consumer is None:
         return {}
     if "derived_indicators" not in {str(item) for item in available_inputs or ()}:
         return {}
 
     reference_type, reference = _market_signal_reference(runtime_settings)
     if reference is None:
         if bool(getattr(runtime_settings, "market_signal_required", False)):
-            raise RuntimeError("IBIT external market signal is required but no signal reference is configured")
-        return {"derived_indicators": {}}
+            raise RuntimeError(
+                f"{normalized_profile} external market signal is required "
+                "but no signal reference is configured"
+            )
+        if normalized_profile == IBIT_SMART_DCA_PROFILE:
+            return {"derived_indicators": {}}
+        return {}
 
     market_inputs, metadata = extract_consumer_market_signal_inputs_from_reference(
         reference,
         reference_type=reference_type,
-        consumer=IBIT_SMART_DCA_MARKET_SIGNAL_CONSUMER,
+        consumer=consumer,
         cache_dir=_market_signal_cache_dir(runtime_settings),
         as_of=_market_signal_as_of(as_of),
         client_factory=client_factory,

tests/test_market_signal_runtime.py

@@ -8,12 +8,12 @@
 import market_signal_runtime
 
 
-def test_non_ibit_profile_does_not_load_market_signal():
+def test_unsupported_profile_does_not_load_market_signal():
     settings = SimpleNamespace(market_signal_required=True)
 
     assert (
         market_signal_runtime.resolve_external_market_signal_inputs(
-            strategy_profile="soxl_soxx_trend_income",
+            strategy_profile="tqqq_growth_income",
             available_inputs={"derived_indicators"},
             runtime_settings=settings,
         )
@@ -42,6 +42,30 @@ def test_ibit_required_reference_missing_raises():
         )
 
 
+def test_soxl_without_reference_preserves_legacy_inputs():
+    settings = SimpleNamespace(market_signal_required=False)
+
+    assert market_signal_runtime.resolve_external_market_signal_inputs(
+        strategy_profile="soxl_soxx_trend_income",
+        available_inputs={"derived_indicators"},
+        runtime_settings=settings,
+    ) == {}
+
+
+def test_soxl_required_reference_missing_raises():
+    settings = SimpleNamespace(market_signal_required=True)
+
+    with pytest.raises(
+        RuntimeError,
+        match="soxl_soxx_trend_income external market signal is required",
+    ):
+        market_signal_runtime.resolve_external_market_signal_inputs(
+            strategy_profile="soxl_soxx_trend_income",
+            available_inputs={"derived_indicators"},
+            runtime_settings=settings,
+        )
+
+
 def test_ibit_handoff_index_reference_is_extracted(monkeypatch, tmp_path):
     calls: dict[str, object] = {}
 
@@ -103,3 +127,70 @@ def fake_extract(
         "last_valid",
         5,
     )
+
+
+def test_soxl_handoff_index_reference_is_extracted(monkeypatch, tmp_path):
+    calls: dict[str, object] = {}
+
+    def fake_extract(
+        reference,
+        *,
+        reference_type,
+        consumer,
+        cache_dir,
+        as_of,
+        client_factory=None,
+        fallback_mode=None,
+        fallback_max_stale_days=None,
+    ):
+        calls["extract"] = (
+            reference,
+            reference_type,
+            consumer,
+            cache_dir,
+            as_of,
+            client_factory,
+            fallback_mode,
+            fallback_max_stale_days,
+        )
+        return {
+            "derived_indicators": {
+                "SOXL": {"price": 25.0, "ma_trend": 24.0},
+                "SOXX": {"price": 400.0, "ma_trend": 390.0},
+            }
+        }, {
+            "reference_type": reference_type,
+            "source_uri": reference,
+            "materialized_count": 2,
+        }
+
+    monkeypatch.setattr(
+        market_signal_runtime,
+        "extract_consumer_market_signal_inputs_from_reference",
+        fake_extract,
+    )
+    settings = SimpleNamespace(
+        market_signal_handoff_index_uri="gs://signals/platform_handoffs/index.json",
+        market_signal_cache_dir=str(tmp_path),
+        market_signal_required=True,
+        market_signal_fallback_mode="none",
+    )
+
+    assert market_signal_runtime.resolve_external_market_signal_inputs(
+        strategy_profile="soxl_soxx_trend_income",
+        available_inputs={"derived_indicators"},
+        runtime_settings=settings,
+        as_of=datetime(2026, 6, 19, tzinfo=timezone.utc),
+        logger=lambda _message: None,
+        client_factory=object,
+    )["derived_indicators"]["SOXL"]["price"] == 25.0
+    assert calls["extract"] == (
+        "gs://signals/platform_handoffs/index.json",
+        "platform_handoff_index",
+        "us_equity:soxl_soxx_trend_income",
+        tmp_path,
+        "2026-06-19",
+        object,
+        "none",
+        3,
+    )

tests/test_strategy_registry.py

@@ -22,4 +22,3 @@ def test_profile_status_matrix_reports_firstrade_without_bridge_metadata():
     assert all(row["platform"] == FIRSTRADE_PLATFORM for row in rows)
     assert all("strategy_adapter_source_platform" not in row for row in rows)
     assert "global_etf_rotation" in get_supported_profiles_for_platform(FIRSTRADE_PLATFORM)
-