From c82b31618eee5caf039c18d7f18b319d4c48153c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 7 May 2026 18:53:51 +0800 Subject: [PATCH 1/2] Add dynamic RSI semiconductor indicators --- pyproject.toml | 2 +- .../common/runtime_inputs.py | 37 ++++++++++++++++++- src/quant_platform_kit/ibkr/runtime_inputs.py | 10 ++++- .../longbridge/market_data.py | 4 +- tests/test_ibkr_runtime_inputs.py | 7 +++- tests/test_longbridge_market_data.py | 7 ++-- 6 files changed, 58 insertions(+), 9 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index bce970d..4f8b554 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "quant-platform-kit" -version = "0.7.19" +version = "0.7.20" description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies." readme = "README.md" requires-python = ">=3.9" diff --git a/src/quant_platform_kit/common/runtime_inputs.py b/src/quant_platform_kit/common/runtime_inputs.py index 3a1971b..d927007 100644 --- a/src/quant_platform_kit/common/runtime_inputs.py +++ b/src/quant_platform_kit/common/runtime_inputs.py @@ -47,10 +47,19 @@ def build_semiconductor_rotation_indicators_from_history( soxl_history: Iterable[float] | pd.Series, soxx_history: Iterable[float] | pd.Series, trend_ma_window: int = 140, + dynamic_rsi_quantile_window: int = 252, + dynamic_rsi_quantile: float = 0.90, + dynamic_rsi_floor: float = 70.0, ) -> dict[str, dict[str, float]]: window = int(trend_ma_window) if window <= 0: raise ValueError("trend_ma_window must be positive") + rsi_quantile_window = int(dynamic_rsi_quantile_window) + if rsi_quantile_window <= 0: + raise ValueError("dynamic_rsi_quantile_window must be positive") + rsi_quantile = float(dynamic_rsi_quantile) + if not 0.0 < rsi_quantile < 1.0: + raise ValueError("dynamic_rsi_quantile must be between 0 and 1") soxl_close = _normalize_numeric_history(soxl_history, label="SOXL") soxx_close = _normalize_numeric_history(soxx_history, label="SOXX") @@ -61,7 +70,26 @@ def build_semiconductor_rotation_indicators_from_history( 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]) - soxx_rsi14 = float(_compute_rsi(soxx_close, window=14).iloc[-1]) + soxx_rsi_history = _compute_rsi(soxx_close, window=14) + soxx_rsi14 = float(soxx_rsi_history.iloc[-1]) + rsi_threshold_history = ( + soxx_rsi_history.rolling( + rsi_quantile_window, + min_periods=max(60, min(rsi_quantile_window, 126) // 2), + ) + .quantile(rsi_quantile) + .shift(1) + ) + soxx_dynamic_rsi_threshold = float( + max( + float(dynamic_rsi_floor), + ( + rsi_threshold_history.iloc[-1] + if pd.notna(rsi_threshold_history.iloc[-1]) + else float(dynamic_rsi_floor) + ), + ) + ) soxx_bb_mid = float(soxx_close.rolling(20).mean().iloc[-1]) soxx_bb_std = float(soxx_close.rolling(20).std(ddof=0).iloc[-1]) return { @@ -75,6 +103,7 @@ def build_semiconductor_rotation_indicators_from_history( "ma20": soxx_ma20, "ma20_slope": soxx_ma20_slope, "rsi14": soxx_rsi14, + "rsi14_dynamic_threshold": soxx_dynamic_rsi_threshold, "bb_mid": soxx_bb_mid, "bb_upper": soxx_bb_mid + 2.0 * soxx_bb_std, "bb_lower": soxx_bb_mid - 2.0 * soxx_bb_std, @@ -87,12 +116,18 @@ def build_semiconductor_rotation_inputs_from_history( soxl_history: Iterable[float] | pd.Series, soxx_history: Iterable[float] | pd.Series, trend_ma_window: int = 140, + dynamic_rsi_quantile_window: int = 252, + dynamic_rsi_quantile: float = 0.90, + dynamic_rsi_floor: float = 70.0, ) -> 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, + dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, + dynamic_rsi_quantile=dynamic_rsi_quantile, + dynamic_rsi_floor=dynamic_rsi_floor, ) } diff --git a/src/quant_platform_kit/ibkr/runtime_inputs.py b/src/quant_platform_kit/ibkr/runtime_inputs.py index 5846428..7e91d68 100644 --- a/src/quant_platform_kit/ibkr/runtime_inputs.py +++ b/src/quant_platform_kit/ibkr/runtime_inputs.py @@ -74,8 +74,13 @@ def build_semiconductor_rotation_indicators( *, trend_ma_window: int = 140, lookback_buffer: int = 20, + dynamic_rsi_quantile_window: int = 252, ) -> dict[str, dict[str, float]]: - effective_lookback = max(220, int(trend_ma_window) + int(lookback_buffer)) + effective_lookback = max( + 420, + int(trend_ma_window) + int(lookback_buffer), + int(dynamic_rsi_quantile_window) + int(lookback_buffer) + 90, + ) soxl_history = historical_close_loader( ib, "SOXL", @@ -92,6 +97,7 @@ def build_semiconductor_rotation_indicators( soxl_history=soxl_history, soxx_history=soxx_history, trend_ma_window=trend_ma_window, + dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, ) @@ -101,6 +107,7 @@ def build_semiconductor_rotation_inputs( *, trend_ma_window: int = 140, lookback_buffer: int = 20, + dynamic_rsi_quantile_window: int = 252, ) -> dict[str, dict[str, dict[str, float]]]: return { "derived_indicators": build_semiconductor_rotation_indicators( @@ -108,5 +115,6 @@ def build_semiconductor_rotation_inputs( historical_close_loader, trend_ma_window=trend_ma_window, lookback_buffer=lookback_buffer, + dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, ) } diff --git a/src/quant_platform_kit/longbridge/market_data.py b/src/quant_platform_kit/longbridge/market_data.py index 51868bf..30f8462 100644 --- a/src/quant_platform_kit/longbridge/market_data.py +++ b/src/quant_platform_kit/longbridge/market_data.py @@ -21,10 +21,11 @@ def calculate_rotation_indicators( *, trend_window: int, lookback: int | None = None, + dynamic_rsi_quantile_window: int = 252, ) -> dict[str, dict[str, float]] | None: from longport.openapi import AdjustType, Period - effective_lookback = lookback if lookback is not None else max(220, trend_window + 20) + effective_lookback = lookback if lookback is not None else max(280, trend_window + 20, dynamic_rsi_quantile_window + 28) soxl_bars = q_ctx.candlesticks("SOXL.US", Period.Day, effective_lookback, AdjustType.ForwardAdjust) soxx_bars = q_ctx.candlesticks("SOXX.US", Period.Day, effective_lookback, AdjustType.ForwardAdjust) if not soxl_bars or not soxx_bars: @@ -39,4 +40,5 @@ def calculate_rotation_indicators( soxl_history=df_soxl["close"], soxx_history=df_soxx["close"], trend_ma_window=trend_window, + dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, ) diff --git a/tests/test_ibkr_runtime_inputs.py b/tests/test_ibkr_runtime_inputs.py index ae2e543..fd2ebf4 100644 --- a/tests/test_ibkr_runtime_inputs.py +++ b/tests/test_ibkr_runtime_inputs.py @@ -94,8 +94,8 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): trend_ma_window=140, ) - self.assertEqual(observed[0], ("SOXL", "220 D", "1 day")) - self.assertEqual(observed[1], ("SOXX", "220 D", "1 day")) + self.assertEqual(observed[0], ("SOXL", "420 D", "1 day")) + self.assertEqual(observed[1], ("SOXX", "420 D", "1 day")) self.assertEqual(indicators["soxl"]["price"], 269.0) self.assertAlmostEqual( indicators["soxl"]["ma_trend"], @@ -112,6 +112,7 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): ) self.assertGreater(indicators["soxx"]["ma20_slope"], 0.0) self.assertEqual(indicators["soxx"]["rsi14"], 100.0) + self.assertGreaterEqual(indicators["soxx"]["rsi14_dynamic_threshold"], 70.0) self.assertGreater(indicators["soxx"]["bb_upper"], indicators["soxx"]["price"]) self.assertLess(indicators["soxx"]["bb_lower"], indicators["soxx"]["price"]) @@ -133,6 +134,7 @@ def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) - sum(200.0 + idx for idx in range(30, 170)) / 140, ) self.assertEqual(indicators["soxx"]["rsi14"], 100.0) + self.assertGreaterEqual(indicators["soxx"]["rsi14_dynamic_threshold"], 70.0) self.assertGreater(indicators["soxx"]["bb_upper"], indicators["soxx"]["price"]) wrapped = build_semiconductor_rotation_inputs_from_history( soxl_history=[100.0 + idx for idx in range(170)], @@ -161,6 +163,7 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): self.assertEqual(payload["derived_indicators"]["soxx"]["price"], 200.0) self.assertEqual(payload["derived_indicators"]["soxx"]["ma20"], 200.0) self.assertEqual(payload["derived_indicators"]["soxx"]["rsi14"], 50.0) + self.assertEqual(payload["derived_indicators"]["soxx"]["rsi14_dynamic_threshold"], 70.0) def test_build_semiconductor_rotation_indicators_requires_sufficient_history(self) -> None: def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): diff --git a/tests/test_longbridge_market_data.py b/tests/test_longbridge_market_data.py index b15a381..b5fd002 100644 --- a/tests/test_longbridge_market_data.py +++ b/tests/test_longbridge_market_data.py @@ -42,11 +42,12 @@ def test_calculate_rotation_indicators(self) -> None: indicators = calculate_rotation_indicators(FakeQuoteContext(), trend_window=150) self.assertIsNotNone(indicators) - self.assertEqual(indicators["soxl"]["price"], 319.0) - self.assertEqual(indicators["soxx"]["price"], 419.0) - self.assertAlmostEqual(indicators["soxx"]["ma20"], sum(200.0 + i for i in range(200, 220)) / 20) + self.assertEqual(indicators["soxl"]["price"], 379.0) + self.assertEqual(indicators["soxx"]["price"], 479.0) + self.assertAlmostEqual(indicators["soxx"]["ma20"], sum(200.0 + i for i in range(260, 280)) / 20) self.assertGreater(indicators["soxx"]["ma20_slope"], 0.0) self.assertEqual(indicators["soxx"]["rsi14"], 100.0) + self.assertGreaterEqual(indicators["soxx"]["rsi14_dynamic_threshold"], 70.0) self.assertGreater(indicators["soxx"]["bb_upper"], indicators["soxx"]["price"]) self.assertLess(indicators["soxx"]["bb_lower"], indicators["soxx"]["price"]) From ed129e5967e5549900b3762750db18483a2d4dad Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 7 May 2026 19:27:46 +0800 Subject: [PATCH 2/2] Address dynamic RSI indicator review --- src/quant_platform_kit/__init__.py | 2 +- src/quant_platform_kit/common/runtime_inputs.py | 2 +- tests/test_ibkr_runtime_inputs.py | 10 ++++++++++ 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/__init__.py b/src/quant_platform_kit/__init__.py index 82f40fe..a20694f 100644 --- a/src/quant_platform_kit/__init__.py +++ b/src/quant_platform_kit/__init__.py @@ -1,6 +1,6 @@ """QuantPlatformKit public package surface.""" -__version__ = "0.7.19" +__version__ = "0.7.20" from .common.models import ( ExecutionReport, diff --git a/src/quant_platform_kit/common/runtime_inputs.py b/src/quant_platform_kit/common/runtime_inputs.py index d927007..06f6a5a 100644 --- a/src/quant_platform_kit/common/runtime_inputs.py +++ b/src/quant_platform_kit/common/runtime_inputs.py @@ -75,7 +75,7 @@ def build_semiconductor_rotation_indicators_from_history( rsi_threshold_history = ( soxx_rsi_history.rolling( rsi_quantile_window, - min_periods=max(60, min(rsi_quantile_window, 126) // 2), + min_periods=min(rsi_quantile_window, max(60, min(rsi_quantile_window, 126) // 2)), ) .quantile(rsi_quantile) .shift(1) diff --git a/tests/test_ibkr_runtime_inputs.py b/tests/test_ibkr_runtime_inputs.py index fd2ebf4..1d5cfb6 100644 --- a/tests/test_ibkr_runtime_inputs.py +++ b/tests/test_ibkr_runtime_inputs.py @@ -144,6 +144,16 @@ def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) - self.assertEqual(set(wrapped), {"derived_indicators"}) self.assertEqual(wrapped["derived_indicators"]["soxl"]["price"], 269.0) + def test_build_semiconductor_rotation_indicators_accepts_short_dynamic_rsi_window(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, + dynamic_rsi_quantile_window=20, + ) + + self.assertEqual(indicators["soxx"]["rsi14_dynamic_threshold"], 100.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":