From b9e3388f7668e8a7193d08cd9897032601e24976 Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Thu, 18 Jun 2026 14:25:34 +0000 Subject: [PATCH] =?UTF-8?q?feat(data):=20clean=5Ffeed=20=E2=80=94=20ripara?= =?UTF-8?q?=20spike-print=20del=20feed=20Deribit=20coi=20dati=20reali=20Bi?= =?UTF-8?q?nance?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit La ricerca Price Ladder ha rivelato spike-print nei parquet storici (es. BTC 2024-02-13 low 38.580 / high fasulli, BTC reale ~50k) = stesso problema TP_PHANTOM/feed testnet di CLAUDE.md, che avvelena i backtest (stop/entry su prezzi mai avvenuti) e gonfia le code. Tool conservativo: DETECT (wick >15% oltre il cluster close locale) -> ARBITRA con Binance spot (ccxt, gia' cablato): sostituisce O/H/L/C solo se Binance dissente >2% su high/low (un wick VERO confermato da Binance resta intatto), backup in data/_feed_backup/ + scrittura atomica + validazione. RIPARATE 254 barre (BTC 132, ETH 122) su 8 file BTC/ETH x TF; 2 wick reali confermati da Binance e TENUTI (ETH 30m/1h flash-crash veri). Impatto validato: il BTC ladder che dava FULL DD 53.69% (artefatto) ora ne da 10.8% (la "coda di trend" era spike fasulli); PORT06 baseline FULL Sharpe 8.13->8.18, DD/OOS invariati (canonici stabili). data/raw e' gitignored: e' committato il TOOL, non i parquet (rigenerabili + backup locale). Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/analysis/clean_feed.py | 140 +++++++++++++++++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 scripts/analysis/clean_feed.py diff --git a/scripts/analysis/clean_feed.py b/scripts/analysis/clean_feed.py new file mode 100644 index 0000000..e21742b --- /dev/null +++ b/scripts/analysis/clean_feed.py @@ -0,0 +1,140 @@ +"""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()