spb/hfmarketdata

1#!/usr/bin/env python32"""Walk-forward moving-average crossover backtest on HF Market Data bars.34  ma_crossover.py --asset etf --ticker SPY --start 2010-01-01 [--grid 10,20,50 --grid-slow 100,150,200]5                  [--train-years 3 --test-years 1] [--cost-bps 5] [--allow-short] [--fixed 50,200] [--plot out.png]67The reported metrics are OUT-OF-SAMPLE: for each test block the (fast, slow) pair is chosen on the8preceding training window only. Positions apply to the next bar (no look-ahead).9"""10from __future__ import annotations1112import argparse13import itertools14import sys15from pathlib import Path1617import numpy as np18import pandas as pd1920sys.path.insert(0, str(Path(__file__).parent))21import hfmd  # noqa: E402222324def signal_ma_cross(close: pd.Series, fast: int, slow: int, allow_short: bool) -> pd.Series:25    """Position in {0,1} (or {-1,0,1}) from past data only. Replace to test another idea."""26    f, s = close.rolling(fast).mean(), close.rolling(slow).mean()27    pos = (f > s).astype(float)28    if allow_short:29        pos = pos - (f < s).astype(float)30    return pos313233def run(close: pd.Series, pos: pd.Series, cost_bps: float) -> pd.DataFrame:34    ret = close.pct_change().fillna(0.0)35    p = pos.shift(1).fillna(0.0)  # decide at t, hold from t+136    trades = p.diff().abs().fillna(0.0)37    strat = p * ret - trades * cost_bps / 1e438    return pd.DataFrame({"ret": ret, "pos": p, "strat": strat, "trades": trades})394041def periods_per_year(dt: pd.Series) -> int:42    step = dt.diff().dropna().median()43    if step >= pd.Timedelta(days=1):44        return 252 if dt.dt.dayofweek.max() <= 4 else 36545    per_day = int(pd.Timedelta(hours=6.5) / step) if step < pd.Timedelta(hours=1) else int(pd.Timedelta(hours=24) / step)46    return max(per_day, 1) * 252474849def metrics(r: pd.Series, ppy: int) -> dict:50    if len(r) < 2:51        return {"cagr": np.nan, "vol": np.nan, "sharpe": np.nan, "max_dd": np.nan}52    eq = (1 + r).cumprod()53    years = len(r) / ppy54    return {55        "cagr": eq.iloc[-1] ** (1 / years) - 1 if years > 0 else np.nan,56        "vol": r.std() * np.sqrt(ppy),57        "sharpe": r.mean() / r.std() * np.sqrt(ppy) if r.std() > 0 else np.nan,58        "max_dd": (eq / eq.cummax() - 1).min(),59    }606162def main() -> int:63    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)64    ap.add_argument("--asset", required=True, choices=["stock", "etf", "crypto", "index", "fx", "futures"])65    ap.add_argument("--ticker", required=True)66    ap.add_argument("--timeframe", default="1day")67    ap.add_argument("--start", default="2010-01-01")68    ap.add_argument("--end")69    ap.add_argument("--adjustment", help="futures: prefer contin_adj_ratio")70    ap.add_argument("--grid", default="10,20,50", help="fast MA candidates")71    ap.add_argument("--grid-slow", default="100,150,200", help="slow MA candidates")72    ap.add_argument("--fixed", help="fast,slow — skip optimisation, evaluate one pair on the whole history")73    ap.add_argument("--train-years", type=float, default=3.0)74    ap.add_argument("--test-years", type=float, default=1.0)75    ap.add_argument("--cost-bps", type=float, default=5.0, help="cost per side per unit of position change")76    ap.add_argument("--allow-short", action="store_true")77    ap.add_argument("--show-insample", action="store_true", help="also print the best pair on the full history (for contrast)")78    ap.add_argument("--plot")79    ap.add_argument("--out-csv", help="write the out-of-sample equity curve")80    a = ap.parse_args()8182    adj = a.adjustment or ("contin_adj_ratio" if a.asset == "futures" else None)83    df = hfmd.bars(a.asset, a.ticker.upper(), a.timeframe, a.start, a.end, adj)84    if len(df) < 300:85        print(f"only {len(df)} bars — too short for a walk-forward", file=sys.stderr)86        return 187    df = df.set_index("datetime")88    close = df["close"].astype(float)89    ppy = periods_per_year(pd.Series(df.index))90    print(f"{a.ticker.upper()} {a.timeframe} {adj or ''}: {len(close):,} bars {close.index[0].date()} → {close.index[-1].date()} · periods/year={ppy} · cost={a.cost_bps} bps/side")9192    fasts = [int(x) for x in a.grid.split(",")]93    slows = [int(x) for x in a.grid_slow.split(",")]94    pairs = [(f, s) for f, s in itertools.product(fasts, slows) if f < s]9596    if a.fixed:97        f, s = (int(x) for x in a.fixed.split(","))98        res = run(close, signal_ma_cross(close, f, s, a.allow_short), a.cost_bps).iloc[s:]99        blocks = pd.DataFrame([{"test_start": res.index[0].date(), "test_end": res.index[-1].date(), "fast": f, "slow": s, **{f"oos_{k}": v for k, v in metrics(res["strat"], ppy).items()}}])100    else:101        train_n, test_n = int(a.train_years * ppy), int(a.test_years * ppy)102        if train_n < max(slows) + 20:103            print(f"train window ({train_n} bars) too short for slow MA {max(slows)}", file=sys.stderr)104            return 1105        pieces, rows = [], []106        start = train_n107        while start + 20 < len(close):108            tr = close.iloc[start - train_n:start]109            best, best_sh = None, -np.inf110            for f, s in pairs:111                sh = metrics(run(tr, signal_ma_cross(tr, f, s, a.allow_short), a.cost_bps)["strat"].iloc[s:], ppy)["sharpe"]112                if np.isfinite(sh) and sh > best_sh:113                    best, best_sh = (f, s), sh114            f, s = best or pairs[0]115            # compute the signal on history + test block so the MAs are warm at the block start116            seg = close.iloc[max(0, start - s - 5):start + test_n]117            res = run(seg, signal_ma_cross(seg, f, s, a.allow_short), a.cost_bps).loc[close.index[start]:]118            pieces.append(res)119            m = metrics(res["strat"], ppy)120            rows.append({"test_start": res.index[0].date(), "test_end": res.index[-1].date(), "fast": f, "slow": s, "train_sharpe": round(best_sh, 2), **{f"oos_{k}": v for k, v in m.items()}})121            start += test_n122        if not pieces:123            print("not enough data for one test block", file=sys.stderr)124            return 1125        res = pd.concat(pieces)126        blocks = pd.DataFrame(rows)127128    strat, bench = metrics(res["strat"], ppy), metrics(res["ret"], ppy)129    n_trades = int((res["trades"] > 0).sum())130    print(f"\nOUT-OF-SAMPLE {res.index[0].date()} → {res.index[-1].date()} ({len(res):,} bars, {len(blocks)} block(s))")131    print(f"{'':14}{'strategy':>12}{'buy&hold':>12}")132    for k in ("cagr", "vol", "sharpe", "max_dd"):133        print(f"{k:<14}{strat[k]:>12.3f}{bench[k]:>12.3f}")134    print(f"{'exposure':<14}{res['pos'].abs().mean():>12.2f}")135    print(f"{'trades':<14}{n_trades:>12d}   turnover/yr={res['trades'].sum() / (len(res) / ppy):.1f}")136    print("\nper block (parameters chosen on the preceding training window):")137    with pd.option_context("display.width", 160, "display.float_format", "{:.3f}".format):138        print(blocks.to_string(index=False))139140    if a.show_insample and not a.fixed:141        best = max(pairs, key=lambda p: metrics(run(close, signal_ma_cross(close, *p, a.allow_short), a.cost_bps)["strat"].iloc[p[1]:], ppy)["sharpe"] or -np.inf)142        m = metrics(run(close, signal_ma_cross(close, *best, a.allow_short), a.cost_bps)["strat"].iloc[best[1]:], ppy)143        print(f"\nIN-SAMPLE best pair on full history (optimistic, for contrast): {best} sharpe={m['sharpe']:.2f} cagr={m['cagr']:.3f} max_dd={m['max_dd']:.3f}")144145    if a.out_csv:146        out = res.copy()147        out["equity"] = (1 + out["strat"]).cumprod()148        out["benchmark"] = (1 + out["ret"]).cumprod()149        out.to_csv(a.out_csv)150        print(f"wrote {a.out_csv}")151152    if a.plot:153        try:154            import matplotlib155            matplotlib.use("Agg")156            import matplotlib.pyplot as plt157        except ImportError:158            print("matplotlib not installed", file=sys.stderr)159            return 0160        eq, bm = (1 + res["strat"]).cumprod(), (1 + res["ret"]).cumprod()161        fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(11, 6.5), sharex=True, gridspec_kw={"height_ratios": [3, 1]})162        ax1.plot(eq.index, eq, lw=1.3, label=f"MA cross walk-forward (Sharpe {strat['sharpe']:.2f})")163        ax1.plot(bm.index, bm, lw=1.0, alpha=0.7, label=f"buy & hold (Sharpe {bench['sharpe']:.2f})")164        if not a.fixed:165            for _, b in blocks.iterrows():166                ax1.axvline(pd.Timestamp(b["test_start"]), color="grey", alpha=0.25, lw=0.8)167                ax1.text(pd.Timestamp(b["test_start"]), ax1.get_ylim()[0], f"{b['fast']}/{b['slow']}", fontsize=7, rotation=90, va="bottom", alpha=0.7)168        ax1.set_yscale("log")169        ax1.grid(alpha=0.25)170        ax1.legend(loc="upper left")171        ax1.set_title(f"{a.ticker.upper()} — out-of-sample equity (log), cost {a.cost_bps} bps/side")172        ax2.fill_between(eq.index, (eq / eq.cummax() - 1) * 100, 0, alpha=0.4)173        ax2.set_ylabel("drawdown %")174        ax2.grid(alpha=0.25)175        fig.tight_layout()176        fig.savefig(a.plot, dpi=130)177        print(f"wrote {a.plot}")178    return 0179180181if __name__ == "__main__":182    sys.exit(main())183