ml_overlay.py

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