"""r0822b_depeg.py — IL FALSIFICATORE DI VENUE PER XS01-OOS (2026-08-22, ondata wave-0822). IL PUNTO FRAGILE, DICHIARATO DALL'AUTORE DEL RISULTATO ----------------------------------------------------- `r0822b_xs01_oos.py` (RESULTS-0822 §21) ha misurato che il meccanismo congelato di XS01 ha un edge FUORI dalla sua finestra di scoperta — 2021-2023 su **Binance spot USDT** — e che li' e' PIU' GRANDE (mediana di fase +1,12 contro +0,37, differenza appaiata +0,668 in 10/10 fasi). Quel risultato riscrive il gate del canale funded, quindi merita il suo falsificatore. Lo stesso autore lo ha nominato: > "La validazione di venue Binance-vs-Hyperliquid gira SOLO sulla finestra comune 2024+, cioe' > proprio il periodo SENZA depeg USDT. Il fuori campione contiene il depeg di maggio 2022. Se i > close Binance/USDT 2021-2023 divergessero da un venue USD in modo NON COMUNE fra gli asset, il > ranking cambierebbe e il risultato con esso. Test che lo deciderebbe: rifare il meccanismo sui > close Coinbase USD per il sottoinsieme quotato li', 2021-2023, e chiedere corr fra i due sleeve > >= 0.99 come sul 2024+." Questo script esegue esattamente quel test. PERCHE' LA DOMANDA NON E' OZIOSA (il meccanismo del danno, prima dei numeri) --------------------------------------------------------------------------- La regola del progetto — "Binance non e' la verita', e' USDT, ~10 bps fuori, e sotto depeg fino al 3% off" — e' stata **correttamente** giudicata non-binding da §21: uno z-score cross-sezionale sottrae la media di sezione, quindi e' invariante a un fattore **COMUNE**. Ma un depeg non e' garantito comune: nei giorni di stress la liquidita' si sposta in modo diverso per asset (i major restano scambiati contro USDT, i minori no; i flussi di fuga passano per coppie diverse), e una divergenza **IDIOSINCRATICA** cambia il RANKING — che e' esattamente cio' che la strategia legge. Maggio 2022 (UST/LUNA, con USDT sceso a ~0,95) e' il caso peggiore possibile. Percio' la divergenza va misurata in DUE pezzi, e sono due domande diverse: (a) COMUNE = mediana per giorno dello scarto fra i venue -> quella che lo z-score annulla; (b) IDIOSINCRATICA = scarto per asset DOPO aver tolto la mediana del giorno -> quella che uccide. Riportare solo (a) darebbe la risposta rassicurante e sbagliata. IL VINCOLO DEL DATO, E COME E' STATO RISOLTO (dichiarato PRIMA di misurare) -------------------------------------------------------------------------- Su Coinbase USD, dei 13 asset di U13: BNB -> quotato solo dal 2025-10 = FUORI (nessun venue USD ce l'ha nel 2021-23: e' l'asset nativo del venue che stiamo mettendo in discussione) XRP -> SOSPESO su Coinbase dal 2021-01-19 al 2023-07 (causa SEC) = buco proprio sulla finestra NEAR/INJ -> quotati su Coinbase solo dal 2022-09 AVAX -> dal 2021-09-30 (il vincolo che decide l'inizio della finestra) XRP e' recuperato da **Bitstamp USD** (l'altra referenza USD del `venue_watch` del progetto, che non ha mai sospeso la coppia). Restano fuori BNB, NEAR, INJ. Ne escono DUE test, e il secondo copre cio' che il primo non copre: TEST 1 "U10-USD" = universo FISSO di 10 gambe (BTC ETH LTC LINK AAVE ADA DOGE SOL AVAX + XRP), 2021-09-30 -> 2023-12-31, meccanismo **CONGELATO** (k=5, min_assets=10: dieci gambe sono ESATTAMENTE il minimo, nessun parametro toccato). Contiene il depeg di maggio 2022 e FTX. NON contiene il 2021. TEST 2 "U13-IBRIDO" = i 13 asset e la finestra INTERA 2021-01 -> 2023-12, con il close USD sostituito a quello USDT **dove esiste** (9-12 gambe su 13 a seconda della data) e Binance dove non esiste. Non e' un confronto puro di venue, e' la domanda operativa: *se sostituisco tutto il prezzo USD che esiste, il risultato di §21 si muove?* In entrambi i test il confronto e' **APPAIATO**: stesso sottoinsieme, stessa finestra, stesse 10 fasi, stesso codice del motore. Confrontare "19 gambe su Binance" con "10 su Coinbase" misurerebbe l'UNIVERSO, non il venue (regola del progetto: un confronto appaiato eredita TUTTI i parametri della riga in cui compare) — e per questo il TEST 1 riporta anche cosa fa U10 **su Binance** rispetto a U13 su Binance, cioe' quanto del titolo di §21 questo sottoinsieme riproduce. MECCANISMO: CONGELATO. Il motore e' importato da `r0822b_xs01_oos.py` (che a sua volta replica `sleeves._xsec_returns` a max|diff| = 0.0). Nessuna riga di strategia riscritta qui: se il motore fosse riscritto, una differenza di venue e una differenza di implementazione sarebbero indistinguibili. Uso: nice -n 19 timeout 900 uv run python scripts/research/r0822b_depeg.py Rete: sola lettura, <= 2 req/s, MAI nella finestra :25-:29 (rate limit per-IP condiviso col collettore della catena opzioni; guasto del 29/07). Download in cache nello scratchpad: la seconda corsa e' offline. """ from __future__ import annotations import sys import time from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts" / "research")) sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) import altlib as A # noqa: E402 import r0822b_xs01_oos as X # noqa: E402 (motore CONGELATO + cache Binance) from src.portfolio.sleeves import XS_CFG # noqa: E402 CACHE = X.CACHE.parent / "r0822b_depeg" CACHE.mkdir(parents=True, exist_ok=True) # --------------------------------------------------------------------------- # UNIVERSI E FINESTRE — dichiarati PRIMA di guardare i numeri # --------------------------------------------------------------------------- U13 = X.U13 # l'universo di §21 # venue USD per asset. `None` = nessun venue USD con storia sulla finestra. USD_SRC: dict[str, tuple[str, str] | None] = { "BTC": ("coinbase", "BTC/USD"), "ETH": ("coinbase", "ETH/USD"), "LTC": ("coinbase", "LTC/USD"), "LINK": ("coinbase", "LINK/USD"), "AAVE": ("coinbase", "AAVE/USD"), "ADA": ("coinbase", "ADA/USD"), "DOGE": ("coinbase", "DOGE/USD"), "SOL": ("coinbase", "SOL/USD"), "AVAX": ("coinbase", "AVAX/USD"), "NEAR": ("coinbase", "NEAR/USD"), "INJ": ("coinbase", "INJ/USD"), "XRP": ("bitstamp", "XRP/USD"), # Coinbase l'ha sospeso 2021-01 -> 2023-07 "BNB": None, # nessun venue USD prima del 2025-10 } U10 = ["BTC", "ETH", "LTC", "LINK", "AAVE", "ADA", "DOGE", "SOL", "AVAX", "XRP"] T1_START, T1_END = "2021-09-30", "2024-01-01" # AVAX su Coinbase e' il vincolo T2_START, T2_END = "2021-01-01", "2024-01-01" # la finestra fuori campione di §21 DEPEG = ("2022-05-07", "2022-05-21") # UST/LUNA + USDT a ~0.95 (estremo incluso) PHASES = list(range(XS_CFG["H"])) # 10 fasi: la stima onesta e' la MEDIANA CORR_SOGLIA = 0.99 # soglia DICHIARATA DALL'AUTORE, non da me # =========================================================================== # 1. FETCH USD (Coinbase / Bitstamp) + certificazione locale # =========================================================================== def _fetch_paged(ex, market: str, tf: str, start_ms: int, end_ms: int, step_bars: int, bar_ms: int) -> list: """Paginazione DETERMINISTICA sulla finestra: avanza anche quando una pagina torna vuota. ⚠️ Non e' un dettaglio: XRP su Coinbase ha un buco di 2,5 anni (sospensione SEC) e un loop che si ferma alla prima pagina vuota — il modo ovvio di scrivere questo ciclo — troncherebbe la serie li', producendo un 'asset che non esiste' invece di un 'asset con un buco'. Sono due fatti diversi e portano a due decisioni diverse. """ rows, cur, span = [], start_ms, step_bars * bar_ms while cur < end_ms: X._outside_forbidden_window() batch = [] for attempt in range(3): try: batch = ex.fetch_ohlcv(market, tf, since=cur, limit=step_bars) break except Exception as e: # noqa: BLE001 if attempt == 2: print(f" [rete] {market} @ {cur}: {type(e).__name__} {str(e)[:70]}") time.sleep(2 ** attempt) time.sleep(0.55) # <= 2 req/s rows += [b for b in (batch or []) if cur <= b[0] < cur + span] cur += span return rows def fetch_usd_daily(asset: str) -> pd.DataFrame: """Close giornalieri USD dal venue dichiarato in USD_SRC. Cache su disco.""" src = USD_SRC.get(asset) if src is None: return pd.DataFrame(columns=["timestamp", "close", "volume"]) venue, market = src f = CACHE / f"{venue[:2]}_{asset.lower()}_1d.parquet" if f.exists(): return pd.read_parquet(f) import ccxt ex = getattr(ccxt, venue)({"enableRateLimit": True}) ex.load_markets() if market not in ex.markets: d = pd.DataFrame(columns=["timestamp", "open", "high", "low", "close", "volume"]) d.to_parquet(f) return d start = ex.parse8601("2019-01-01T00:00:00Z") end = int(time.time() * 1000) step = 300 if venue == "coinbase" else 1000 rows = _fetch_paged(ex, market, "1d", start, end, step, 86_400_000) d = (pd.DataFrame(rows, columns=["timestamp", "open", "high", "low", "close", "volume"]) .drop_duplicates("timestamp").sort_values("timestamp").reset_index(drop=True)) d.to_parquet(f) return d def series_of(d: pd.DataFrame) -> pd.Series: if not len(d): # indice tz-aware anche da vuoto: un asset senza venue USD (BNB) deve poter essere # confrontato/reindicizzato come gli altri, non far cadere lo script return pd.Series(dtype=float, index=pd.DatetimeIndex([], tz="UTC")) return pd.Series(d["close"].values.astype(float), index=pd.to_datetime(d["timestamp"], unit="ms", utc=True)) def certify_usd(asset: str, d: pd.DataFrame, a: str, b: str) -> dict: """Stessa forma della certificazione di §21, ma valutata SULLA FINESTRA DEL TEST. Certificare "tutta la serie" direbbe la cosa sbagliata: XRP su Coinbase ha 900 giorni di buco che sulla finestra 2021-09/2023-12 sono TUTTI dentro. Il numero utile e' la copertura della finestra in cui il meccanismo gira, non della serie. """ s = series_of(d) s = s[(s.index >= pd.Timestamp(a, tz="UTC")) & (s.index < pd.Timestamp(b, tz="UTC"))] if not len(s): return dict(sym=asset, n=0, ok=False, start="-", end="-", gap=-1, flat=float("nan"), zero_vol=float("nan")) v = series_of(d.assign(close=d["volume"])) v = v[(v.index >= pd.Timestamp(a, tz="UTC")) & (v.index < pd.Timestamp(b, tz="UTC"))] attesi = (pd.Timestamp(b, tz="UTC") - s.index.min()).days c = s.values flat = float(np.mean(c[1:] == c[:-1])) if len(c) > 1 else float("nan") zv = float(np.mean(v.values == 0.0)) if len(v) else float("nan") gap = int(attesi - len(s)) return dict(sym=asset, n=len(s), start=str(s.index.min().date()), end=str(s.index.max().date()), gap=gap, flat=flat, zero_vol=zv, ok=bool(gap == 0 and flat < 0.02 and (not np.isfinite(zv) or zv < 0.01))) # =========================================================================== # 2. pannelli # =========================================================================== def bin_panel(syms: list[str]) -> pd.DataFrame: """Pannello Binance USDT — dalla cache di §21, cosi' il lato USDT del confronto e' BIT-IDENTICO a quello che ha prodotto il risultato che sto falsificando.""" return pd.concat({s: series_of(X.fetch_binance_daily(s)) for s in syms}, axis=1).sort_index() def usd_panel(syms: list[str]) -> pd.DataFrame: return pd.concat({s: series_of(fetch_usd_daily(s)) for s in syms}, axis=1).sort_index() def splice(usd: pd.Series, usdt: pd.Series) -> pd.Series: """Serie IBRIDA: USD dove esiste, USDT prima — RI-SCALANDO il tratto USDT al livello USD nel giorno del giunto. ⚠️ Senza la ri-scalatura il giunto crea un rendimento fantasma pari allo scarto fra i venue (pochi bps, ma esattamente il tipo di artefatto che questo script deve misurare, non fabbricare). Il meccanismo legge RAPPORTI di prezzo a 30 e 90 giorni: dopo la ri-scalatura la serie e' continua e i rapporti che attraversano il giunto sono ben definiti. """ u = usd.dropna() if not len(u): return usdt t0 = u.index[0] if t0 not in usdt.index or not np.isfinite(usdt.get(t0, np.nan)): return usd.reindex(usdt.index) k = float(u.iloc[0]) / float(usdt.loc[t0]) pre = usdt[usdt.index < t0] * k return pd.concat([pre, u]).sort_index().reindex(usdt.index) def hybrid_panel(syms: list[str]) -> tuple[pd.DataFrame, dict]: B, U = bin_panel(syms), usd_panel(syms) cols, meta = {}, {} for s in syms: u = U[s] if s in U.columns else pd.Series(dtype=float) u = u.reindex(B.index) cols[s] = splice(u, B[s]) cov = u.dropna() meta[s] = (str(cov.index[0].date()) if len(cov) else "mai") return pd.concat(cols, axis=1).sort_index(), meta # =========================================================================== # 3. metriche e allineamento appaiato # =========================================================================== sh, dd, cagr, win, alive, line = X.sh, X.dd, X.cagr, X.win, X.alive, X.line def pair(a: pd.Series, b: pd.Series) -> tuple[pd.Series, pd.Series]: """Allinea due sleeve sullo stesso indice: un confronto di venue non puo' confrontare finestre.""" j = pd.concat({"a": a, "b": b}, axis=1, join="inner").dropna() return j["a"], j["b"] def ranking_scan(PB: pd.DataFrame, PU: pd.DataFrame, phases: list[int], label: str) -> pd.DataFrame: """IL CANALE DEL DANNO, misurato dove passa: il venue puo' far male SOLO cambiando le gambe scelte. Scansione di TUTTE le fasi (non solo la canonica: un ribaltamento su una fase sola sarebbe invisibile all'ancora 0) con self-check dello specchio contro il motore.""" from scipy.stats import spearmanr lb = XS_CFG["lookbacks"] mlb, H, k = max(lb), XS_CFG["H"], XS_CFG["k"] rows, sp, checked, tot = [], [], 0, 0 for ph in phases: det: dict = {} X.xsec_engine(PB, phase=ph, detail=det) WB = det["W"] for i in range(mlb, len(PB)): if i % H != ph: continue iB, sB = score_at(PB, i, lb) iU, sU = score_at(PU, i, lb) if len(iB) < 10 or len(iU) < 10: continue tot += 1 oB = np.argsort(sB) loB, hiB = set(iB[oB[:k]]), set(iB[oB[-k:]]) mh, ml = set(np.flatnonzero(WB[i] > 0)), set(np.flatnonzero(WB[i] < 0)) if mh or ml: # self-check solo dove il motore ha una posizione checked += 1 if (mh, ml) != (hiB, loB): raise AssertionError(f"specchio != motore ({label}, fase {ph}, i={i}): " "misurerei la mia riscrittura invece del venue") oU = np.argsort(sU) loU, hiU = set(iU[oU[:k]]), set(iU[oU[-k:]]) ov = (len(hiB & hiU) + len(loB & loU)) / (2 * k) sp.append(float(spearmanr(sB, sU).statistic)) if ov < 1.0: diff = sorted(set(PB.columns[list(hiB ^ hiU)]) | set(PB.columns[list(loB ^ loU)])) rows.append(dict(fase=ph, data=PB.index[i], overlap=ov, gambe=",".join(diff))) D = pd.DataFrame(rows) print(f" [{label}] {tot} ribilanciamenti su {len(phases)} fasi; specchio verificato " f"contro il motore in {checked}/{checked} con posizione") print(f" [{label}] Spearman dello score fra i venue: mediana {np.median(sp):.4f} " f"min {np.min(sp):.4f}") print(f" [{label}] ribilanciamenti con gambe DIVERSE: {len(D)}/{tot} " f"({len(D)/max(tot,1)*100:.2f}%)") for _, r in D.iterrows(): print(f" fase {int(r['fase'])} {r['data'].date()} " f"sovrapposizione {r['overlap']*100:3.0f}% gambe scambiate: {r['gambe']}") return D def score_at(P: pd.DataFrame, i: int, lookbacks) -> tuple[np.ndarray, np.ndarray]: """Specchio della riga di score del motore (z-score medio sui lookback). Serve per confrontare i RANKING fra venue; la sezione [4] verifica che i top-k/bottom-k che ne derivano coincidano ESATTAMENTE con quelli scelti dal motore — se lo specchio divergesse, ogni numero di ranking misurerebbe la mia riscrittura invece del venue.""" px = P.values.astype(float) elig = np.isfinite(px[i]) for L in lookbacks: elig = elig & np.isfinite(px[i - L]) idx = np.flatnonzero(elig) sc, cnt = np.zeros(len(idx)), 0 for L in lookbacks: r = px[i, idx] / px[i - L, idx] - 1.0 if r.std() > 0: sc += (r - r.mean()) / r.std() cnt += 1 return idx, (sc / cnt if cnt else sc) # =========================================================================== def main() -> None: t0 = time.time() pd.set_option("display.width", 200) print("=" * 100) print("DEPEG — il close USDT 2021-2023 puo' aver cambiato il RANKING di XS01-OOS?") print(f"XS_CFG (CONGELATO) = {XS_CFG} soglia dichiarata dall'autore: corr >= {CORR_SOGLIA}") print("=" * 100) # ------------------------------------------------------------- [0] sanity print("\n[0] SANITY — il motore importato e' quello che ha prodotto §21?") from src.portfolio.sleeves import _xsec_returns off = _xsec_returns() rep = X.xsec_engine(X.hl_panel(), phase=0) j = pd.concat({"o": off, "r": rep}, axis=1, join="inner").dropna() md = float(np.max(np.abs(j["o"] - j["r"]))) print(f" motore vs sleeve UFFICIALE su HL: n={len(j)} max|diff| = {md:.2e} " f"-> {'IDENTICO' if md < 1e-12 else 'DIVERGE — STOP'}") if md >= 1e-12: return # ------------------------------------------------------- [1] dato USD print("\n[1] DATO USD — quanti dei 13 asset di §21 hanno un prezzo in DOLLARI nel 2021-2023?") print(f" {'sym':<6}{'venue USD':>11}{'n':>6}{'start':>12}{'end':>12}{'gap':>5}" f"{'flat%':>8}{'vol0%':>8} (finestra {T1_START} .. {T1_END})") certs = {} for s in U13: d = fetch_usd_daily(s) c = certify_usd(s, d, T1_START, T1_END) certs[s] = c ven = USD_SRC[s][0] if USD_SRC[s] else "—" flag = "" if c["ok"] else " <-- non usabile su questa finestra" print(f" {s:<6}{ven:>11}{c['n']:>6}{c['start']:>12}{c['end']:>12}{c['gap']:>5}" f"{c['flat']*100:>8.2f}{c['zero_vol']*100:>8.2f}{flag}") usabili = [s for s in U13 if certs[s]["ok"]] print(f" -> usabili sulla finestra del TEST 1: {len(usabili)}/13 {usabili}") print(f" -> ESCLUSI: BNB (nessun venue USD prima del 2025-10 — e' l'asset nativo del venue") print(f" sotto esame), NEAR e INJ (Coinbase solo dal 2022-09).") print(f" ⚠️ LIMITE DEL TEST, da dire subito: 10 gambe su 13 e finestra che parte dal") print(f" {T1_START} (AVAX su Coinbase) -> il TEST 1 NON copre il 2021, che in §21 e'") print(f" l'anno piu' forte (+2,14). Copre pero' ESATTAMENTE cio' che l'obiezione nomina:") print(f" il depeg di maggio 2022 e FTX. Il 2021 lo copre il TEST 2 (ibrido).") # controllo di ALLINEAMENTO: le barre USD sono sulla stessa griglia di quelle USDT? print("\n Controllo di ALLINEAMENTO — se le barre USD e USDT fossero su griglie sfasate,") print(" ogni numero sotto sarebbe uno sfasamento di calendario travestito da divergenza") print(" di venue. corr dei RITORNI giornalieri, 2024+ (finestra dove esistono entrambi):") B24 = bin_panel(U13) U24 = usd_panel(U13) al, shifted = [], [] for s in U13: if s not in U24.columns or not U24[s].notna().any(): continue k = pd.concat({"u": U24[s], "b": B24[s]}, axis=1, join="inner").dropna() k = k[k.index >= pd.Timestamp("2024-01-01", tz="UTC")].pct_change().dropna() if len(k) > 200: al.append((s, float(k["u"].corr(k["b"])))) shifted.append(float(k["u"].shift(1).corr(k["b"]))) # controllo POSITIVO print(" " + " ".join(f"{s}:{c:.4f}" for s, c in al)) lo_al = min(c for _, c in al) # ⚠️ Un controllo che non fallisce mai non e' un controllo: il modo di fallire e' uno # sfasamento di UNA barra, e va mostrato che il numero se ne accorge. Con la serie USD # ritardata di un giorno la corr collassa: e' quella la scala del guasto che sto escludendo. print(f" corr minima {lo_al:.4f} | controllo POSITIVO (USD ritardato di 1 giorno): " f"corr mediana {np.median(shifted):+.4f}, massima {np.max(shifted):+.4f}") if lo_al < 0.95: print(" -> SFASAMENTO: i due venue non sono sulla stessa griglia. STOP.") return print(" -> allineati (uno sfasamento darebbe ~0, non 0,99). Gli asset sotto 0,999") print(f" ({', '.join(s for s, c in al if c < 0.999)}) sono solo piu' RUMOROSI in USD") print(" (book piu' sottile sul venue USD), non sfasati.") # --------------------------------------------- [2] la divergenza, in DUE pezzi print("\n[2] LA DIVERGENZA USD-vs-USDT, scomposta — e il pezzo che conta e' il SECONDO") Bp = bin_panel(U10).loc[T1_START:T1_END] Up = usd_panel(U10).reindex(Bp.index) both = Bp.notna() & Up.notna() D = 1e4 * np.log(Up / Bp) # scarto di livello, bps D = D.where(both) common = D.median(axis=1) # (a) componente COMUNE (il peg + il venue) idio = D.sub(common, axis=0) # (b) componente IDIOSINCRATICA (cambia il rango) print(f" {len(Bp)} giorni, {len(U10)} asset, {int(both.values.sum())} osservazioni appaiate") print(f" (a) COMUNE |mediana per giorno|: med {common.abs().median():7.1f} bps " f"p95 {common.abs().quantile(.95):7.1f} max {common.abs().max():7.1f}") print(f" (b) IDIOSINCRATICO |scarto de-medianato|: med {idio.abs().stack().median():7.1f} bps " f"p95 {idio.abs().stack().quantile(.95):7.1f} max {idio.abs().stack().max():7.1f}") print(" -> (a) e' cio' che lo z-score cross-sezionale annulla per costruzione; (b) no.") print(f"\n LA FINESTRA DEL DEPEG {DEPEG[0]} .. {DEPEG[1]} (UST/LUNA; USDT a ~0,95):") dm = (D.index >= pd.Timestamp(DEPEG[0], tz="UTC")) & (D.index < pd.Timestamp(DEPEG[1], tz="UTC")) print(f" {'giorno':<12}{'comune':>9}" + "".join(f"{s:>8}" for s in U10)) for t in D.index[dm]: print(f" {str(t.date()):<12}{common.loc[t]:>9.0f}" + "".join(f"{idio.loc[t, s]:>8.0f}" if np.isfinite(idio.loc[t, s]) else f"{'—':>8}" for s in U10)) print(" (numeri in bps; colonna 'comune' = mediana del giorno, le altre = scarto DOPO") print(" averla tolta, cioe' la parte che puo' spostare il ranking)") idm = idio[dm].abs().stack() print(f" idiosincratico NELLA finestra: med {idm.median():.1f} bps p95 " f"{idm.quantile(.95):.1f} max {idm.max():.1f} " f"(FUORI: med {idio[~dm].abs().stack().median():.1f}, " f"p95 {idio[~dm].abs().stack().quantile(.95):.1f})") print("\n Ma il ranking non legge il LIVELLO: legge i rapporti a 30 e 90 giorni. Divergenza") print(" idiosincratica degli STESSI rendimenti che entrano nello score (bps):") for L in XS_CFG["lookbacks"]: rB = (Bp / Bp.shift(L) - 1.0) * 1e4 rU = (Up / Up.shift(L) - 1.0) * 1e4 dR = (rU - rB).where(both) dRi = dR.sub(dR.median(axis=1), axis=0) w = dRi[dm].abs().stack() print(f" r{L:<3d}: idiosincratico med {dRi.abs().stack().median():6.1f} bps " f"p95 {dRi.abs().stack().quantile(.95):7.1f} max {dRi.abs().stack().max():8.1f}" f" | nel depeg med {w.median():6.1f} max {w.max():8.1f}") # ------------------------------------------- [3] IL TEST 1 — U10, k=5 congelato print(f"\n[3] TEST 1 — meccanismo CONGELATO su U10, {T1_START} .. {T1_END}, USD vs USDT") print(f" 10 gambe = ESATTAMENTE il minimo del meccanismo (min_assets=10, k=5): nessun") print(f" parametro toccato, e il confronto e' appaiato su TUTTO tranne il venue.") PB = Bp[U10].dropna() PU = Up[U10].dropna() idxc = PB.index.intersection(PU.index) PB, PU = PB.loc[idxc], PU.loc[idxc] print(f" pannello comune: {len(idxc)} giorni, {PB.index[0].date()} .. {PB.index[-1].date()}") rows = [] for ph in PHASES: a = alive(X.xsec_engine(PU, phase=ph)) b = alive(X.xsec_engine(PB, phase=ph)) a, b = pair(a, b) rows.append(dict(ph=ph, corr=float(a.corr(b)), sh_usd=sh(a), sh_usdt=sh(b), d=sh(a) - sh(b), dd_usd=dd(a), dd_usdt=dd(b), n=len(a))) R = pd.DataFrame(rows) print(f" {'fase':>5}{'corr':>9}{'Sh USD':>9}{'Sh USDT':>9}{'dSharpe':>10}" f"{'DD USD':>9}{'DD USDT':>9}") for _, r in R.iterrows(): print(f" {int(r['ph']):>5}{r['corr']:>9.4f}{r['sh_usd']:>9.2f}{r['sh_usdt']:>9.2f}" f"{r['d']:>+10.3f}{r['dd_usd']*100:>8.1f}%{r['dd_usdt']*100:>8.1f}%") print(f" {'MEDIANA':>5}{R['corr'].median():>9.4f}{R['sh_usd'].median():>9.2f}" f"{R['sh_usdt'].median():>9.2f}{R['d'].median():>+10.3f}" f"{R['dd_usd'].median()*100:>8.1f}%{R['dd_usdt'].median()*100:>8.1f}%") print(f" corr MINIMA fra le 10 fasi: {R['corr'].min():.4f} (fase {int(R.loc[R['corr'].idxmin(),'ph'])})" f" | soglia dichiarata {CORR_SOGLIA} -> " f"{'PASS' if R['corr'].min() >= CORR_SOGLIA else 'FAIL'}") print(f" dSharpe appaiato (USD - USDT): mediana {R['d'].median():+.3f} " f"banda [{R['d'].min():+.3f},{R['d'].max():+.3f}] " f"fasi in cui USD e' MEGLIO {(R['d'] > 0).sum()}/{len(R)}") print(f" ⚠️ la statistica e' la MEDIANA DELLE DIFFERENZE APPAIATE, non la differenza delle") print(f" mediane (lezione 26/07: quest'ultima confronta fasi diverse e puo' ribaltare).") # quanto di §21 riproduce questo sottoinsieme? (o si misura l'universo, non il venue) p13 = bin_panel(U13)[U13].dropna() s13_w = win(alive(X.xsec_engine(p13, phase=0)), T1_START, T1_END) s10_w = win(alive(X.xsec_engine(PB, phase=0)), T1_START, T1_END) print(f"\n Contesto obbligatorio — quanto di §21 vive in questo sottoinsieme (Binance, fase 0,") print(f" stessa finestra): U13 Sh {sh(s13_w):+.2f} | U10 Sh {sh(s10_w):+.2f}") print(f" -> il TEST 1 giudica il VENUE su U10; NON e' una replica del titolo di §21.") # ------------------------------------------------ [4] i RANKING, non i P&L print("\n[4] IL CANALE DEL DANNO — cambia il RANKING? (e' cio' che la strategia legge)") print(" Un P&L simile puo' nascere da due ranking diversi che si compensano: la domanda") print(" diretta e' se le GAMBE SCELTE cambiano. Scansione su tutte e 10 le fasi.") D1 = ranking_scan(PB, PU, PHASES, "TEST 1 / U10") if len(D1): ind = D1[(D1["data"] >= pd.Timestamp(DEPEG[0], tz="UTC")) & (D1["data"] < pd.Timestamp(DEPEG[1], tz="UTC"))] print(f" di cui dentro la finestra del depeg: {len(ind)}") # ------------------------------- [5] TEST 2 — U13 ibrido, finestra INTERA di §21 print(f"\n[5] TEST 2 — U13 IBRIDO (USD dove esiste, USDT altrove), {T2_START} .. {T2_END}") print(" Domanda operativa: sostituendo TUTTO il prezzo in dollari che esiste, il titolo") print(" di §21 (mediana di fase +1,12 fuori campione) si muove?") HB, meta = hybrid_panel(U13) print(" prima data USD per asset: " + " ".join(f"{s}:{meta[s]}" for s in U13)) PBh = p13 PUh = HB[U13].dropna() ic = PBh.index.intersection(PUh.index) PBh, PUh = PBh.loc[ic], PUh.loc[ic] rows2 = [] for ph in PHASES: a = win(alive(X.xsec_engine(PUh, phase=ph)), T2_START, T2_END) b = win(alive(X.xsec_engine(PBh, phase=ph)), T2_START, T2_END) a, b = pair(a, b) rows2.append(dict(ph=ph, corr=float(a.corr(b)), sh_usd=sh(a), sh_usdt=sh(b), d=sh(a) - sh(b), dd_usd=dd(a), dd_usdt=dd(b))) R2 = pd.DataFrame(rows2) print(f" {'fase':>5}{'corr':>9}{'Sh IBRIDO':>11}{'Sh USDT':>10}{'dSharpe':>10}" f"{'DD IBR':>9}{'DD USDT':>9}") for _, r in R2.iterrows(): print(f" {int(r['ph']):>5}{r['corr']:>9.4f}{r['sh_usd']:>11.2f}{r['sh_usdt']:>10.2f}" f"{r['d']:>+10.3f}{r['dd_usd']*100:>8.1f}%{r['dd_usdt']*100:>8.1f}%") print(f" {'MEDIANA':>5}{R2['corr'].median():>9.4f}{R2['sh_usd'].median():>11.2f}" f"{R2['sh_usdt'].median():>10.2f}{R2['d'].median():>+10.3f}" f"{R2['dd_usd'].median()*100:>8.1f}%{R2['dd_usdt'].median()*100:>8.1f}%") print(f" corr minima {R2['corr'].min():.4f} -> " f"{'PASS' if R2['corr'].min() >= CORR_SOGLIA else 'FAIL'} (soglia {CORR_SOGLIA})") print(f" 📌 IL NUMERO DI TESTA DI §21, RICALCOLATO IN DOLLARI: mediana di fase fuori") print(f" campione {R2['sh_usdt'].median():+.2f} (USDT, replica) -> " f"{R2['sh_usd'].median():+.2f} (IBRIDO USD).") # differenza appaiata con la macchineria del progetto (stesso gate di §21) dlt = A.anchor_luck_delta( lambda ph: win(alive(X.xsec_engine(PUh, phase=ph)), T2_START, T2_END), lambda ph: win(alive(X.xsec_engine(PBh, phase=ph)), T2_START, T2_END), PHASES, metric=sh) print(f" anchor_luck_delta (IBRIDO - USDT): mediana appaiata {dlt['median_paired']:+.3f} " f"[{dlt['lo']:+.3f},{dlt['hi']:+.3f}], fasi con IBRIDO>USDT " f"{dlt['n_positive']}/{dlt['n_anchors']}") print("\n E il ranking, sull'ibrido (la fase con la corr piu' bassa va GUARDATA, non") print(" riassunta: una corr di 0,97 su una fase o e' un ribaltamento vero o e' un giunto):") D2 = ranking_scan(PBh, PUh, PHASES, "TEST 2 / U13-IBRIDO") # ------------------------------- [5b] le gambe che NESSUN venue USD puo' controllare print("\n[5b] LE TRE GAMBE NON CONTROLLABILI — quanto del risultato di §21 ci passa?") print(" BNB e' il token nativo del venue sotto esame e NON esiste in USD nel 2021-23;") print(" NEAR e INJ hanno un prezzo USD solo dal 2022-09. Se il titolo di §21 vivesse su") print(" di loro, questo falsificatore non avrebbe potuto vedere niente, e andrebbe detto.") base_b = A.anchor_luck_band( lambda ph: win(alive(X.xsec_engine(PBh, phase=ph)), T2_START, T2_END), PHASES, canonical=0, metric=sh) print(f" base U13 (Binance, fuori campione, mediana di fase): {base_b['median']:+.2f}") for a_ in ("BNB", "NEAR", "INJ"): sub = p13[[c for c in U13 if c != a_]].dropna() bb = A.anchor_luck_band( lambda ph, sub=sub: win(alive(X.xsec_engine(sub, phase=ph, min_assets=10)), T2_START, T2_END), PHASES, canonical=0, metric=sh) print(f" senza {a_:<5s}: mediana di fase {bb['median']:+.2f} " f"({bb['median']-base_b['median']:+.2f}) banda [{bb['lo']:+.2f},{bb['hi']:+.2f}]") print(" (leave-one-out a 12 gambe: il meccanismo congelato regge, min_assets=10)") # ---------------------------------------------------- [6] IL CASO PEGGIORE print("\n[6] IL CASO PEGGIORE — non la media: dove i due venue divergono di piu'?") print(f" {'anno':>6}" + "".join(f"{k:>12}" for k in ("corr T1", "dSh T1", "corr T2", "dSh T2"))) for y in (2021, 2022, 2023): a1 = b1 = a2 = b2 = None cs1, ds1, cs2, ds2 = [], [], [], [] for ph in PHASES: a1 = win(alive(X.xsec_engine(PU, phase=ph)), f"{y}-01-01", f"{y+1}-01-01") b1 = win(alive(X.xsec_engine(PB, phase=ph)), f"{y}-01-01", f"{y+1}-01-01") if len(a1) > 60 and len(b1) > 60: a1, b1 = pair(a1, b1) cs1.append(float(a1.corr(b1))); ds1.append(sh(a1) - sh(b1)) a2 = win(alive(X.xsec_engine(PUh, phase=ph)), f"{y}-01-01", f"{y+1}-01-01") b2 = win(alive(X.xsec_engine(PBh, phase=ph)), f"{y}-01-01", f"{y+1}-01-01") if len(a2) > 60 and len(b2) > 60: a2, b2 = pair(a2, b2) cs2.append(float(a2.corr(b2))); ds2.append(sh(a2) - sh(b2)) f1 = f"{np.median(cs1):.4f}" if cs1 else "—" g1 = f"{np.median(ds1):+.3f}" if ds1 else "—" f2 = f"{np.median(cs2):.4f}" if cs2 else "—" g2 = f"{np.median(ds2):+.3f}" if ds2 else "—" print(f" {y:>6}{f1:>12}{g1:>12}{f2:>12}{g2:>12}") worst1 = R.loc[R["corr"].idxmin()] worst2 = R2.loc[R2["corr"].idxmin()] wd1 = R.loc[R["d"].abs().idxmax()] wd2 = R2.loc[R2["d"].abs().idxmax()] print(f" peggior FASE per corr: T1 fase {int(worst1['ph'])} corr {worst1['corr']:.4f} | " f"T2 fase {int(worst2['ph'])} corr {worst2['corr']:.4f}") print(f" peggior FASE per dSharpe: T1 fase {int(wd1['ph'])} {wd1['d']:+.3f} | " f"T2 fase {int(wd2['ph'])} {wd2['d']:+.3f}") st = idio.abs().stack().sort_values(ascending=False) print(" le 8 divergenze IDIOSINCRATICHE piu' grandi della finestra (bps di livello):") for (t, s), v in st.head(8).items(): indep = "DEPEG" if pd.Timestamp(DEPEG[0], tz="UTC") <= t < pd.Timestamp(DEPEG[1], tz="UTC") else "" print(f" {str(t.date()):<12}{s:<6}{v:8.0f} bps (comune quel giorno " f"{common.loc[t]:+.0f} bps) {indep}") print(" -> se le peggiori NON cadono nel depeg, il rumore di venue e' book sottile,") print(" non contaminazione da stablecoin: due cause diverse, due implicazioni diverse.") # il giorno peggiore: dove il P&L dei due sleeve si stacca di piu' a0 = alive(X.xsec_engine(PU, phase=0)); b0 = alive(X.xsec_engine(PB, phase=0)) a0, b0 = pair(a0, b0) gap = (a0 - b0) print(f" scarto GIORNALIERO fra i due sleeve (fase 0): med " f"{gap.abs().median()*1e4:.1f} bps p99 {gap.abs().quantile(.99)*1e4:.1f} bps " f"max {gap.abs().max()*1e4:.1f} bps il {gap.abs().idxmax().date()}") cum = (1 + a0).cumprod().iloc[-1] / (1 + b0).cumprod().iloc[-1] - 1 print(f" divario CUMULATO su tutta la finestra del TEST 1: {cum*100:+.2f}% " f"(equity USD vs equity USDT)") # --------------------------------------------------------------- [7] gate print("\n[7] GATE") print(f" soglia di venue (dichiarata dall'autore PRIMA, non da me): corr >= {CORR_SOGLIA}") print(f" TEST 1 U10 congelato : corr mediana {R['corr'].median():.4f}, minima " f"{R['corr'].min():.4f} -> {'PASS' if R['corr'].min() >= CORR_SOGLIA else 'FAIL'}") print(f" TEST 2 U13 ibrido : corr mediana {R2['corr'].median():.4f}, minima " f"{R2['corr'].min():.4f} -> {'PASS' if R2['corr'].min() >= CORR_SOGLIA else 'FAIL'}") print(f" riferimento 2024+ di §21 (Binance vs Hyperliquid): 0,9991 / dSharpe -0,025") su = alive(X.xsec_engine(PUh, phase=0)) mg = A.marginal_vs_tp01(win(su, T2_START, T2_END)) print(f" marginal_vs_tp01 sullo sleeve IBRIDO fuori campione: {mg.get('marginal_verdict')} " f"(corr {mg.get('corr_full')}, is_hedge {mg.get('is_hedge')}, " f"insample_edge {mg.get('has_insample_edge')})") imp = A.implausible_sharpe(win(su, T2_START, T2_END)) print(f" implausible_sharpe (ibrido OOS): {imp['implausible']} Sh {imp.get('sharpe'):+.2f} " f"perdite su barre attive {imp.get('loss_frac', float('nan'))*100:.1f}%") print(" causalita': NON RI-GIRATA — il motore e' importato invariato da r0822b_xs01_oos.py,") print(" che la verifica (prezzi futuri randomizzati -> max|diff| = 0). Cambiare il venue") print(" cambia i dati, non l'ordine temporale in cui il motore li legge.") print(" deflated_sharpe: NON PERTINENTE — qui non si SELEZIONA una cella, si confronta due") print(" volte lo STESSO meccanismo su due fonti di prezzo. Nessun trial nuovo di strategia.") print(" null de-levering: NON PERTINENTE (nessun claim di 'meno drawdown').") print(" anchor: girato per costruzione — ogni numero sopra e' su tutte e 10 le fasi.") print(f"\n[fine] {time.time()-t0:.1f}s") if __name__ == "__main__": main()