Add SOXX realized volatility indicator (#35)

Author: Pigbibi

src/quant_platform_kit/common/runtime_inputs.py

@@ -51,6 +51,15 @@ def _std(values: Iterable[float]) -> float:
     return float(sqrt(variance))
 
 
+def _sample_std(values: Iterable[float]) -> float:
+    values = tuple(values)
+    if len(values) < 2:
+        raise ValueError("sample std requires at least two values")
+    mean_value = _mean(values)
+    variance = sum((value - mean_value) ** 2 for value in values) / (len(values) - 1)
+    return float(sqrt(variance))
+
+
 def _tail_mean(values: tuple[float, ...], window: int) -> float:
     if len(values) < window:
         raise ValueError("insufficient history for rolling mean")
@@ -63,6 +72,18 @@ def _tail_std(values: tuple[float, ...], window: int) -> float:
     return _std(values[-window:])
 
 
+def _tail_realized_volatility(values: tuple[float, ...], window: int) -> float:
+    if len(values) < window + 1:
+        raise ValueError("insufficient history for realized volatility")
+    tail_values = values[-(window + 1):]
+    returns: list[float] = []
+    for previous, current in zip(tail_values, tail_values[1:]):
+        if previous == 0.0:
+            raise ValueError("realized volatility requires non-zero prices")
+        returns.append((current / previous) - 1.0)
+    return float(_sample_std(returns) * sqrt(252))
+
+
 def _compute_rsi(values: tuple[float, ...], *, window: int = 14) -> tuple[float, ...]:
     if len(values) < window + 1:
         raise ValueError("insufficient history for RSI")
@@ -161,6 +182,7 @@ def build_semiconductor_rotation_indicators_from_history(
     )
     soxx_bb_mid = _tail_mean(soxx_close, 20)
     soxx_bb_std = _tail_std(soxx_close, 20)
+    soxx_realized_volatility_20 = _tail_realized_volatility(soxx_close, 20)
     return {
         "soxl": {
             "price": float(soxl_close[-1]),
@@ -176,6 +198,8 @@ def build_semiconductor_rotation_indicators_from_history(
             "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,
+            "realized_volatility": soxx_realized_volatility_20,
+            "realized_volatility_20": soxx_realized_volatility_20,
         },
     }
 

tests/test_ibkr_runtime_inputs.py

@@ -115,6 +115,11 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
         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"])
+        self.assertIn("realized_volatility_20", indicators["soxx"])
+        self.assertEqual(
+            indicators["soxx"]["realized_volatility"],
+            indicators["soxx"]["realized_volatility_20"],
+        )
 
     def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) -> None:
         indicators = build_semiconductor_rotation_indicators_from_history(
@@ -136,6 +141,7 @@ def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) -
         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.assertIn("realized_volatility_20", indicators["soxx"])
         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)],
@@ -174,6 +180,7 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
         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)
+        self.assertEqual(payload["derived_indicators"]["soxx"]["realized_volatility_20"], 0.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

@@ -79,6 +79,11 @@ def test_calculate_rotation_indicators(self) -> None:
         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"])
+        self.assertIn("realized_volatility_20", indicators["soxx"])
+        self.assertEqual(
+            indicators["soxx"]["realized_volatility"],
+            indicators["soxx"]["realized_volatility_20"],
+        )
 
 
 if __name__ == "__main__":