diff --git a/src/quant_platform_kit/common/runtime_inputs.py b/src/quant_platform_kit/common/runtime_inputs.py index 22d181c..7eae9a8 100644 --- a/src/quant_platform_kit/common/runtime_inputs.py +++ b/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, }, } diff --git a/src/quant_platform_kit/longbridge/market_data.py b/src/quant_platform_kit/longbridge/market_data.py index 3dd27c9..51868bf 100644 --- a/src/quant_platform_kit/longbridge/market_data.py +++ b/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, + ) diff --git a/tests/test_ibkr_runtime_inputs.py b/tests/test_ibkr_runtime_inputs.py index b039d2b..ae2e543 100644 --- a/tests/test_ibkr_runtime_inputs.py +++ b/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"): diff --git a/tests/test_longbridge_market_data.py b/tests/test_longbridge_market_data.py index 427e054..b15a381 100644 --- a/tests/test_longbridge_market_data.py +++ b/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__":