ML overlay experiment: walk-forward HGBT filter/scale, verdict = no material edge over baseline

ml_overlay.py

@@ -0,0 +1,163 @@
+"""ML overlay on the PASSing Donchian long-only trend strategy.
+
+Design (leakage-controlled):
+  - Features at day t use ONLY data through t-1.
+  - Target: next-20d return > 0 (binary), label window [t, t+20].
+  - Walk-forward: expanding train window, refit every 60 days,
+    PURGE = 20 days gap between train end and t (label horizon).
+  - Position = Donchian20_long_only signal (0/1) MODULATED by the model:
+      filter : position = donch if p > 0.5 else 0
+      scale  : position = donch * clip(2*(p-0.25), 0, 1)
+  - Model: HistGradientBoostingClassifier (sklearn), tiny configs only.
+
+Every ML variant is a counted trial. The verdict is the ML variant's
+purged-CPCV OOS Sharpe vs the Donchian baseline's (1.48 from sweep2).
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+from sklearn.ensemble import HistGradientBoostingClassifier
+
+from crypto_eval.engine2 import run_panel_backtest
+from crypto_eval.grid2 import donchian_panel
+from crypto_eval.cv import make_cpcv
+from crypto_eval.stats import eval_backtest
+
+ADV = {"BTC": 20e9, "ETH": 10e9, "LTC": 1e9, "XRP": 1.5e9,
+       "ADA": 0.5e9, "DOGE": 1e9, "ETC": 0.3e9, "DASH": 0.1e9}
+
+TARGET_HORIZON = 20
+REFIT_EVERY = 60
+MIN_TRAIN = 750
+
+
+def build_features(panel: pd.DataFrame) -> pd.DataFrame:
+    """Feature rows indexed by decision date t; all inputs end at t-1."""
+    c = panel.shift(1)  # data through t-1
+    ret = c.pct_change()
+    feats = {}
+    for lb in (10, 20, 50, 100):
+        feats[f"mom_{lb}"] = c / c.shift(lb) - 1
+    feats["vol_20"] = ret.rolling(20).std() * np.sqrt(365)
+    feats["vol_60"] = ret.rolling(60).std() * np.sqrt(365)
+    feats["ma_dist_100"] = c / c.rolling(100).mean() - 1
+    feats["donch_rank"] = c / c.rolling(20).max()  # near-high vs not
+    X = pd.concat(feats, axis=1)  # MultiIndex columns: (feature, symbol)
+    return X
+
+
+def build_target(panel: pd.DataFrame) -> pd.Series:
+    """next-20d return > 0, indexed by decision date t (label window
+    [t, t+20]; consumers must not train on rows within 20d of t)."""
+    fwd = panel.shift(-TARGET_HORIZON) / panel - 1
+    return (fwd["ETH"] > 0).astype(float)
+
+
+def walk_forward_positions(panel: pd.DataFrame, X: pd.DataFrame,
+                           y: pd.Series, sym: str, mode: str, depth: int,
+                           lr: float) -> pd.Series:
+    """Walk-forward predictions -> position series on the asset's index."""
+    idx = X.index
+    donch = donchian_panel(panel[[sym]], window=20, long_only=True)[sym]
+    pos = pd.Series(0.0, index=idx)
+    y_arr = y.reindex(idx).to_numpy()
+    X_sym = X.xs(sym, axis=1, level=1)
+    valid = X_sym.notna().all(axis=1) & ~np.isnan(y_arr)
+
+    for start in range(MIN_TRAIN, len(idx) - 1, REFIT_EVERY):
+        train_end = start - TARGET_HORIZON  # purge: labels end before t
+        tr = np.where(valid.to_numpy()[:train_end])[0]
+        if len(tr) < MIN_TRAIN:
+            continue
+        model = HistGradientBoostingClassifier(
+            max_depth=depth, learning_rate=lr, max_iter=150, random_state=0)
+        model.fit(X_sym.to_numpy()[:train_end][tr], y_arr[:train_end][tr])
+        # predict one refit window at a time
+        end = min(start + REFIT_EVERY, len(idx) - 1)
+        p = model.predict_proba(X_sym.to_numpy()[start:end])[:, 1]
+        for j, t in enumerate(range(start, end)):
+            if not valid.iloc[t]:
+                continue
+            d = donch.iloc[t]
+            if mode == "filter":
+                pos.iloc[t] = d if p[j] > 0.5 else 0.0
+            else:  # scale
+                pos.iloc[t] = d * float(np.clip(2.0 * (p[j] - 0.25), 0.0, 1.0))
+    # leading rows before first refit: plain donch (documented: warmup)
+    pos.iloc[:MIN_TRAIN] = donch.iloc[:MIN_TRAIN]
+    return pos
+
+
+def main() -> int:
+    panel = pd.read_parquet("/tmp/data/panel_daily_close.parquet")
+    panel.index = pd.to_datetime(panel.index, utc=True)
+    panel = panel.sort_index()["2016-01-01":]  # ML needs longer warmup
+    eth = panel[["ETH"]]
+    print(f"[ml] panel {panel.shape[0]} days from {panel.index[0].date()}",
+          flush=True)
+
+    X = build_features(eth)
+    y = build_target(eth)
+
+    baseline_pos = donchian_panel(eth, window=20, long_only=True)
+    results = {}
+
+    variants = [
+        ("ml_filter_d2_lr005", "filter", 2, 0.05),
+        ("ml_filter_d3_lr005", "filter", 3, 0.05),
+        ("ml_scale_d2_lr005", "scale", 2, 0.05),
+        ("ml_scale_d3_lr01", "scale", 3, 0.10),
+    ]
+    for name, mode, depth, lr in variants:
+        print(f"[ml] {name}: walk-forward refit ...", flush=True)
+        pos = walk_forward_positions(eth, X, y, "ETH", mode, depth, lr)
+        pos_df = pos.to_frame(name="ETH")
+        bt = run_panel_backtest(eth, pos_df, {"ETH": ADV["ETH"]})
+        rets = bt["portfolio_return"].to_numpy()
+        _, combos, audit = make_cpcv(bt["portfolio_return"].index,
+                                     horizon=1, embargo=5, freq="1D")
+        assert audit.ok
+        oos = [eval_backtest(rets[t], n_trials=1, trial_sr_variance=0.0,
+                             periods_per_year=365) for _, t in combos]
+        srs = [c["sharpe_annualized"] for c in oos]
+        full_sr = float(np.nanmean(rets) / (np.nanstd(rets, ddof=1) + 1e-12)
+                        * np.sqrt(365))
+        results[name] = {
+            "full_sample_sr": full_sr,
+            "oos_mean_sr": float(np.mean(srs)),
+            "oos_combos_positive": int(sum(1 for s in srs if s > 0)),
+            "turnover_pct_exposure": float(pos.mean()),
+        }
+        print(f"[ml] {name}: full {full_sr:.2f}  OOS {np.mean(srs):.2f} "
+              f"({results[name]['oos_combos_positive']}/15)",
+              flush=True)
+
+    # baseline on the same window for a fair comparison
+    bt = run_panel_backtest(eth, baseline_pos, {"ETH": ADV["ETH"]})
+    rets = bt["portfolio_return"].to_numpy()
+    _, combos, _ = make_cpcv(bt["portfolio_return"].index, horizon=1,
+                             embargo=5, freq="1D")
+    oos = [eval_backtest(rets[t], n_trials=1, trial_sr_variance=0.0,
+                         periods_per_year=365) for _, t in combos]
+    srs = [c["sharpe_annualized"] for c in oos]
+    results["baseline_donch20_no_ml"] = {
+        "full_sample_sr": float(np.nanmean(rets) / np.nanstd(rets, ddof=1)
+                                * np.sqrt(365)),
+        "oos_mean_sr": float(np.mean(srs)),
+        "oos_combos_positive": int(sum(1 for s in srs if s > 0)),
+        "turnover_pct_exposure": float(baseline_pos["ETH"].mean()),
+    }
+    print(f"[ml] baseline: {results['baseline_donch20_no_ml']}", flush=True)
+
+    Path("/tmp/ml_report.json").write_text(json.dumps(results, indent=2))
+    print("[ml] done", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())