PythagorasGoal/Old/scripts/analysis/clean_feed.py

"""CLEAN FEED — ripara gli spike-print del feed Deribit/Cerbero coi dati reali di Binance.

Motivo (2026-06-18): la ricerca Price Ladder ha rivelato che data/raw/btc_1h.parquet (e gli
altri TF/asset) contengono barre con WICK FASULLI (es. BTC 2024-02-13: low 38.580 con
close ~49.968, BTC reale ~50k) — lo stesso spike-print testnet documentato in CLAUDE.md
(TP_PHANTOM / feed congelato). Sono pochi (decine per file) ma avvelenano i backtest
(stop/entry su prezzi mai avvenuti) e gonfiano le code (la "FULL DD BTC ~54%" del ladder era
in gran parte questo).

Metodo (conservativo, fonte di verita' = Binance spot via ccxt, gia' cablato nel progetto):
  1. DETECT: barra sospetta = high/low che sfora >15% il cluster di close locale [i-1,i,i+1]
     (close sano + wick fasullo). Soglia larga: tanto e' Binance ad arbitrare.
  2. ARBITRA: per ogni sospetta, scarica la barra Binance reale (BTC/USDT, ETH/USDT) allo
     stesso tf/timestamp. Sostituisce O/H/L/C SOLO se Binance dissente materialmente (>2% su
     high o low) -> un wick VERO confermato da Binance resta intatto. Volume/timestamp invariati.
  3. BACKUP (data/_feed_backup/) + scrittura atomica + VALIDAZIONE (re-scan = 0 sospette,
     n righe invariato). Log dettagliato di ogni barra riparata (old OHLC -> new).

    uv run python scripts/analysis/clean_feed.py [ASSET_TF ...]   # default: tutti BTC/ETH x TF
    uv run python scripts/analysis/clean_feed.py BTC_1h           # un solo file
"""
from __future__ import annotations

import shutil
import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from src.data.downloader import _parquet_path, DATA_DIR

BACKUP = PROJECT_ROOT / "data" / "_feed_backup"
SYMBOL = {"BTC": "BTC/USDT", "ETH": "ETH/USDT"}
WICK_THR = 0.15        # detect: wick oltre 15% il cluster di close locale
REPLACE_THR = 0.02     # arbitra: sostituisci solo se Binance dissente >2% su high/low
CLOSE_THR = 0.01       # close-aware: sostituisci la barra se il CLOSE Deribit dista >1% da Binance
TF_MS = {"5m": 5, "8m": 8, "13m": 13, "15m": 15, "19m": 19, "30m": 30, "1h": 60}
_EX = None


def _binance():
    global _EX
    if _EX is None:
        import ccxt
        _EX = ccxt.binance({"enableRateLimit": True})
    return _EX


def suspect_mask(df: pd.DataFrame) -> np.ndarray:
    c = df["close"].to_numpy(float); h = df["high"].to_numpy(float); l = df["low"].to_numpy(float)
    cp = np.roll(c, 1); cp[0] = c[0]; cn = np.roll(c, -1); cn[-1] = c[-1]
    locmax = np.maximum.reduce([c, cp, cn]); locmin = np.minimum.reduce([c, cp, cn])
    return (h > locmax * (1 + WICK_THR)) | (l < locmin * (1 - WICK_THR))


def _binance_bar(symbol: str, tf: str, ts_ms: int):
    """OHLC reale Binance alla barra ts_ms (None se assente)."""
    try:
        rows = _binance().fetch_ohlcv(symbol, tf, since=ts_ms - 1, limit=3)
    except Exception as e:
        print(f"      ! binance err: {type(e).__name__}: {str(e)[:80]}")
        return None
    for r in rows:
        if int(r[0]) == ts_ms:
            return float(r[1]), float(r[2]), float(r[3]), float(r[4])
    return None


def clean_file(asset: str, tf: str) -> dict:
    path = _parquet_path(asset, tf)
    if not path.exists():
        return {"file": f"{asset}_{tf}", "skip": "no-file"}
    df = pd.read_parquet(path)
    mask = suspect_mask(df)
    idx = np.where(mask)[0]
    n0 = len(df)
    if len(idx) == 0:
        return {"file": f"{asset}_{tf}", "suspect": 0, "repaired": 0, "kept_real": 0,
                "missing_binance": 0, "rows_before": n0, "rows_after": n0,
                "still_suspect": 0, "log": []}
    repaired, kept, missing = 0, 0, 0
    log = []
    for i in idx:
        ts = int(df.iloc[i]["timestamp"])
        b = _binance_bar(SYMBOL[asset], tf, ts)
        oh, ol = float(df.iloc[i]["high"]), float(df.iloc[i]["low"])
        if b is None:
            missing += 1
            continue
        bo, bh, bl, bc = b
        if abs(oh - bh) / bh > REPLACE_THR or abs(ol - bl) / max(bl, 1e-9) > REPLACE_THR:
            df.iat[i, df.columns.get_loc("open")] = bo
            df.iat[i, df.columns.get_loc("high")] = bh
            df.iat[i, df.columns.get_loc("low")] = bl
            df.iat[i, df.columns.get_loc("close")] = bc
            repaired += 1
            ts_s = pd.to_datetime(ts, unit="ms", utc=True).strftime("%Y-%m-%d %H:%M")
            log.append(f"      {ts_s}  H {oh:,.0f}->{bh:,.0f}  L {ol:,.0f}->{bl:,.0f}")
        else:
            kept += 1                      # Binance conferma il wick: barra reale, intatta
    if repaired:
        BACKUP.mkdir(parents=True, exist_ok=True)
        shutil.copy2(path, BACKUP / f"{asset.lower()}_{tf}.parquet.bak")
        tmp = path.with_suffix(".parquet.tmp")
        df.to_parquet(tmp, index=False)
        tmp.replace(path)
    # validazione
    df2 = pd.read_parquet(path)
    still = int(suspect_mask(df2).sum())
    return {"file": f"{asset}_{tf}", "suspect": len(idx), "repaired": repaired,
            "kept_real": kept, "missing_binance": missing, "rows_before": n0,
            "rows_after": len(df2), "still_suspect": still, "log": log}


def _binance_series(asset: str, tf: str, start_ms: int, end_ms: int) -> dict:
    """OHLC reale Binance per l'intero range -> dict ts_ms -> (o,h,l,c). Bulk paginato."""
    ex = _binance()
    tf_ms = TF_MS[tf] * 60 * 1000
    out: dict[int, tuple] = {}
    since = start_ms
    while since <= end_ms:
        try:
            rows = ex.fetch_ohlcv(SYMBOL[asset], tf, since=since, limit=1000)
        except Exception as e:
            print(f"      ! binance err: {type(e).__name__}: {str(e)[:80]}")
            break
        if not rows:
            break
        for r in rows:
            out[int(r[0])] = (float(r[1]), float(r[2]), float(r[3]), float(r[4]))
        nxt = int(rows[-1][0]) + tf_ms
        if nxt <= since:
            break
        since = nxt
        if len(rows) < 1000 and since > end_ms:
            break
    return out


def clean_file_close(asset: str, tf: str, thr: float = CLOSE_THR, backup_dir: Path | None = None) -> dict:
    """CLOSE-AWARE: sostituisce O/H/L/C con Binance per ogni barra il cui CLOSE Deribit
    dista > thr da Binance (1% default). Cattura i print 'silenziosi' che il wick-check
    >15% non vede (close fantasma su barra di range piccolo). Fonte di verita' = Binance
    spot (il feed storico e' perp testnet -> inaffidabile; lo spot ~ mainnet via arbitraggio)."""
    if tf not in TF_MS:
        return {"file": f"{asset}_{tf}", "skip": "tf-non-binance"}
    path = _parquet_path(asset, tf)
    if not path.exists():
        return {"file": f"{asset}_{tf}", "skip": "no-file"}
    df = pd.read_parquet(path)
    n0 = len(df)
    tms = df["timestamp"].to_numpy("int64")
    c = df["close"].to_numpy(float)
    bz = _binance_series(asset, tf, int(tms[0]), int(tms[-1]))
    col = {k: df.columns.get_loc(k) for k in ("open", "high", "low", "close")}
    fixed, by_year, missing = 0, {}, 0
    log = []
    for i in range(n0):
        b = bz.get(int(tms[i]))
        if b is None:
            missing += 1
            continue
        bo, bh, bl, bc = b
        if bc <= 0:
            continue
        orig = float(c[i])    # cattura PRIMA della scrittura (to_numpy puo' essere una view)
        if abs(orig - bc) / bc > thr:
            df.iat[i, col["open"]] = bo
            df.iat[i, col["high"]] = bh
            df.iat[i, col["low"]] = bl
            df.iat[i, col["close"]] = bc
            fixed += 1
            y = pd.to_datetime(int(tms[i]), unit="ms", utc=True).year
            by_year[y] = by_year.get(y, 0) + 1
            if len(log) < 10:
                ts_s = pd.to_datetime(int(tms[i]), unit="ms", utc=True).strftime("%Y-%m-%d %H:%M")
                log.append(f"      {ts_s}  C {orig:,.2f}->{bc:,.2f}  ({abs(orig-bc)/bc*100:.1f}%)")
    if fixed:
        bdir = backup_dir or BACKUP
        bdir.mkdir(parents=True, exist_ok=True)
        shutil.copy2(path, bdir / f"{asset.lower()}_{tf}.parquet.bak")
        tmp = path.with_suffix(".parquet.tmp")
        df.to_parquet(tmp, index=False)
        tmp.replace(path)
    # validazione: ri-scan, 0 barre residue oltre soglia (fra quelle coperte da Binance)
    df2 = pd.read_parquet(path)
    c2 = df2["close"].to_numpy(float)
    still = sum(1 for i in range(len(df2))
                if (b := bz.get(int(tms[i]))) and b[3] > 0 and abs(c2[i] - b[3]) / b[3] > thr)
    return {"file": f"{asset}_{tf}", "covered": n0 - missing, "fixed": fixed,
            "missing_binance": missing, "rows_before": n0, "rows_after": len(df2),
            "still_over_thr": still, "by_year": by_year, "log": log}


def main():
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    close_mode = "--close" in sys.argv
    dry = "--dry" in sys.argv
    if close_mode:
        targets = args or [f"{a}_{tf}" for a in ("BTC", "ETH") for tf in ("5m", "15m", "1h")]
        stamp = time.strftime("%Y%m%d-%H%M%S")
        bdir = BACKUP / f"pre_close_clean_{stamp}"
        print(f"CLEAN FEED (close-aware vs Binance, thr={CLOSE_THR*100:.0f}%) — "
              f"{'DRY-RUN (nessuna scrittura)' if dry else f'backup in {bdir}'}\n")
        grand = 0
        for t in targets:
            asset, tf = t.split("_", 1)
            if dry:
                # dry: conta soltanto, niente scrittura
                r = _dry_close(asset, tf)
            else:
                r = clean_file_close(asset, tf, backup_dir=bdir)
            if r.get("skip"):
                print(f"  {t:<9} SKIP ({r['skip']})"); continue
            grand += r.get("fixed", 0)
            yr = " ".join(f"{y}:{n}" for y, n in sorted(r.get("by_year", {}).items()))
            print(f"  {r['file']:<9} coperte={r.get('covered',0):>7} riparate={r.get('fixed',0):>4} "
                  f"no-binance={r.get('missing_binance',0):>5} | righe {r['rows_before']}=={r.get('rows_after',r['rows_before'])} "
                  f"residue>thr={r.get('still_over_thr','-')}")
            if yr:
                print(f"      per anno: {yr}")
            for line in r.get("log", []):
                print(line)
        print(f"\n  TOTALE barre riparate (close-aware): {grand}")
        return
    targets = args or [f"{a}_{tf}" for a in ("BTC", "ETH") for tf in ("5m", "15m", "30m", "1h")]
    print(f"CLEAN FEED — backup in {BACKUP}\n")
    grand = 0
    for t in targets:
        asset, tf = t.split("_", 1)
        r = clean_file(asset, tf)
        if r.get("skip"):
            print(f"  {t:<9} SKIP ({r['skip']})"); continue
        grand += r.get("repaired", 0)
        print(f"  {r['file']:<9} sospette={r['suspect']:>3} riparate={r['repaired']:>3} "
              f"reali-tenute={r.get('kept_real',0):>3} no-binance={r.get('missing_binance',0):>2} "
              f"| righe {r['rows_before']}=={r['rows_after']}  residue-sospette={r['still_suspect']}")
        for line in r.get("log", [])[:8]:
            print(line)
        if len(r.get("log", [])) > 8:
            print(f"      ... (+{len(r['log'])-8} altre)")
    print(f"\n  TOTALE barre riparate: {grand}")


def _dry_close(asset: str, tf: str, thr: float = CLOSE_THR) -> dict:
    """Conta soltanto quante barre verrebbero riparate (nessuna scrittura)."""
    if tf not in TF_MS:
        return {"file": f"{asset}_{tf}", "skip": "tf-non-binance"}
    path = _parquet_path(asset, tf)
    if not path.exists():
        return {"file": f"{asset}_{tf}", "skip": "no-file"}
    df = pd.read_parquet(path)
    tms = df["timestamp"].to_numpy("int64"); c = df["close"].to_numpy(float)
    bz = _binance_series(asset, tf, int(tms[0]), int(tms[-1]))
    fixed, by_year, missing = 0, {}, 0
    for i in range(len(df)):
        b = bz.get(int(tms[i]))
        if b is None:
            missing += 1; continue
        if b[3] > 0 and abs(c[i] - b[3]) / b[3] > thr:
            fixed += 1
            y = pd.to_datetime(int(tms[i]), unit="ms", utc=True).year
            by_year[y] = by_year.get(y, 0) + 1
    return {"file": f"{asset}_{tf}", "covered": len(df) - missing, "fixed": fixed,
            "missing_binance": missing, "rows_before": len(df), "by_year": by_year, "log": []}


if __name__ == "__main__":
    main()