From b6d406c1a95eebc9230af50f6ff80e9d939678a5 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 22 Apr 2026 00:51:15 +0800 Subject: [PATCH] Extract generic semiconductor runtime inputs --- src/quant_platform_kit/__init__.py | 4 ++ .../common/runtime_inputs.py | 61 +++++++++++++++++++ src/quant_platform_kit/ibkr/runtime_inputs.py | 58 ++++++------------ tests/test_ibkr_runtime_inputs.py | 29 +++++++++ 4 files changed, 112 insertions(+), 40 deletions(-) diff --git a/src/quant_platform_kit/__init__.py b/src/quant_platform_kit/__init__.py index b046f6e..82f40fe 100644 --- a/src/quant_platform_kit/__init__.py +++ b/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", diff --git a/src/quant_platform_kit/common/runtime_inputs.py b/src/quant_platform_kit/common/runtime_inputs.py index 87b55dd..22d181c 100644 --- a/src/quant_platform_kit/common/runtime_inputs.py +++ b/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, *, diff --git a/src/quant_platform_kit/ibkr/runtime_inputs.py b/src/quant_platform_kit/ibkr/runtime_inputs.py index f861cd6..5846428 100644 --- a/src/quant_platform_kit/ibkr/runtime_inputs.py +++ b/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( diff --git a/tests/test_ibkr_runtime_inputs.py b/tests/test_ibkr_runtime_inputs.py index 47ffc2a..b039d2b 100644 --- a/tests/test_ibkr_runtime_inputs.py +++ b/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":