Tutorial 05 - Pair spread decomposition and stat-arb

import matplotlib.pyplot as plt
import pandas as pd

from quant_trading.data import load_bundled_real_ohlcv_panel, ohlcv_audit_report
from quant_trading.strategy_pairs import (
    walkforward_pair_spread_features, pair_diagnostics, pair_feature_snapshot,
    make_classic_pair_weight_grid, make_detime_pair_weight_grid,
    compare_pair_suites, run_classical_pair_baselines, run_detime_pair_baselines,
)
from quant_trading.validation import turnover_report, compare_weight_strategies

tickers = ["AUDUSD=X", "NZDUSD=X", "EURUSD=X", "GBPUSD=X"]
pairs = [("AUDUSD=X", "NZDUSD=X"), ("EURUSD=X", "GBPUSD=X")]
ohlcv = load_bundled_real_ohlcv_panel(tickers, min_observations=120)
ohlcv = {field: table.tail(760).copy() for field, table in ohlcv.items()}
prices = ohlcv["Close"]
ohlcv_audit_report(ohlcv)

spread_features, spread_panel, beta_panel, pair_specs = walkforward_pair_spread_features(
    prices,
    pairs,
    hedge_window=126,
    method="STL",
    period=126,
    train_window=504,
    step=5,
    z_window=63,
)
spread_panel.tail()

pair_diagnostics(prices, pair_specs, hedge_window=90)

snapshot = pair_feature_snapshot(spread_features, tail=2)
snapshot.query("feature in ['trend_slope', 'cycle_slope', 'residual_z', 'residual_abs_z']").tail(16)

classic_weights = make_classic_pair_weight_grid(prices, pairs=pairs, lookback=90)
detime_weights = make_detime_pair_weight_grid(prices, pair_specs, spread_features, spread_panel=spread_panel, beta_panel=beta_panel)
all_weights = {**classic_weights, **detime_weights}
comparison, results = compare_weight_strategies(prices, all_weights, fee_bps=1.0, slippage_bps=2.0)
comparison.insert(0, "strategy_group", ["detime_pair" if str(idx).startswith("detime") else "classical_pair" for idx in comparison.index])
comparison[["strategy_group", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4)

leaders = comparison.sort_values("sharpe", ascending=False).head(4).index
fig, ax = plt.subplots(figsize=(10, 4))
for name in leaders:
    results[name].equity.plot(ax=ax, linewidth=1.2, label=name)
ax.set_title("Pair-spread equity curves on real FX samples")
ax.set_ylabel("Equity")
ax.legend(loc="best", fontsize=8)
ax.grid(True, alpha=0.25)
plt.tight_layout()
plt.show()

image/png

turnover_report(all_weights).round(4)