"""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()