"""CERTIFICAZIONE STORICO — dimostrare che data/raw (rebuild Deribit mainnet) e' PULITO. Dopo che il feed contaminato (print fantasma testnet) ha invalidato l'intera libreria, NESSUNA nuova ricerca vale finche' lo storico non e' CERTIFICATO. Ricostruire non basta: va dimostrato pulito, su TUTTA la storia (non solo il recente) e su tutte le dimensioni. Quattro blocchi, read-only (nessuna scrittura): (1) INTEGRITA' STRUTTURALE (locale, tutti asset/TF): invarianti OHLC (high>=low, high>=max(o,c), low<=min(o,c)), prezzi>0, volume>=0, NaN, ts monotono/dup, GAP (barre mancanti vs griglia attesa), run di barre FLAT (O=H=L=C). (2) COERENZA RESAMPLE (locale): ricalcolo 15m/1h dal 5m e confronto col salvato -> la "base unica + resample" e' davvero internamente coerente (errore ~0). (3) SCAN SPIKE / FAKE-WICK (locale): firma della contaminazione (high/low che sfora >THR il cluster di close vicino). Sul feed pulito deve essere ~0. (4) ACCORDO CROSS-VENUE (rete): close 1h vs COINBASE USD (venue indipendente, USD non USDT), STORIA INTERA per-anno: bps med/p95/max + %barre >50bps/>1%/>3%. E' il test di VERITA' del prezzo, specie sui vecchi anni dove viveva la contaminazione. uv run python scripts/analysis/certify_feed.py # tutto (locale + rete) uv run python scripts/analysis/certify_feed.py --local # solo blocchi 1-3 (veloce) uv run python scripts/analysis/certify_feed.py --asset BTC ETH """ from __future__ import annotations import sys import time from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) import numpy as np import pandas as pd import ccxt RAW = PROJECT_ROOT / "data" / "raw" ASSETS = ["BTC", "ETH", "SOL", "ADA", "XRP", "LTC", "DOGE", "BNB"] TFS = ["5m", "15m", "1h"] TF_MIN = {"5m": 5, "15m": 15, "1h": 60} SPIKE_THR = 0.10 # high/low oltre +-10% dal cluster di close vicino = sospetto # Coinbase USD: venue indipendente da Deribit, denominato USD (non USDT). BNB non listato. COINBASE_SYM = {"BTC": "BTC/USD", "ETH": "ETH/USD", "SOL": "SOL/USD", "ADA": "ADA/USD", "XRP": "XRP/USD", "LTC": "LTC/USD", "DOGE": "DOGE/USD", "BNB": None} def load(asset, tf): p = RAW / f"{asset.lower()}_{tf}.parquet" if not p.exists(): return None return pd.read_parquet(p) # ----------------------------- (1) INTEGRITA' ----------------------------- def integrity(asset, tf, df): o, h, l, c = (df[x].values for x in ("open", "high", "low", "close")) v = df["volume"].values ts = df["timestamp"].values issues = [] nan = int(df[["open", "high", "low", "close", "volume"]].isna().sum().sum()) if nan: issues.append(f"NaN={nan}") nonpos = int((np.minimum.reduce([o, h, l, c]) <= 0).sum()) if nonpos: issues.append(f"prezzo<=0={nonpos}") if (v < 0).sum(): issues.append(f"vol<0={int((v<0).sum())}") hl = int((h < l).sum()) if hl: issues.append(f"high np.minimum(o, c)).sum()) if lo: issues.append(f"low>min(o,c)={lo}") dts = np.diff(ts) dup = int((dts == 0).sum()) if dup: issues.append(f"ts-dup={dup}") if (dts < 0).sum(): issues.append(f"ts-non-monotono={int((dts<0).sum())}") step = TF_MIN[tf] * 60_000 gaps = dts[dts > step] miss = int((gaps // step - 1).sum()) if len(gaps) else 0 flat = int(((o == h) & (h == l) & (l == c)).sum()) # run flat piu' lunga fmask = (o == h) & (h == l) & (l == c) longest = cur = 0 for f in fmask: cur = cur + 1 if f else 0 longest = max(longest, cur) status = "OK" if not issues else "**" + " ".join(issues) return dict(n=len(df), gaps=len(gaps), miss=miss, flat=flat, flatpct=flat/len(df)*100, flatrun=longest, status=status) # ----------------------------- (2) RESAMPLE ----------------------------- def resample(base, tf): g = base.copy() g.index = pd.to_datetime(g["timestamp"], unit="ms", utc=True) rule = {"15m": "15min", "1h": "1h"}[tf] out = g.resample(rule, label="left", closed="left").agg( {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) out = out.dropna(subset=["open"]) epoch = pd.Timestamp("1970-01-01", tz="UTC") out.insert(0, "timestamp", ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")) return out.reset_index(drop=True) def resample_coherence(asset): base = load(asset, "5m") if base is None: return "no 5m" res = [] for tf in ("15m", "1h"): stored = load(asset, tf) if stored is None: res.append(f"{tf}:no-file"); continue rb = resample(base, tf) m = stored.merge(rb, on="timestamp", suffixes=("_s", "_r"), how="inner") if len(m) == 0: res.append(f"{tf}:0-overlap"); continue maxerr = 0.0 for col in ("open", "high", "low", "close"): d = (m[f"{col}_s"] - m[f"{col}_r"]).abs() / m[f"{col}_r"].replace(0, np.nan) maxerr = max(maxerr, float(d.max(skipna=True)) if len(d) else 0.0) nmiss = abs(len(stored) - len(rb)) res.append(f"{tf}: maxΔ {maxerr*1e4:.2f}bps Δrows {nmiss}") return " | ".join(res) # ----------------------------- (3) SPIKE ----------------------------- def spike_scan(df, thr=SPIKE_THR): c = df["close"].values.astype(float) h = df["high"].values.astype(float) l = df["low"].values.astype(float) n = len(c) s = pd.Series(c) # cluster di close vicino (mediana finestra 7 centrata, causale-agnostica = audit) cluster = s.rolling(7, center=True, min_periods=3).median().values hi = h > cluster * (1 + thr) lo = l < cluster * (1 - thr) susp = hi | lo idx = np.where(susp)[0] worst = [] if len(idx): dev = np.where(hi, h / cluster - 1, np.where(lo, 1 - l / cluster, 0.0)) order = idx[np.argsort(-np.abs(dev[idx]))][:3] ts = pd.to_datetime(df["timestamp"].values, unit="ms", utc=True) for i in order: worst.append(f"{ts[i].date()} {dev[i]*100:+.0f}%") return int(susp.sum()), worst # ----------------------------- (4) CROSS-VENUE ----------------------------- def fetch_coinbase_1h(sym, start_ms, end_ms): ex = ccxt.coinbase({"enableRateLimit": True}) if not ex.markets: ex.load_markets() if sym not in ex.markets: return None, f"simbolo {sym} assente su Coinbase" out, since, guard = {}, start_ms, 0 while since <= end_ms and guard < 4000: guard += 1 for attempt in range(3): try: r = ex.fetch_ohlcv(sym, "1h", since=since, limit=300) break except Exception as e: if attempt == 2: return (pd.Series(out) if out else None), f"err {type(e).__name__}" time.sleep(2 ** attempt) r = [x for x in r if int(x[0]) >= since] if not r: break for x in r: if start_ms <= int(x[0]) <= end_ms and x[4]: out[int(x[0])] = float(x[4]) nxt = int(r[-1][0]) + 3600_000 if nxt <= since: break since = nxt return (pd.Series(out) if out else None), f"{len(out)} barre" def cross_venue(asset): sym = COINBASE_SYM[asset] if sym is None: return None, "no-coinbase-ref (BNB non listato)" df1h = load(asset, "1h") if df1h is None: return None, "no 1h locale" ts = pd.to_datetime(df1h["timestamp"], unit="ms", utc=True) s_ms = int(df1h["timestamp"].iloc[0]) e_ms = int(df1h["timestamp"].iloc[-1]) ref, note = fetch_coinbase_1h(sym, s_ms, e_ms) if ref is None: return None, note a = df1h.set_index("timestamp")["close"] m = pd.concat([a.rename("a"), ref.rename("b")], axis=1, join="inner").dropna() if len(m) == 0: return None, "0 overlap" m["bps"] = (m["a"] - m["b"]).abs() / m["b"] * 1e4 m["year"] = pd.to_datetime(m.index, unit="ms", utc=True).year rows = [] for y, g in m.groupby("year"): rows.append((y, len(g), g["bps"].median(), g["bps"].quantile(.95), g["bps"].max(), (g["bps"] > 50).mean()*100, (g["bps"] > 100).mean()*100, (g["bps"] > 300).mean()*100)) return rows, f"overlap {len(m)} barre vs Coinbase {sym}" def main(): argv = sys.argv[1:] local_only = "--local" in argv assets = ASSETS if "--asset" in argv: i = argv.index("--asset") assets = [a.upper() for a in argv[i+1:] if not a.startswith("--")] print("=" * 100) print(" (1) INTEGRITA' STRUTTURALE — per asset/TF") print("=" * 100) print(f" {'file':<11s}{'barre':>9s}{'gap':>6s}{'mancanti':>9s}{'flat':>8s}{'flat%':>7s}{'flatRun':>8s} status") print(" " + "-" * 96) for a in assets: for tf in TFS: df = load(a, tf) if df is None: print(f" {a.lower()+'_'+tf:<11s} (assente)"); continue r = integrity(a, tf, df) print(f" {a.lower()+'_'+tf:<11s}{r['n']:>9d}{r['gaps']:>6d}{r['miss']:>9d}{r['flat']:>8d}" f"{r['flatpct']:>6.1f}%{r['flatrun']:>8d} {r['status']}") print("\n" + "=" * 100) print(" (2) COERENZA RESAMPLE (5m -> 15m/1h ricalcolato == salvato)") print("=" * 100) for a in assets: print(f" {a:<5s} {resample_coherence(a)}") print("\n" + "=" * 100) print(f" (3) SCAN SPIKE / FAKE-WICK (high/low oltre +-{SPIKE_THR*100:.0f}% dal cluster close vicino)") print("=" * 100) for a in assets: for tf in TFS: df = load(a, tf) if df is None: continue n, worst = spike_scan(df) tag = "OK" if n == 0 else f"** {n} sospetti — {', '.join(worst)}" print(f" {a.lower()+'_'+tf:<11s} {tag}") if local_only: print("\n (--local: salto il blocco rete cross-venue)") return print("\n" + "=" * 100) print(" (4) ACCORDO CROSS-VENUE — close 1h vs COINBASE USD, per anno (bps; 1bps=0.01%)") print("=" * 100) for a in assets: rows, note = cross_venue(a) print(f"\n {a} ({note})") if rows is None: continue print(f" {'anno':>5s}{'barre':>7s}{'med':>7s}{'p95':>7s}{'max':>8s}{'>50bp%':>8s}{'>1%':>7s}{'>3%':>7s}") for y, n, med, p95, mx, g50, g100, g300 in rows: print(f" {y:>5d}{n:>7d}{med:>7.1f}{p95:>7.1f}{mx:>8.1f}{g50:>8.1f}{g100:>7.1f}{g300:>7.1f}") if __name__ == "__main__": main()