PythagorasGoal/scripts/research/fetch_ib_ucits.py

"""FETCH + CERTIFY equivalenti UCITS dei 6 ETF di GTAA01 -> data/raw/eqx_<sym>_1d.parquet.

PERCHE' ESISTE. Il 2026-07-26 il conto reale ha RIFIUTATO l'ordine sui 6 ETF USA di GTAA01
(blocco PRIIPs: ETF domiciliati USA, nessun KID, vietati al retail UE). Lo sleeve, cosi' com'e',
non e' deployabile. L'unica via d'uscita e' negoziare gli equivalenti UCITS — e per sapere se
quella via regge servono i DATI dei veicoli, non le loro schede prodotto.

SCELTE DICHIARATE
  * Tutti i candidati sono presi su **LSEETF in USD**. Non e' una comodita': e' cio' che elimina
    un livello di costo (conversione valutaria per ordine) che su un CAGR del 3.65% non sarebbe
    trascurabile. Le linee Xetra in EUR (SXR8/SXRV/ZPRR) NON risolvono su questo gateway e
    comunque introdurrebbero l'FX.
  * Namespace **eqx_** separato da eq_: i veicoli USA restano su disco perche' sono la base della
    VALIDAZIONE (30 anni), che nessun UCITS puo' fornire. Il progetto valida sull'indice/veicolo
    lungo e negozia il veicolo corto; sovrascrivere eq_ cancellerebbe la meta' lunga.
  * Piu' candidati per gamba dove l'equivalenza NON e' ovvia (small cap: RTWO e' un fattore
    *quality*, CSUSS e' *ESG* -> non sono IWM). La scelta si fa dopo, sui dati.

CERTIFICAZIONE: riusa `certify()` di fetch_ib_equities.py — stesse regole, split non aggiustati
inclusi (la lezione IWM/EFA del 25/07 vale identica su questi veicoli).

  uv run --with ib_async python scripts/research/fetch_ib_ucits.py
  uv run --with ib_async python scripts/research/fetch_ib_ucits.py --force
"""
from __future__ import annotations

import sys
import time
from pathlib import Path

import pandas as pd

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "research"))

from fetch_ib_equities import certify  # noqa: E402

RAW = ROOT / "data" / "raw"
RAW.mkdir(parents=True, exist_ok=True)

EXCH, CUR = "LSEETF", "USD"

# gamba USA -> candidati UCITS, in ordine di preferenza a priori (equivalenza dell'INDICE prima
# della storia: un fattore 'quality'/'ESG' non e' la stessa esposizione anche se ha piu' barre).
#
# ⚠️ SI SCARICANO DUE INSIEMI, e la ragione e' il BIGLIETTO MINIMO. Il prezzo di UNA azione e' una
# scelta dell'emittente, non una proprieta' dell'indice: CSPX costa $802 e VUAA $144 sullo STESSO
# S&P 500. Senza frazionamento un ordine e' almeno una azione, quindi a capitale piccolo il prezzo
# unitario decide se la gamba e' rappresentabile. Ma i veicoli economici sono anche i piu' RECENTI
# (XNAS: 4.3 anni contro i 7.5 di EQQQ) -> non si puo' scegliere una cosa sola:
#   * insieme STORIA  -> finestra comune piu' lunga, serve a MISURARE la deviazione del veicolo;
#   * insieme DEPLOY  -> prezzo unitario basso, serve a sapere se si puo' ESEGUIRE.
CANDIDATI: dict[str, tuple[str, ...]] = {
    "SPY": ("CSPX", "VUAA"),                # S&P 500: CSPX $802 (storia) · VUAA $144 (deploy)
    "QQQ": ("CNDX", "EQQQ", "XNAS"),        # Nasdaq-100: EQQQ $693 (storia) · XNAS $66 (deploy)
    "IWM": ("XRSU", "R2US", "IDP6", "SPY4"),  # Russell 2000: XRSU $438 · R2US $86 (piu' economico E lungo)
                                            # SPY4 = S&P 400 MID cap: indice DIVERSO, tenuto solo
                                            # per misurare il costo di una sostituzione forzata
                                            # (e' cio' che alcuni broker offrono al posto dello small cap)
    "TLT": ("IDTL", "DTLA"),                # US Treasury 20+ — indice identico, IDTL $3.09
    "GLD": ("IGLN",),                       # oro fisico (ETC, non UCITS: ha comunque il KID)
    "HYG": ("IHYU",),                       # USD High Yield corp — indice equivalente
}
UNIVERSE = [s for v in CANDIDATI.values() for s in v]


def main() -> int:
    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_ucits.py")
        return 2

    ib = IB()
    try:
        ib.connect("127.0.0.1", 4002, clientId=91, timeout=20)
    except Exception as e:
        print(f"[CONNESSIONE FALLITA] 127.0.0.1:4002 -> {repr(e)[:120]}\n  Avvia: docker compose up -d ib-gateway")
        return 1

    print("=" * 108)
    print(f"  FETCH + CERTIFY UCITS ({EXCH}/{CUR}, ADJUSTED_LAST) -> data/raw/eqx_* | acct {ib.managedAccounts()}")
    print("=" * 108)
    force = "--force" in sys.argv[1:]
    ok, rep = [], []
    for sym in UNIVERSE:
        out_path = RAW / f"eqx_{sym.lower()}_1d.parquet"
        if out_path.exists() and not force:
            print(f"  {sym:6} GIA' SU DISCO -> skip (--force per riscaricare)")
            ok.append(sym)
            continue
        con = Stock(sym, EXCH, CUR)
        try:
            det = ib.reqContractDetails(con)
        except Exception as e:
            det = None
            print(f"  {sym:6} ERR contratto {type(e).__name__} {str(e)[:50]}")
        if not det:
            print(f"  {sym:6} CONTRATTO ASSENTE su {EXCH}/{CUR}")
            rep.append({"sym": sym, "status": "ASSENTE"})
            time.sleep(1.2)
            continue
        try:
            bars = ib.reqHistoricalData(det[0].contract, endDateTime="", durationStr="30 Y",
                                        barSizeSetting="1 day", whatToShow="ADJUSTED_LAST",
                                        useRTH=True, formatDate=1, timeout=120)
        except Exception as e:
            print(f"  {sym:6} ERR storico {repr(e)[:70]}")
            rep.append({"sym": sym, "status": "ERR"})
            time.sleep(1.2)
            continue
        if not bars:
            print(f"  {sym:6} 0 barre (market data non sottoscritto?)")
            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)
        out = df.copy()
        out["timestamp"] = out.index.astype("datetime64[ms]").astype("int64")
        out.reset_index(drop=True).to_parquet(out_path)
        if c["status"] == "OK":
            ok.append(sym)
        print(f"  {sym:6} 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','?')} "
              f"adj={c.get('adj_first/last','?')} [{c['status']}]"
              + (f" split={c['splits']}" if c.get("splits") else ""))
        time.sleep(1.2)

    print("-" * 108)
    print(f"  CERTIFICATI OK ({len(ok)}/{len(UNIVERSE)}): {ok}")
    # Il vincolo che decide tutto: la finestra COMUNE alle 6 gambe scelte.
    print("\n  ⚠️ La profondita' per gamba conta meno della FINESTRA COMUNE: il candidato piu' corto")
    print("     determina il campione su cui si potra' misurare qualunque cosa.")
    ib.disconnect()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())