PythagorasGoal/scripts/research/fetch_ib_equities.py

"""FETCH + CERTIFY universo azioni/ETF da IB (ADJUSTED_LAST) -> data/raw/eq_<sym>_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);
  (5) SPLIT NON AGGIUSTATI (aggiunto 2026-07-25): ADJUSTED_LAST non sempre aggiusta gli split, e
      il check (3) NON li vede — uno split 2:1 fa esattamente -50%, sul filo della soglia. IWM
      (gamba di GTAA01 in produzione) ed EFA avevano uno split non aggiustato il 2005-06-09 e
      passavano come OK. Rilevatore in src/data/eq_splits.py -> status SPLIT-NON-AGG.
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]
sys.path.insert(0, str(ROOT))
from src.data.eq_splits import detect_unadjusted_splits  # noqa: E402
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"]
# espansione "diversi mercati" (intl / bond / credito / commodity / settori extra) per il lead-lag crypto
BROAD2 = ["DIA", "EFA", "EEM", "FXI", "EWJ", "AGG", "LQD", "IEF", "USO", "SLV", "DBC", "VNQ"]
UNIVERSE = SECTORS + BROAD + BROAD2


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)
    # SPLIT NON AGGIUSTATI: IB ADJUSTED_LAST non sempre li aggiusta. NB la guardia `maxret > 0.5`
    # NON li vede: uno split 2:1 non aggiustato fa ESATTAMENTE -50% e cade sul filo della soglia
    # (IWM 2005-06-09 passava a 49.5% con status OK, ed e' una gamba di GTAA01 in produzione).
    # Vedi src/data/eq_splits.py per il discriminante split-vs-crollo (range intraday).
    splits = detect_unadjusted_splits(df)

    status = "OK"
    if dup or not mono:
        status = "INTEGRITA'"
    elif splits:
        status = "SPLIT-NON-AGG"
    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,
            "splits": [f"{s['date'].date()} 1:{s['factor']:g}" for s in splits]}


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:]
    universe = UNIVERSE
    if "--only" in sys.argv[1:]:                      # refresh mirato (es. solo i 6 ETF GTAA per il cron)
        universe = sys.argv[sys.argv.index("--only") + 1].upper().split(",")
        force = True                                  # --only implica refresh dei simboli indicati
    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']}]"
              + (f" split={c['splits']}" if c.get("splits") else ""))
        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_<sym>_1d.parquet (ADJUSTED_LAST, namespace dedicato).")
    ib.disconnect()


if __name__ == "__main__":
    main()