research(cross-market): futures overnight non-sovrapposto -> edge ~0 su SP500, soffio su small-cap
Test ONESTO dell'idea "monitor Deribit/trade IB": entra mid-notte sul future indice, cattura il moto SUCCESSIVO (finestre non sovrapposte, no look-ahead). Dati: ES/NQ/RTY orari da IB (fut_*_1h, ~3y). RISULTATO: ES (S&P500) nessun edge (Sharpe ~0/neg, t_crypto 0-1.5); NQ momentum del future non crypto; RTY (small-cap) unico con t_crypto incrementale 2.0-2.7 e crypto che aggiunge oltre il moto proprio del future, ma Sharpe 0.4-0.5, 24 config (multiple-testing), 2.3y, per-anno incoerente (2026 negativo). VERDETTO: l'idea NON da' edge tradabile, men che meno su SP500. Il forte crypto<->equity e' co-movimento contemporaneo (risk-beta), non anticipazione: imposta una finestra causale non-sovrapposta e svanisce. Il "Sharpe 5" del gap era look-ahead. RTY -> forward-monitor al piu'. Coerente col soffitto del progetto. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,61 @@
|
||||
"""FETCH futures indice ORARI (ES/NQ/RTY) da IB -> data/raw/fut_<sym>_1h.parquet (UTC).
|
||||
|
||||
Per il test onesto dell'idea "monitor Deribit / trade IB": serve il path INTRADAY del future indice
|
||||
(che si trada di notte) per misurare finestre overnight NON sovrapposte col segnale crypto.
|
||||
ContFuture orario, in chunk da 1 anno (IB limita le durate intraday). Convertito in UTC.
|
||||
Resumable (salta i parquet gia' scritti). Per RETURNS/lead-lag il back-adjust del ContFuture e' ok
|
||||
(i ritorni infra-contratto sono preservati; i gap di roll ~4/anno sono trascurabili).
|
||||
|
||||
uv run --with ib_async python scripts/research/fetch_ib_futures.py
|
||||
"""
|
||||
import sys, time
|
||||
from pathlib import Path
|
||||
import numpy as np, pandas as pd
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
RAW = ROOT / "data" / "raw"
|
||||
SYMS = ["ES", "NQ", "RTY"]
|
||||
N_CHUNKS = 5 # ~5 anni se disponibili
|
||||
|
||||
|
||||
def main():
|
||||
from ib_async import IB, ContFuture
|
||||
ib = IB()
|
||||
try:
|
||||
ib.connect("127.0.0.1", 4002, clientId=144, timeout=15)
|
||||
except Exception as e:
|
||||
print(f"[CONNESSIONE FALLITA] {repr(e)[:100]}"); sys.exit(1)
|
||||
print(f" acct {ib.managedAccounts()} | fetch futures orari -> data/raw/fut_*")
|
||||
for sym in SYMS:
|
||||
out = RAW / f"fut_{sym.lower()}_1h.parquet"
|
||||
if out.exists():
|
||||
print(f" {sym}: gia' su disco -> skip"); continue
|
||||
cf = ContFuture(sym, exchange="CME")
|
||||
try:
|
||||
ib.qualifyContracts(cf)
|
||||
except Exception as e:
|
||||
print(f" {sym}: qualify ERR {repr(e)[:60]}"); continue
|
||||
# ContFuture NON accetta endDateTime (Error 10339) -> chiamata singola, durata massima (~3y orari)
|
||||
try:
|
||||
b = ib.reqHistoricalData(cf, endDateTime="", durationStr="4 Y", barSizeSetting="1 hour",
|
||||
whatToShow="TRADES", useRTH=False, formatDate=1, timeout=150)
|
||||
except Exception as e:
|
||||
print(f" {sym}: ERR {repr(e)[:60]}"); continue
|
||||
if not b:
|
||||
print(f" {sym}: VUOTO"); continue
|
||||
D = pd.DataFrame([(pd.Timestamp(x.date), x.open, x.high, x.low, x.close, x.volume) for x in b],
|
||||
columns=["ts", "open", "high", "low", "close", "volume"]).drop_duplicates("ts").sort_values("ts").reset_index(drop=True)
|
||||
# -> UTC ms (robusto alla risoluzione us/ns: naive-UTC -> datetime64[ms] -> int64)
|
||||
ts = pd.to_datetime(D["ts"], utc=True).dt.tz_convert("UTC").dt.tz_localize(None)
|
||||
D["timestamp"] = ts.values.astype("datetime64[ms]").astype("int64")
|
||||
D = D.drop(columns=["ts"])
|
||||
D.to_parquet(out)
|
||||
u = pd.to_datetime(D["timestamp"], unit="ms", utc=True)
|
||||
print(f" {sym}: SCRITTO {len(D)} barre {u.iloc[0]} .. {u.iloc[-1]} -> {out.name}")
|
||||
time.sleep(1.5)
|
||||
ib.disconnect()
|
||||
print(" done.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,116 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user