#!/usr/bin/env python3 """ Backtest prototype strategies from Ernest P. Chan, *Algorithmic Trading* (2013). Fidelity modes (``--fidelity``): * **book** — matches MATLAB examples where noted: full-sample hedge / Johansen on expanding history, half-life lookback, uncapped linear units (Ch.2–3). Includes look-ahead on hedge ratios (Chan acknowledges this in Ch.3). * **causal** — rolling OLS hedge, quarterly Johansen refit, capped linear units. Strategies mapped to book examples: * Ch.2–3 linear MR (Ex 2.8) — EWA/EWC, Johansen triplets * Ch.3 Bollinger on spread (Ex 3.2) — ETF pairs, half-life lookback * Ch.3 Kalman hedge (Ex 3.3) — EWA/EWC * Ch.4 buy-on-gap / short-on-gap (Ex 4.1) * Ch.6 time-series momentum (Ex 6.1) **Not from Chan's book** (kept for comparison, tagged in output): * ``connors_rsi_*`` — Connors RSI pullback (repo ``SwingEngine``) Example:: cd /Users/robzingale/trading_bot && PYTHONUNBUFFERED=1 \\ .venv/bin/python RenTech/strategy_stack/run_chan_algorithmic_trading_benchmark.py \\ --start 2016-01-04 --end 2026-04-02 --fidelity book """ from __future__ import annotations import argparse import json import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Literal import numpy as np import pandas as pd _REPO = Path(__file__).resolve().parents[2] if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) import universe_scanner as us # type: ignore[import-not-found] from RenTech.strategy_stack.data_loader import DataLoader from RenTech.strategy_stack.statarb_engine import ( causal_ols_spread, engle_granger_spread, rolling_zscore, ) from RenTech.strategy_stack.swing_engine import SwingEngine LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT_PREFIX = LOGS / "chan_algorithmic_trading_benchmark" Fidelity = Literal["book", "causal"] try: from statsmodels.tsa.vector_ar.vecm import coint_johansen except ImportError: # pragma: no cover coint_johansen = None # type: ignore[misc, assignment] @dataclass(frozen=True) class StrategyResult: sid: str title: str chapter: str book_example: str fidelity: str daily_ret: pd.Series meta: dict[str, Any] def _close_series(df: pd.DataFrame) -> pd.Series: if "close" in df.columns: return df["close"].astype(np.float64) if "Close" in df.columns: return df["Close"].astype(np.float64) raise KeyError("DataFrame missing close/Close column") def _metrics(r: pd.Series, *, capital: float) -> dict[str, float]: r = r.fillna(0.0).astype(np.float64) n = len(r) if n < 2: return { "n_days": float(n), "total_return_pct": float("nan"), "cagr_pct": float("nan"), "sharpe": float("nan"), "max_drawdown_pct": float("nan"), "ending_equity_usd": float(capital), } eq = capital * (1.0 + r).cumprod() years = n / 252.0 end_eq = float(eq.iloc[-1]) tot = end_eq / capital - 1.0 cagr = (end_eq / capital) ** (1.0 / years) - 1.0 if years > 0 else float("nan") sd = float(r.std(ddof=1)) sharpe = float(r.mean() / sd * np.sqrt(252.0)) if sd > 1e-12 else float("nan") mdd = float((eq / eq.cummax() - 1.0).min()) return { "n_days": float(n), "total_return_pct": float(tot * 100.0), "cagr_pct": float(cagr * 100.0), "sharpe": sharpe, "max_drawdown_pct": float(mdd * 100.0), "ending_equity_usd": end_eq, } def _clip_window(r: pd.Series, start: pd.Timestamp, end: pd.Timestamp | None) -> pd.Series: r = r.copy() r.index = pd.to_datetime(r.index).tz_localize(None) mask = r.index >= start if end is not None: mask &= r.index <= end return r.loc[mask].fillna(0.0).astype(np.float64) def _half_life_lookback(spread: pd.Series, *, default: int = 20, cap: int = 120) -> int: hl = us.calculate_half_life(spread) if np.isfinite(hl) and 1 <= hl <= cap: return max(5, int(round(hl))) return default def _fill_missing_data(values: np.ndarray, *, initial: float = 0.0) -> np.ndarray: """Chan ``fillMissingData``: forward-fill NaNs from row 1 onward.""" out = values.copy() if out.size == 0: return out if np.isnan(out[0]): out[0] = initial for i in range(1, len(out)): if np.isnan(out[i]): out[i] = out[i - 1] return out def _chan_dollar_portfolio_returns( prices: pd.DataFrame, unit_weights: pd.DataFrame, num_units: pd.Series, ) -> pd.Series: """ Chan Ex 2.8 / 3.1 PnL: positions_{t,i} = numUnits_t × weight_{t,i} × price_{t,i} ret_t = Σ_i lag(positions) × Δprice/price / Σ_i |lag(positions)| """ nu = num_units.astype(np.float64) w = unit_weights.reindex(prices.index).astype(np.float64) px = prices.reindex(prices.index).astype(np.float64) dollar_pos = w.mul(px, axis=0).mul(nu, axis=0) pct = px.pct_change() pnl = (dollar_pos.shift(1) * pct).sum(axis=1) gross = dollar_pos.shift(1).abs().sum(axis=1).replace(0.0, np.nan) return (pnl / gross).fillna(0.0) def _spread_and_hedge( cy: pd.Series, cx: pd.Series, *, fidelity: Fidelity, hedge_window: int, min_train: int, ) -> tuple[pd.Series, pd.Series]: """Price spread y - β·x and per-bar hedge β (book: expanding OLS; causal: rolling).""" if fidelity == "book": spread, _, _, beta_fs, _ = engle_granger_spread(cy, cx, use_log=False) beta = pd.Series(float(beta_fs), index=cy.index) return spread, beta spread, _, _, beta = causal_ols_spread( cy, cx, use_log=False, hedge_window=hedge_window if hedge_window > 0 else None, min_train=min_train, ) return spread, beta def _linear_zscore_units(spread: pd.Series, lookback: int, *, fidelity: Fidelity) -> pd.Series: """Chan linear MR: numUnits = -(spread - MA) / STD.""" mu = spread.rolling(lookback, min_periods=max(2, lookback // 2)).mean() sig = spread.rolling(lookback, min_periods=max(2, lookback // 2)).std(ddof=1) z = (spread - mu) / sig.replace(0.0, np.nan) units = -z if fidelity == "causal": units = units.clip(-3.0, 3.0) return units.fillna(0.0) def _bollinger_units( spread: pd.Series, lookback: int, *, entry_z: float = 1.0, exit_z: float = 0.0, ) -> pd.Series: """ Chan Ex 3.2 (``bollinger.m``): separate long/short unit tracks with fill-forward. """ z = rolling_zscore(spread, lookback) zv = z.to_numpy(dtype=np.float64) n = len(zv) long_u = np.full(n, np.nan) short_u = np.full(n, np.nan) long_entry = zv < -entry_z long_exit = zv >= -exit_z short_entry = zv > entry_z short_exit = zv <= exit_z long_u[0] = 0.0 short_u[0] = 0.0 long_u[long_entry] = 1.0 long_u[long_exit] = 0.0 short_u[short_entry] = -1.0 short_u[short_exit] = 0.0 long_u = _fill_missing_data(long_u, initial=0.0) short_u = _fill_missing_data(short_u, initial=0.0) return pd.Series(long_u + short_u, index=spread.index) def _pair_bollinger_chan( df_y: pd.DataFrame, df_x: pd.DataFrame, *, fidelity: Fidelity, entry_z: float = 1.0, hedge_window: int = 60, min_train: int = 40, ) -> tuple[pd.Series, dict[str, Any]]: cy = _close_series(df_y) cx = _close_series(df_x) spread, beta = _spread_and_hedge( cy, cx, fidelity=fidelity, hedge_window=hedge_window, min_train=min_train ) lb = _half_life_lookback(spread) units = _bollinger_units(spread, lb, entry_z=entry_z, exit_z=0.0) weights = pd.DataFrame({cy.name or "y": 1.0, cx.name or "x": -beta}, index=cy.index) prices = pd.DataFrame({cy.name or "y": cy, cx.name or "x": cx}) ret = _chan_dollar_portfolio_returns(prices, weights, units) return ret, {"lookback": lb, "entry_z": entry_z, "spread": "price"} def _pair_linear_chan( df_y: pd.DataFrame, df_x: pd.DataFrame, *, fidelity: Fidelity, hedge_window: int = 60, min_train: int = 40, ) -> tuple[pd.Series, dict[str, Any]]: cy = _close_series(df_y) cx = _close_series(df_x) spread, beta = _spread_and_hedge( cy, cx, fidelity=fidelity, hedge_window=hedge_window, min_train=min_train ) lb = _half_life_lookback(spread) units = _linear_zscore_units(spread, lb, fidelity=fidelity) weights = pd.DataFrame({cy.name or "y": 1.0, cx.name or "x": -beta}, index=cy.index) prices = pd.DataFrame({cy.name or "y": cy, cx.name or "x": cx}) ret = _chan_dollar_portfolio_returns(prices, weights, units) return ret, {"lookback": lb, "spread": "price"} def _kalman_filter_hedge( y: np.ndarray, x: np.ndarray, *, delta: float = 0.0001, ve: float = 0.001, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: """ Chan Ex 3.3 (``KF_beta_EWA_EWC.m``): 2-state Kalman (slope + intercept). Returns beta (2×T), forecast error e, variance Q, and spread z-scores e/sqrt(Q). """ x_aug = np.column_stack([x.astype(np.float64), np.ones(len(x), dtype=np.float64)]) n = len(y) beta = np.zeros((2, n), dtype=np.float64) e = np.zeros(n, dtype=np.float64) q = np.zeros(n, dtype=np.float64) p = np.zeros((2, 2), dtype=np.float64) vw = delta / (1.0 - delta) * np.eye(2) yv = y.astype(np.float64) for t in range(n): if t > 0: beta[:, t] = beta[:, t - 1] r_mat = p + vw if t > 0 else p yhat = float(x_aug[t] @ beta[:, t]) qt = float(x_aug[t] @ r_mat @ x_aug[t] + ve) et = yv[t] - yhat k = r_mat @ x_aug[t] / qt beta[:, t] = beta[:, t] + k * et p = r_mat - np.outer(k, x_aug[t]) @ r_mat e[t] = et q[t] = qt return beta, e, q, e / np.sqrt(np.maximum(q, 1e-18)) def _pair_kalman_chan( df_y: pd.DataFrame, df_x: pd.DataFrame, ) -> tuple[pd.Series, dict[str, Any]]: """Chan Ex 3.3: Kalman hedge + entry when |e| > sqrt(Q).""" cy = _close_series(df_y).rename("y") cx = _close_series(df_x).rename("x") idx = cy.index.intersection(cx.index) cy = cy.reindex(idx) cx = cx.reindex(idx) beta, e, q, _ = _kalman_filter_hedge(cy.to_numpy(), cx.to_numpy()) sqrt_q = np.sqrt(np.maximum(q, 1e-18)) z = e / sqrt_q n = len(idx) long_u = np.full(n, np.nan) short_u = np.full(n, np.nan) long_u[0] = 0.0 short_u[0] = 0.0 long_u[z < -1.0] = 1.0 long_u[z >= -1.0] = 0.0 short_u[z > 1.0] = -1.0 short_u[z <= 1.0] = 0.0 long_u = _fill_missing_data(long_u, initial=0.0) short_u = _fill_missing_data(short_u, initial=0.0) units = pd.Series(long_u + short_u, index=idx) hedge = pd.Series(beta[0], index=idx) weights = pd.DataFrame({"y": 1.0, "x": -hedge}, index=idx) prices = pd.DataFrame({"y": cy, "x": cx}) ret = _chan_dollar_portfolio_returns(prices, weights, units) return ret, {"delta": 0.0001, "ve": 0.001} def _johansen_eigenvector( log_px: pd.DataFrame, t: int, *, fidelity: Fidelity, min_train: int, refit_bars: int, ) -> np.ndarray | None: if coint_johansen is None: raise ImportError("statsmodels required for Johansen tests") if fidelity == "book": window = log_px.iloc[: t + 1] else: start = max(0, t - min_train + 1) if (t - min_train) % refit_bars == 0 or t == min_train else None if start is None: return None window = log_px.iloc[start : t + 1] if len(window) < min_train: return None res = coint_johansen(window.to_numpy(), det_order=0, k_ar_diff=1) ev = res.evec[:, 0].astype(np.float64) if ev[0] != 0: ev = ev / ev[0] return ev def _johansen_linear_triplet( prices: pd.DataFrame, *, fidelity: Fidelity, refit_bars: int = 63, min_train: int = 252, ) -> tuple[pd.Series, dict[str, Any]]: """Chan Ex 2.7–2.8: Johansen eigenvector unit portfolio + linear z-score.""" px = prices.dropna(how="any").astype(np.float64) cols = list(px.columns) log_px = np.log(px) n = len(px) evec = np.full((n, len(cols)), np.nan) last_ev: np.ndarray | None = None for i in range(min_train, n): ev = _johansen_eigenvector( log_px, i, fidelity=fidelity, min_train=min_train, refit_bars=refit_bars ) if ev is not None: last_ev = ev if last_ev is not None: evec[i] = last_ev evec_df = pd.DataFrame(evec, index=px.index, columns=cols).ffill() yport = (px * evec_df).sum(axis=1) lb = _half_life_lookback(yport) units = _linear_zscore_units(yport, lb, fidelity=fidelity) ret = _chan_dollar_portfolio_returns(px, evec_df, units) return ret, {"lookback": lb, "min_train": min_train, "fidelity": fidelity} def _chan_tsmom(close: pd.Series, *, lookback: int = 250, hold_days: int = 25) -> pd.Series: """Chan Ex 6.1 (``TU_mom.m``) — exact return formula.""" c = close.astype(np.float64) lag_c = c.shift(1) daily_ret = (c - lag_c) / lag_c.replace(0.0, np.nan) longs = (c > c.shift(lookback)).fillna(False) shorts = (c < c.shift(lookback)).fillna(False) pos = np.zeros(len(c), dtype=np.float64) for h in range(hold_days): pos += longs.shift(h).fillna(False).astype(float).to_numpy() pos -= shorts.shift(h).fillna(False).astype(float).to_numpy() pos_s = pd.Series(pos, index=c.index) return (pos_s.shift(1) * daily_ret / float(hold_days)).fillna(0.0) def _gap_strategy( ohlc: dict[str, pd.DataFrame], *, side: Literal["long", "short"], top_n: int = 10, entry_zscore: float = 1.0, ma_lookback: int = 20, std_window: int = 90, ) -> pd.Series: """ Chan Ex 4.1 (``bog.m``): buy-on-gap long or short-on-gap mirror. Book normalizes daily PnL by ``topN`` even when fewer names qualify. """ closes: dict[str, pd.Series] = {} opens: dict[str, pd.Series] = {} lows: dict[str, pd.Series] = {} highs: dict[str, pd.Series] = {} for t, df in ohlc.items(): if df is None or df.empty: continue d = df.copy() d.index = pd.to_datetime(d.index).tz_localize(None) for src, dst in (("Close", "close"), ("Open", "open"), ("Low", "low"), ("High", "high")): if src in d.columns and dst not in d.columns: d[dst] = d[src] if not {"close", "open", "low"}.issubset(d.columns): continue closes[t] = d["close"].astype(np.float64) opens[t] = d["open"].astype(np.float64) lows[t] = d["low"].astype(np.float64) highs[t] = d.get("high", d["close"]).astype(np.float64) if not closes: raise ValueError("No OHLC data for gap strategy") cl = pd.DataFrame(closes).sort_index() op = pd.DataFrame(opens).reindex(cl.index) lo = pd.DataFrame(lows).reindex(cl.index) hi = pd.DataFrame(highs).reindex(cl.index) c2c = cl.pct_change() std90 = c2c.rolling(std_window, min_periods=std_window).std(ddof=1).shift(1) ma = cl.rolling(ma_lookback, min_periods=ma_lookback).mean().shift(1) ret_o2c = (cl - op) / op if side == "long": buy_price = lo.shift(1) * (1.0 - entry_zscore * std90) ret_gap = (op - lo.shift(1)) / lo.shift(1) else: buy_price = hi.shift(1) * (1.0 + entry_zscore * std90) ret_gap = (op - hi.shift(1)) / hi.shift(1) port_ret = pd.Series(0.0, index=cl.index) for t in range(1, len(cl)): row_gap = ret_gap.iloc[t] row_op = op.iloc[t] row_trig = buy_price.iloc[t] row_ma = ma.iloc[t] valid = row_gap.notna() & row_op.notna() & row_trig.notna() & row_ma.notna() if side == "long": valid &= row_op < row_trig valid &= row_op > row_ma picks = row_gap[valid].sort_values() else: valid &= row_op > row_trig valid &= row_op < row_ma picks = row_gap[valid].sort_values(ascending=False) if picks.empty: continue chosen = picks.index[: min(top_n, len(picks))] day_pnl = ret_o2c.iloc[t][chosen].mean() if np.isfinite(day_pnl): port_ret.iloc[t] = float(day_pnl) / float(top_n) return port_ret.fillna(0.0) def _pair_with_uso_gate( df_y: pd.DataFrame, df_x: pd.DataFrame, df_uso: pd.DataFrame, *, fidelity: Fidelity, oil_percentile: float = 0.90, ) -> tuple[pd.Series, dict[str, Any]]: """ Chan Ch.4: pause GLD/GDX when oil (USO) is elevated — use rolling 252d percentile gate. """ ret, meta = _pair_bollinger_chan(df_y, df_x, fidelity=fidelity) uso = _close_series(df_uso).reindex(ret.index) thresh = uso.rolling(252, min_periods=126).quantile(oil_percentile) gate = (uso <= thresh).astype(float).shift(1).fillna(0.0) return ret * gate, {**meta, "uso_gate_pct": oil_percentile} def _download_panel( tickers: list[str], *, lookback_years: float, refresh: bool, ) -> dict[str, pd.DataFrame]: return us.download_and_cache_data( tickers, timeframe="1d", lookback_years=float(lookback_years), max_age_hours=0.0 if refresh else 24.0, ) def _run_all( *, start: str, end: str, fidelity: Fidelity, lookback_years: float, refresh_cache: bool, bog_universe: str, skip_bog: bool, ) -> list[StrategyResult]: t0 = pd.Timestamp(start) t1 = pd.Timestamp(end) if end.strip() else None loader = DataLoader() results: list[StrategyResult] = [] etf_pairs = [ ("ewa_ewc_bollinger", "EWA/EWC Bollinger spread", "Ch.3", "Ex 3.2", "EWA", "EWC"), ("gld_gdx_bollinger", "GLD/GDX Bollinger spread", "Ch.4", "Ex 3.2", "GLD", "GDX"), ("rth_xlp_bollinger", "RTH/XLP Bollinger spread", "Ch.4", "Ex 3.2", "RTH", "XLP"), ] etf_tickers = sorted({t for row in etf_pairs for t in (row[4], row[5])}) etf_tickers += ["USO", "SPY", "QQQ", "IGE", "TLT", "GLD"] etf_data = _download_panel(etf_tickers, lookback_years=lookback_years, refresh=refresh_cache) for sid, title, chapter, example, y_t, x_t in etf_pairs: if y_t not in etf_data or x_t not in etf_data: print(f" Skip {sid}: missing {y_t} or {x_t}", flush=True) continue try: r, meta = _pair_bollinger_chan(etf_data[y_t], etf_data[x_t], fidelity=fidelity) results.append( StrategyResult( sid=sid, title=title, chapter=chapter, book_example=example, fidelity=fidelity, daily_ret=_clip_window(r, t0, t1), meta={**meta, "pair": f"{y_t}/{x_t}"}, ) ) print(f" OK {sid}", flush=True) except Exception as exc: print(f" Skip {sid}: {exc}", flush=True) if "EWA" in etf_data and "EWC" in etf_data: try: r, meta = _pair_linear_chan(etf_data["EWA"], etf_data["EWC"], fidelity=fidelity) results.append( StrategyResult( sid="ewa_ewc_linear", title="EWA/EWC linear z-score", chapter="Ch.2–3", book_example="Ex 2.8", fidelity=fidelity, daily_ret=_clip_window(r, t0, t1), meta={**meta, "pair": "EWA/EWC"}, ) ) print(" OK ewa_ewc_linear", flush=True) except Exception as exc: print(f" Skip ewa_ewc_linear: {exc}", flush=True) try: r, meta = _pair_kalman_chan(etf_data["EWC"], etf_data["EWA"]) results.append( StrategyResult( sid="ewa_ewc_kalman", title="EWA/EWC Kalman hedge MR", chapter="Ch.3", book_example="Ex 3.3", fidelity="book", daily_ret=_clip_window(r, t0, t1), meta={**meta, "pair": "EWC/Y EWA/X"}, ) ) print(" OK ewa_ewc_kalman", flush=True) except Exception as exc: print(f" Skip ewa_ewc_kalman: {exc}", flush=True) if "GLD" in etf_data and "GDX" in etf_data and "USO" in etf_data: try: r, meta = _pair_with_uso_gate( etf_data["GLD"], etf_data["GDX"], etf_data["USO"], fidelity=fidelity ) results.append( StrategyResult( sid="gld_gdx_uso_gate", title="GLD/GDX Bollinger + USO stress gate", chapter="Ch.4", book_example="Ch.4 GLD/GDX/USO", fidelity=fidelity, daily_ret=_clip_window(r, t0, t1), meta=meta, ) ) print(" OK gld_gdx_uso_gate", flush=True) except Exception as exc: print(f" Skip gld_gdx_uso_gate: {exc}", flush=True) triplets = [ ("ewa_ewc_ige_johansen", "EWA-EWC-IGE Johansen linear", "Ch.2–3", "Ex 2.7–2.8", ["EWA", "EWC", "IGE"]), ("gld_gdx_uso_johansen", "GLD-GDX-USO Johansen linear", "Ch.4", "Ex 2.7 + Ch.4", ["GLD", "GDX", "USO"]), ] for sid, title, chapter, example, legs in triplets: missing = [t for t in legs if t not in etf_data] if missing: print(f" Skip {sid}: missing {missing}", flush=True) continue try: px = pd.DataFrame({t: _close_series(etf_data[t]) for t in legs}).dropna(how="any") r, meta = _johansen_linear_triplet(px, fidelity=fidelity) results.append( StrategyResult( sid=sid, title=title, chapter=chapter, book_example=example, fidelity=fidelity, daily_ret=_clip_window(r, t0, t1), meta={**meta, "legs": legs}, ) ) print(f" OK {sid}", flush=True) except Exception as exc: print(f" Skip {sid}: {exc}", flush=True) # Not in Chan — labeled explicitly for tkr, sid in (("SPY", "connors_rsi_spy"), ("QQQ", "connors_rsi_qqq")): if tkr not in etf_data: etf_data[tkr] = loader.fetch_daily(tkr, period="max") try: eng = SwingEngine() out = eng.transform(pd.DataFrame({"close": _close_series(etf_data[tkr])})) close = out["close"].astype(np.float64) r = (out["micro_position"].astype(float).shift(1) * close.pct_change()).fillna(0.0) results.append( StrategyResult( sid=sid, title=f"Connors RSI pullback on {tkr} (NOT in Chan)", chapter="—", book_example="not in book", fidelity="n/a", daily_ret=_clip_window(r, t0, t1), meta={"ticker": tkr, "note": "repo SwingEngine; Connors not Chan 2013"}, ) ) print(f" OK {sid}", flush=True) except Exception as exc: print(f" Skip {sid}: {exc}", flush=True) for tkr, lb, hold, sid in ( ("SPY", 250, 25, "tsmom_spy_250_25"), ("TLT", 250, 25, "tsmom_tlt_250_25"), ("GLD", 250, 25, "tsmom_gld_250_25"), ): if tkr not in etf_data: etf_data[tkr] = loader.fetch_daily(tkr, period="max") try: r = _chan_tsmom(_close_series(etf_data[tkr]), lookback=lb, hold_days=hold) results.append( StrategyResult( sid=sid, title=f"TSMOM {tkr} ({lb}d/{hold}d)", chapter="Ch.6", book_example="Ex 6.1", fidelity="book", daily_ret=_clip_window(r, t0, t1), meta={"ticker": tkr, "lookback": lb, "hold_days": hold}, ) ) print(f" OK {sid}", flush=True) except Exception as exc: print(f" Skip {sid}: {exc}", flush=True) if not skip_bog: if bog_universe == "sp100": bog_tickers = us.get_sp100_tickers(universe="sp100") else: sym_path = _REPO / "alpaca_symbols.txt" bog_tickers = ( [ln.strip() for ln in sym_path.read_text().splitlines() if ln.strip()] if sym_path.is_file() else us.get_sp100_tickers(universe="sp100") ) print(f"Gap strategies: downloading {len(bog_tickers)} names …", flush=True) bog_data = _download_panel(bog_tickers, lookback_years=lookback_years, refresh=refresh_cache) for side, sid, title in ( ("long", "buy_on_gap", "Buy-on-gap long O→C"), ("short", "short_on_gap", "Short-on-gap mirror O→C"), ): try: r = _gap_strategy(bog_data, side=side) results.append( StrategyResult( sid=sid, title=f"{title} ({bog_universe})", chapter="Ch.4", book_example="Ex 4.1", fidelity="book", daily_ret=_clip_window(r, t0, t1), meta={"universe": bog_universe, "side": side, "top_n": 10}, ) ) print(f" OK {sid}", flush=True) except Exception as exc: print(f" Skip {sid}: {exc}", flush=True) return results def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--start", default="2016-01-04") ap.add_argument("--end", default="2026-04-02") ap.add_argument("--capital", type=float, default=100_000.0) ap.add_argument( "--fidelity", choices=("book", "causal"), default="book", help="book = Chan MATLAB-style (in-sample hedge); causal = rolling hedge", ) ap.add_argument("--lookback-years", type=float, default=12.0) ap.add_argument("--out-prefix", type=Path, default=DEFAULT_OUT_PREFIX) ap.add_argument("--refresh-cache", action="store_true") ap.add_argument("--bog-universe", choices=("sp100", "sp500"), default="sp100") ap.add_argument("--skip-bog", action="store_true") args = ap.parse_args() cap = float(args.capital) fidelity: Fidelity = str(args.fidelity) # type: ignore[assignment] print("Chan Algorithmic Trading benchmark", flush=True) print( f" window: {args.start} → {args.end or 'latest'} " f"capital=${cap:,.0f} fidelity={fidelity}", flush=True, ) results = _run_all( start=args.start, end=args.end, fidelity=fidelity, lookback_years=float(args.lookback_years), refresh_cache=bool(args.refresh_cache), bog_universe=str(args.bog_universe), skip_bog=bool(args.skip_bog), ) if not results: raise SystemExit("No strategies produced results.") rows: list[dict[str, Any]] = [] daily_cols: dict[str, pd.Series] = {} for res in results: m = _metrics(res.daily_ret, capital=cap) row = { "sid": res.sid, "title": res.title, "chapter": res.chapter, "book_example": res.book_example, "fidelity": res.fidelity, **m, **{f"meta_{k}": v for k, v in res.meta.items()}, } rows.append(row) daily_cols[res.sid] = res.daily_ret print( f"{res.sid:26s} ret={m['total_return_pct']:7.1f}% " f"CAGR={m['cagr_pct']:6.1f}% Sharpe={m['sharpe']:5.2f} " f"maxDD={m['max_drawdown_pct']:6.1f}%", flush=True, ) prefix = args.out_prefix.expanduser().resolve() prefix.parent.mkdir(parents=True, exist_ok=True) summary_path = Path(f"{prefix}_summary.csv") daily_path = Path(f"{prefix}_daily.csv") meta_path = Path(f"{prefix}_meta.json") pd.DataFrame(rows).sort_values("sharpe", ascending=False).to_csv(summary_path, index=False) daily = pd.DataFrame(daily_cols) daily.index.name = "date" daily = daily.reset_index() daily["date"] = pd.to_datetime(daily["date"]).dt.strftime("%Y-%m-%d") daily.to_csv(daily_path, index=False) meta_path.write_text( json.dumps( { "source": "Ernest P. Chan, Algorithmic Trading (2013)", "command": " ".join(sys.argv), "fidelity": fidelity, "implementation_notes": { "pair_spread": "price spread y - beta*x (Ex 3.2 PriceSpread.m), not log spread", "linear_units": "numUnits = -(spread-MA)/STD; book mode uncapped", "bollinger": "separate long/short tracks + fillMissingData (Ex 3.2)", "kalman": "delta=0.0001, Ve=0.001, |e|>sqrt(Q) entries (Ex 3.3)", "johansen_book": "expanding-window eigenvector; causal=quarterly refit", "buy_on_gap": "pnl/topN normalization per bog.m", "tsmom": "ret = lag(pos)*(cl-lag(cl))/lag(cl)/holddays per TU_mom.m", "not_in_book": ["connors_rsi_spy", "connors_rsi_qqq"], "still_missing": [ "cross-sectional MR (Ch.4)", "SPY component index arb (Ex 4.2)", "futures roll-return momentum (Ch.5–6)", "transaction costs", ], }, "strategies": rows, }, indent=2, default=str, ) ) print(f"\nWrote {summary_path}", flush=True) print(f"Wrote {daily_path}", flush=True) print(f"Wrote {meta_path}", flush=True) if __name__ == "__main__": main()