PythagorasGoal/scripts/research/fut_overnight_leadlag.py

"""TEST ONESTO finestre NON sovrapposte: crypto[notte presto] -> future indice[notte tardi].

L'idea "monitor Deribit / trade IB" e' valida SOLO se il crypto anticipa il moto SUCCESSIVO del future
(non lo stesso intervallo = look-ahead). Qui:
  entrata a T (ora UTC nella notte, equity cash chiuso): osservato crypto fino a T.
  segnale  = crypto[P 21:00 -> T]            (info nota a T)
  controllo= future[P 21:00 -> T]            (il moto PROPRIO del future fino a T)
  cattura  = future[T -> D 13:00]            (cio' che catturi entrando a T, fino a ~apertura cash)
TEST: il segnale crypto predice la cattura INCREMENTALMENTE al controllo? (crypto vede prima del future?)
TRADE: long/short future su sign(crypto[P21:00->T]), cattura future[T->open], net ~2bps. Sharpe/anno/OOS.
Confronto col baseline sign(future[P21:00->T]) (momentum proprio del future) per isolare il valore crypto.

Dati: data/raw/fut_<sym>_1h.parquet (UTC) + crypto 1h. sqrt(252).
"""
import sys, argparse, json
from pathlib import Path
import numpy as np, pandas as pd

ROOT = Path(__file__).resolve().parents[2]
RAW = ROOT / "data" / "raw"
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts" / "research"))
from crypto_lead_harness import crypto_hourly, at

OOS = pd.Timestamp("2024-01-01", tz="UTC"); COST = 0.0002
CLOSE_H = 21; OPEN_H = 13


def fut_hourly(sym):
    d = pd.read_parquet(RAW / f"fut_{sym.lower()}_1h.parquet")
    return pd.Series(d["close"].astype(float).values, index=pd.to_datetime(d["timestamp"], unit="ms", utc=True)).sort_index()


def _sh(r):
    r = np.asarray(r, float); r = r[np.isfinite(r)]
    return float(np.mean(r) / np.std(r) * np.sqrt(252)) if len(r) > 5 and np.std(r) > 0 else 0.0


def run(sym="ES", lead="BTC", h_entry=6):
    fut = fut_hourly(sym); bc = crypto_hourly(lead)
    days = pd.date_range(fut.index[0].normalize(), fut.index[-1].normalize(), freq="B", tz="UTC")
    rows = []
    for k in range(1, len(days)):
        D = days[k]; P = days[k-1]
        t_pc = P + pd.Timedelta(hours=CLOSE_H)        # P close 21:00
        t_e = D + pd.Timedelta(hours=h_entry)          # entrata T
        t_o = D + pd.Timedelta(hours=OPEN_H)           # ~apertura cash 13:00
        fp, fe, fo = at(fut, t_pc), at(fut, t_e), at(fut, t_o)
        cp, ce = at(bc, t_pc), at(bc, t_e)
        if not all(np.isfinite(v) and v > 0 for v in (fp, fe, fo, cp, ce)):
            continue
        crypto_sig = ce / cp - 1.0          # crypto P21:00 -> T
        fut_ctrl = fe / fp - 1.0            # future P21:00 -> T (controllo)
        capture = fo / fe - 1.0             # future T -> open (cio' che catturi)
        rows.append((D, crypto_sig, fut_ctrl, capture))
    if len(rows) < 100:
        return {"sym": sym, "lead": lead, "h_entry": h_entry, "err": f"n={len(rows)}"}
    Dd = pd.DataFrame(rows, columns=["d", "csig", "fctrl", "cap"]).set_index("d")
    x, ctrl, y = Dd["csig"].values, Dd["fctrl"].values, Dd["cap"].values

    def z(a):
        s = a.std(); return (a - a.mean()) / s if s > 0 else a * 0
    # incrementale: cap ~ crypto + future_own_move
    X = np.column_stack([np.ones(len(y)), z(x), z(ctrl)])
    beta, *_ = np.linalg.lstsq(X, z(y), rcond=None)
    resid = z(y) - X @ beta; dof = max(len(y) - 3, 1)
    se = np.sqrt(np.sum(resid**2) / dof * np.diag(np.linalg.inv(X.T @ X)))
    t_crypto = float(beta[1] / se[1]) if se[1] > 0 else 0.0
    t_futown = float(beta[2] / se[2]) if se[2] > 0 else 0.0
    # trade: crypto-signal vs future-own-signal
    C_crypto = np.sign(x) * y - COST
    C_futown = np.sign(ctrl) * y - COST
    m = Dd.index >= OOS
    yrs = Dd.index.year.values
    py = {int(yv): round(_sh(C_crypto[yrs == yv]), 2) for yv in sorted(set(yrs)) if (yrs == yv).sum() >= 30}
    return {"sym": sym, "lead": lead, "h_entry": h_entry, "n": len(Dd),
            "corr_crypto_cap": round(float(np.corrcoef(x, y)[0, 1]), 3),
            "t_crypto_incr": round(t_crypto, 2), "t_futown_incr": round(t_futown, 2),
            "sh_crypto_full": round(_sh(C_crypto), 2), "sh_crypto_oos": round(_sh(C_crypto[m]), 2),
            "sh_futown_full": round(_sh(C_futown), 2),
            "ann_crypto_pct": round(float(np.nanmean(C_crypto) * 252 * 100), 1),
            "per_year": py}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--syms", default="ES,NQ,RTY")
    ap.add_argument("--leads", default="BTC,ETH")
    ap.add_argument("--entries", default="0,3,6,9")
    args = ap.parse_args()
    print("=" * 96)
    print("  FUTURE OVERNIGHT — crypto[P21:00->T] predice future[T->open]? (finestre NON sovrapposte)")
    print("=" * 96)
    print("  cattura = future[T->open]; controllo = future[P21:00->T]; t_crypto_incr = crypto AL NETTO del future. net 2bps\n")
    best = []
    for sym in args.syms.split(","):
        for lead in args.leads.split(","):
            for h in [int(x) for x in args.entries.split(",")]:
                r = run(sym, lead, h)
                if "err" in r:
                    print(f"  {sym}<-{lead} T={h}h: {r['err']}"); continue
                print(f"  {sym}<-{lead} T={h:>2}h UTC: n={r['n']} corr {r['corr_crypto_cap']:+.3f} | "
                      f"t_crypto(incr) {r['t_crypto_incr']:+.1f} t_futOwn {r['t_futown_incr']:+.1f} | "
                      f"trade crypto Sh {r['sh_crypto_full']:+.2f} (OOS {r['sh_crypto_oos']:+.2f}) ann {r['ann_crypto_pct']:+.1f}% | "
                      f"futOwn Sh {r['sh_futown_full']:+.2f}")
                best.append(r)
    best.sort(key=lambda r: r.get("sh_crypto_oos", -9), reverse=True)
    print("\n  --- migliori per Sharpe OOS (trade su segnale crypto) ---")
    for r in best[:5]:
        print(f"    {r['sym']}<-{r['lead']} T={r['h_entry']}h: OOS {r['sh_crypto_oos']} full {r['sh_crypto_full']} "
              f"t_crypto {r['t_crypto_incr']} per-anno {r['per_year']}")
    print("\n  NB: t_crypto_incr alto E sh_crypto > sh_futOwn => il crypto anticipa il future (idea valida).")
    print("      t_crypto ~0 o sh_crypto ~ futOwn => e' solo momentum del future, il crypto non aggiunge.")


if __name__ == "__main__":
    main()