From e18adba4a66cec3db7135c8faeede90f9a09c98d Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Fri, 12 Jun 2026 06:08:42 +0000 Subject: [PATCH] fix(data): downloader su endpoint v2 + guard anti-troncamento (post-mortem refresh notturno) Il refresh notturno e' rimasto 10h appeso e ha TRONCATO i parquet BTC al 2018: cerbero-mcp ha rimosso l'endpoint legacy /mcp-deribit/tools/get_historical (404), il downloader skippava in silenzio OGNI chunk (3 retry + sleep esponenziale x migliaia di chunk) e a fine giro scriveva comunque il file con la sola Fase 1 storica. Aggravante: TIMEFRAMES includeva '1m' (1 giorno/richiesta = ~3000 richieste/asset). ETH salvato in tempo (kill prima della sovrascrittura); BTC ripristinato via v2 (1h/15m/5m completi 2018->oggi in ~30 min). - _fetch_deribit -> endpoint v2 /mcp/tools/get_historical (lo stesso del runner) - guard chunk: >50% skippati = endpoint rotto -> RuntimeError, niente parziali - guard anti-regressione in download_asset: mai sovrascrivere un parquet con dati che finiscono PRIMA dell'esistente - '1m' fuori da TIMEFRAMES (refresh torna 5m/15m/1h; il 1m ad-hoc se serve) Co-Authored-By: Claude Fable 5 --- src/data/downloader.py | 43 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 38 insertions(+), 5 deletions(-) diff --git a/src/data/downloader.py b/src/data/downloader.py index ab35ffb..79e3d05 100644 --- a/src/data/downloader.py +++ b/src/data/downloader.py @@ -32,7 +32,10 @@ ASSETS = { }, } -TIMEFRAMES = ["1m", "5m", "15m", "1h"] +# NB: "1m" RIMOSSO da download_all (2026-06-12): con MAX_DAYS_PER_REQUEST=1 sono +# ~3000 richieste/asset — il refresh notturno e' rimasto 10h su BTC. Il 1m si +# scarica ad-hoc con download_asset("BTC", "1m") quando serve davvero. +TIMEFRAMES = ["5m", "15m", "1h"] DERIBIT_RESOLUTION = {"1m": "1", "5m": "5", "15m": "15", "1h": "60"} @@ -45,15 +48,20 @@ def _parquet_path(asset: str, tf: str) -> Path: return DATA_DIR / f"{asset.lower()}_{tf}.parquet" -def _fetch_deribit(instrument: str, resolution: str, start: str, end: str) -> list[dict]: +def _fetch_deribit(instrument: str, tf: str, start: str, end: str) -> list[dict]: + """Endpoint v2 unificato (2026-06-12): il legacy /mcp-deribit/tools/get_historical + e' stato RIMOSSO da cerbero-mcp (404) — il refresh notturno ha skippato in + silenzio ogni chunk e scritto parquet troncati al 2018. Il v2 e' lo stesso + endpoint usato dal runner live (interval '1h'/'15m'/...).""" resp = requests.post( - f"{CERBERO_URL}/mcp-deribit/tools/get_historical", + f"{CERBERO_URL}/mcp/tools/get_historical", headers=CERBERO_HEADERS, json={ + "exchange": "deribit", "instrument": instrument, + "interval": tf, "start_date": start, "end_date": end, - "resolution": resolution, }, timeout=30, ) @@ -84,6 +92,7 @@ def _download_cerbero_range( ) all_candles: list[dict] = [] max_days = MAX_DAYS_PER_REQUEST[tf] + chunks = skipped = 0 current = datetime.fromisoformat(start_date) end = datetime.fromisoformat(end_date) @@ -100,19 +109,29 @@ def _download_cerbero_range( for attempt in range(3): try: - candles = _fetch_deribit(instrument, resolution, start_str, end_str) + candles = _fetch_deribit(instrument, tf, start_str, end_str) all_candles.extend(candles) break except Exception as e: if attempt == 2: print(f" SKIP {start_str}→{end_str}: {e}") + skipped += 1 time.sleep(2 ** attempt) + chunks += 1 pbar.update(max_days) current = chunk_end pbar.close() + # GUARD (2026-06-12): se la maggioranza dei chunk e' stata skippata l'endpoint + # e' rotto, non il singolo range — ritornare il parziale farebbe scrivere un + # parquet troncato in silenzio (successo: 10h di SKIP e file fermi al 2018) + if chunks and skipped > chunks // 2: + raise RuntimeError( + f"{instrument} {tf}: {skipped}/{chunks} chunk falliti — endpoint rotto? " + "NON scrivo dati parziali.") + if not all_candles: return pd.DataFrame() @@ -210,6 +229,20 @@ def download_asset(asset: str, tf: str) -> pd.DataFrame: df = pd.concat(parts).drop_duplicates(subset="timestamp").sort_values("timestamp").reset_index(drop=True) + # GUARD anti-regressione (2026-06-12): MAI sovrascrivere un parquet buono con + # dati che finiscono PRIMA dell'esistente (fase Cerbero vuota per endpoint + # rotto = file troncato al 2018 scritto in silenzio; il bootstrap SH01 e i + # backtest leggono questi file) + if path.exists(): + try: + old_last = int(pd.read_parquet(path, columns=["timestamp"])["timestamp"].max()) + except Exception: + old_last = 0 + if int(df["timestamp"].max()) < old_last: + raise RuntimeError( + f"{path.name}: i nuovi dati finiscono PRIMA dell'esistente — " + "NON sovrascrivo (endpoint rotto?)") + path.parent.mkdir(parents=True, exist_ok=True) df.to_parquet(path, index=False)