foxhunt/scripts/surfer/crypto_sweep.py

#!/usr/bin/env python3
"""Deflated signal sweep — Batch 2 (crypto perps): broaden + robustness-check.

Cross-sectional, market-neutral, unit-gross signals on major Binance USDT perps. Funding
baked into the return (Reff = log-return − daily funding; long pays positive funding). ~10bp
taker cost on turnover. Deflated Sharpe deflated by the CUMULATIVE search (Batch1 17 + these).
For any DEVELOP-grade signal (CPCVmed>0 & IS>0 & OOS>0): also report 2× cost + per-year Sharpe
(regime robustness). Survivorship caveat: v1 = currently-liquid majors over history.
"""
import glob
import math
import os
import sys

import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from signal_sweep import xs_weights, pnl_w, validate, sharpe_t  # noqa: E402
import torch  # noqa: E402

PRIOR_TRIALS = 17  # Batch 1 (futures factor zoo)
DEV = "cuda" if torch.cuda.is_available() else "cpu"


MIN_FUNDCOV = float(os.environ.get("MIN_FUNDCOV", "0.0"))   # filter coins by funding coverage
MIN_DAYS = int(os.environ.get("MIN_DAYS", "0"))             # ...and minimum history


def load_crypto():
    syms, data = [], {}
    for p in sorted(glob.glob("data/surfer/crypto/*.npz")):
        d = np.load(p); s = p.split("/")[-1][:-4]
        fund = d["funding"].astype(float)
        if len(d["day"]) < MIN_DAYS or np.mean(fund != 0) < MIN_FUNDCOV:
            continue
        data[s] = (d["day"].astype(np.int64), d["close"].astype(float), fund)
        syms.append(s)
    syms = sorted(syms)
    days = np.array(sorted(set().union(*[set(data[s][0].tolist()) for s in syms])))
    di = {d: i for i, d in enumerate(days.tolist())}
    T, N = len(days), len(syms)
    close = np.full((T, N), np.nan); fund = np.full((T, N), np.nan)
    for j, s in enumerate(syms):
        dd, cc, ff = data[s]
        for k in range(len(dd)):
            r = di[int(dd[k])]; close[r, j] = cc[k]; fund[r, j] = ff[k]
    return syms, days, close, fund


def trailing(lc, L):
    out = np.full_like(lc, np.nan); out[L:] = lc[L:] - lc[:-L]; return out


def zc(x):
    mu = np.nanmean(x, axis=1, keepdims=True); sd = np.nanstd(x, axis=1, keepdims=True)
    return np.nan_to_num((x - mu) / np.where(sd > 0, sd, 1))


def main():
    syms, days, close, fund = load_crypto()
    T, N = close.shape
    lc = np.log(close)
    R = np.zeros((T, N)); R[1:] = lc[1:] - lc[:-1]; R = np.where(np.isfinite(R), R, 0.0)
    fund = np.where(np.isfinite(fund), fund, 0.0)
    Reff = R - fund
    vol30 = np.full_like(lc, np.nan)
    for t in range(30, T):
        vol30[t] = np.nanstd(R[t - 30:t], axis=0)

    def tmf(K):  # trailing mean funding over K days
        out = np.full_like(fund, np.nan)
        for t in range(K, T):
            out[t] = np.nanmean(fund[t - K:t], axis=0)
        return out

    f7, f3, f14 = tmf(7), tmf(3), tmf(14)
    W = {}  # name -> weights[T,N]
    for K in [3, 7, 14, 30]:
        W[f"XS_carry_{K}"] = xs_weights(-tmf(K))
    for L in [7, 30, 90]:
        W[f"XS_mom_{L}"] = xs_weights(trailing(lc, L))
    W["XS_rev_3"] = xs_weights(-trailing(lc, 3))
    W["XS_fundmom"] = xs_weights(f3 - f14)               # rising funding (positioning building)
    W["XS_lowvol"] = xs_weights(-vol30)
    W["XS_carry+mom30"] = xs_weights(zc(-f7) + zc(trailing(lc, 30)))
    W["XS_carry+rev3"] = xs_weights(zc(-f7) + zc(-trailing(lc, 3)))
    W["XS_carry+lowvol"] = xs_weights(zc(-f7) + zc(-vol30))

    NT = PRIOR_TRIALS + len(W)
    year = (1970 + days / 365.25).astype(int)
    rows = []
    pnls = {}
    for nm, w in W.items():
        pnl = pnl_w(w, Reff, cost_bp=10)
        pnls[nm] = (w, pnl)
        rows.append((nm, validate(pnl, days, NT)))
    rows.sort(key=lambda r: -(r[1]["oos"] if not math.isnan(r[1]["oos"]) else -9))
    print(f"\n===== CRYPTO SWEEP (broadened) — {N} perps, {T}d ({days.min()}..{days.max()}), fundcov={np.mean(fund!=0):.2f} =====")
    print(f"N_trials(cumulative deflation) = {NT}")
    print(f"{'signal':>16} {'full':>6} {'IS':>6} {'OOS':>6} {'CPCVmed':>8} {'5%':>6} {'DSR':>5}")
    for nm, v in rows:
        print(f"{nm:>16} {v['full']:>+6.2f} {v['is_']:>+6.2f} {v['oos']:>+6.2f} {v['med']:>+8.2f} {v['p5']:>+6.2f} {v['dsr']:>5.2f}")

    print("\n----- ROBUSTNESS for develop-grade signals (CPCVmed>0 & IS>0 & OOS>0) -----")
    yrs = sorted(set(year[1:].tolist()))
    print(f"{'signal':>16} {'full10bp':>9} {'full20bp':>9} | per-year Sharpe: " + " ".join(f"{y}" for y in yrs))
    for nm, v in rows:
        if v["med"] > 0 and v["is_"] > 0 and v["oos"] > 0:
            w, pnl = pnls[nm]
            pnl2 = pnl_w(w, Reff, cost_bp=20)
            s10 = sharpe_t(torch.tensor(pnl, device=DEV, dtype=torch.float64))
            s20 = sharpe_t(torch.tensor(pnl2, device=DEV, dtype=torch.float64))
            py = []
            for y in yrs:
                m = year[1:] == y
                py.append(sharpe_t(torch.tensor(pnl[m], device=DEV, dtype=torch.float64)) if m.sum() > 30 else float("nan"))
            print(f"{nm:>16} {s10:>+9.2f} {s20:>+9.2f} | " + " ".join(f"{p:>+5.2f}" if not math.isnan(p) else "  n/a" for p in py))

    surv = [nm for nm, v in rows if v["dsr"] > 0.95 and v["oos"] > 0 and v["med"] > 0]
    print(f"\nDEPLOY survivors (DSR>0.95 & OOS>0 & CPCVmed>0): {surv if surv else 'NONE'}")
    print("Develop-grade = robust real edge worth building; deploy-grade = DSR>0.95 (harsh, deflated by all trials).")
    print("Caveat: survivorship (current majors); confirm point-in-time next.")


if __name__ == "__main__":
    main()