PythagorasGoal/scripts/analysis/certify_feed.py

"""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<low={hl}")
    ho = int((h < np.maximum(o, c)).sum())
    if ho: issues.append(f"high<max(o,c)={ho}")
    lo = int((l > 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()