PythagorasGoal/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
_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 main():
    targets = sys.argv[1:] 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}")


if __name__ == "__main__":
    main()