PythagorasGoal/scripts/research/crypto_lead_harness.py

"""HARNESS parametrizzato — anticipazione crypto -> mercato (lead-lag eseguibile, onesto).

Generalizza l'effetto weekend: la finestra-LEAD e' l'intervallo in cui l'equity e' CHIUSO e la crypto
no (prev close 21:00 UTC -> next open 13:30 UTC). Il weekend e' il caso lungo (Ven 21:00 -> Lun 13:30).
Per ogni sessione equity D (con sessione precedente P):
  lead = crypto return su [lead_start, D 13:00]   (lead_start = P 21:00 se hours='overnight', else D13:00-hours)
  predict target: gap = open[D]/close[P]-1 ; intraday = close[D]/open[D]-1 ; full = close[D]/close[P]-1
  control = rendimento sessione precedente equity (close[P]/close[P_prev]-1)  -> test INCREMENTALE
Filtro giorni: all | mon (solo lunedi'/weekend) | nonmon.

Output JSON per config: n, corr, beta+t-stat del lead AL NETTO del control (incrementale), Sharpe
settimanale/annualizzato del trade eseguibile (sign(lead)*predict - costo) FULL/IS/OOS(2022+),
hit-rate, e PER-ANNO (hit e mean) per la robustezza multi-anno.

  uv run python scripts/research/crypto_lead_harness.py --configs '[{"lead":"BTC","target":"QQQ","day":"mon","predict":"intraday","hours":"overnight"}]'
Dati: cache su disco (crypto 1h, ETF eq_*). Nessun IB online. Vettoriale, veloce.
"""
import sys, json, argparse
from pathlib import Path
import numpy as np, pandas as pd

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts" / "research"))
from src.data.downloader import load_data
import eqlib

OOS_DEFAULT = "2022-01-01"
OPEN_H = 13   # ~apertura US 13:30 UTC -> uso barra 13:00 (info nota prima dell'open per il lead)
CLOSE_H = 21  # ~chiusura US 21:00 UTC

_CRYPTO = {}
def crypto_hourly(asset):
    if asset not in _CRYPTO:
        s = load_data(asset, "1h").set_index("datetime")["close"].astype(float)
        full = pd.date_range(s.index[0].floor("h"), s.index[-1].ceil("h"), freq="h", tz="UTC")
        _CRYPTO[asset] = s.reindex(s.index.union(full)).ffill().reindex(full)
    return _CRYPTO[asset]


def at(series, ts):
    try:
        return float(series.asof(ts))
    except Exception:
        return np.nan


def evaluate(cfg, cost_rt=0.0004, oos=OOS_DEFAULT):
    OOS = pd.Timestamp(oos, tz="UTC")
    lead = cfg["lead"]; tgt = cfg["target"]; day = cfg.get("day", "all")
    predict = cfg.get("predict", "intraday"); hours = cfg.get("hours", "overnight")
    bc = crypto_hourly(lead)
    try:
        oc = eqlib.load_eq(tgt)["open"].astype(float); cc = eqlib.load_eq(tgt)["close"].astype(float)
    except Exception as e:
        return {**cfg, "err": f"no data {tgt}"}
    idx = cc.index
    rows = []
    for j in range(2, len(idx)):
        D = idx[j]; P = idx[j-1]; Pp = idx[j-2]
        if day == "mon" and D.weekday() != 0: continue
        if day == "nonmon" and D.weekday() == 0: continue
        d_open = D.normalize() + pd.Timedelta(hours=OPEN_H)
        p_close = P.normalize() + pd.Timedelta(hours=CLOSE_H)
        lead_start = p_close if hours == "overnight" else d_open - pd.Timedelta(hours=int(hours))
        c1 = at(bc, d_open); c0 = at(bc, lead_start)
        if not (np.isfinite(c1) and np.isfinite(c0) and c0 > 0): continue
        ld = c1 / c0 - 1.0
        gap = oc[D] / cc[P] - 1.0
        intr = cc[D] / oc[D] - 1.0
        full = cc[D] / cc[P] - 1.0
        ctrl = cc[P] / cc[Pp] - 1.0
        rows.append((D, ld, gap, intr, full, ctrl))
    if len(rows) < 60:
        return {**cfg, "err": f"n={len(rows)}"}
    D_ = pd.DataFrame(rows, columns=["d", "lead", "gap", "intraday", "full", "ctrl"]).set_index("d")
    y = D_[predict].values; x = D_["lead"].values; ctrl = D_["ctrl"].values

    def z(a):
        sd = a.std(); return (a - a.mean()) / sd if sd > 0 else a * 0
    corr = float(np.corrcoef(x, y)[0, 1])
    # incrementale vs control (OLS standardizzato)
    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_inc = float(beta[1] / se[1]) if se[1] > 0 else 0.0
    # trade eseguibile: long-short e long-flat su segno del lead, intraday/predict, net costi
    sign = np.sign(x)
    def sharpe(r):
        r = r[np.isfinite(r)]
        return float(np.mean(r) / np.std(r) * np.sqrt(52)) if len(r) > 5 and np.std(r) > 0 else 0.0
    ls = sign * y - cost_rt
    lf = np.where(x > 0, y, 0.0) - np.where(x > 0, cost_rt, 0.0)
    yrs = D_.index.year.values
    def per_year(r):
        out = {}
        for yv in sorted(set(yrs)):
            m = yrs == yv
            if m.sum() >= 8:
                out[int(yv)] = round(float(np.mean(np.sign(x[m]) == np.sign(y[m]))), 2)
        return out
    is_m = D_.index < OOS; oos_m = D_.index >= OOS
    py = per_year(ls)
    return {**cfg, "n": len(D_), "corr": round(corr, 3), "t_incremental": round(t_inc, 2),
            "hit": round(float(np.mean(sign == np.sign(y))), 3),
            "sh_ls_full": round(sharpe(ls), 2), "sh_ls_is": round(sharpe(ls[is_m]), 2),
            "sh_ls_oos": round(sharpe(ls[oos_m]), 2),
            "sh_lf_full": round(sharpe(lf), 2), "sh_lf_oos": round(sharpe(lf[oos_m]), 2),
            "ann_ls_pct": round(float(np.nanmean(ls) * 52 * 100), 1),
            "years_pos": int(sum(1 for v in py.values() if v > 0.5)), "years_tot": len(py),
            "per_year_hit": py}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--configs", required=True)
    ap.add_argument("--cost", type=float, default=0.0004)
    ap.add_argument("--oos", default=OOS_DEFAULT)
    args = ap.parse_args()
    cfgs = json.loads(args.configs)
    print(json.dumps([evaluate(c, cost_rt=args.cost, oos=args.oos) for c in cfgs]))


if __name__ == "__main__":
    main()