Skip to content

Add dynamic RSI semiconductor indicators #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 2 commits into from
May 7, 2026
Merged

Add dynamic RSI semiconductor indicators #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c82b316
Add dynamic RSI semiconductor indicators
Pigbibi May 7, 2026
ed129e5
Address dynamic RSI indicator review
Pigbibi May 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "quant-platform-kit"
version = "0.7.19"
version = "0.7.20"

@chatgpt-codex-connector chatgpt-codex-connector Bot May 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep runtime version constant aligned with project version

This bumps the package metadata to 0.7.20, but the exported quant_platform_kit.__version__ constant remains 0.7.19 in src/quant_platform_kit/__init__.py. That creates conflicting version sources (metadata vs runtime constant), which can break version-gated behavior, diagnostics, or compatibility checks that read __version__.

Useful? React with 👍 / 👎.

description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies."
readme = "README.md"
requires-python = ">=3.9"
Expand Down
2 changes: 1 addition & 1 deletion src/quant_platform_kit/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""QuantPlatformKit public package surface."""

__version__ = "0.7.19"
__version__ = "0.7.20"

from .common.models import (
ExecutionReport,
Expand Down
37 changes: 36 additions & 1 deletion src/quant_platform_kit/common/runtime_inputs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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 {
Expand All @@ -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,
Expand All @@ -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,
)
}

Expand Down
10 changes: 9 additions & 1 deletion src/quant_platform_kit/ibkr/runtime_inputs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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,
)


Expand All @@ -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,
)
}
4 changes: 3 additions & 1 deletion src/quant_platform_kit/longbridge/market_data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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,
)
17 changes: 15 additions & 2 deletions tests/test_ibkr_runtime_inputs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"],
Expand All @@ -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"])

Expand All @@ -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)],
Expand All @@ -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":
Expand All @@ -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"):
Expand Down
7 changes: 4 additions & 3 deletions tests/test_longbridge_market_data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"])

Expand Down