"""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())