Files
PythagorasGoal/tests/test_r0909_xrp_terza_gamba.py
T
Adriano Dal Pastro bb40e87d13 crolli: il libro perde nel giorno e chiude > 0 in 8/8 finestre dal 2022 (SKH01 short); crollo catturato a vol bassa, XRP diluisce come SOL
Quattro misure (r0909_*), nessun cambio a libro/pesi/config:
- libro nei crolli: −0,48%/g nei 160 giorni ≤ −5%, positivo per finestra dal 2022
  per la gamba short di SKH01 (108% dei guadagni); beta 0,0769 riprodotto (§46);
  il peso di SKH01 resta chiuso dal gate (in-sample).
- crollo catturato 1-5/06/2026 a quote vere: f_net 0,74 = rally; put δ−0,10
  1,92× il modello; a vol bassa e fuori dal gate di VRP01 → §3 non si riapre.
- XRP terza gamba (harness r0822_sol_leg): hold-out −0,169 in 0/24, un anno buono.
- universo Deribit: liquidi solo BTC/ETH/XRP/SOL; XS01 13/19; BTCDVOL future
  non negoziabile; PAXG non misurato.
Revisione fable: due conclusioni smontate (MTM 2,6× era un artefatto di quote;
«8/12 guadagna» era l'ordine delle classi). Debito §5.18: cblib.spot_series
+ asof guarda un'ora avanti (feed 1h etichettato all'apertura).
Test Opus: 92 nuovi, suite 1008/1008; due difetti del verdetto XRP corretti.
Docs: diario, RESULTS §74-77, CLAUDE.md, memoria 20/50, README; journal 07-08/09.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017zJjHUS7mf4pnGE6pq9RTt
2026-09-09 16:33:54 +00:00

352 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Test di `scripts/research/r0909_xrp_terza_gamba.py` (2026-09-09) — XRP come terza gamba.
Tre cose vanno protette, e nessuna e' il numero finale:
1. **Il paginatore.** `_fetch_1h` esiste nella forma che ha perche' Coinbase ha un BUCO su XRP
(delistato 2021-01, rilistato 2023-07 dopo la sentenza SEC): un batch vuoto NON deve
fermare il ciclo. Un paginatore che si ferma al primo vuoto avrebbe certificato XRP su
mezzo campione — e la certificazione decide QUALE lente si usa, cioe' decide il risultato.
2. **La regola L-PULITA.** "Dal primo anno in cui tutti i successivi restano sotto soglia":
un anno pulito ISOLATO non e' un regime. E' la regola che sceglie la finestra, ed e'
dichiarata prima di guardare l'esito — quindi va congelata prima che qualcuno la ritocchi.
3. **I tre rami del verdetto**, coi loro AND: AGGIUNGE richiede hold-out E de-levering E anni.
Piu' il guardrail del namespace: XRP vive in `alt_xrp_*`, e `load_data("XRP")` deve continuare
a fallire (il "solo dati certi" di CLAUDE.md non e' codice, e' l'assenza del file).
⚠️ NESSUN test va in rete: il riferimento Coinbase/Bitstamp o e' in cache o il test si salta.
"""
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"))
import r0909_xrp_terza_gamba as M # noqa: E402
RAW = ROOT / "data" / "raw"
ORA_MS = 3_600_000
T0 = pd.Timestamp("2026-01-01", tz="UTC")
# ===========================================================================================
# (a) _fetch_1h — un batch vuoto NON ferma il ciclo
# ===========================================================================================
class ExFinto:
"""Exchange finto e DETERMINISTICO: restituisce un batch vuoto alle chiamate elencate in
`vuote` (indici 0-based), altrove `limit` barre orarie a partire da `since`."""
def __init__(self, vuote=(), righe_prima_di_since=False):
self.vuote = set(vuote)
self.righe_prima_di_since = righe_prima_di_since
self.since_visti: list[int] = []
def fetch_ohlcv(self, symbol, tf, since=None, limit=None):
n = len(self.since_visti)
self.since_visti.append(since)
assert tf == "1h" and symbol and limit
if n in self.vuote:
return []
base = since - 10 * ORA_MS if self.righe_prima_di_since else since
return [[base + i * ORA_MS, 1.0, 1.0, 1.0, 10.0 + n + i, 1.0] for i in range(limit)]
@pytest.fixture()
def orologio_fermo(monkeypatch):
"""`_fetch_1h` chiude il ciclo su `time.time()`: qui lo si ferma a T0+72h, cosi' il numero
di pagine e' un fatto e non il meteo della macchina. (Il patch vive solo dentro il test.)"""
fine = T0 + pd.Timedelta(hours=72)
monkeypatch.setattr(M.time, "time", lambda: fine.timestamp())
return fine
def test_fetch_1h_AVANZA_dopo_un_batch_vuoto(orologio_fermo):
"""IL test di questo file. Coinbase su XRP restituisce vuoto per due anni e mezzo: se il
ciclo si fermasse (break) o non facesse avanzare `since` (loop infinito), la certificazione
girerebbe su un campione mutilato — e la lente L-PULITA la sceglie la certificazione.
DEVE FALLIRE SE: qualcuno sostituisce `since += limit*3_600_000; continue` con un `break`."""
ex = ExFinto(vuote={0})
s = M._fetch_1h(ex, "XRP/USD", T0, 24)
assert len(ex.since_visti) >= 2, "il ciclo si e' fermato al primo batch vuoto"
assert ex.since_visti[1] == ex.since_visti[0] + 24 * ORA_MS, (
"dopo un batch vuoto `since` deve avanzare di un'intera pagina")
assert len(s) == 48
assert s.index.min() == T0 + pd.Timedelta(hours=24), "la prima pagina era il buco"
assert s.index.is_monotonic_increasing and s.index.is_unique
def test_fetch_1h_scavalca_un_buco_IN_MEZZO_e_riprende(orologio_fermo):
"""Il buco vero di XRP e' in mezzo (2021-01 → 2023-07), non in testa: il dato PRIMA e quello
DOPO devono stare tutti e due nella serie, col buco che resta un buco (D2: una riga assente
non si inventa)."""
ex = ExFinto(vuote={1})
s = M._fetch_1h(ex, "XRP/USD", T0, 24)
assert len(s) == 48
assert s.index.min() == T0
assert s.index.max() == T0 + pd.Timedelta(hours=71)
buco = s[(s.index >= T0 + pd.Timedelta(hours=24)) & (s.index < T0 + pd.Timedelta(hours=48))]
assert len(buco) == 0, "il buco e' stato riempito con qualcosa: non c'era niente da riempire"
def test_fetch_1h_scarta_le_righe_PRIMA_di_since_e_avanza_lo_stesso(orologio_fermo):
"""Alcuni venue ignorano `since` e ritornano sempre la stessa pagina: il filtro
`x[0] >= since` la svuota, e a quel punto vale la stessa regola del batch vuoto."""
ex = ExFinto(righe_prima_di_since=True)
s = M._fetch_1h(ex, "XRP/USD", T0, 24)
# le prime 10 barre di ogni pagina cadono prima di `since` e vengono scartate
assert len(ex.since_visti) >= 2
assert ex.since_visti[1] > ex.since_visti[0], "il ciclo si e' impantanato"
assert (s.index >= T0).all()
def test_fetch_1h_non_si_impianta_su_una_pagina_da_una_riga(monkeypatch):
"""La guardia `since = nxt if nxt > since else since + pagina` esiste per non girare a
vuoto: qui ogni chiamata rende UNA barra a `since`, e il ciclo deve comunque terminare."""
fine = T0 + pd.Timedelta(hours=5)
monkeypatch.setattr(M.time, "time", lambda: fine.timestamp())
class UnaRiga:
def __init__(self):
self.n = 0
def fetch_ohlcv(self, symbol, tf, since=None, limit=None):
self.n += 1
assert self.n < 50, "ciclo infinito"
return [[since, 1.0, 1.0, 1.0, 10.0, 1.0]]
s = M._fetch_1h(UnaRiga(), "XRP/USD", T0, 24)
assert len(s) == 5
assert list(s.index) == list(pd.date_range(T0, periods=5, freq="h"))
# ===========================================================================================
# (b) la regola L-PULITA dentro certifica()
# ===========================================================================================
ANNI = (2022, 2023, 2024, 2025, 2026)
def _finto_feed(sporchi: dict[int, int], n: int = 700):
"""Feed sintetico: per ogni anno `n` barre orarie, di cui `sporchi[y]` a +2% dal
riferimento (cioe' sopra la soglia >1% che definisce una barra sporca)."""
dd, rr = [], []
for y in ANNI:
idx = pd.date_range(f"{y}-01-01", periods=n, freq="h", tz="UTC")
rif = np.full(n, 100.0)
mio = rif.copy()
mio[:sporchi.get(y, 0)] = 102.0
dd.append(pd.DataFrame(dict(datetime=idx, close=mio, high=mio + 1.0, low=mio - 1.0)))
rr.append(pd.DataFrame(dict(ts=idx, close=rif, fonte="CB")))
D = pd.concat(dd, ignore_index=True)
R = pd.concat(rr, ignore_index=True).set_index("ts")
return D, R
def _certifica(monkeypatch, sporchi: dict[int, int], n: int = 700) -> dict:
D, R = _finto_feed(sporchi, n)
monkeypatch.setattr(M, "dati", lambda asset, tf: D)
monkeypatch.setattr(M, "riferimento_1h", lambda start: R)
return M.certifica()
def test_L_PULITA_parte_dal_primo_anno_con_TUTTI_i_successivi_puliti(monkeypatch):
"""Lo scenario vero: 2022 e 2023 sporchi (0,57% nel 2023), 2024-2026 puliti → 2024-01-01."""
c = _certifica(monkeypatch, {2022: 8, 2023: 4})
assert c["per_anno"][2023]["q_1pct"] == pytest.approx(4 / 700, abs=1e-6)
assert c["per_anno"][2022]["q_1pct"] > M.SOGLIA_PULITA
assert c["puliti"] == [2024, 2025, 2026]
assert c["pulita_da"] == pd.Timestamp("2024-01-01", tz="UTC")
def test_L_PULITA_ignora_un_anno_pulito_ISOLATO(monkeypatch):
"""La regressione che la regola esiste per impedire: 2022 pulito, 2023 sporco, 2024+ puliti.
Un anno pulito isolato NON e' un regime — la lente parte dalla coda pulita finale (2024),
non dal 2022. DEVE FALLIRE SE: qualcuno semplifica in `min(puliti)`."""
c = _certifica(monkeypatch, {2023: 8})
assert c["puliti"] == [2022, 2024, 2025, 2026], "il 2022 e' pulito, e va detto"
assert c["pulita_da"] == pd.Timestamp("2024-01-01", tz="UTC"), (
"la lente non puo' partire dal 2022: il 2023 in mezzo e' sporco")
def test_L_PULITA_e_tutta_la_storia_se_non_c_e_niente_di_sporco(monkeypatch):
c = _certifica(monkeypatch, {})
assert c["puliti"] == list(ANNI)
assert c["pulita_da"] == pd.Timestamp("2022-01-01", tz="UTC")
def test_L_PULITA_e_NONE_se_nessun_anno_e_pulito(monkeypatch):
"""P5: "non c'e' una finestra pulita" ha uno stato suo. `misura()` allora non aggiunge la
lente e `verdetto()` ripiega sul 2024+ dichiarato (la lente di SOL)."""
c = _certifica(monkeypatch, {y: 8 for y in ANNI})
assert c["puliti"] == []
assert c["pulita_da"] is None
def test_la_soglia_di_pulizia_e_INCLUSIVA_e_vale_SOGLIA_PULITA(monkeypatch):
"""Il confronto e' `<=`: esattamente in soglia l'anno e' pulito. Con n=1000 barre la
soglia 0,5% cade su un intero (5 barre), quindi il confine e' asseribile."""
c = _certifica(monkeypatch, {2022: 5, 2023: 5, 2024: 5, 2025: 5, 2026: 5}, n=1000)
assert c["per_anno"][2022]["q_1pct"] == pytest.approx(M.SOGLIA_PULITA)
assert c["puliti"] == list(ANNI)
c2 = _certifica(monkeypatch, {2022: 6, 2023: 5, 2024: 5, 2025: 5, 2026: 5}, n=1000)
assert 2022 not in c2["puliti"]
assert c2["pulita_da"] == pd.Timestamp("2023-01-01", tz="UTC")
def test_certifica_riporta_per_ogni_anno_le_misure_che_la_tabella_stampa(monkeypatch):
c = _certifica(monkeypatch, {2022: 8})
for y, r in c["per_anno"].items():
assert set(r) == {"n", "q_1pct", "med_bps", "fonte", "flat_1h", "flat_5m"}
assert r["n"] == 700
assert r["flat_1h"] == 0.0, "high != low per costruzione"
# ===========================================================================================
# (c) verdetto() — i tre rami, tutti con AND
# ===========================================================================================
NOME = "L-PULITA (2024+)"
def _m(hold, sh3, sh2k, anni_dsh, nome: str = NOME, k: float = 0.827) -> dict:
h = np.asarray(hold, float)
z = np.zeros(len(h))
return dict(lenti={nome: dict(sh=z, hold=h, dd=z, cagr=z)},
delev={nome: dict(k=k, sh3=sh3, sh2k=sh2k, cagr3=0.146, cagr2k=0.134)},
anno={2024 + i: dict(dsh=float(v)) for i, v in enumerate(anni_dsh)})
def test_verdetto_AGGIUNGE_solo_se_passano_tutte_e_tre_le_condizioni():
v = M.verdetto(_m([0.10] * 24, sh3=1.60, sh2k=1.50, anni_dsh=[0.5, 0.4, -0.1]))
assert v.startswith("XRP AGGIUNGE")
def test_verdetto_DILUISCE_quando_cadono_hold_out_E_de_levering():
v = M.verdetto(_m([-0.169] * 24, sh3=1.54, sh2k=1.53, anni_dsh=[0.54, -0.44, -0.25]))
assert v.startswith("XRP DILUISCE")
def test_verdetto_PARI_quando_le_condizioni_si_contraddicono():
"""N5/M8: quando una condizione passa e l'altra no, il verdetto non sceglie la piu' comoda —
dichiara che non e' selezionabile."""
solo_hold = M.verdetto(_m([0.10] * 24, sh3=1.50, sh2k=1.60, anni_dsh=[0.5, 0.4, -0.1]))
assert solo_hold.startswith("XRP: PARI / NON SELEZIONABILE")
solo_delev = M.verdetto(_m([-0.10] * 24, sh3=1.60, sh2k=1.50, anni_dsh=[0.5, 0.4, -0.1]))
assert solo_delev.startswith("XRP: PARI / NON SELEZIONABILE")
def test_verdetto_chiede_il_75_percento_delle_ancore_non_la_sola_mediana():
"""M7/§2: la banda d'ancora NON e' un intervallo di confidenza, ma la robustezza alla scelta
dell'ancora e' una condizione a se'. Mediana > 0 con solo il 70% delle ancore positive non
basta. DEVE FALLIRE SE: qualcuno lascia solo la mediana."""
hold = [0.10] * 16 + [-0.10] * 8 # mediana +0,10, ma >0 solo nel 67% delle ancore
v = M.verdetto(_m(hold, sh3=1.60, sh2k=1.50, anni_dsh=[0.5, 0.4, -0.1]))
assert not v.startswith("XRP AGGIUNGE")
hold_ok = [0.10] * 18 + [-0.10] * 6 # esattamente 75%
assert M.verdetto(_m(hold_ok, sh3=1.60, sh2k=1.50,
anni_dsh=[0.5, 0.4, -0.1])).startswith("XRP AGGIUNGE")
def test_verdetto_il_de_levering_chiede_un_margine_di_0_02():
"""M5: il null del de-levering non si passa "per un pelo" — un vantaggio dentro 0,02 di
Sharpe e' de-levering travestito, ed e' esattamente il caso di XRP (1,54 contro 1,53)."""
quasi = M.verdetto(_m([0.10] * 24, sh3=1.52, sh2k=1.50, anni_dsh=[0.5, 0.4, -0.1]))
assert not quasi.startswith("XRP AGGIUNGE")
appena = M.verdetto(_m([0.10] * 24, sh3=1.521, sh2k=1.50, anni_dsh=[0.5, 0.4, -0.1]))
assert appena.startswith("XRP AGGIUNGE")
def test_verdetto_ripiega_sulla_lente_di_SOL_se_non_c_e_una_coda_pulita():
"""Quando `pulita_da` e' None, `misura()` non crea la lente L-PULITA: il verdetto deve
leggere quella dichiarata di riserva, non la prima che capita."""
m = _m([-0.10] * 24, sh3=1.50, sh2k=1.60, anni_dsh=[-0.1, -0.2, -0.3],
nome="2024+ (lente di SOL)")
m["lenti"]["L-FULL"] = dict(sh=np.zeros(24), hold=np.full(24, 9.0),
dd=np.zeros(24), cagr=np.zeros(24))
m["delev"]["L-FULL"] = dict(k=1.0, sh3=9.0, sh2k=0.0, cagr3=0.0, cagr2k=0.0)
v = M.verdetto(m)
assert "lente 2024+ (lente di SOL)" in v
assert v.startswith("XRP DILUISCE"), "ha letto la L-FULL invece della lente dichiarata"
def test_verdetto_riproduce_ALLA_LETTERA_la_riga_pubblicata_il_09_09():
"""M23: i valori sono quelli usciti dalla misura a 24 ancore del 09/09."""
m = _m([-0.169] * 24, sh3=1.54, sh2k=1.53, anni_dsh=[0.54, -0.44, -0.25])
assert M.verdetto(m) == (
"XRP DILUISCE — lente L-PULITA (2024+): dSharpe hold-out mediano -0.169 "
"(>0 nel 0% di 24 ancore); de-levering: Sh 3 gambe 1.54 vs 2 gambe×0.827 1.53; "
"anni 2024+ con dSh>0: 1/3")
# ===========================================================================================
# (d) il GUARDRAIL del namespace
# ===========================================================================================
def test_il_divieto_su_data_raw_xrp_e_gia_DERIVATO_dal_codice():
"""`tests/test_sol_leg.py::test_nessun_asset_fuori_universo_nel_namespace_NUDO` deriva gli
asset vietati da `rebuild_history.DERIBIT_INSTR`: finche' XRP e' li' dentro, quel test
copre gia' `data/raw/xrp_*.parquet` e qui NON si duplica. Questo test congela la premessa
— se qualcuno togliesse XRP dal dict, la guardia sparirebbe in silenzio."""
from scripts.analysis.rebuild_history import DERIBIT_INSTR
assert "XRP" in DERIBIT_INSTR, (
"XRP fuori da DERIBIT_INSTR: il divieto su data/raw/xrp_*.parquet non e' piu' derivato "
"da nessuna parte — riscrivere la guardia in test_sol_leg.py prima di toglierlo")
def test_load_data_XRP_continua_a_FALLIRE():
"""L'altra meta': il dato XRP dello studio vive in `alt_xrp_*` e il feed attivo non lo vede.
Se un giorno `load_data("XRP")` funzionasse, il cron (`--asset BTC ETH`) non lo
rinfrescherebbe comunque -> dato stantio che sembra attivo."""
from src.data.downloader import load_data
with pytest.raises(FileNotFoundError):
load_data("XRP", "1h")
def test_lo_studio_legge_il_namespace_di_RICERCA():
"""`dati()` importata da r0822_sol_leg instrada XRP su `alt_xrp_*`: se non c'e' il file,
l'errore deve dire come rigenerarlo, non fingere che l'asset non esista."""
assert M.ASSET == "XRP" and M.TRE == ("BTC", "ETH", "XRP")
if (RAW / "alt_xrp_1h.parquet").exists():
d = M.dati("XRP", "1h")
assert {"close", "datetime"} <= set(d.columns) and len(d) > 0
# ===========================================================================================
# (e) INTEGRAZIONE (opzionale: salta senza i dati o senza la cache del riferimento)
# ===========================================================================================
def test_integrazione_xrp_diluisce_anche_a_due_ancore(monkeypatch):
"""~2 s. Non riproduce i numeri a 24 ancore (2 ancore sono un'altra misura): riproduce la
DIREZIONE, che e' la cosa che decide — dSharpe hold-out mediano negativo e verdetto
"XRP DILUISCE", cioe' l'ipotesi a priori confermata."""
if not all((RAW / f"alt_xrp_{tf}.parquet").exists() for tf in ("1h", "5m")):
pytest.skip("namespace di ricerca alt_xrp_* assente (rigenerabile, gitignored)")
if not all((RAW / f"{a}_{tf}.parquet").exists() for a in ("btc", "eth") for tf in ("1h", "5m")):
pytest.skip("feed certificato BTC/ETH assente")
if not M.CACHE.exists():
pytest.skip(f"cache del riferimento assente ({M.CACHE}): il test NON va in rete")
# cintura: anche se la cache sparisse a meta' test, nessuno tocca la rete
def _vietato(*a, **k):
raise AssertionError("il test ha provato ad andare in rete (_fetch_1h)")
monkeypatch.setattr(M, "_fetch_1h", _vietato)
m = M.misura(2)
assert m["n_anchors"] == 2
assert m["cert"]["pulita_da"] == pd.Timestamp("2024-01-01", tz="UTC")
nome = ([n for n in m["lenti"] if n.startswith("L-PULITA")] or ["2024+ (lente di SOL)"])[0]
assert np.median(m["lenti"][nome]["hold"]) < 0, "il dSharpe hold-out ha cambiato segno"
assert M.verdetto(m).startswith("XRP DILUISCE")
assert m["corr"]["L-FULL"] < 0.74, "la correlazione col book supera la soglia del 19/06"