Adriano Dal Pastro
09c8bb7de8
Esteso il test crypto-lead a ZN(bond), ESTX50/DAX(Europa), NKD(Nikkei) via futures orari IB (commodity GC/CL/HG bloccate da subscription). Test non-sovrapposto crypto[T-8h->T]->future[T->T+6h]. ES/NQ/RTY niente (gia'); ZN negativo; NKD debole (~overnight drift). ESTX50/DAX SEMBRANO fortissimi (t_crypto 7.8, Sharpe 2.5, 3/3 anni) MA e' artefatto di confine UTC: picco a coltello a T=00:00, morto a T=1h; GAP di 1h uccide l'effetto (Sharpe 2.45->-0.52); tutto l'edge nella singola barra 00:00->01:00 (Sh +2.93) vs ora dopo (-1.02). Firma esatta di day_boundary_robust (CLAUDE.md). VERDETTO: nessuna anticipazione crypto->mercato sfruttabile, ne' SP500 ne' altro. Sempre co-movimento contemporaneo (risk-beta) o artefatto di confine. Resta valido solo il diversificatore TP01+GTAA. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
109 lines
4.9 KiB
Python
109 lines
4.9 KiB
Python
"""LEAD-LAG GENERICO non-sovrapposto: crypto[T-S -> T] predice future[T -> T+H]? (ogni mercato/fuso).
|
|
|
|
Market-agnostic. Per ogni giorno, entrata a ora T (UTC): segnale = crypto nella finestra [T-S, T]
|
|
(finisce all'entrata), cattura = future nella finestra SUCCESSIVA [T, T+H] (non sovrapposta).
|
|
Controllo = moto PROPRIO del future [T-S, T] -> isola se il crypto AGGIUNGE (anticipa) oltre il
|
|
momentum del future. Sweep su T (copre apertura Europa ~07h, USA ~13h, Asia ~00h) e H.
|
|
|
|
TRADE: sign(crypto[T-S,T]) * future[T,T+H] - costo. Sharpe sqrt(252), per-anno, OOS 2024+.
|
|
Dati: data/raw/fut_*_1h.parquet (UTC) + crypto 1h. Solo cache, nessun IB.
|
|
"""
|
|
import sys, glob
|
|
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
|
|
|
|
|
|
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(fut, bc, T, S, H):
|
|
days = pd.date_range(fut.index[0].normalize(), fut.index[-1].normalize(), freq="D", tz="UTC")
|
|
rows = []
|
|
for D in days:
|
|
te = D + pd.Timedelta(hours=T)
|
|
ts = te - pd.Timedelta(hours=S)
|
|
tx = te + pd.Timedelta(hours=H)
|
|
f0, f1 = at(fut, ts), at(fut, te); f2 = at(fut, tx)
|
|
c0, c1 = at(bc, ts), at(bc, te)
|
|
if not all(np.isfinite(v) and v > 0 for v in (f0, f1, f2, c0, c1)):
|
|
continue
|
|
rows.append((D, c1/c0 - 1, f1/f0 - 1, f2/f1 - 1))
|
|
if len(rows) < 120:
|
|
return None
|
|
Dd = pd.DataFrame(rows, columns=["d", "csig", "fctrl", "cap"]).set_index("d")
|
|
# filtra giorni in cui la cattura e' identicamente 0 (mercato chiuso, prezzo stale)
|
|
Dd = Dd[Dd["cap"].abs() > 1e-9]
|
|
if len(Dd) < 120:
|
|
return None
|
|
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
|
|
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_c = float(beta[1] / se[1]) if se[1] > 0 else 0.0
|
|
C = np.sign(x) * y - COST
|
|
m = Dd.index >= OOS
|
|
yrs = Dd.index.year.values
|
|
py = {int(v): round(_sh(C[yrs == v]), 2) for v in sorted(set(yrs)) if (yrs == v).sum() >= 30}
|
|
pos = sum(1 for v in py.values() if v > 0)
|
|
return dict(T=T, H=H, n=len(Dd), t_crypto=round(t_c, 2), sh_full=round(_sh(C), 2),
|
|
sh_oos=round(_sh(C[m]), 2), ann=round(float(np.nanmean(C) * 252 * 100), 1),
|
|
years_pos=pos, years_tot=len(py), per_year=py)
|
|
|
|
|
|
def main():
|
|
syms = sorted(p.name[4:-11].upper() for p in RAW.glob("fut_*_1h.parquet"))
|
|
print("=" * 98)
|
|
print(f" LEAD-LAG GENERICO non-sovrapposto — futures: {syms}")
|
|
print("=" * 98)
|
|
print(" crypto[T-8h -> T] -> future[T -> T+6h], controllo=moto proprio future. net 2bps, OOS2024+\n")
|
|
bcs = {l: crypto_hourly(l) for l in ("BTC", "ETH")}
|
|
winners = []
|
|
for sym in syms:
|
|
fut = fut_hourly(sym)
|
|
best = None
|
|
for lead in ("BTC", "ETH"):
|
|
for T in (0, 4, 8, 12, 16, 20):
|
|
r = run(fut, bcs[lead], T, 8, 6)
|
|
if not r:
|
|
continue
|
|
r["sym"] = sym; r["lead"] = lead
|
|
if best is None or r["sh_oos"] > best["sh_oos"]:
|
|
best = r
|
|
# raccogli i forti (crypto significativo E robusto)
|
|
if r["t_crypto"] >= 2.5 and r["sh_oos"] > 0.5 and r["years_pos"] == r["years_tot"]:
|
|
winners.append(r)
|
|
if best:
|
|
print(f" {sym:7} miglior: {best['lead']}->T{best['T']}h: t_crypto {best['t_crypto']:+.1f} "
|
|
f"Sh full {best['sh_full']:+.2f} OOS {best['sh_oos']:+.2f} ann {best['ann']:+.1f}% "
|
|
f"{best['years_pos']}/{best['years_tot']}y {best['per_year']}")
|
|
print("\n --- CANDIDATI FORTI (t_crypto>=2.5, OOS>0.5, tutti gli anni positivi) ---")
|
|
if not winners:
|
|
print(" NESSUNO. -> nessuna anticipazione crypto->future robusta oltre il rumore/beta.")
|
|
else:
|
|
winners.sort(key=lambda r: r["sh_oos"], reverse=True)
|
|
for w in winners[:10]:
|
|
print(f" {w['sym']:7} {w['lead']}->T{w['T']}h H{w['H']}: t_crypto {w['t_crypto']} "
|
|
f"Sh OOS {w['sh_oos']} full {w['sh_full']} ann {w['ann']}% {w['years_pos']}/{w['years_tot']}y {w['per_year']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|