Add semiconductor runtime indicators (#9)

Author: Pigbibi

src/quant_platform_kit/ibkr/connection.py

@@ -67,5 +67,14 @@ def connect_ib(
         ib_factory = IB
 
     ib = ib_factory()
-    ib.connect(host, port, clientId=client_id, timeout=timeout)
+    try:
+        ib.connect(host, port, clientId=client_id, timeout=timeout)
+    except TimeoutError as exc:
+        raise TimeoutError(
+            "IBKR API handshake timed out after TCP preflight succeeded "
+            f"for {host}:{port} clientId={client_id}. "
+            "Check that IB Gateway/TWS is fully logged in, API access is enabled, "
+            "the paper/live port matches the session, no login/API prompt is blocking, "
+            "and the client ID is not already stuck in another session."
+        ) from exc
     return ib

src/quant_platform_kit/ibkr/runtime_inputs.py

@@ -87,7 +87,7 @@ def build_semiconductor_rotation_indicators(
         historical_close_loader(
             ib,
             "SOXX",
-            duration="20 D",
+            duration=f"{effective_lookback} D",
             bar_size="1 day",
         )
     )
@@ -96,17 +96,23 @@ def build_semiconductor_rotation_indicators(
 
     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 soxx_close.empty:
+    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,
         },
     }
 

src/quant_platform_kit/longbridge/market_data.py

@@ -22,22 +22,28 @@ def calculate_rotation_indicators(
 
     effective_lookback = lookback if lookback is not None else max(220, trend_window + 20)
     soxl_bars = q_ctx.candlesticks("SOXL.US", Period.Day, effective_lookback, AdjustType.ForwardAdjust)
-    soxx_bars = q_ctx.candlesticks("SOXX.US", Period.Day, 20, AdjustType.ForwardAdjust)
+    soxx_bars = q_ctx.candlesticks("SOXX.US", Period.Day, effective_lookback, AdjustType.ForwardAdjust)
     if not soxl_bars or not soxx_bars:
         return None
 
     df_soxl = pd.DataFrame([{"close": float(k.close)} for k in soxl_bars])
     df_soxx = pd.DataFrame([float(k.close) for k in soxx_bars], columns=["close"])
-    if len(df_soxl) < trend_window or len(df_soxx) < 1:
+    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]),
         },
     }

tests/test_ibkr_connection.py

@@ -50,6 +50,26 @@ def fake_socket_create_connection(address, timeout):
         self.assertIsInstance(ib, FakeIB)
         self.assertEqual(observed["args"], ("127.0.0.1", 4001, 9, 20))
 
+    def test_connect_ib_wraps_api_handshake_timeout(self) -> None:
+        class FakeConnection:
+            def close(self):
+                pass
+
+        def fake_socket_create_connection(_address, _timeout):
+            return FakeConnection()
+
+        class FakeIB:
+            def connect(self, host, port, clientId, timeout):
+                raise TimeoutError()
+
+        with self.assertRaisesRegex(TimeoutError, "API handshake timed out"):
+            connect_ib(
+                "10.0.0.8",
+                4002,
+                9,
+                socket_create_connection=fake_socket_create_connection,
+                ib_factory=FakeIB,
+            )
 
     def test_probe_tcp_endpoint_wraps_timeout(self) -> None:
         def fake_socket_create_connection(_address, _timeout):

tests/test_ibkr_runtime_inputs.py

@@ -81,7 +81,7 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
             if symbol == "SOXL":
                 return [100.0 + idx for idx in range(170)]
             if symbol == "SOXX":
-                return [200.0 + idx for idx in range(20)]
+                return [200.0 + idx for idx in range(170)]
             raise AssertionError(symbol)
 
         indicators = build_semiconductor_rotation_indicators(
@@ -91,20 +91,29 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
         )
 
         self.assertEqual(observed[0], ("SOXL", "220 D", "1 day"))
-        self.assertEqual(observed[1], ("SOXX", "20 D", "1 day"))
+        self.assertEqual(observed[1], ("SOXX", "220 D", "1 day"))
         self.assertEqual(indicators["soxl"]["price"], 269.0)
         self.assertAlmostEqual(
             indicators["soxl"]["ma_trend"],
             sum(100.0 + idx for idx in range(20, 170)) / 150,
         )
-        self.assertEqual(indicators["soxx"]["price"], 219.0)
+        self.assertEqual(indicators["soxx"]["price"], 369.0)
+        self.assertAlmostEqual(
+            indicators["soxx"]["ma_trend"],
+            sum(200.0 + idx for idx in range(20, 170)) / 150,
+        )
+        self.assertAlmostEqual(
+            indicators["soxx"]["ma20"],
+            sum(200.0 + idx for idx in range(150, 170)) / 20,
+        )
+        self.assertGreater(indicators["soxx"]["ma20_slope"], 0.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":
                 return [100.0] * 170
             if symbol == "SOXX":
-                return [200.0] * 20
+                return [200.0] * 170
             raise AssertionError(symbol)
 
         payload = build_semiconductor_rotation_inputs(
@@ -116,6 +125,7 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
         self.assertEqual(set(payload), {"derived_indicators"})
         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)
 
     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

@@ -25,7 +25,7 @@ def quote(self, symbols):
     def candlesticks(self, symbol, period, count, adjust_type):
         if symbol == "SOXL.US":
             return [FakeBar(100 + i) for i in range(count)]
-        return [FakeBar(200.0) for _ in range(20)]
+        return [FakeBar(200.0 + i) for i in range(count)]
 
 
 class LongBridgeMarketDataTests(unittest.TestCase):
@@ -43,7 +43,9 @@ def test_calculate_rotation_indicators(self) -> None:
 
         self.assertIsNotNone(indicators)
         self.assertEqual(indicators["soxl"]["price"], 319.0)
-        self.assertEqual(indicators["soxx"]["price"], 200.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.assertGreater(indicators["soxx"]["ma20_slope"], 0.0)
 
 
 if __name__ == "__main__":