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>
87 lines
4.4 KiB
Python
87 lines
4.4 KiB
Python
"""DEEP-DIVE: crypto[sessione USA] -> indice EUROPEO[overnight successivo]. Reale o artefatto?
|
|
|
|
Lo sweep ha trovato un forte segnale: crypto[T-8h->T] (T~00:00 UTC, fine sessione USA) predice il
|
|
future europeo (ESTX50/DAX)[T->T+6h] con t_crypto incrementale ~8, Sharpe 2.5, 3/3 anni. Verifiche:
|
|
(1) ROBUSTEZZA su T (ora entrata): un effetto vero non e' a coltello su una sola ora.
|
|
(2) vs SEMPRE-LONG: l'overnight ha un drift? il crypto AGGIUNGE oltre il long incondizionato?
|
|
(3) GAP 1-2h tra fine-segnale e inizio-cattura: se sopravvive, niente contaminazione di bordo.
|
|
(4) vs FUTURE-OWN: il crypto batte il momentum proprio del future (gia' nel t incrementale).
|
|
Dati: cache (fut europei + crypto 1h). sqrt(252), net 2bps.
|
|
"""
|
|
import sys
|
|
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
|
|
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 series(fut, bc, T, S=8, H=6, gap=0):
|
|
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)
|
|
f0, f1 = at(fut, te - pd.Timedelta(hours=S)), at(fut, te)
|
|
fcs, fce = at(fut, te + pd.Timedelta(hours=gap)), at(fut, te + pd.Timedelta(hours=gap + H))
|
|
c0, c1 = at(bc, te - pd.Timedelta(hours=S)), at(bc, te)
|
|
if not all(np.isfinite(v) and v > 0 for v in (f0, f1, fcs, fce, c0, c1)):
|
|
continue
|
|
rows.append((D, c1/c0 - 1, f1/f0 - 1, fce/fcs - 1))
|
|
Dd = pd.DataFrame(rows, columns=["d", "csig", "fctrl", "cap"]).set_index("d")
|
|
return Dd[Dd["cap"].abs() > 1e-9]
|
|
|
|
|
|
def stats(Dd):
|
|
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
|
|
crypto = np.sign(x)*y - COST
|
|
futown = np.sign(ctrl)*y - COST
|
|
always = y - COST
|
|
yrs = Dd.index.year.values
|
|
py = {int(v): round(_sh(crypto[yrs==v]),2) for v in sorted(set(yrs)) if (yrs==v).sum()>=30}
|
|
return dict(n=len(Dd), t_crypto=round(t_c,2), sh_crypto=round(_sh(crypto),2),
|
|
sh_futown=round(_sh(futown),2), sh_always=round(_sh(always),2),
|
|
ann_crypto=round(float(np.nanmean(crypto)*252*100),1), per_year=py)
|
|
|
|
|
|
def main():
|
|
print("="*96); print(" DEEP-DIVE crypto -> indice EUROPEO overnight (reale o artefatto?)"); print("="*96)
|
|
bcs = {l: crypto_hourly(l) for l in ("BTC","ETH")}
|
|
for sym in ("ESTX50","DAX"):
|
|
fut = fut_hourly(sym)
|
|
print(f"\n ===== {sym} =====")
|
|
print(f" (1) ROBUSTEZZA su T (lead=BTC, S=8h H=6h gap=0): t_crypto | Sh crypto vs futown vs always-long")
|
|
for T in (20,21,22,23,0,1,2,3,4):
|
|
s = stats(series(fut, bcs["BTC"], T))
|
|
print(f" T={T:>2}h: n={s['n']} t {s['t_crypto']:+.1f} | crypto {s['sh_crypto']:+.2f} futown {s['sh_futown']:+.2f} always {s['sh_always']:+.2f} ann {s['ann_crypto']:+.1f}% {s['per_year']}")
|
|
print(f" (3) GAP test (T=0h, cattura inizia +1h e +2h dopo il segnale):")
|
|
for g in (0,1,2):
|
|
s = stats(series(fut, bcs["BTC"], 0, gap=g))
|
|
print(f" gap={g}h: t {s['t_crypto']:+.1f} | crypto {s['sh_crypto']:+.2f} always {s['sh_always']:+.2f}")
|
|
print(f" (lead ETH, T=0h): ", end="")
|
|
s = stats(series(fut, bcs["ETH"], 0)); print(f"t {s['t_crypto']:+.1f} crypto {s['sh_crypto']:+.2f} always {s['sh_always']:+.2f} {s['per_year']}")
|
|
print("\n LETTURA: se 'crypto' >> 'always' E >> 'futown' E regge su T e col gap -> lead REALE (crypto")
|
|
print(" anticipa il catch-up europeo). Se crypto ~ always -> e' solo overnight-drift europeo.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|