Add dynamic RSI semiconductor indicators (#26)

* Add dynamic RSI semiconductor indicators

* Address dynamic RSI indicator review

Author: Pigbibi

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"

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,

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=min(rsi_quantile_window, 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,
         )
     }
 

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,12 +107,14 @@ 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(
             ib,
             historical_close_loader,
             trend_ma_window=trend_ma_window,
             lookback_buffer=lookback_buffer,
+            dynamic_rsi_quantile_window=dynamic_rsi_quantile_window,
         )
     }

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,
     )

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)],
@@ -142,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":
@@ -161,6 +173,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"):

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"])