PythagorasGoal/src/data/downloader.py

"""Download historical OHLCV data. Primary: Cerbero MCP. Fallback: Binance/ccxt."""
from __future__ import annotations

import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pandas as pd
import requests
from tqdm import tqdm

DATA_DIR = Path(__file__).resolve().parents[2] / "data" / "raw"

CERBERO_URL = "https://cerbero-mcp.tielogic.xyz"
CERBERO_TOKEN = "_hm0FkyC67P9OXJTy7R9SE2lfhGz_Wa6i89KqH_uXrk"
CERBERO_HEADERS = {
    "Authorization": f"Bearer {CERBERO_TOKEN}",
    "X-Bot-Tag": "pythagoras-downloader",
    "Content-Type": "application/json",
}

ASSETS = {
    "BTC": {
        "deribit": {"instrument": "BTC-PERPETUAL", "start": "2018-09-01"},
        "binance_symbol": "BTC/USDT",
        "binance_start": "2018-01-01",
    },
    "ETH": {
        "deribit": {"instrument": "ETH-PERPETUAL", "start": "2019-06-01"},
        "binance_symbol": "ETH/USDT",
        "binance_start": "2018-01-01",
    },
}

# 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"}

TF_SECONDS = {"1m": 60, "5m": 300, "15m": 900, "1h": 3600}

MAX_DAYS_PER_REQUEST = {"1m": 1, "5m": 5, "15m": 15, "1h": 30}


def _parquet_path(asset: str, tf: str) -> Path:
    return DATA_DIR / f"{asset.lower()}_{tf}.parquet"


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/tools/get_historical",
        headers=CERBERO_HEADERS,
        json={
            "exchange": "deribit",
            "instrument": instrument,
            "interval": tf,
            "start_date": start,
            "end_date": end,
        },
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()
    return data.get("candles", [])


def _fetch_binance(symbol: str, tf: str, since_ms: int, limit: int = 1000) -> list[list]:
    import ccxt
    exchange = ccxt.binance({"enableRateLimit": True, "options": {"defaultType": "spot"}})
    return exchange.fetch_ohlcv(symbol, tf, since=since_ms, limit=limit)


def _download_cerbero_range(
    instrument: str, resolution: str, tf: str, start_date: str, end_date: str,
    allow_unvalidated: bool = False,
) -> pd.DataFrame:
    # Gate: si raccolgono dati SOLO per strumenti validati nel registry.
    # Esegui `python -m src.data.instruments` per (ri)costruirlo.
    if not allow_unvalidated:
        from src.data.instruments import is_validated
        if not is_validated(instrument, tf, "deribit"):
            raise ValueError(
                f"Strumento non validato: {instrument} @ {tf}. "
                f"Costruisci il registry (python -m src.data.instruments) o passa "
                f"allow_unvalidated=True per forzare."
            )
    all_candles: list[dict] = []
    max_days = MAX_DAYS_PER_REQUEST[tf]
    chunks = skipped = 0
    current = datetime.fromisoformat(start_date)
    end = datetime.fromisoformat(end_date)

    pbar = tqdm(
        total=(end - current).days,
        desc=f"  Cerbero {instrument} {tf}",
        unit="days",
    )

    while current < end:
        chunk_end = min(current + timedelta(days=max_days), end)
        start_str = current.strftime("%Y-%m-%d")
        end_str = chunk_end.strftime("%Y-%m-%d")

        for attempt in range(3):
            try:
                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()

    df = pd.DataFrame(all_candles)
    df = df.rename(columns={"timestamp": "timestamp"})
    df["timestamp"] = df["timestamp"].astype("int64")
    return df.drop_duplicates(subset="timestamp").sort_values("timestamp").reset_index(drop=True)


def _download_binance_range(
    symbol: str, tf: str, start_date: str, end_date: str
) -> pd.DataFrame:
    import ccxt

    exchange = ccxt.binance({
        "enableRateLimit": True,
        "timeout": 30000,
        "options": {"defaultType": "spot", "adjustForTimeDifference": True},
    })
    exchange.load_markets()

    start_ms = int(datetime.fromisoformat(start_date).replace(tzinfo=timezone.utc).timestamp() * 1000)
    end_ms = int(datetime.fromisoformat(end_date).replace(tzinfo=timezone.utc).timestamp() * 1000)
    tf_ms = TF_SECONDS[tf] * 1000
    all_rows: list[list] = []

    pbar = tqdm(desc=f"  Binance {symbol} {tf}", unit=" candles")

    since = start_ms
    while since < end_ms:
        for attempt in range(3):
            try:
                ohlcv = exchange.fetch_ohlcv(symbol, tf, since=since, limit=1000)
                break
            except ccxt.RateLimitExceeded:
                time.sleep(10)
                ohlcv = []
            except Exception as e:
                if attempt == 2:
                    print(f"  ERR: {e}")
                time.sleep(2 ** attempt)
                ohlcv = []

        if not ohlcv:
            break

        filtered = [c for c in ohlcv if c[0] < end_ms]
        all_rows.extend(filtered)
        pbar.update(len(filtered))
        since = ohlcv[-1][0] + tf_ms

        if len(ohlcv) < 1000 or ohlcv[-1][0] >= end_ms:
            break

    pbar.close()

    if not all_rows:
        return pd.DataFrame()

    cols = ["timestamp", "open", "high", "low", "close", "volume"]
    df = pd.DataFrame(all_rows, columns=cols)
    df["timestamp"] = df["timestamp"].astype("int64")
    return df.drop_duplicates(subset="timestamp").sort_values("timestamp").reset_index(drop=True)


def download_asset(asset: str, tf: str) -> pd.DataFrame:
    path = _parquet_path(asset, tf)
    info = ASSETS[asset]
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    resolution = DERIBIT_RESOLUTION[tf]

    parts: list[pd.DataFrame] = []

    binance_start = info["binance_start"]
    deribit_start = info["deribit"]["start"]

    if binance_start < deribit_start:
        print(f"\n  Fase 1: Binance {binance_start} → {deribit_start}")
        df_binance = _download_binance_range(
            info["binance_symbol"], tf, binance_start, deribit_start
        )
        if not df_binance.empty:
            parts.append(df_binance)

    print(f"\n  Fase 2: Cerbero/Deribit {deribit_start} → {today}")
    df_deribit = _download_cerbero_range(
        info["deribit"]["instrument"], resolution, tf, deribit_start, today
    )
    if not df_deribit.empty:
        parts.append(df_deribit)

    if not parts:
        print(f"  VUOTO: {asset} {tf}")
        return 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)

    first = datetime.fromtimestamp(df["timestamp"].iloc[0] / 1000, tz=timezone.utc)
    last = datetime.fromtimestamp(df["timestamp"].iloc[-1] / 1000, tz=timezone.utc)
    print(f"  ✓ {path.name}: {len(df)} candele [{first.date()} → {last.date()}]")

    return df


def download_all() -> dict[str, dict[str, pd.DataFrame]]:
    result: dict[str, dict[str, pd.DataFrame]] = {}
    for asset in ASSETS:
        print(f"\n{'='*60}")
        print(f"  {asset}")
        print(f"{'='*60}")
        result[asset] = {}
        for tf in TIMEFRAMES:
            result[asset][tf] = download_asset(asset, tf)
    return result


def load_data(asset: str = "BTC", tf: str = "1h") -> pd.DataFrame:
    path = _parquet_path(asset, tf)
    if not path.exists():
        raise FileNotFoundError(f"Dati non trovati: {path}. Esegui download_all().")
    df = pd.read_parquet(path)
    df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    return df


def data_summary() -> pd.DataFrame:
    rows = []
    for asset in ASSETS:
        for tf in TIMEFRAMES:
            path = _parquet_path(asset, tf)
            if path.exists():
                df = pd.read_parquet(path)
                first = datetime.fromtimestamp(df["timestamp"].iloc[0] / 1000, tz=timezone.utc)
                last = datetime.fromtimestamp(df["timestamp"].iloc[-1] / 1000, tz=timezone.utc)
                rows.append({
                    "asset": asset,
                    "tf": tf,
                    "candles": len(df),
                    "from": first.date(),
                    "to": last.date(),
                    "size_mb": round(path.stat().st_size / 1024 / 1024, 1),
                })
    return pd.DataFrame(rows)


if __name__ == "__main__":
    download_all()
    print("\n" + "=" * 60)
    print(data_summary().to_string(index=False))