research(gtaa): la via UCITS e' aperta e costa ~zero — il broker non e' la variabile

Domanda dell'operatore: "usiamo revolut o degiro". Risposta misurata: cambiare broker non sblocca nulla (il PRIIPs e' una norma, non una politica di IB); cambiare VEICOLO si', e costa ~zero. CORREZIONE A UNA MIA AFFERMAZIONE. La nota in gtaa.py diceva che gli UCITS fanno perdere la validazione a 30 anni. Falso: il PRIIPs vieta di COMPRARE, non di GUARDARE — i prezzi dei 6 ETF USA restano leggibili, quindi il segnale gira sui 30 anni per sempre e cambia solo il veicolo su cui si incassa. Misure (3 lenti, un grado di liberta' per volta, 6.3 anni comuni): L0 segnale USA + rend. USA Sh 0.81 / CAGR 3.96% L2 segnale UCITS + rend. UCITS Sh 0.84 / CAGR 4.08% drag del veicolo +0.10%/anno EW, coerente coi TER; ritenuta USA ~35bps a FAVORE dell'UCITS e non inclusa nel drag. Lo stimatore ovvio sbagliava: la media delle differenze giornaliere dava -0.47%/anno su CSPX contro -0.06% vero (SE ~7%/anno = 15x la quantita' stimata, piu' drag di varianza). La deviazione fra veicoli sullo stesso indice si misura sul RAPPORTO CUMULATO. Il vincolo non e' il broker ma il prezzo di UNA azione, che e' una scelta: CSPX $802 vs VUAA $144 sullo stesso S&P 500. A $3.000 con azioni intere l'insieme STORIA tiene 4/6 gambe (a mercato il 33%), l'insieme DEPLOY 6/6 (65%) -> il frazionamento non serve. Letto su gambe-vive+vol, non su Sharpe: il vincolo intero ALZA lo Sharpe perche' de-leveraggia (null de-levering, 4a occorrenza). Resta da verificare una cosa sola: 77-149 ordini/anno contro soglie $0.90 ($3k) / $2.23 ($10k) / $7.62 ($50k) per ordine. Raccomandazione: restare su IB. Feed equity: aggiunto il CROSS-CHECK che mancava (src/data/eq_crosscheck.py). Il primo veicolo estero ha trovato subito CSPX 2012-01-13 con open/high in USD e low/close in EUR (fattore 1.2797 = EURUSD del giorno), invisibile alla guardia maxret>50% — stesso schema dello split 2:1 del 25/07. Soglia non tarabile sulla deviazione (rumore 9.90%, margine 2.2x): cambiata statistica in |dev|/movimento del gemello -> margine 5.3x. Limite EURUSD 1.09 dichiarato e chiuso sul DANNO (dSharpe mediano -0.003), congelato in un test. Book, pesi, cron, config INVARIATI. 435 test verdi (+24). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XBbYmiXqbUNuGpK9sfbsGp
2026-07-26 23:01:54 +00:00
parent 6664575e1c
commit be4690375f
8 changed files with 1481 additions and 16 deletions
@@ -0,0 +1,206 @@
+"""Test del cross-check equity contro il gemello (src/data/eq_crosscheck.py).
+
+Il rilevatore nasce da un difetto VERO trovato il 2026-07-26: CSPX 2012-01-13 con open/high in
+USD e low/close in EUR. La certificazione esistente non lo vedeva perche' la sua unica guardia
+sui movimenti era `maxret > 50% -> SPIKE?` e una contaminazione EUR/USD vale ~22-28%.
+
+I test che contano davvero sono i CONTROLLI su dati reali:
+  * il difetto noto viene segnalato (senza questo il rilevatore potrebbe essere semplicemente rotto);
+  * gli eventi di mercato VERI non vengono segnalati — IDTL marzo 2020 (liquidazione dei treasury)
+    e IGLN aprile 2013 (crollo dell'oro) sono andate-e-ritorni identici nella FORMA a una stampa
+    sbagliata, e una scansione fatta sui prezzi invece che sul rapporto col gemello li segnalava.
+E il test `test_limite_dichiarato_*` congela il LIMITE noto: se qualcuno abbassa la soglia, il
+test dice esattamente cosa e' cambiato invece di lasciar credere che il rilevatore sia completo.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from src.data.eq_crosscheck import (  # noqa: E402
+    MIN_DEV, MIN_UNEXPLAINED, REVERT_BARS, TWIN_MOVE_FLOOR,
+    detect_twin_outliers, repair_twin_outliers,
+)
+
+RAW = ROOT / "data" / "raw"
+
+
+def _load(prefix: str, sym: str) -> pd.DataFrame:
+    p = RAW / f"{prefix}_{sym.lower()}_1d.parquet"
+    if not p.exists():
+        pytest.skip(f"{p.name} assente (data/raw e' gitignored)")
+    d = pd.read_parquet(p)
+    d.index = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
+    return d.sort_index()
+
+
+def _close(prefix: str, sym: str) -> pd.Series:
+    d = _load(prefix, sym)
+    return pd.Series(d["close"].astype(float).values, index=d.index)
+
+
+def _serie(vals, start="2020-01-01"):
+    return pd.Series(np.asarray(vals, float),
+                     index=pd.bdate_range(start, periods=len(vals)))
+
+
+# ===========================================================================
+# controlli su DATI REALI — un rilevatore che non segnala nulla puo' essere rotto
+# ===========================================================================
+def test_controllo_positivo_il_difetto_noto_di_cspx_viene_segnalato():
+    """IL test. Se questo fallisce, il rilevatore non serve a niente."""
+    bad = detect_twin_outliers(_close("eqx", "CSPX"), _close("eq", "SPY"))
+    date = {pd.Timestamp(b["date"]).date().isoformat() for b in bad}
+    assert "2012-01-13" in date, f"difetto noto non rilevato (segnalate: {sorted(date)})"
+
+
+def test_il_difetto_ha_la_firma_valutaria_attesa():
+    bad = [b for b in detect_twin_outliers(_close("eqx", "CSPX"), _close("eq", "SPY"))
+           if pd.Timestamp(b["date"]).date().isoformat() == "2012-01-13"]
+    assert len(bad) == 1
+    b = bad[0]
+    assert b["dev"] < -0.15, "la deviazione non ha la scala di una contaminazione valutaria"
+    assert abs(b["twin_move"]) < 0.01, "il gemello si era mosso: allora non e' inspiegato"
+    assert b["unexplained"] > 3 * MIN_UNEXPLAINED / 2
+
+
+def test_controllo_negativo_gli_eventi_di_mercato_veri_non_sono_segnalati():
+    """Marzo 2020 sui treasury e aprile 2013 sull'oro sono andate-e-ritorni VERI: nella forma
+    identici a una stampa sbagliata. Li distingue solo il fatto che li fa anche il gemello."""
+    for veic, gem, quando in [("IDTL", "TLT", "2020-03"), ("IGLN", "GLD", "2013-04")]:
+        bad = detect_twin_outliers(_close("eqx", veic), _close("eq", gem))
+        colpiti = [pd.Timestamp(b["date"]).strftime("%Y-%m") for b in bad]
+        assert quando not in colpiti, f"{veic}: evento di mercato vero scambiato per difetto"
+
+
+def test_nessun_falso_positivo_sui_veicoli_puliti():
+    for veic, gem in [("EQQQ", "QQQ"), ("XRSU", "IWM"), ("R2US", "IWM"),
+                      ("IDTL", "TLT"), ("IGLN", "GLD"), ("IHYU", "HYG")]:
+        bad = detect_twin_outliers(_close("eqx", veic), _close("eq", gem))
+        assert bad == [], f"{veic}/{gem}: falsi positivi {[str(b['date']) for b in bad]}"
+
+
+# ===========================================================================
+# le tre condizioni, una alla volta
+# ===========================================================================
+def test_una_deviazione_che_il_gemello_SPIEGA_non_e_segnalata():
+    """La condizione decisiva: un disallineamento d'orario non puo' superare il movimento del
+    mercato. Se il gemello si e' mosso quanto il veicolo, non c'e' niente da segnalare."""
+    n = 80
+    twin = _serie(np.full(n, 100.0))
+    twin.iloc[40] = 80.0          # il gemello CROLLA del 20%...
+    twin.iloc[41] = 100.0
+    veh = twin * 1.0              # ...e il veicolo fa lo stesso
+    veh.iloc[40] = 78.0           # con un piccolo scarto d'orario
+    assert detect_twin_outliers(veh, twin) == []
+
+
+def test_una_divergenza_che_PERSISTE_non_e_una_stampa():
+    """Un cambio di livello permanente (fusione, cambio di indice, ridenominazione) va tenuto:
+    scartarlo cancellerebbe un fatto invece di un errore."""
+    n = 80
+    twin = _serie(np.full(n, 100.0))
+    veh = _serie(np.concatenate([np.full(40, 100.0), np.full(n - 40, 78.0)]))
+    assert detect_twin_outliers(veh, twin) == []
+
+
+def test_una_deviazione_troppo_piccola_non_e_segnalata():
+    n = 80
+    twin = _serie(np.full(n, 100.0))
+    veh = twin.copy()
+    veh.iloc[40] = 100.0 * (1 - MIN_DEV / 2)
+    assert detect_twin_outliers(veh, twin) == []
+
+
+# ===========================================================================
+# iniezione: cosa si cattura e cosa NO (il limite dichiarato, congelato in un test)
+# ===========================================================================
+def _inietta(fx: float):
+    n = 200
+    rng = np.random.default_rng(7)
+    base = 100 * np.exp(np.cumsum(rng.normal(0, 0.01, n)))
+    twin = _serie(base)
+    veh = _serie(base * 1.0)
+    veh.iloc[120] = veh.iloc[120] / fx
+    return veh, twin
+
+
+def test_contaminazione_a_cambio_alto_gbp_viene_catturata():
+    veh, twin = _inietta(1.27)
+    bad = detect_twin_outliers(veh, twin)
+    assert len(bad) == 1 and pd.Timestamp(bad[0]["date"]) == veh.index[120]
+
+
+def test_limite_dichiarato_contaminazione_a_cambio_basso_eur_sfugge_a_mercato_mosso():
+    """⚠️ Questo test NON descrive un comportamento desiderato: congela un LIMITE noto.
+
+    EURUSD ~1.09 produce ~8% di deviazione, che sta dentro il rumore legittimo dei giorni
+    violenti (misurato 9.9% il 2025-04-09). Abbassare la soglia per catturarlo genererebbe
+    falsi positivi sui giorni veri. Il punto e' chiuso altrove misurando il DANNO
+    (r0726_gtaa_ucits.py sez. 5: dSharpe mediano -0.003, peggiore -0.06).
+    Se qualcuno cambia le soglie, questo test dice esattamente cosa e' cambiato.
+    """
+    n = 200
+    rng = np.random.default_rng(11)
+    base = 100 * np.exp(np.cumsum(rng.normal(0, 0.01, n)))
+    twin = _serie(base)
+    veh = _serie(base.copy())
+    veh.iloc[120] /= 1.09
+    twin.iloc[120] *= 1.02        # il gemello quel giorno si muove: la deviazione diventa 'spiegabile'
+    assert detect_twin_outliers(veh, twin) == [], \
+        "il limite dichiarato non vale piu': aggiornare la docstring del modulo e il diario"
+
+
+# ===========================================================================
+# riparazione
+# ===========================================================================
+def test_la_riparazione_scarta_la_barra_e_non_ne_inventa_una():
+    d = _load("eqx", "CSPX")
+    pulito, bad = repair_twin_outliers(d, _close("eq", "SPY"))
+    assert len(bad) == 1
+    assert len(pulito) == len(d) - 1
+    rimasta = pd.to_datetime(pulito.index).tz_localize(None).normalize()
+    assert pd.Timestamp("2012-01-13") not in set(rimasta)
+    # nessun prezzo ricostruito: i close superstiti sono identici agli originali
+    orig = d["close"]
+    assert (pulito["close"].values == orig.loc[pulito.index].values).all()
+
+
+def test_la_riparazione_rende_sano_il_rendimento_a_cavallo():
+    d = _load("eqx", "CSPX")
+    pulito, _ = repair_twin_outliers(d, _close("eq", "SPY"))
+    r = pulito["close"].pct_change()
+    w = r[(r.index >= "2012-01-10") & (r.index <= "2012-01-20")]
+    assert w.abs().max() < 0.05, f"resta un movimento assurdo: {w.abs().max():.1%}"
+
+
+def test_riparazione_idempotente():
+    d = _load("eqx", "CSPX")
+    p1, b1 = repair_twin_outliers(d, _close("eq", "SPY"))
+    p2, b2 = repair_twin_outliers(p1, _close("eq", "SPY"))
+    assert b2 == [] and len(p2) == len(p1)
+
+
+def test_serie_vuota_o_cortissima_non_esplode():
+    vuoto = pd.DataFrame({"close": []}, index=pd.DatetimeIndex([]))
+    assert repair_twin_outliers(vuoto, _serie([1.0, 2.0])) == (vuoto, [])
+    assert detect_twin_outliers(_serie([1.0] * 5), _serie([1.0] * 5)) == []
+
+
+# ===========================================================================
+# la taratura e' un valore, non un dettaglio
+# ===========================================================================
+def test_le_soglie_sono_congelate():
+    """Tarate su misura: rumore massimo 8.1, difetto 42.7 -> soglia 18.0 equidistante in scala
+    logaritmica. Cambiarle senza rifare la taratura invalida ogni conclusione del 26/07."""
+    assert MIN_UNEXPLAINED == 18.0
+    assert MIN_DEV == 0.05
+    assert TWIN_MOVE_FLOOR == 0.005
+    assert REVERT_BARS == 5
@@ -0,0 +1,170 @@
+"""Test della rivalidazione di GTAA01 sui veicoli UCITS (scripts/research/r0726_gtaa_ucits.py).
+
+Tre cose vanno bloccate qui, e sono di natura diversa:
+  1. il MOTORE generalizzato deve essere la produzione (bit-exact), altrimenti le tre lenti
+     misurano una strategia che non e' quella del progetto;
+  2. lo STIMATORE del drag: la media delle differenze giornaliere NON e' la deviazione fra due
+     veicoli sullo stesso indice, e il test lo dimostra su un caso costruito dove la risposta
+     giusta e' nota;
+  3. il FINDING operativo: a $3.000 l'insieme di veicoli a prezzo unitario alto perde due gambe,
+     quello a prezzo basso no. E' la ragione per cui il frazionamento del broker non serve.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "scripts" / "research"))
+
+from src.portfolio import gtaa as G  # noqa: E402
+
+import r0726_gtaa_ucits as U  # noqa: E402
+
+RAW = ROOT / "data" / "raw"
+
+
+def _need(prefix: str, syms) -> None:
+    for s in syms:
+        if not (RAW / f"{prefix}_{s.lower()}_1d.parquet").exists():
+            pytest.skip(f"{prefix}_{s.lower()}_1d.parquet assente (data/raw e' gitignored)")
+
+
+# ===========================================================================
+# 1. il motore E' la produzione
+# ===========================================================================
+def test_il_motore_generalizzato_riproduce_gtaa_returns_bit_exact():
+    """IL test di questo file. `sleeve_native` deve dare ESATTAMENTE gtaa.gtaa_returns: se
+    diverge, ogni confronto fra lenti sta misurando un'altra strategia."""
+    _need("eq", U.US)
+    us = {s: U.close_us(s) for s in U.US}
+    mine = U.sleeve_native(us, us, G.GTAA_DEFAULT_CAPITAL, G.REBAL_BAND_USD, G.REBAL_EVERY,
+                           U.cost_ib_us)
+    prod = U._norm(G.gtaa_returns(capital=G.GTAA_DEFAULT_CAPITAL))
+    both = pd.concat([mine, prod], axis=1).dropna()
+    assert len(both) > 5000
+    assert float((both.iloc[:, 0] - both.iloc[:, 1]).abs().max()) == 0.0
+
+
+def test_segnale_e_veicolo_separati_coincidono_quando_sono_la_stessa_serie():
+    _need("eq", ["SPY"])
+    _need("eqx", ["CSPX"])
+    spy, cspx = U.close_us("SPY"), U.close_uc("CSPX", U.close_us("SPY"))[0]
+    cal = U.common_cal([spy, cspx])
+    a = U.leg_returns(spy, spy, cal, 1000.0, 50.0, 5, U.cost_ib_us)[0]
+    b = U.leg_returns(spy, spy, cal, 1000.0, 50.0, 5, U.cost_ib_us)[0]
+    assert float((a - b).abs().max()) == 0.0
+    # con veicolo diverso il risultato DEVE cambiare, altrimenti il parametro e' ignorato
+    c = U.leg_returns(spy, cspx, cal, 1000.0, 50.0, 5, U.cost_ib_us)[0]
+    assert float((a - c).abs().max()) > 1e-6
+
+
+# ===========================================================================
+# 2. lo stimatore del drag
+# ===========================================================================
+def _due_veicoli(n=1500, drag_annuo=0.0, rumore=0.0, seed=3):
+    """Due veicoli sullo STESSO indice: b = a x (drift noto) x (rumore di campionamento).
+    Il rumore e' i.i.d. e stazionario — e' l'analogo del fatto che Londra chiude 4h30 prima."""
+    rng = np.random.default_rng(seed)
+    idx = pd.bdate_range("2015-01-01", periods=n)
+    p = 100 * np.exp(np.cumsum(rng.normal(0.0003, 0.011, n)))
+    a = pd.Series(p, index=idx)
+    b = a * (1 + drag_annuo) ** (np.arange(n) / 252.0)
+    if rumore:
+        b = b * (1 + rng.normal(0, rumore, n))
+    return a, b
+
+
+def test_il_drag_cumulato_recupera_un_drag_noto():
+    a, b = _due_veicoli(drag_annuo=-0.0030)
+    assert U.drag_cumulato(a, b, a.index) == pytest.approx(-0.0030, abs=2e-4)
+
+
+def test_il_drag_cumulato_e_immune_al_rumore_di_campionamento():
+    """Il rumore stazionario non e' un drag: uno stimatore corretto non lo deve vedere."""
+    a, b = _due_veicoli(drag_annuo=0.0, rumore=0.01)
+    assert abs(U.drag_cumulato(a, b, a.index)) < 0.003
+
+
+def test_la_media_delle_differenze_giornaliere_NON_e_lo_stimatore_giusto():
+    """⚠️ Il test che giustifica `drag_cumulato`. Due veicoli con lo STESSO drift geometrico ma
+    campionati diversamente: la media delle differenze semplici vede un drag che non esiste,
+    perche' e' contaminata dalla diversa varianza giornaliera (drag di varianza), mentre il
+    rapporto cumulato — che e' la domanda vera, 'quanto mi costa il veicolo' — vede ~0.
+
+    E' il difetto che nella prima stesura dava -0.47%/anno su CSPX invece di -0.06%.
+    """
+    a, b = _due_veicoli(drag_annuo=0.0, rumore=0.012)
+    cum = U.drag_cumulato(a, b, a.index)
+    media = float((b.pct_change() - a.pct_change()).dropna().mean() * 252)
+    assert abs(cum) < 0.003, "il rapporto cumulato dovrebbe vedere ~0"
+    assert abs(media) > 5 * abs(cum), \
+        f"il caso costruito non separa i due stimatori (cum {cum:.4%}, media {media:.4%})"
+
+
+# ===========================================================================
+# 3. il finding operativo: prezzo unitario e gambe vive
+# ===========================================================================
+def test_azioni_intere_non_possono_aumentare_l_esposizione():
+    """Un vincolo arrotonda IN GIU': se l'esposizione sale, il modello e' sbagliato (e uno
+    Sharpe piu' alto sarebbe null de-levering travestito da miglioramento)."""
+    _need("eqx", set(U.SET_DEPLOY.values()))
+    _need("eq", U.US)
+    us = {s: U.close_us(s) for s in U.US}
+    dep = {s: U.close_uc(u, us[s])[0] for s, u in U.SET_DEPLOY.items()}
+    cal = U.common_cal(list(dep.values()))[-800:]
+    for cap in (3_000.0, 20_000.0):
+        _, pf = U.sleeve(dep, dep, cal, cap, G.REBAL_BAND_USD, G.REBAL_EVERY,
+                         U.cost_form(fixed=1.0), want_pos=True)
+        _, pi = U.sleeve(dep, dep, cal, cap, G.REBAL_BAND_USD, G.REBAL_EVERY,
+                         U.cost_form(fixed=1.0), integer_shares=True, want_pos=True)
+        assert pi.sum().sum() <= pf.sum().sum() + 1e-9
+
+
+def test_a_3000_l_insieme_a_prezzo_alto_perde_gambe_e_quello_a_prezzo_basso_no():
+    """IL finding operativo, bloccato. CSPX ($802) ed EQQQ ($693) non sono comprabili con una
+    gamba da $500: senza frazionamento quelle due gambe semplicemente non esistono."""
+    _need("eq", U.US)
+    _need("eqx", set(U.SET_STORIA.values()) | set(U.SET_DEPLOY.values()))
+    us = {s: U.close_us(s) for s in U.US}
+    storia = {s: U.close_uc(u, us[s])[0] for s, u in U.SET_STORIA.items()}
+    deploy = {s: U.close_uc(u, us[s])[0] for s, u in U.SET_DEPLOY.items()}
+    cal = U.common_cal(list(storia.values()) + list(deploy.values()))[-700:]
+
+    def gambe_vive(v):
+        _, pos = U.sleeve(v, v, cal, 3_000.0, G.REBAL_BAND_USD, G.REBAL_EVERY,
+                          U.cost_form(fixed=1.0), integer_shares=True, want_pos=True)
+        return int((pos.abs().sum(axis=0) > 0).sum())
+
+    assert gambe_vive(storia) < 6, "l'insieme a prezzo alto dovrebbe perdere gambe a $3.000"
+    assert gambe_vive(deploy) == 6, "l'insieme a prezzo basso deve tenerle tutte e sei"
+
+
+def test_i_veicoli_deploy_costano_meno_per_azione_di_quelli_storia():
+    _need("eq", U.US)
+    _need("eqx", set(U.SET_STORIA.values()) | set(U.SET_DEPLOY.values()))
+    us = {s: U.close_us(s) for s in U.US}
+    for s in U.US:
+        a = float(U._norm(U.close_uc(U.SET_STORIA[s], us[s])[0]).iloc[-1])
+        b = float(U._norm(U.close_uc(U.SET_DEPLOY[s], us[s])[0]).iloc[-1])
+        assert b <= a + 1e-9, f"{s}: il veicolo 'deploy' non e' piu' economico ({b} > {a})"
+
+
+# ===========================================================================
+# scelte dichiarate
+# ===========================================================================
+def test_i_veicoli_scelti_sono_indici_equivalenti_non_fattori():
+    """RTWO e' Russell 2000 *quality*, CSUSS e' MSCI USA Small Cap *ESG*, IDP6 e' S&P 600:
+    tutti hanno piu' storia dei veicoli scelti, e nessuno e' la stessa esposizione."""
+    scelti = set(U.SET_STORIA.values()) | set(U.SET_DEPLOY.values())
+    assert not (scelti & {"RTWO", "CSUSS", "IDP6"})
+
+
+def test_le_due_mappe_coprono_le_stesse_sei_gambe():
+    assert tuple(U.SET_STORIA) == U.US and tuple(U.SET_DEPLOY) == U.US
+    assert len(U.US) == 6