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

5c5ebe1 verified about 9 hours ago

6.8 kB

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

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