Add SOXL overlay runtime indicators

src/quant_platform_kit/common/runtime_inputs.py

@@ -28,6 +28,20 @@ def _normalize_numeric_history(
     return normalized.astype(float)
 
 
+def _compute_rsi(close: pd.Series, *, window: int = 14) -> pd.Series:
+    delta = close.diff()
+    gains = delta.clip(lower=0.0)
+    losses = -delta.clip(upper=0.0)
+    avg_gain = gains.ewm(alpha=1 / window, adjust=False, min_periods=window).mean()
+    avg_loss = losses.ewm(alpha=1 / window, adjust=False, min_periods=window).mean()
+    rs = avg_gain / avg_loss.replace(0.0, pd.NA)
+    rsi = 100.0 - (100.0 / (1.0 + rs))
+    rsi = rsi.mask((avg_loss == 0.0) & (avg_gain > 0.0), 100.0)
+    rsi = rsi.mask((avg_gain == 0.0) & (avg_loss > 0.0), 0.0)
+    rsi = rsi.mask((avg_gain == 0.0) & (avg_loss == 0.0), 50.0)
+    return rsi
+
+
 def build_semiconductor_rotation_indicators_from_history(
     *,
     soxl_history: Iterable[float] | pd.Series,
@@ -47,6 +61,9 @@ 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_bb_mid = float(soxx_close.rolling(20).mean().iloc[-1])
+    soxx_bb_std = float(soxx_close.rolling(20).std(ddof=0).iloc[-1])
     return {
         "soxl": {
             "price": float(soxl_close.iloc[-1]),
@@ -57,6 +74,10 @@ def build_semiconductor_rotation_indicators_from_history(
             "ma_trend": soxx_ma_trend,
             "ma20": soxx_ma20,
             "ma20_slope": soxx_ma20_slope,
+            "rsi14": soxx_rsi14,
+            "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,
         },
     }
 

src/quant_platform_kit/longbridge/market_data.py

@@ -4,6 +4,10 @@
 
 import pandas as pd
 
+from quant_platform_kit.common.runtime_inputs import (
+    build_semiconductor_rotation_indicators_from_history,
+)
+
 
 def fetch_last_price(q_ctx: Any, symbol: str) -> float | None:
     quotes = q_ctx.quote([symbol])
@@ -31,19 +35,8 @@ def calculate_rotation_indicators(
     if len(df_soxl) < trend_window or len(df_soxx) < trend_window:
         return None
 
-    df_soxl["ma_trend"] = df_soxl["close"].rolling(trend_window).mean()
-    df_soxx["ma_trend"] = df_soxx["close"].rolling(trend_window).mean()
-    df_soxx["ma20"] = df_soxx["close"].rolling(20).mean()
-    df_soxx["ma20_slope"] = df_soxx["ma20"].diff()
-    return {
-        "soxl": {
-            "price": float(df_soxl["close"].iloc[-1]),
-            "ma_trend": float(df_soxl["ma_trend"].iloc[-1]),
-        },
-        "soxx": {
-            "price": float(df_soxx["close"].iloc[-1]),
-            "ma_trend": float(df_soxx["ma_trend"].iloc[-1]),
-            "ma20": float(df_soxx["ma20"].iloc[-1]),
-            "ma20_slope": float(df_soxx["ma20_slope"].iloc[-1]),
-        },
-    }
+    return build_semiconductor_rotation_indicators_from_history(
+        soxl_history=df_soxl["close"],
+        soxx_history=df_soxx["close"],
+        trend_ma_window=trend_window,
+    )

tests/test_ibkr_runtime_inputs.py

@@ -111,6 +111,9 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
             sum(200.0 + idx for idx in range(150, 170)) / 20,
         )
         self.assertGreater(indicators["soxx"]["ma20_slope"], 0.0)
+        self.assertEqual(indicators["soxx"]["rsi14"], 100.0)
+        self.assertGreater(indicators["soxx"]["bb_upper"], indicators["soxx"]["price"])
+        self.assertLess(indicators["soxx"]["bb_lower"], indicators["soxx"]["price"])
 
     def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) -> None:
         indicators = build_semiconductor_rotation_indicators_from_history(
@@ -129,6 +132,8 @@ def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) -
             indicators["soxx"]["ma_trend"],
             sum(200.0 + idx for idx in range(30, 170)) / 140,
         )
+        self.assertEqual(indicators["soxx"]["rsi14"], 100.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)],
             soxx_history=[200.0 + idx for idx in range(170)],
@@ -155,6 +160,7 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
         self.assertEqual(payload["derived_indicators"]["soxl"]["price"], 100.0)
         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)
 
     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

@@ -46,6 +46,9 @@ def test_calculate_rotation_indicators(self) -> None:
         self.assertEqual(indicators["soxx"]["price"], 419.0)
         self.assertAlmostEqual(indicators["soxx"]["ma20"], sum(200.0 + i for i in range(200, 220)) / 20)
         self.assertGreater(indicators["soxx"]["ma20_slope"], 0.0)
+        self.assertEqual(indicators["soxx"]["rsi14"], 100.0)
+        self.assertGreater(indicators["soxx"]["bb_upper"], indicators["soxx"]["price"])
+        self.assertLess(indicators["soxx"]["bb_lower"], indicators["soxx"]["price"])
 
 
 if __name__ == "__main__":