From f8b3cc16138167be9f65d32cb77cfce8ad8b1dd0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 9 Jun 2026 14:50:27 +0800 Subject: [PATCH] Add SOXX dynamic volatility threshold indicators --- pyproject.toml | 2 +- setup.py | 2 +- src/quant_platform_kit/__init__.py | 2 +- .../common/runtime_inputs.py | 188 ++++++++++++++++-- src/quant_platform_kit/ibkr/runtime_inputs.py | 10 + .../longbridge/market_data.py | 26 ++- tests/test_ibkr_runtime_inputs.py | 72 ++++++- tests/test_longbridge_market_data.py | 32 ++- 8 files changed, 295 insertions(+), 39 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a7d58fc..fccf698 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "quant-platform-kit" -version = "0.7.37" +version = "0.7.38" description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies." readme = "README.md" requires-python = ">=3.9" diff --git a/setup.py b/setup.py index d4256dd..3221eaf 100644 --- a/setup.py +++ b/setup.py @@ -3,7 +3,7 @@ setup( name="quant-platform-kit", - version="0.7.37", + version="0.7.38", description="Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies.", package_dir={"": "src"}, packages=find_packages(where="src"), diff --git a/src/quant_platform_kit/__init__.py b/src/quant_platform_kit/__init__.py index 2fddb4b..1458b69 100644 --- a/src/quant_platform_kit/__init__.py +++ b/src/quant_platform_kit/__init__.py @@ -4,7 +4,7 @@ used by older strategy repositories. """ -__version__ = "0.7.37" +__version__ = "0.7.38" from .common.models import ( ExecutionReport, diff --git a/src/quant_platform_kit/common/runtime_inputs.py b/src/quant_platform_kit/common/runtime_inputs.py index d7b225d..852b1cf 100644 --- a/src/quant_platform_kit/common/runtime_inputs.py +++ b/src/quant_platform_kit/common/runtime_inputs.py @@ -34,7 +34,9 @@ def _normalize_numeric_history( continue normalized.append(numeric) if not normalized: - raise ValueError(f"Semiconductor rotation inputs require non-empty {label} history") + raise ValueError( + f"Semiconductor rotation inputs require non-empty {label} history" + ) return tuple(normalized) @@ -75,10 +77,10 @@ def _tail_std(values: tuple[float, ...], window: int) -> float: return _std(values[-window:]) -def _tail_realized_volatility(values: tuple[float, ...], window: int) -> float: +def _sample_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):] + tail_values = values[-(window + 1) :] returns: list[float] = [] for previous, current in zip(tail_values, tail_values[1:]): if previous == 0.0: @@ -87,6 +89,23 @@ def _tail_realized_volatility(values: tuple[float, ...], window: int) -> float: return float(_sample_std(returns) * sqrt(252)) +def _tail_realized_volatility(values: tuple[float, ...], window: int) -> float: + return _sample_realized_volatility(values, window) + + +def _realized_volatility_history( + values: tuple[float, ...], *, window: int +) -> tuple[float | None, ...]: + if window <= 0: + raise ValueError("window must be positive") + result: list[float | None] = [None] * len(values) + for index in range(window, len(values)): + result[index] = _sample_realized_volatility( + values[index - window : index + 1], window + ) + return tuple(result) + + def _compute_rsi(values: tuple[float, ...], *, window: int = 14) -> tuple[float, ...]: if len(values) < window + 1: raise ValueError("insufficient history for RSI") @@ -122,24 +141,61 @@ def _rsi_from_avg(avg_gain_value: float, avg_loss_value: float) -> float: return tuple(rsis) -def _rolling_quantile(values: tuple[float, ...], *, window: int, quantile: float) -> tuple[float | None, ...]: +def _rolling_count(values: tuple[float | None, ...], *, window: int) -> tuple[int, ...]: + if window <= 0: + raise ValueError("window must be positive") + result: list[int] = [0] * len(values) + for index in range(len(values)): + start = max(0, index - window + 1) + result[index] = sum( + 1 for value in values[start : index + 1] if value is not None + ) + return tuple(result) + + +def _rolling_quantile( + values: tuple[float | None, ...], + *, + window: int, + quantile: float, + min_periods: int | None = None, +) -> tuple[float | None, ...]: if window <= 0: raise ValueError("window must be positive") if not 0.0 < quantile < 1.0: raise ValueError("quantile must be between 0 and 1") + effective_min_periods = window if min_periods is None else max(1, int(min_periods)) result: list[float | None] = [None] * len(values) - for index in range(window - 1, len(values)): - chunk = sorted(values[index - window + 1 : index + 1]) - if not chunk: + for index in range(len(values)): + start = max(0, index - window + 1) + chunk = sorted( + value for value in values[start : index + 1] if value is not None + ) + if len(chunk) < effective_min_periods: continue position = (len(chunk) - 1) * quantile lower_index = int(position) upper_index = min(lower_index + 1, len(chunk) - 1) fraction = position - lower_index - result[index] = chunk[lower_index] * (1.0 - fraction) + chunk[upper_index] * fraction + result[index] = ( + chunk[lower_index] * (1.0 - fraction) + chunk[upper_index] * fraction + ) return tuple(result) +def _bounded_threshold( + value: float | None, *, floor: float | None, cap: float | None +) -> float | None: + if value is None: + return None + threshold = float(value) + if floor is not None: + threshold = max(float(floor), threshold) + if cap is not None: + threshold = min(float(cap), threshold) + return threshold + + def build_semiconductor_rotation_indicators_from_history( *, soxl_history: Iterable[float], @@ -148,6 +204,12 @@ def build_semiconductor_rotation_indicators_from_history( dynamic_rsi_quantile_window: int = 252, dynamic_rsi_quantile: float = 0.90, dynamic_rsi_floor: float = 70.0, + dynamic_volatility_delever_window: int = 10, + dynamic_volatility_delever_quantile_window: int = 252, + dynamic_volatility_delever_quantile: float = 0.95, + dynamic_volatility_delever_min_periods: int = 126, + dynamic_volatility_delever_floor: float = 0.50, + dynamic_volatility_delever_cap: float = 0.75, ) -> dict[str, dict[str, float]]: window = int(trend_ma_window) if window <= 0: @@ -158,11 +220,26 @@ def build_semiconductor_rotation_indicators_from_history( 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") + volatility_window = int(dynamic_volatility_delever_window) + if volatility_window <= 0: + raise ValueError("dynamic_volatility_delever_window must be positive") + volatility_quantile_window = int(dynamic_volatility_delever_quantile_window) + if volatility_quantile_window <= 0: + raise ValueError("dynamic_volatility_delever_quantile_window must be positive") + volatility_quantile = float(dynamic_volatility_delever_quantile) + if not 0.0 < volatility_quantile < 1.0: + raise ValueError("dynamic_volatility_delever_quantile must be between 0 and 1") + volatility_min_periods = max( + 1, + min(volatility_quantile_window, int(dynamic_volatility_delever_min_periods)), + ) soxl_close = _normalize_numeric_history(soxl_history, label="SOXL") soxx_close = _normalize_numeric_history(soxx_history, label="SOXX") if len(soxl_close) < window or len(soxx_close) < window: - raise ValueError("Semiconductor rotation inputs require sufficient SOXL/SOXX history") + raise ValueError( + "Semiconductor rotation inputs require sufficient SOXL/SOXX history" + ) soxl_ma_trend = _tail_mean(soxl_close, window) soxx_ma_trend = _tail_mean(soxx_close, window) @@ -176,17 +253,68 @@ def build_semiconductor_rotation_indicators_from_history( window=rsi_quantile_window, quantile=rsi_quantile, ) - previous_threshold = rsi_threshold_history[-2] if len(rsi_threshold_history) >= 2 else None + previous_threshold = ( + rsi_threshold_history[-2] if len(rsi_threshold_history) >= 2 else None + ) soxx_dynamic_rsi_threshold = float( max( float(dynamic_rsi_floor), - float(previous_threshold) if previous_threshold is not None else float(dynamic_rsi_floor), + float(previous_threshold) + if previous_threshold is not None + else float(dynamic_rsi_floor), ) ) soxx_bb_mid = _tail_mean(soxx_close, 20) soxx_bb_std = _tail_std(soxx_close, 20) soxx_realized_volatility_10 = _tail_realized_volatility(soxx_close, 10) soxx_realized_volatility_20 = _tail_realized_volatility(soxx_close, 20) + soxx_volatility_history = _realized_volatility_history( + soxx_close, + window=volatility_window, + ) + volatility_threshold_history = _rolling_quantile( + soxx_volatility_history, + window=volatility_quantile_window, + quantile=volatility_quantile, + min_periods=volatility_min_periods, + ) + soxx_dynamic_volatility_threshold = _bounded_threshold( + volatility_threshold_history[-1], + floor=dynamic_volatility_delever_floor, + cap=dynamic_volatility_delever_cap, + ) + soxx_dynamic_volatility_sample_count = _rolling_count( + soxx_volatility_history, + window=volatility_quantile_window, + )[-1] + volatility_threshold_fields = { + f"realized_volatility_{volatility_window}_dynamic_threshold": soxx_dynamic_volatility_threshold, + f"realized_volatility_{volatility_window}_dynamic_sample_count": float( + soxx_dynamic_volatility_sample_count + ), + f"realized_volatility_{volatility_window}_dynamic_lookback": float( + volatility_quantile_window + ), + f"realized_volatility_{volatility_window}_dynamic_percentile": volatility_quantile, + f"realized_volatility_{volatility_window}_dynamic_min_periods": float( + volatility_min_periods + ), + f"realized_volatility_{volatility_window}_dynamic_floor": float( + dynamic_volatility_delever_floor + ), + f"realized_volatility_{volatility_window}_dynamic_cap": float( + dynamic_volatility_delever_cap + ), + "realized_volatility_dynamic_threshold": soxx_dynamic_volatility_threshold, + "realized_volatility_dynamic_sample_count": float( + soxx_dynamic_volatility_sample_count + ), + "realized_volatility_dynamic_lookback": float(volatility_quantile_window), + "realized_volatility_dynamic_percentile": volatility_quantile, + "realized_volatility_dynamic_min_periods": float(volatility_min_periods), + "realized_volatility_dynamic_floor": float(dynamic_volatility_delever_floor), + "realized_volatility_dynamic_cap": float(dynamic_volatility_delever_cap), + } return { "soxl": { "price": float(soxl_close[-1]), @@ -205,6 +333,11 @@ def build_semiconductor_rotation_indicators_from_history( "realized_volatility": soxx_realized_volatility_20, "realized_volatility_10": soxx_realized_volatility_10, "realized_volatility_20": soxx_realized_volatility_20, + **{ + key: value + for key, value in volatility_threshold_fields.items() + if value is not None + }, }, } @@ -213,12 +346,17 @@ def required_semiconductor_rotation_history_lookback( *, trend_ma_window: int = 140, dynamic_rsi_quantile_window: int = 252, + dynamic_volatility_delever_window: int = 10, + dynamic_volatility_delever_quantile_window: int = 252, minimum_lookback: int = DEFAULT_SEMICONDUCTOR_ROTATION_HISTORY_LOOKBACK, ) -> int: return max( int(minimum_lookback), int(trend_ma_window) + 20, int(dynamic_rsi_quantile_window) + 28, + int(dynamic_volatility_delever_quantile_window) + + int(dynamic_volatility_delever_window) + + 1, ) @@ -230,6 +368,12 @@ def build_semiconductor_rotation_inputs_from_history( dynamic_rsi_quantile_window: int = 252, dynamic_rsi_quantile: float = 0.90, dynamic_rsi_floor: float = 70.0, + dynamic_volatility_delever_window: int = 10, + dynamic_volatility_delever_quantile_window: int = 252, + dynamic_volatility_delever_quantile: float = 0.95, + dynamic_volatility_delever_min_periods: int = 126, + dynamic_volatility_delever_floor: float = 0.50, + dynamic_volatility_delever_cap: float = 0.75, ) -> dict[str, dict[str, dict[str, float]]]: return { "derived_indicators": build_semiconductor_rotation_indicators_from_history( @@ -239,6 +383,12 @@ def build_semiconductor_rotation_inputs_from_history( dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, dynamic_rsi_quantile=dynamic_rsi_quantile, dynamic_rsi_floor=dynamic_rsi_floor, + dynamic_volatility_delever_window=dynamic_volatility_delever_window, + dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window, + dynamic_volatility_delever_quantile=dynamic_volatility_delever_quantile, + dynamic_volatility_delever_min_periods=dynamic_volatility_delever_min_periods, + dynamic_volatility_delever_floor=dynamic_volatility_delever_floor, + dynamic_volatility_delever_cap=dynamic_volatility_delever_cap, ) } @@ -250,7 +400,9 @@ def build_account_state_from_portfolio_snapshot( liquid_cash: float | None = None, ) -> dict[str, Any]: metadata = getattr(snapshot, "metadata", {}) or {} - raw_sellable_quantities = metadata.get("sellable_quantities") if isinstance(metadata, Mapping) else None + raw_sellable_quantities = ( + metadata.get("sellable_quantities") if isinstance(metadata, Mapping) else None + ) resolved_sellable_quantities: dict[str, float] = {} if isinstance(raw_sellable_quantities, Mapping): resolved_sellable_quantities = { @@ -281,7 +433,9 @@ def build_account_state_from_portfolio_snapshot( quantity = float(position.quantity) quantities[symbol] = quantity - sellable_quantities[symbol] = float(resolved_sellable_quantities.get(symbol, quantity)) + sellable_quantities[symbol] = float( + resolved_sellable_quantities.get(symbol, quantity) + ) market_values[symbol] = float(position.market_value) resolved_liquid_cash = liquid_cash @@ -301,7 +455,9 @@ def build_account_state_from_portfolio_snapshot( "sellable_quantities": sellable_quantities, "total_strategy_equity": float(snapshot.total_equity), } - raw_cash_by_currency = metadata.get("cash_by_currency") if isinstance(metadata, Mapping) else None + raw_cash_by_currency = ( + metadata.get("cash_by_currency") if isinstance(metadata, Mapping) else None + ) if isinstance(raw_cash_by_currency, Mapping): account_state["cash_by_currency"] = { str(currency).strip().upper(): float(amount) @@ -321,7 +477,9 @@ def build_portfolio_snapshot_from_account_state( normalized_symbols = _normalize_symbols(strategy_symbols) market_values = dict(account_state["market_values"]) quantities = dict(account_state["quantities"]) - symbols = normalized_symbols or tuple(sorted(str(symbol) for symbol in market_values)) + symbols = normalized_symbols or tuple( + sorted(str(symbol) for symbol in market_values) + ) positions: list[Position] = [] for symbol in symbols: diff --git a/src/quant_platform_kit/ibkr/runtime_inputs.py b/src/quant_platform_kit/ibkr/runtime_inputs.py index ce1aeaa..01cddbb 100644 --- a/src/quant_platform_kit/ibkr/runtime_inputs.py +++ b/src/quant_platform_kit/ibkr/runtime_inputs.py @@ -75,10 +75,14 @@ def build_semiconductor_rotation_indicators( *, trend_ma_window: int = 140, dynamic_rsi_quantile_window: int = 252, + dynamic_volatility_delever_window: int = 10, + dynamic_volatility_delever_quantile_window: int = 252, ) -> dict[str, dict[str, float]]: effective_lookback = required_semiconductor_rotation_history_lookback( trend_ma_window=trend_ma_window, dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, + dynamic_volatility_delever_window=dynamic_volatility_delever_window, + dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window, ) soxl_history = historical_close_loader( ib, @@ -97,6 +101,8 @@ def build_semiconductor_rotation_indicators( soxx_history=soxx_history, trend_ma_window=trend_ma_window, dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, + dynamic_volatility_delever_window=dynamic_volatility_delever_window, + dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window, ) @@ -106,6 +112,8 @@ def build_semiconductor_rotation_inputs( *, trend_ma_window: int = 140, dynamic_rsi_quantile_window: int = 252, + dynamic_volatility_delever_window: int = 10, + dynamic_volatility_delever_quantile_window: int = 252, ) -> dict[str, dict[str, dict[str, float]]]: return { "derived_indicators": build_semiconductor_rotation_indicators( @@ -113,5 +121,7 @@ def build_semiconductor_rotation_inputs( historical_close_loader, trend_ma_window=trend_ma_window, dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, + dynamic_volatility_delever_window=dynamic_volatility_delever_window, + dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window, ) } diff --git a/src/quant_platform_kit/longbridge/market_data.py b/src/quant_platform_kit/longbridge/market_data.py index fc85c37..2604d7b 100644 --- a/src/quant_platform_kit/longbridge/market_data.py +++ b/src/quant_platform_kit/longbridge/market_data.py @@ -44,7 +44,9 @@ def fetch_last_price(q_ctx: Any, symbol: str) -> float | None: return fetch_last_prices(q_ctx, [symbol]).get(_normalize_symbol(symbol)) -def fetch_last_prices(q_ctx: Any, symbols: list[str] | tuple[str, ...]) -> dict[str, float]: +def fetch_last_prices( + q_ctx: Any, symbols: list[str] | tuple[str, ...] +) -> dict[str, float]: normalized_symbols = [] for symbol in symbols: normalized_symbol = _normalize_symbol(symbol) @@ -57,8 +59,12 @@ def fetch_last_prices(q_ctx: Any, symbols: list[str] | tuple[str, ...]) -> dict[ quotes = _quote_with_retry(q_ctx, normalized_symbols) prices: dict[str, float] = {} for index, quote in enumerate(quotes): - fallback_symbol = normalized_symbols[index] if index < len(normalized_symbols) else "" - quoted_symbol = _normalize_symbol(getattr(quote, "symbol", "") or fallback_symbol) + fallback_symbol = ( + normalized_symbols[index] if index < len(normalized_symbols) else "" + ) + quoted_symbol = _normalize_symbol( + getattr(quote, "symbol", "") or fallback_symbol + ) if not quoted_symbol: continue last_done = getattr(quote, "last_done", None) @@ -77,6 +83,8 @@ def calculate_rotation_indicators( trend_window: int, lookback: int | None = None, dynamic_rsi_quantile_window: int = 252, + dynamic_volatility_delever_window: int = 10, + dynamic_volatility_delever_quantile_window: int = 252, ) -> dict[str, dict[str, float]] | None: from longport.openapi import AdjustType, Period @@ -86,10 +94,16 @@ def calculate_rotation_indicators( else required_semiconductor_rotation_history_lookback( trend_ma_window=trend_window, dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, + dynamic_volatility_delever_window=dynamic_volatility_delever_window, + dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window, ) ) - 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) + 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: return None @@ -103,4 +117,6 @@ def calculate_rotation_indicators( soxx_history=df_soxx["close"], trend_ma_window=trend_window, dynamic_rsi_quantile_window=dynamic_rsi_quantile_window, + dynamic_volatility_delever_window=dynamic_volatility_delever_window, + dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window, ) diff --git a/tests/test_ibkr_runtime_inputs.py b/tests/test_ibkr_runtime_inputs.py index c2d068a..c93adc3 100644 --- a/tests/test_ibkr_runtime_inputs.py +++ b/tests/test_ibkr_runtime_inputs.py @@ -2,7 +2,10 @@ import unittest -from quant_platform_kit.strategy_contracts import StrategyManifest, StrategyRuntimeAdapter +from quant_platform_kit.strategy_contracts import ( + StrategyManifest, + StrategyRuntimeAdapter, +) from quant_platform_kit import ( build_semiconductor_rotation_indicators_from_history, build_semiconductor_rotation_inputs_from_history, @@ -42,7 +45,9 @@ def loader(_ib, symbol, duration="2 Y", bar_size="1 day"): self.assertEqual(observed["call"], ("QQQ", "2 Y", "1 day")) self.assertEqual(payload["benchmark_history"][0]["close"], 1.0) - def test_build_ibkr_strategy_context_uses_required_inputs_and_portfolio(self) -> None: + def test_build_ibkr_strategy_context_uses_required_inputs_and_portfolio( + self, + ) -> None: entrypoint = type( "Entrypoint", (), @@ -52,11 +57,15 @@ def test_build_ibkr_strategy_context_uses_required_inputs_and_portfolio(self) -> domain="us_equity", display_name="SOXL/SOXX Semiconductor Trend Income", description="test", - required_inputs=frozenset({"derived_indicators", "portfolio_snapshot"}), + required_inputs=frozenset( + {"derived_indicators", "portfolio_snapshot"} + ), ) }, )() - runtime_adapter = StrategyRuntimeAdapter(portfolio_input_name="portfolio_snapshot") + runtime_adapter = StrategyRuntimeAdapter( + portfolio_input_name="portfolio_snapshot" + ) portfolio_snapshot = object() ctx = build_ibkr_strategy_context( @@ -77,7 +86,9 @@ def test_build_ibkr_strategy_context_uses_required_inputs_and_portfolio(self) -> self.assertEqual(ctx.capabilities["broker_client"], "fake-ib") self.assertEqual(ctx.runtime_config["translator"], "noop") - def test_build_semiconductor_rotation_indicators_uses_soxl_and_soxx_history(self) -> None: + def test_build_semiconductor_rotation_indicators_uses_soxl_and_soxx_history( + self, + ) -> None: observed = [] def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): @@ -117,12 +128,23 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): self.assertLess(indicators["soxx"]["bb_lower"], indicators["soxx"]["price"]) self.assertIn("realized_volatility_10", indicators["soxx"]) self.assertIn("realized_volatility_20", indicators["soxx"]) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_threshold"], 0.50 + ) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_sample_count"], 160.0 + ) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_percentile"], 0.95 + ) self.assertEqual( indicators["soxx"]["realized_volatility"], indicators["soxx"]["realized_volatility_20"], ) - def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) -> None: + def test_build_semiconductor_rotation_indicators_from_history_is_generic( + 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)], @@ -144,6 +166,12 @@ def test_build_semiconductor_rotation_indicators_from_history_is_generic(self) - self.assertGreater(indicators["soxx"]["bb_upper"], indicators["soxx"]["price"]) self.assertIn("realized_volatility_10", indicators["soxx"]) self.assertIn("realized_volatility_20", indicators["soxx"]) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_threshold"], 0.50 + ) + self.assertEqual( + indicators["soxx"]["realized_volatility_dynamic_threshold"], 0.50 + ) 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)], @@ -152,7 +180,9 @@ 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: + 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)], @@ -181,11 +211,31 @@ 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) - self.assertEqual(payload["derived_indicators"]["soxx"]["realized_volatility_10"], 0.0) - self.assertEqual(payload["derived_indicators"]["soxx"]["realized_volatility_20"], 0.0) + self.assertEqual( + payload["derived_indicators"]["soxx"]["rsi14_dynamic_threshold"], 70.0 + ) + self.assertEqual( + payload["derived_indicators"]["soxx"]["realized_volatility_10"], 0.0 + ) + self.assertEqual( + payload["derived_indicators"]["soxx"]["realized_volatility_20"], 0.0 + ) + self.assertEqual( + payload["derived_indicators"]["soxx"][ + "realized_volatility_10_dynamic_threshold" + ], + 0.50, + ) + self.assertEqual( + payload["derived_indicators"]["soxx"][ + "realized_volatility_10_dynamic_sample_count" + ], + 160.0, + ) - def test_build_semiconductor_rotation_indicators_requires_sufficient_history(self) -> None: + def test_build_semiconductor_rotation_indicators_requires_sufficient_history( + self, + ) -> None: def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): if symbol == "SOXL": return [100.0] * 100 diff --git a/tests/test_longbridge_market_data.py b/tests/test_longbridge_market_data.py index 102368c..361db30 100644 --- a/tests/test_longbridge_market_data.py +++ b/tests/test_longbridge_market_data.py @@ -5,7 +5,11 @@ import unittest from unittest.mock import patch -from quant_platform_kit.longbridge.market_data import calculate_rotation_indicators, fetch_last_price, fetch_last_prices +from quant_platform_kit.longbridge.market_data import ( + calculate_rotation_indicators, + fetch_last_price, + fetch_last_prices, +) class FakeQuote: @@ -55,7 +59,9 @@ def quote(self, symbols): return super().quote(symbols) quote_context = RateLimitedQuoteContext() - with patch("quant_platform_kit.longbridge.market_data.time.sleep") as sleep_mock: + with patch( + "quant_platform_kit.longbridge.market_data.time.sleep" + ) as sleep_mock: self.assertEqual(fetch_last_price(quote_context, "SOXL.US"), 123.45) self.assertEqual(quote_context.calls, 2) @@ -67,13 +73,20 @@ def test_calculate_rotation_indicators(self) -> None: openapi_module.Period = types.SimpleNamespace(Day="Day") openapi_module.AdjustType = types.SimpleNamespace(ForwardAdjust="ForwardAdjust") - with patch.dict(sys.modules, {"longport": longport_module, "longport.openapi": openapi_module}): - indicators = calculate_rotation_indicators(FakeQuoteContext(), trend_window=150) + with patch.dict( + sys.modules, + {"longport": longport_module, "longport.openapi": openapi_module}, + ): + indicators = calculate_rotation_indicators( + FakeQuoteContext(), trend_window=150 + ) self.assertIsNotNone(indicators) self.assertEqual(indicators["soxl"]["price"], 519.0) self.assertEqual(indicators["soxx"]["price"], 619.0) - self.assertAlmostEqual(indicators["soxx"]["ma20"], sum(200.0 + i for i in range(400, 420)) / 20) + self.assertAlmostEqual( + indicators["soxx"]["ma20"], sum(200.0 + i for i in range(400, 420)) / 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) @@ -81,6 +94,15 @@ def test_calculate_rotation_indicators(self) -> None: self.assertLess(indicators["soxx"]["bb_lower"], indicators["soxx"]["price"]) self.assertIn("realized_volatility_10", indicators["soxx"]) self.assertIn("realized_volatility_20", indicators["soxx"]) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_threshold"], 0.50 + ) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_sample_count"], 252.0 + ) + self.assertEqual( + indicators["soxx"]["realized_volatility_10_dynamic_percentile"], 0.95 + ) self.assertEqual( indicators["soxx"]["realized_volatility"], indicators["soxx"]["realized_volatility_20"],