Extract generic semiconductor runtime inputs

src/quant_platform_kit/__init__.py

@@ -48,6 +48,8 @@
 from .common.runtime_inputs import (
     build_account_state_from_portfolio_snapshot,
     build_portfolio_snapshot_from_account_state,
+    build_semiconductor_rotation_indicators_from_history,
+    build_semiconductor_rotation_inputs_from_history,
     build_strategy_evaluation_inputs,
 )
 from .common.strategy_plugins import (
@@ -128,6 +130,8 @@
     "build_allocation_payload",
     "build_account_state_from_portfolio_snapshot",
     "build_portfolio_snapshot_from_account_state",
+    "build_semiconductor_rotation_indicators_from_history",
+    "build_semiconductor_rotation_inputs_from_history",
     "build_strategy_evaluation_inputs",
     "build_value_target_portfolio_inputs_from_account_state",
     "build_value_target_portfolio_inputs_from_snapshot",

src/quant_platform_kit/common/runtime_inputs.py

@@ -4,6 +4,8 @@
 from datetime import datetime, timezone
 from typing import Any, Callable
 
+import pandas as pd
+
 from .models import PortfolioSnapshot, Position
 
 
@@ -15,6 +17,65 @@ def _normalize_symbols(strategy_symbols: Iterable[str]) -> tuple[str, ...]:
     )
 
 
+def _normalize_numeric_history(
+    history: Iterable[float] | pd.Series,
+    *,
+    label: str,
+) -> pd.Series:
+    normalized = pd.to_numeric(pd.Series(history), errors="coerce").dropna()
+    if normalized.empty:
+        raise ValueError(f"Semiconductor rotation inputs require non-empty {label} history")
+    return normalized.astype(float)
+
+
+def build_semiconductor_rotation_indicators_from_history(
+    *,
+    soxl_history: Iterable[float] | pd.Series,
+    soxx_history: Iterable[float] | pd.Series,
+    trend_ma_window: int = 140,
+) -> dict[str, dict[str, float]]:
+    window = int(trend_ma_window)
+    if window <= 0:
+        raise ValueError("trend_ma_window must be positive")
+
+    soxl_close = _normalize_numeric_history(soxl_history, label="SOXL")
+    soxx_close = _normalize_numeric_history(soxx_history, label="SOXX")
+    if len(soxl_close) < window or len(soxx_close) < window:
+        raise ValueError("Semiconductor rotation inputs require sufficient SOXL/SOXX history")
+
+    soxl_ma_trend = float(soxl_close.rolling(window).mean().iloc[-1])
+    soxx_ma_trend = float(soxx_close.rolling(window).mean().iloc[-1])
+    soxx_ma20 = float(soxx_close.rolling(20).mean().iloc[-1])
+    soxx_ma20_slope = float(soxx_close.rolling(20).mean().diff().iloc[-1])
+    return {
+        "soxl": {
+            "price": float(soxl_close.iloc[-1]),
+            "ma_trend": soxl_ma_trend,
+        },
+        "soxx": {
+            "price": float(soxx_close.iloc[-1]),
+            "ma_trend": soxx_ma_trend,
+            "ma20": soxx_ma20,
+            "ma20_slope": soxx_ma20_slope,
+        },
+    }
+
+
+def build_semiconductor_rotation_inputs_from_history(
+    *,
+    soxl_history: Iterable[float] | pd.Series,
+    soxx_history: Iterable[float] | pd.Series,
+    trend_ma_window: int = 140,
+) -> dict[str, dict[str, dict[str, float]]]:
+    return {
+        "derived_indicators": build_semiconductor_rotation_indicators_from_history(
+            soxl_history=soxl_history,
+            soxx_history=soxx_history,
+            trend_ma_window=trend_ma_window,
+        )
+    }
+
+
 def build_account_state_from_portfolio_snapshot(
     snapshot: Any,
     *,

src/quant_platform_kit/ibkr/runtime_inputs.py

@@ -3,14 +3,15 @@
 from collections.abc import Callable
 from typing import Any
 
-import pandas as pd
-
 from quant_platform_kit.strategy_contracts import (
     StrategyEntrypoint,
     StrategyRuntimeAdapter,
     build_strategy_context_from_available_inputs,
     build_strategy_evaluation_inputs,
 )
+from quant_platform_kit.common.runtime_inputs import (
+    build_semiconductor_rotation_indicators_from_history,
+)
 
 
 def build_market_history_inputs(
@@ -75,46 +76,23 @@ def build_semiconductor_rotation_indicators(
     lookback_buffer: int = 20,
 ) -> dict[str, dict[str, float]]:
     effective_lookback = max(220, int(trend_ma_window) + int(lookback_buffer))
-    soxl_series = pd.Series(
-        historical_close_loader(
-            ib,
-            "SOXL",
-            duration=f"{effective_lookback} D",
-            bar_size="1 day",
-        )
+    soxl_history = historical_close_loader(
+        ib,
+        "SOXL",
+        duration=f"{effective_lookback} D",
+        bar_size="1 day",
     )
-    soxx_series = pd.Series(
-        historical_close_loader(
-            ib,
-            "SOXX",
-            duration=f"{effective_lookback} D",
-            bar_size="1 day",
-        )
+    soxx_history = historical_close_loader(
+        ib,
+        "SOXX",
+        duration=f"{effective_lookback} D",
+        bar_size="1 day",
+    )
+    return build_semiconductor_rotation_indicators_from_history(
+        soxl_history=soxl_history,
+        soxx_history=soxx_history,
+        trend_ma_window=trend_ma_window,
     )
-    if soxl_series.empty or soxx_series.empty:
-        raise ValueError("IBKR semiconductor runtime requires SOXL/SOXX price history")
-
-    soxl_close = pd.to_numeric(soxl_series, errors="coerce").dropna()
-    soxx_close = pd.to_numeric(soxx_series, errors="coerce").dropna()
-    if len(soxl_close) < int(trend_ma_window) or len(soxx_close) < int(trend_ma_window):
-        raise ValueError("IBKR semiconductor runtime requires sufficient SOXL/SOXX history")
-
-    soxl_ma_trend = float(soxl_close.rolling(int(trend_ma_window)).mean().iloc[-1])
-    soxx_ma_trend = float(soxx_close.rolling(int(trend_ma_window)).mean().iloc[-1])
-    soxx_ma20 = float(soxx_close.rolling(20).mean().iloc[-1])
-    soxx_ma20_slope = float(soxx_close.rolling(20).mean().diff().iloc[-1])
-    return {
-        "soxl": {
-            "price": float(soxl_close.iloc[-1]),
-            "ma_trend": soxl_ma_trend,
-        },
-        "soxx": {
-            "price": float(soxx_close.iloc[-1]),
-            "ma_trend": soxx_ma_trend,
-            "ma20": soxx_ma20,
-            "ma20_slope": soxx_ma20_slope,
-        },
-    }
 
 
 def build_semiconductor_rotation_inputs(

tests/test_ibkr_runtime_inputs.py

@@ -3,6 +3,10 @@
 import unittest
 
 from quant_platform_kit.strategy_contracts import StrategyManifest, StrategyRuntimeAdapter
+from quant_platform_kit import (
+    build_semiconductor_rotation_indicators_from_history,
+    build_semiconductor_rotation_inputs_from_history,
+)
 from quant_platform_kit.ibkr.runtime_inputs import (
     build_benchmark_history_inputs,
     build_ibkr_strategy_context,
@@ -108,6 +112,31 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
         )
         self.assertGreater(indicators["soxx"]["ma20_slope"], 0.0)
 
+    def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) -> None:
+        indicators = build_semiconductor_rotation_indicators_from_history(
+            soxl_history=[100.0 + idx for idx in range(170)],
+            soxx_history=[200.0 + idx for idx in range(170)],
+            trend_ma_window=140,
+        )
+
+        self.assertEqual(indicators["soxl"]["price"], 269.0)
+        self.assertAlmostEqual(
+            indicators["soxl"]["ma_trend"],
+            sum(100.0 + idx for idx in range(30, 170)) / 140,
+        )
+        self.assertEqual(indicators["soxx"]["price"], 369.0)
+        self.assertAlmostEqual(
+            indicators["soxx"]["ma_trend"],
+            sum(200.0 + idx for idx in range(30, 170)) / 140,
+        )
+        wrapped = build_semiconductor_rotation_inputs_from_history(
+            soxl_history=[100.0 + idx for idx in range(170)],
+            soxx_history=[200.0 + idx for idx in range(170)],
+            trend_ma_window=140,
+        )
+        self.assertEqual(set(wrapped), {"derived_indicators"})
+        self.assertEqual(wrapped["derived_indicators"]["soxl"]["price"], 269.0)
+
     def test_build_semiconductor_rotation_inputs_wraps_derived_indicators(self) -> None:
         def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
             if symbol == "SOXL":