Tutorial 03 - Residual mean reversion: RSI, Bollinger and residual bands

from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd
from IPython.display import display

from examples.quant_trading.data import load_sample_goog_ohlcv, market_data_manifest, ohlcv_audit_report
from examples.quant_trading.features import build_feature_table, decompose_one_series, walkforward_decompose_ohlcv
from examples.quant_trading.classic_indicators import bollinger_bands, rsi
from examples.quant_trading.strategy_mean_reversion import (
    compare_mean_reversion_suites,
    make_classic_mean_reversion_weight_grid,
    make_detime_mean_reversion_weight_grid,
    run_classical_mean_reversion_baselines,
    run_detime_mean_reversion_baselines,
)
from examples.quant_trading.validation import compare_weight_strategies, turnover_report, write_run_audit, write_run_manifest

pd.set_option("display.max_columns", 30)
report_dir = Path("examples/quant_trading/reports")
report_dir.mkdir(parents=True, exist_ok=True)

ohlcv_single = load_sample_goog_ohlcv(trim_start="2014-01-01")
ticker = ohlcv_single.attrs.get("symbol", "GOOG")
ohlcv = {field: ohlcv_single[[field]].rename(columns={field: ticker}) for field in ["Open", "High", "Low", "Close", "Volume"]}
prices = ohlcv["Close"]
manifest = market_data_manifest(
    tickers=[ticker],
    start=str(prices.index.min().date()),
    end=str(prices.index.max().date()),
    interval="1d",
    source=ohlcv_single.attrs.get("source", "bundled real sample"),
)
audit = ohlcv_audit_report(ohlcv)
display(audit)

classic_weights = make_classic_mean_reversion_weight_grid(prices, allow_short=False)
classic_table, classic_results = compare_weight_strategies(prices, classic_weights, fee_bps=1.0, slippage_bps=2.0)
display(classic_table[["total_return", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4))

bands = bollinger_bands(prices, window=20, num_std=2.0)
raw_rsi = rsi(prices, window=14)
fig, ax = plt.subplots(figsize=(10, 4))
prices[ticker].plot(ax=ax, linewidth=1.0, label="close")
bands.middle[ticker].plot(ax=ax, linewidth=0.9, label="Bollinger middle")
bands.upper[ticker].plot(ax=ax, linewidth=0.8, label="upper")
bands.lower[ticker].plot(ax=ax, linewidth=0.8, label="lower")
ax.set_title("Classical Bollinger input: raw price relative to a rolling price mean")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

fig, ax = plt.subplots(figsize=(10, 2.8))
raw_rsi[ticker].plot(ax=ax, linewidth=1.0, label="RSI(14)")
for level in (30, 50, 70):
    ax.axhline(level, linestyle="--", linewidth=0.8)
ax.set_title("Classical RSI input")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

image/png

image/png

features = walkforward_decompose_ohlcv(
    ohlcv,
    method="STL",
    period="auto",
    period_candidates=(63, 126, 252),
    train_window=504,
    step=5,
    z_window=63,
)
feature_tail = build_feature_table(prices, features).tail(120)
display(feature_tail.tail(5).round(4))
feature_tail.to_csv(report_dir / "column_03_feature_table_tail.csv")

diagnostic = decompose_one_series(
    prices[ticker],
    method="STL",
    period="auto",
    period_candidates=(63, 126, 252),
    z_window=63,
    transform="log",
)
volume_diagnostic = decompose_one_series(
    ohlcv["Volume"][ticker],
    method="STL",
    period=int(diagnostic.attrs.get("period", 126)),
    z_window=63,
    transform="log1p",
)

fig, ax = plt.subplots(figsize=(10, 3.5))
diagnostic["residual_z"].plot(ax=ax, linewidth=1.1, color="#2563eb", label="residual z-score")
for level in (-1.5, 0, 1.5):
    ax.axhline(level, linestyle="--" if level else "-", linewidth=0.8, color="black" if level == 0 else "#64748b")
ax.set_title("Continuous residual z-score after trend and cycle removal")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

fig, ax = plt.subplots(figsize=(10, 3.5))
diagnostic["cycle_z"].plot(ax=ax, linewidth=1.0, color="#16a34a", label="cycle z")
diagnostic["residual_z"].plot(ax=ax, linewidth=1.0, color="#2563eb", alpha=0.75, label="residual z")
volume_diagnostic["residual_z"].plot(ax=ax, linewidth=0.8, color="#f97316", alpha=0.70, label="volume residual z")
ax.axhline(0, color="black", linewidth=0.8)
ax.set_title("Continuous context filters used by residual reversion")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

image/png

image/png

detime_weights = make_detime_mean_reversion_weight_grid(prices, features, allow_short=False)
detime_table, detime_results = compare_weight_strategies(prices, detime_weights, fee_bps=1.0, slippage_bps=2.0)
display(detime_table[["total_return", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4))

classical = run_classical_mean_reversion_baselines(prices, allow_short=False, fee_bps=1.0, slippage_bps=2.0)
detime = run_detime_mean_reversion_baselines(prices, features, allow_short=False, fee_bps=1.0, slippage_bps=2.0)
comparison = compare_mean_reversion_suites(classical, detime)
display(comparison[["strategy_group", "cagr", "sharpe", "max_drawdown", "average_turnover", "hit_rate"]].round(4))
comparison.to_csv(report_dir / "column_03_strategy_comparison.csv")
turnover_report({**classic_weights, **detime_weights}).to_csv(report_dir / "column_03_turnover_report.csv")
write_run_audit(report_dir, data_manifest=manifest, audit=audit, strategy_stats=comparison, prefix="column_03")
manifest_path = write_run_manifest(
    report_dir / "column_03_run_manifest.json",
    command="notebook:03_residual_mean_reversion_rsi_bollinger",
    dataset="bundled_real_GOOG",
    strategies=list(comparison.index),
    result_file=str(report_dir / "column_03_strategy_comparison.csv"),
)
manifest_path.as_posix()

'examples/quant_trading/reports/column_03_run_manifest.json'

chosen = "detime_residual_z_1p0_cycle_volume"
fig, ax1 = plt.subplots(figsize=(10, 4))
prices[ticker].plot(ax=ax1, linewidth=1.0, label="close")
ax1.set_title(f"Position overlay: {chosen}")
ax2 = ax1.twinx()
detime_weights[chosen][ticker].plot(ax=ax2, linewidth=0.9, alpha=0.7, label="position")
ax2.set_ylabel("position")
ax1.grid(True, alpha=0.3)
plt.show()

image/png