From 4353fdafb5551de271155f43a8c74a786cc0a47c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 9 Jun 2026 14:23:09 +0800 Subject: [PATCH] Add SOXL dynamic volatility threshold research --- docs/tqqq-soxl-optimization-research.md | 57 ++++ ...l_dynamic_volatility_delever_thresholds.py | 310 ++++++++++++++++++ .../soxl_soxx_trend_income_backtest.py | 109 +++++- tests/test_soxl_soxx_trend_income_backtest.py | 35 ++ 4 files changed, 508 insertions(+), 3 deletions(-) create mode 100644 scripts/research_soxl_dynamic_volatility_delever_thresholds.py diff --git a/docs/tqqq-soxl-optimization-research.md b/docs/tqqq-soxl-optimization-research.md index 3ef5683..d88ef69 100644 --- a/docs/tqqq-soxl-optimization-research.md +++ b/docs/tqqq-soxl-optimization-research.md @@ -87,6 +87,63 @@ Interpretation: - Promote `dynamic_p90_floor24_cap36` as the live default, with fixed 28 kept as the warm-up fallback. +## 2026-06-09 Dynamic SOXL Volatility Threshold Recheck + +Follow-up question: can the fixed `SOXX 10d realized volatility >= 55%` SOXL +delever gate be replaced by a dynamic threshold? + +Output directory: + +`data/output/soxl_dynamic_volatility_delever_threshold_research_20260609` + +Method: + +- Real SOXL/SOXX/BOXX price replay from 2016-06-06 through 2026-06-02. +- Income layer disabled to isolate the SOXL/SOXX core gate. +- Dynamic RSI stays on, matching the current SOXL live profile. +- Turnover cost is 5 bps. +- The research overlay was first validated by replaying fixed 55%; the + `overlay_fixed55_replay` output matches `current_core_fixed55` exactly. + +Full-sample summary: + +| Candidate | CAGR | Max Drawdown | Rebalances/Year | Turnover/Year | SOXL Delever Stops | Effective Threshold | Decision | +| --- | ---: | ---: | ---: | ---: | ---: | --- | --- | +| `current_core_fixed55` | 72.15% | -40.51% | 34.63 | 8.76 | 17 | fixed 55% | baseline | +| `dynamic_p95_cap75` | 81.50% | -37.41% | 35.48 | 9.42 | 31 | p95, cap 75%, no floor | research watch; threshold can drop too low | +| `dynamic_p95_floor50_cap75` | 77.12% | -40.51% | 34.31 | 8.47 | 7 | p95 bounded 50%-75% | best promotion candidate | +| `dynamic_p95_floor55_cap75` | 75.11% | -40.51% | 34.52 | 8.47 | 5 | p95 bounded 55%-75% | conservative backup | +| `dynamic_p90_floor55_cap75` | 75.05% | -40.51% | 34.52 | 8.61 | 7 | p90 bounded 55%-75% | similar to p95/floor55 | +| `no_vol_delever` | 74.12% | -40.51% | 34.10 | 8.12 | 0 | n/a | not preferred; post-2022 drawdown worsens | + +Key window checks for `dynamic_p95_floor50_cap75` versus current fixed 55: + +| Window | CAGR Delta | Max Drawdown Delta | Interpretation | +| --- | ---: | ---: | --- | +| YTD through 2026-06-02 | +96.28 pp | +0.00 pp | better recent rebound | +| Last 3 months | +690.87 pp | +0.00 pp | better recent rebound; annualized short-window number is not a stable long-term expectation | +| Last 1 year | +10.11 pp | +0.00 pp | better, no drawdown regression | +| Post-2022 bull | +9.62 pp | +0.00 pp | better, no drawdown regression | +| 2022 rate bear | +0.00 pp | +0.00 pp | neutral | +| COVID crash | +0.00 pp | +0.00 pp | neutral | +| 2018-2019 trade war | +0.00 pp | +0.00 pp | neutral | +| Long real-product window | +4.97 pp | +0.00 pp | better, no drawdown regression | + +Interpretation: + +- Dynamic SOXL volatility gating does help, but the useful percentile is higher + than TQQQ. The current fixed 55% gate is already fairly loose, so p80/p85 and + lower floors over-delever and reduce CAGR. +- `dynamic_p95_cap75` has the best headline result, but the effective threshold + can fall to about 27.6%, increasing SOXL stop count to 31 and turnover to + 9.42/year. Treat it as research-watch rather than a live default. +- `dynamic_p95_floor50_cap75` is the cleanest live candidate: it improves CAGR + from 72.15% to 77.12%, leaves max drawdown unchanged, reduces turnover from + 8.76/year to 8.47/year, and reduces actual SOXL delever stops from 17 to 7. +- Do not promote SOXL automatically in this research PR. A later strategy PR + should add production dynamic-threshold fields to `blend_gate_volatility_*` + before changing the live default. + ## 2026-06-04 Current Default Recheck This recheck was designed to answer whether the current live TQQQ/SOXL profiles diff --git a/scripts/research_soxl_dynamic_volatility_delever_thresholds.py b/scripts/research_soxl_dynamic_volatility_delever_thresholds.py new file mode 100644 index 0000000..0558e2e --- /dev/null +++ b/scripts/research_soxl_dynamic_volatility_delever_thresholds.py @@ -0,0 +1,310 @@ +from __future__ import annotations + +import argparse +from pathlib import Path +from typing import Mapping + +import pandas as pd + +from us_equity_snapshot_pipelines.backtest_windows import build_benchmark_returns, build_window_summary +from us_equity_snapshot_pipelines.soxl_soxx_trend_income_backtest import ( + DEFAULT_INITIAL_EQUITY_USD, + DEFAULT_TURNOVER_COST_BPS, + _build_price_frame, + run_backtest, +) + + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_PRICES = ROOT / "data" / "output" / "codex_soxl_rsi_recheck_20260603" / "price_history.csv" +DEFAULT_OUTPUT_DIR = ROOT / "data" / "output" / "soxl_dynamic_volatility_delever_threshold_research" +DEFAULT_BACKTEST_START = "2016-06-06" + + +def _normalize_prices(path: Path) -> pd.DataFrame: + prices = _build_price_frame(pd.read_csv(path)) + symbols = set(prices["symbol"].unique()) + additions = [] + if "BOXX" not in symbols and "BIL" in symbols: + additions.append(prices.loc[prices["symbol"].eq("BIL")].assign(symbol="BOXX")) + if additions: + prices = pd.concat([prices, *additions], ignore_index=True) + return _build_price_frame(prices) + + +def _external_vol_overlay( + *, + threshold: float, + threshold_mode: str = "fixed", + percentile: float | None = None, + floor: float | None = None, + cap: float | None = None, + lookback: int = 252, + min_periods: int = 126, +) -> dict[str, object]: + return { + "soxl_delever_overlay_kind": "volatility", + "soxl_delever_overlay_symbol": "SOXX", + "soxl_delever_overlay_window": 10, + "soxl_delever_overlay_threshold": float(threshold), + "soxl_delever_overlay_threshold_mode": threshold_mode, + "soxl_delever_overlay_threshold_lookback": int(lookback), + "soxl_delever_overlay_threshold_percentile": percentile, + "soxl_delever_overlay_threshold_min_periods": int(min_periods), + "soxl_delever_overlay_threshold_floor": floor, + "soxl_delever_overlay_threshold_cap": cap, + "soxl_delever_overlay_retention_ratio": 0.0, + "soxl_delever_overlay_redirect_symbol": "SOXX", + } + + +def _variants() -> tuple[tuple[str, dict[str, object]], ...]: + return ( + ("current_core_fixed55", {}), + ("overlay_fixed55_replay", _external_vol_overlay(threshold=0.55)), + ( + "no_vol_delever", + {"strategy_overrides": {"blend_gate_volatility_delever_enabled": False}}, + ), + ("fixed50", _external_vol_overlay(threshold=0.50)), + ("fixed60", _external_vol_overlay(threshold=0.60)), + ( + "dynamic_p80_floor45_cap70", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.80, + floor=0.45, + cap=0.70, + ), + ), + ( + "dynamic_p85_floor45_cap70", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.85, + floor=0.45, + cap=0.70, + ), + ), + ( + "dynamic_p90_floor45_cap70", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.90, + floor=0.45, + cap=0.70, + ), + ), + ( + "dynamic_p90_floor50_cap70", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.90, + floor=0.50, + cap=0.70, + ), + ), + ( + "dynamic_p90_floor55_cap75", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.90, + floor=0.55, + cap=0.75, + ), + ), + ( + "dynamic_p95_floor45_cap75", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.95, + floor=0.45, + cap=0.75, + ), + ), + ( + "dynamic_p95_floor45_cap70", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.95, + floor=0.45, + cap=0.70, + ), + ), + ( + "dynamic_p95_floor50_cap75", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.95, + floor=0.50, + cap=0.75, + ), + ), + ( + "dynamic_p95_floor55_cap75", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.95, + floor=0.55, + cap=0.75, + ), + ), + ( + "dynamic_p95_cap75", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.95, + floor=None, + cap=0.75, + ), + ), + ( + "dynamic_p90_cap75", + _external_vol_overlay( + threshold=0.55, + threshold_mode="rolling_percentile", + percentile=0.90, + floor=None, + cap=0.75, + ), + ), + ) + + +def _first_existing_series(frame: pd.DataFrame, *columns: str) -> pd.Series: + for column in columns: + if column in frame.columns: + series = pd.to_numeric(frame[column], errors="coerce") + if not series.dropna().empty: + return series + return pd.Series(dtype=float) + + +def _variant_row(name: str, result: Mapping[str, object]) -> dict[str, object]: + summary = dict(result["summary"]) + signal_history = pd.DataFrame(result["signal_history"]) + core_triggered = ( + signal_history.get("blend_gate_volatility_delever_triggered", pd.Series(dtype=bool)) + .fillna(False) + .astype(bool) + ) + overlay_triggered = ( + signal_history.get("soxl_delever_overlay_triggered", pd.Series(dtype=bool)) + .fillna(False) + .astype(bool) + ) + threshold = _first_existing_series( + signal_history, + "soxl_delever_overlay_threshold", + "blend_gate_volatility_delever_threshold", + ) + dynamic_threshold = _first_existing_series(signal_history, "soxl_delever_overlay_dynamic_threshold") + dynamic_sample_count = _first_existing_series(signal_history, "soxl_delever_overlay_dynamic_sample_count") + threshold_mode = "" + if "soxl_delever_overlay_threshold_mode" in signal_history.columns: + modes = tuple( + str(item) + for item in signal_history["soxl_delever_overlay_threshold_mode"].dropna().unique() + if str(item) + ) + threshold_mode = ",".join(modes) + if not threshold_mode: + threshold_mode = "fixed_core" + return { + "Variant": name, + **summary, + "Core Vol Trigger Days": int(core_triggered.sum()), + "Overlay Vol Trigger Days": int(overlay_triggered.sum()), + "Total Vol Delever Days": int(core_triggered.sum() + overlay_triggered.sum()), + "Threshold Mode": threshold_mode, + "Median Effective Threshold": float(threshold.median()) if not threshold.dropna().empty else float("nan"), + "Min Effective Threshold": float(threshold.min()) if not threshold.dropna().empty else float("nan"), + "Max Effective Threshold": float(threshold.max()) if not threshold.dropna().empty else float("nan"), + "Median Dynamic Threshold": float(dynamic_threshold.median()) + if not dynamic_threshold.dropna().empty + else float("nan"), + "Median Dynamic Sample Count": float(dynamic_sample_count.median()) + if not dynamic_sample_count.dropna().empty + else float("nan"), + } + + +def run_research( + *, + prices_path: Path, + output_dir: Path, + start_date: str, + end_date: str | None, + initial_equity: float, + turnover_cost_bps: float, +) -> Path: + output_dir.mkdir(parents=True, exist_ok=True) + prices = _normalize_prices(prices_path) + prices.to_csv(output_dir / "normalized_price_history.csv", index=False) + benchmark_returns = build_benchmark_returns(prices, symbols=("SOXX", "SOXL")) + summary_rows = [] + window_frames = [] + + for name, kwargs in _variants(): + result = run_backtest( + prices, + initial_equity=float(initial_equity), + start_date=start_date, + end_date=end_date, + turnover_cost_bps=float(turnover_cost_bps), + disable_income_layer=True, + **kwargs, + ) + summary_rows.append(_variant_row(name, result)) + window_summary = build_window_summary( + result["portfolio_returns"], + benchmark_returns=benchmark_returns, + primary_benchmark_symbol="SOXX", + ) + window_summary.insert(0, "Variant", name) + window_frames.append(window_summary) + result["signal_history"].to_csv(output_dir / f"{name}_signal_history.csv", index=False) + result["turnover_history"].rename("turnover").to_csv(output_dir / f"{name}_turnover_history.csv") + + pd.DataFrame(summary_rows).to_csv(output_dir / "variant_summary.csv", index=False) + pd.concat(window_frames, ignore_index=True).to_csv(output_dir / "variant_window_summary.csv", index=False) + return output_dir + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Research SOXL dynamic volatility-delever threshold variants.") + parser.add_argument("--prices", default=str(DEFAULT_PRICES)) + parser.add_argument("--output-dir", default=str(DEFAULT_OUTPUT_DIR)) + parser.add_argument("--start-date", default=DEFAULT_BACKTEST_START) + parser.add_argument("--end-date") + parser.add_argument("--initial-equity", type=float, default=DEFAULT_INITIAL_EQUITY_USD) + parser.add_argument("--turnover-cost-bps", type=float, default=DEFAULT_TURNOVER_COST_BPS) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + output_dir = run_research( + prices_path=Path(args.prices), + output_dir=Path(args.output_dir), + start_date=args.start_date, + end_date=args.end_date, + initial_equity=float(args.initial_equity), + turnover_cost_bps=float(args.turnover_cost_bps), + ) + print(f"wrote SOXL dynamic volatility-delever threshold research -> {output_dir}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/us_equity_snapshot_pipelines/soxl_soxx_trend_income_backtest.py b/src/us_equity_snapshot_pipelines/soxl_soxx_trend_income_backtest.py index f9b32bf..a5a4c44 100644 --- a/src/us_equity_snapshot_pipelines/soxl_soxx_trend_income_backtest.py +++ b/src/us_equity_snapshot_pipelines/soxl_soxx_trend_income_backtest.py @@ -176,6 +176,12 @@ def _build_soxl_delever_overlay_history( fast_window: int | None = None, slow_window: int | None = None, threshold: float | None, + threshold_mode: str = "fixed", + threshold_lookback: int | None = None, + threshold_percentile: float | None = None, + threshold_min_periods: int | None = None, + threshold_floor: float | None = None, + threshold_cap: float | None = None, atr_multiple: float, ) -> pd.DataFrame: kind = _normalize_overlay_kind(kind) @@ -208,10 +214,40 @@ def _build_soxl_delever_overlay_history( triggered = metric <= effective_threshold extra_columns = {} elif kind == "volatility": - effective_threshold = float(threshold if threshold is not None else 0.45) - metric = close.pct_change().rolling(effective_window, min_periods=effective_window).std(ddof=0) * np.sqrt(252) + metric = ( + close.pct_change(fill_method=None).rolling(effective_window, min_periods=effective_window).std() + * np.sqrt(252) + ) + fixed_threshold = float(threshold if threshold is not None else 0.45) + mode = str(threshold_mode or "fixed").strip().lower() + if mode not in {"fixed", "rolling_percentile"}: + raise ValueError("unsupported SOXL volatility delever threshold mode") + dynamic_threshold = pd.Series(np.nan, index=metric.index, dtype=float) + sample_count = pd.Series(0, index=metric.index, dtype=int) + lookback = max(1, int(threshold_lookback or 252)) + min_count = max(1, min(lookback, int(threshold_min_periods or min(126, lookback)))) + percentile = max(0.0, min(1.0, float(threshold_percentile if threshold_percentile is not None else 0.90))) + floor_value = None if threshold_floor is None else float(threshold_floor) + cap_value = None if threshold_cap is None else float(threshold_cap) + if mode == "rolling_percentile": + sample_count = metric.rolling(lookback, min_periods=1).count().astype(int) + dynamic_threshold = metric.rolling(lookback, min_periods=min_count).quantile(percentile) + if floor_value is not None: + dynamic_threshold = dynamic_threshold.clip(lower=floor_value) + if cap_value is not None: + dynamic_threshold = dynamic_threshold.clip(upper=cap_value) + effective_threshold = dynamic_threshold.fillna(fixed_threshold) triggered = metric >= effective_threshold - extra_columns = {} + extra_columns = { + "threshold_mode": mode, + "dynamic_threshold": dynamic_threshold, + "dynamic_sample_count": sample_count, + "dynamic_lookback": lookback, + "dynamic_percentile": percentile, + "dynamic_min_periods": min_count, + "dynamic_floor": floor_value, + "dynamic_cap": cap_value, + } elif kind == "momentum": effective_threshold = float(threshold if threshold is not None else -0.06) metric = close.pct_change(effective_window) @@ -588,6 +624,12 @@ def run_backtest( soxl_delever_overlay_fast_window: int | None = None, soxl_delever_overlay_slow_window: int | None = None, soxl_delever_overlay_threshold: float | None = None, + soxl_delever_overlay_threshold_mode: str = "fixed", + soxl_delever_overlay_threshold_lookback: int | None = None, + soxl_delever_overlay_threshold_percentile: float | None = None, + soxl_delever_overlay_threshold_min_periods: int | None = None, + soxl_delever_overlay_threshold_floor: float | None = None, + soxl_delever_overlay_threshold_cap: float | None = None, soxl_delever_overlay_atr_multiple: float | None = None, soxl_delever_overlay_retention_ratio: float = 0.0, soxl_delever_overlay_redirect_symbol: str = "BOXX", @@ -654,6 +696,12 @@ def run_backtest( fast_window=overlay_fast_window, slow_window=overlay_slow_window, threshold=soxl_delever_overlay_threshold, + threshold_mode=soxl_delever_overlay_threshold_mode, + threshold_lookback=soxl_delever_overlay_threshold_lookback, + threshold_percentile=soxl_delever_overlay_threshold_percentile, + threshold_min_periods=soxl_delever_overlay_threshold_min_periods, + threshold_floor=soxl_delever_overlay_threshold_floor, + threshold_cap=soxl_delever_overlay_threshold_cap, atr_multiple=overlay_atr_multiple, ) if soxl_delever_enabled @@ -836,6 +884,30 @@ def run_backtest( if not delever_row.empty else None, "soxl_delever_overlay_threshold": delever_row.get("threshold") if not delever_row.empty else None, + "soxl_delever_overlay_threshold_mode": delever_row.get("threshold_mode") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_threshold": delever_row.get("dynamic_threshold") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_sample_count": delever_row.get("dynamic_sample_count") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_lookback": delever_row.get("dynamic_lookback") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_percentile": delever_row.get("dynamic_percentile") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_min_periods": delever_row.get("dynamic_min_periods") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_floor": delever_row.get("dynamic_floor") + if not delever_row.empty + else None, + "soxl_delever_overlay_dynamic_cap": delever_row.get("dynamic_cap") + if not delever_row.empty + else None, "soxl_delever_overlay_atr_multiple": overlay_atr_multiple if overlay_kind == "chandelier" else None, "soxl_delever_overlay_retention_ratio": overlay_retention_ratio, "soxl_delever_overlay_redirect_symbol": overlay_redirect_symbol, @@ -991,6 +1063,25 @@ def build_parser() -> argparse.ArgumentParser: type=float, help="Overlay threshold: negative for drawdown/momentum, annualized ratio for volatility", ) + parser.add_argument( + "--soxl-delever-threshold-mode", + default="fixed", + choices=("fixed", "rolling_percentile"), + help="Threshold mode for volatility overlays", + ) + parser.add_argument("--soxl-delever-threshold-lookback", type=int, help="Rolling threshold lookback") + parser.add_argument( + "--soxl-delever-threshold-percentile", + type=float, + help="Rolling threshold percentile, for example 0.90", + ) + parser.add_argument( + "--soxl-delever-threshold-min-periods", + type=int, + help="Minimum samples required before using a rolling threshold", + ) + parser.add_argument("--soxl-delever-threshold-floor", type=float, help="Lower bound for rolling threshold") + parser.add_argument("--soxl-delever-threshold-cap", type=float, help="Upper bound for rolling threshold") parser.add_argument( "--soxl-delever-atr-multiple", type=float, @@ -1059,6 +1150,12 @@ def main(argv: list[str] | None = None) -> int: soxl_delever_overlay_fast_window=args.soxl_delever_fast_window, soxl_delever_overlay_slow_window=args.soxl_delever_slow_window, soxl_delever_overlay_threshold=args.soxl_delever_threshold, + soxl_delever_overlay_threshold_mode=args.soxl_delever_threshold_mode, + soxl_delever_overlay_threshold_lookback=args.soxl_delever_threshold_lookback, + soxl_delever_overlay_threshold_percentile=args.soxl_delever_threshold_percentile, + soxl_delever_overlay_threshold_min_periods=args.soxl_delever_threshold_min_periods, + soxl_delever_overlay_threshold_floor=args.soxl_delever_threshold_floor, + soxl_delever_overlay_threshold_cap=args.soxl_delever_threshold_cap, soxl_delever_overlay_atr_multiple=args.soxl_delever_atr_multiple, soxl_delever_overlay_retention_ratio=float(args.soxl_delever_retention_ratio), soxl_delever_overlay_redirect_symbol=args.soxl_delever_redirect_symbol, @@ -1101,6 +1198,12 @@ def main(argv: list[str] | None = None) -> int: "soxl_delever_fast_window": args.soxl_delever_fast_window, "soxl_delever_slow_window": args.soxl_delever_slow_window, "soxl_delever_threshold": args.soxl_delever_threshold, + "soxl_delever_threshold_mode": args.soxl_delever_threshold_mode, + "soxl_delever_threshold_lookback": args.soxl_delever_threshold_lookback, + "soxl_delever_threshold_percentile": args.soxl_delever_threshold_percentile, + "soxl_delever_threshold_min_periods": args.soxl_delever_threshold_min_periods, + "soxl_delever_threshold_floor": args.soxl_delever_threshold_floor, + "soxl_delever_threshold_cap": args.soxl_delever_threshold_cap, "soxl_delever_atr_multiple": args.soxl_delever_atr_multiple, "soxl_delever_retention_ratio": float(args.soxl_delever_retention_ratio), "soxl_delever_redirect_symbol": args.soxl_delever_redirect_symbol, diff --git a/tests/test_soxl_soxx_trend_income_backtest.py b/tests/test_soxl_soxx_trend_income_backtest.py index 881dfdc..148fe39 100644 --- a/tests/test_soxl_soxx_trend_income_backtest.py +++ b/tests/test_soxl_soxx_trend_income_backtest.py @@ -235,6 +235,41 @@ def test_soxl_soxx_volatility_delever_research_overlay_keeps_partial_soxl() -> N assert triggered["soxl_delever_overlay_retention_ratio"].eq(0.50).all() +def test_soxl_soxx_dynamic_volatility_delever_research_overlay_records_thresholds() -> None: + result = run_backtest( + _build_high_volatility_soxx_prices(), + initial_equity=100_000.0, + start_date="2023-10-02", + end_date="2024-03-29", + turnover_cost_bps=5.0, + soxl_delever_overlay_kind="volatility", + soxl_delever_overlay_symbol="SOXX", + soxl_delever_overlay_window=10, + soxl_delever_overlay_threshold=0.55, + soxl_delever_overlay_threshold_mode="rolling_percentile", + soxl_delever_overlay_threshold_lookback=60, + soxl_delever_overlay_threshold_percentile=0.90, + soxl_delever_overlay_threshold_min_periods=20, + soxl_delever_overlay_threshold_floor=0.20, + soxl_delever_overlay_threshold_cap=0.50, + soxl_delever_overlay_retention_ratio=0.0, + soxl_delever_overlay_redirect_symbol="SOXX", + ) + + signal_history = result["signal_history"] + triggered = signal_history.loc[signal_history["soxl_delever_overlay_triggered"].astype(bool)] + + assert result["summary"]["SOXL Delever Stops"] >= 1 + assert not triggered.empty + assert triggered["soxl_delever_overlay_threshold_mode"].eq("rolling_percentile").all() + assert triggered["soxl_delever_overlay_dynamic_threshold"].notna().all() + assert triggered["soxl_delever_overlay_dynamic_sample_count"].ge(20).all() + assert triggered["soxl_delever_overlay_threshold"].le(0.50).all() + assert ( + triggered["soxl_delever_overlay_metric"] >= triggered["soxl_delever_overlay_threshold"] + ).all() + + def test_soxl_soxx_dual_ma_research_overlay_keeps_partial_soxl() -> None: result = run_backtest( _build_dual_ma_research_prices(),