From 1c0b5f18690529077f427fe777a169869fbdfc8e Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Mon, 22 Jun 2026 21:29:11 +0000 Subject: [PATCH] =?UTF-8?q?research(equities):=20apre=20il=20fronte=20azio?= =?UTF-8?q?ni/ETF=20via=20IB=20=E2=80=94=20dati=20certificati=20+=20cache?= =?UTF-8?q?=20su=20disco?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Branch dedicato. Disciplina v2.0.0: prima il dato certificato, poi la strategia. IB paper (gnzsnz/ib-gateway) da' storia daily ADJUSTED_LAST (div+split) profonda. - ib_equities_probe.py: sonda fattibilita' dati (profondita', adjusted, subscription). - fetch_ib_equities.py: FETCH+CERTIFY universo -> data/raw/eq__1d.parquet (ms epoch, namespace dedicato). RIPARTIBILE (salta i parquet gia' scritti) -> niente refetch da IB. Certifica: integrita', gap lunghi, sanita' ritorni, sanita' adjustment. - eqlib.py: harness ricerca equity. Legge la CACHE su disco (lru_cache) MAI da IB; universi (11 settori SPDR + 9 classici 1998+ + broad), panel allineato, riusa lo scorer indurito altlib. UNIVERSO CERTIFICATO (17, data/raw/eq_* gitignored = cache locale): 9 settori classici dal 1998-12-22 (27.5y) + XLRE(2015)/XLC(2018) + SPY(1996,30y)/QQQ/IWM/ GLD(2004)/HYG(2007)/TLT(2016). Tutti integri (monotoni, no dup, no spike>50%, gap-lunghi 0). Start comune: 9 classici 1998, 11 settori 2018. Prossimo passo: prima ricerca = momentum cross-sectional settoriale, gauntlet onesto. Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/research/eqlib.py | 65 +++++++++++++ scripts/research/fetch_ib_equities.py | 128 ++++++++++++++++++++++++++ scripts/research/ib_equities_probe.py | 95 +++++++++++++++++++ 3 files changed, 288 insertions(+) create mode 100644 scripts/research/eqlib.py create mode 100644 scripts/research/fetch_ib_equities.py create mode 100644 scripts/research/ib_equities_probe.py diff --git a/scripts/research/eqlib.py b/scripts/research/eqlib.py new file mode 100644 index 0000000..7147489 --- /dev/null +++ b/scripts/research/eqlib.py @@ -0,0 +1,65 @@ +"""EQLIB — harness di ricerca EQUITY/ETF (branch research/equities-ib). + +Legge la CACHE su disco (data/raw/eq_*.parquet, ADJUSTED_LAST, scritta una volta da +fetch_ib_equities.py) — MAI da IB. `lru_cache` -> ogni parquet si legge una sola volta per processo. +"Memorizza i dati per non rileggerli ogni volta": la persistenza e' il parquet su disco + questa cache. + +Espone: universi (SECTORS/BROAD), load_eq(sym), panel(universe) allineato, e riusa lo scorer +indurito di altlib (_sh, _dd_ret, _to_daily, marginal_vs_tp01) per giudicare i candidati con la +stessa disciplina del lato crypto. +""" +from functools import lru_cache +from pathlib import Path +import numpy as np, pandas as pd + +ROOT = Path(__file__).resolve().parents[2] +RAW = ROOT / "data" / "raw" + +# 11 SPDR settoriali. I 9 "classici" (sotto SECTORS_CLASSIC) partono 1998; XLRE 2015, XLC 2018. +SECTORS = ["XLK", "XLF", "XLE", "XLV", "XLI", "XLP", "XLY", "XLU", "XLB", "XLRE", "XLC"] +SECTORS_CLASSIC = ["XLK", "XLF", "XLE", "XLV", "XLI", "XLP", "XLY", "XLU", "XLB"] # storia lunga (1998+) +BROAD = ["SPY", "QQQ", "IWM", "TLT", "GLD", "HYG"] + + +@lru_cache(maxsize=64) +def load_eq(sym: str) -> pd.DataFrame: + """OHLCV aggiustato (dividendi+split) per `sym`, indicizzato datetime UTC. Cache su disco -> RAM.""" + p = RAW / f"eq_{sym.lower()}_1d.parquet" + if not p.exists(): + raise FileNotFoundError(f"{p} assente — gira: uv run --with ib_async python scripts/research/fetch_ib_equities.py") + d = pd.read_parquet(p).copy() + d.index = pd.to_datetime(d["timestamp"], unit="ms", utc=True) + return d[["open", "high", "low", "close", "volume"]] + + +@lru_cache(maxsize=16) +def _close_panel(universe: tuple) -> pd.DataFrame: + cols = {s: load_eq(s)["close"].astype(float) for s in universe} + return pd.concat(cols, axis=1).sort_index() + + +def panel(universe=tuple(SECTORS), how: str = "inner") -> pd.DataFrame: + """Prezzi close aggiustati [date x asset]. how='inner' = date comuni a TUTTI (start = ETF piu' giovane); + 'outer' = unione (NaN dove un ETF non esiste ancora).""" + P = _close_panel(tuple(universe)) + return P.dropna(how="any") if how == "inner" else P + + +def describe(universe=None): + universe = universe or (SECTORS + BROAD) + print(f" {'sym':6} {'barre':>6} {'da':>11} {'a':>11} {'anni':>5}") + for s in universe: + try: + d = load_eq(s) + print(f" {s:6} {len(d):>6} {str(d.index[0].date()):>11} {str(d.index[-1].date()):>11} " + f"{(d.index[-1]-d.index[0]).days/365.25:>5.1f}") + except FileNotFoundError: + print(f" {s:6} (assente)") + Pc = panel(SECTORS_CLASSIC); Pa = panel(SECTORS) + print(f"\n panel 9 settori CLASSICI: {Pc.shape[1]}x{len(Pc)} start comune {Pc.index[0].date()}") + print(f" panel 11 settori : {Pa.shape[1]}x{len(Pa)} start comune {Pa.index[0].date()}") + + +if __name__ == "__main__": + print("=" * 70); print(" EQLIB — cache equity su disco (nessun IB)"); print("=" * 70) + describe() diff --git a/scripts/research/fetch_ib_equities.py b/scripts/research/fetch_ib_equities.py new file mode 100644 index 0000000..44dabe1 --- /dev/null +++ b/scripts/research/fetch_ib_equities.py @@ -0,0 +1,128 @@ +"""FETCH + CERTIFY universo azioni/ETF da IB (ADJUSTED_LAST) -> data/raw/eq__1d.parquet. + +Apre il fronte EQUITY (branch research/equities-ib). Disciplina v2.0.0: PRIMA il dato certificato, +POI la strategia. IB dà storia daily aggiustata per dividendi+split (ADJUSTED_LAST), profonda +(SPY dal 1996), sul conto paper. Namespace dedicato 'eq_' (NON tocca i parquet crypto). + +UNIVERSO (prima ricerca = momentum cross-sectional settoriale, l'edge robusto plausibile in equity): + * 11 SPDR settoriali (XLK..XLC); * broad/macro SPY QQQ IWM TLT GLD HYG. +NB: i 9 settori "classici" partono 1998; XLRE 2015, XLC 2018 -> lo start COMUNE a 11 e' 2018. +Per backtest lunghi usare i 9 classici (1998+) o accettare lo start 2018 per gli 11. + +CERTIFICAZIONE (gemello equity di certify_feed.py): + (1) integrità: barre, range, date monotone, duplicati, flat bars (close invariato); + (2) gap: run di giorni-lavorativi mancanti > 5 (festivi normali, buchi lunghi = sospetti); + (3) sanità ritorni: max |daily ret| (un >50% non-evento = errore di adjustment); + (4) sanità adjustment: primo close aggiustato << ultimo (i dividendi abbassano lo storico). +PREREQUISITO: gateway IB paper su 127.0.0.1:4002 (docker compose up -d ib-gateway). + + uv run --with ib_async python scripts/research/fetch_ib_equities.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" +RAW.mkdir(parents=True, exist_ok=True) + +SECTORS = ["XLK", "XLF", "XLE", "XLV", "XLI", "XLP", "XLY", "XLU", "XLB", "XLRE", "XLC"] +BROAD = ["SPY", "QQQ", "IWM", "TLT", "GLD", "HYG"] +UNIVERSE = SECTORS + BROAD + + +def certify(sym: str, df: pd.DataFrame) -> dict: + if df.empty: + return {"sym": sym, "n": 0, "status": "VUOTO"} + idx = df.index + dup = int(idx.duplicated().sum()) + mono = bool(idx.is_monotonic_increasing) + c = df["close"].values.astype(float) + ret = np.diff(c) / c[:-1] + flat = int((ret == 0).sum()) + maxret = float(np.max(np.abs(ret))) if len(ret) else 0.0 + # gap: giorni lavorativi attesi vs presenti, run lunghi mancanti + bdays = pd.bdate_range(idx[0], idx[-1]) + missing = len(bdays) - len(idx.intersection(bdays)) + gaps = bdays.difference(idx) + longgap = 0 + if len(gaps): + g = pd.Series(1, index=gaps).resample("1D").sum().fillna(0) + # conta run consecutivi di bday mancanti + s = (gaps.to_series().diff().dt.days.fillna(1) > 3).cumsum() + longgap = int((gaps.to_series().groupby(s).size() > 5).sum()) + span_y = (idx[-1] - idx[0]).days / 365.25 + adj_ratio = round(float(c[0] / c[-1]), 3) # primo/ultimo: <1 atteso (storico abbassato dai div) + status = "OK" + if dup or not mono: + status = "INTEGRITA'" + elif maxret > 0.5: + status = "SPIKE?" + elif longgap > 0: + status = "GAP-LUNGO" + elif span_y < 1: + status = "corto<1y" + return {"sym": sym, "n": len(df), "primo": idx[0].date(), "ultimo": idx[-1].date(), + "anni": round(span_y, 1), "dup": dup, "mono": mono, "flat": flat, + "maxret%": round(maxret * 100, 1), "miss_bd": missing, "gap_lunghi": longgap, + "adj_first/last": adj_ratio, "status": status} + + +def main(): + try: + from ib_async import IB, Stock + except Exception: + print("ib_async assente. Esegui con: uv run --with ib_async python scripts/research/fetch_ib_equities.py") + sys.exit(2) + ib = IB() + try: + ib.connect("127.0.0.1", 4002, clientId=90, timeout=15) + except Exception as e: + print(f"[CONNESSIONE FALLITA] 127.0.0.1:4002 -> {repr(e)[:120]}\n Avvia: docker compose up -d ib-gateway") + sys.exit(1) + + print("=" * 104) + print(f" FETCH + CERTIFY azioni/ETF (ADJUSTED_LAST) -> data/raw/eq_* | acct {ib.managedAccounts()}") + print("=" * 104) + rep, ok = [], [] + force = "--force" in sys.argv[1:] + for sym in UNIVERSE: + out_path = RAW / f"eq_{sym.lower()}_1d.parquet" + if out_path.exists() and not force: + print(f" {sym:5} GIA' SU DISCO -> skip (usa --force per riscaricare)") + ok.append(sym) + continue + con = Stock(sym, "SMART", "USD") + try: + bars = ib.reqHistoricalData(con, endDateTime="", durationStr="30 Y", barSizeSetting="1 day", + whatToShow="ADJUSTED_LAST", useRTH=True, formatDate=1, timeout=60) + except Exception as e: + print(f" {sym:5} ERR {repr(e)[:70]}"); rep.append({"sym": sym, "status": "ERR"}); time.sleep(1.2); continue + if not bars: + print(f" {sym:5} 0 barre (subscription?)"); rep.append({"sym": sym, "n": 0, "status": "VUOTO"}); time.sleep(1.2); continue + df = pd.DataFrame([(pd.Timestamp(str(b.date)), b.open, b.high, b.low, b.close, b.volume) for b in bars], + columns=["ts", "open", "high", "low", "close", "volume"]).set_index("ts").sort_index() + c = certify(sym, df) + rep.append(c) + if c.get("n", 0) > 0: + out = df.copy() + # ms epoch (come i parquet crypto), robusto alla risoluzione datetime64 (s/us/ns) + out["timestamp"] = out.index.astype("datetime64[ms]").astype("int64") + out.reset_index(drop=True).to_parquet(RAW / f"eq_{sym.lower()}_1d.parquet") + if c["status"] == "OK": + ok.append(sym) + print(f" {sym:5} n={c.get('n',0):>5} {str(c.get('primo','')):>10}->{str(c.get('ultimo',''))} " + f"{c.get('anni','?')}y flat={c.get('flat','?')} maxret={c.get('maxret%','?')}% " + f"miss_bd={c.get('miss_bd','?')} gapL={c.get('gap_lunghi','?')} adj={c.get('adj_first/last','?')} [{c['status']}]") + time.sleep(1.2) # pacing IB + + print("-" * 104) + print(f" CERTIFICATI OK ({len(ok)}/{len(UNIVERSE)}): {ok}") + sec_ok = [s for s in SECTORS if s in ok] + print(f" settori OK: {len(sec_ok)}/11 {sec_ok}") + print(f" -> scritti in data/raw/eq__1d.parquet (ADJUSTED_LAST, namespace dedicato).") + ib.disconnect() + + +if __name__ == "__main__": + main() diff --git a/scripts/research/ib_equities_probe.py b/scripts/research/ib_equities_probe.py new file mode 100644 index 0000000..119b1af --- /dev/null +++ b/scripts/research/ib_equities_probe.py @@ -0,0 +1,95 @@ +"""IB EQUITIES/ETF DATA PROBE — certifica cosa il paper IB dà per la ricerca su azioni/ETF. + +Gemello equity di certify_feed.py: PRIMA il dato (cosa c'è, quanto indietro, aggiustato per +dividendi/split?, cosa costa), POI la strategia. Disciplina v2.0.0. + +Universo candidato per la prima ricerca equity (cross-sectional momentum / trend, l'edge "noioso e +robusto" più plausibile in un mercato efficiente): + * 11 SPDR settoriali (XLK..XLC) — universo canonico del momentum cross-section settoriale; + * ETF broad / macro (SPY QQQ IWM TLT GLD HYG) — per trend e risk-on/off; + * 2 azioni (AAPL MSFT) per tarare profondità/qualità. + +Per ogni simbolo: profondità storica daily con whatToShow=ADJUSTED_LAST (split+dividendi, OBBLIGATORIO +per un backtest equity onesto) e TRADES (raw), + flag se scatta errore di subscription market-data. + + uv run --with ib_async python scripts/research/ib_equities_probe.py +""" +import argparse, sys + +SECTORS = ["XLK", "XLF", "XLE", "XLV", "XLI", "XLP", "XLY", "XLU", "XLB", "XLRE", "XLC"] +BROAD = ["SPY", "QQQ", "IWM", "TLT", "GLD", "HYG"] +STOCKS = ["AAPL", "MSFT"] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--host", default="127.0.0.1") + ap.add_argument("--port", type=int, default=4002) + ap.add_argument("--client-id", type=int, default=88) + ap.add_argument("--years", default="20 Y", help="durata storica da richiedere") + args = ap.parse_args() + + try: + from ib_async import IB, Stock + except Exception: + print("ib_async non importabile. Esegui con: uv run --with ib_async python ...") + sys.exit(2) + + ib = IB() + try: + ib.connect(args.host, args.port, clientId=args.client_id, timeout=15) + except Exception as e: + print(f"[CONNESSIONE FALLITA] {args.host}:{args.port} -> {repr(e)[:140]}") + sys.exit(1) + + print("=" * 96) + print(f" IB EQUITIES/ETF PROBE — {args.host}:{args.port} | acct {ib.managedAccounts()} | depth req {args.years}") + print("=" * 96) + + universe = [("SECTOR", s) for s in SECTORS] + [("BROAD", s) for s in BROAD] + [("STOCK", s) for s in STOCKS] + print(f" {'sym':6} {'tipo':7} {'ADJUSTED_LAST':>26} {'TRADES':>22} note") + rows = [] + for cat, sym in universe: + con = Stock(sym, "SMART", "USD") + try: + cds = ib.reqContractDetails(con) + if not cds: + print(f" {sym:6} {cat:7} {'-- no contract --':>26}") + continue + except Exception as e: + print(f" {sym:6} {cat:7} ERR resolve {repr(e)[:40]}") + continue + + def hist(what): + try: + b = ib.reqHistoricalData(con, endDateTime="", durationStr=args.years, + barSizeSetting="1 day", whatToShow=what, + useRTH=True, formatDate=1, timeout=45) + if not b: + return "0 barre", None + return f"{len(b)}b {b[0].date}..{b[-1].date}", b + except Exception as e: + return f"ERR {repr(e)[:30]}", None + + adj_s, adj_b = hist("ADJUSTED_LAST") + trd_s, _ = hist("TRADES") + note = "" + if "ERR" in adj_s or "0 barre" in adj_s: + note = "subscription? prova delayed" + print(f" {sym:6} {cat:7} {adj_s:>26} {trd_s:>22} {note}") + if adj_b: + rows.append((sym, len(adj_b), str(adj_b[0].date), str(adj_b[-1].date))) + + print("-" * 96) + if rows: + depth = min(r[1] for r in rows); start = max(r[2] for r in rows) + print(f" CERTIFICABILI (ADJUSTED_LAST): {len(rows)}/{len(universe)} | profondità comune ~{depth}b | start comune {start}") + print(f" -> per un backtest cross-sectional servono date allineate: lo start comune e' il limite.") + else: + print(" NESSUN simbolo ha reso storia ADJUSTED — probabile mancanza market-data subscription.") + print(" Ripiego: whatToShow='TRADES' (raw, non adj) o dati 'delayed' / fonte esterna certificabile.") + ib.disconnect() + + +if __name__ == "__main__": + main()