PythagorasGoal/scripts/analysis/rebuild_history.py

"""REBUILD STORICO — fonte di verita' = DERIBIT MAINNET (il venue dove ESEGUIAMO).

Decisione (analisi 2026-06-19, vedi docs/diary/2026-06-19-deribit-history.md):
la contaminazione di data/raw nasce dal downloader che pesca la fase "Deribit" da
Cerbero col token TESTNET (feed farlocco) + la fase storica da Binance/USDT (wedge
USDT ~10bps, fino a 3% sotto depeg). Il cross-check multi-venue ha mostrato che il
venue PIU' vicino al consenso e' Deribit mainnet (0-1 bps), che e' anche dove si
esegue. Quindi: ricostruire lo storico da Deribit MAINNET (ccxt pubblico, NO token).

ARCHITETTURA (la "via migliore"):
  - SORGENTE: ccxt.deribit pubblico = api.deribit.com MAINNET (reale, tokenless;
    il token serve solo per trading/account, NON per gli OHLCV).
  - STRUMENTO per il PREZZO:
      BTC/ETH -> INVERSE perp (BTC/USD:BTC, ETH/USD:ETH): storia lunga (2018-08/2019-03)
                 e ~3 bps dal lineare USDC che eseguiamo (misurato) -> proxy fedele.
      alt     -> LINEARE USDC (X/USDC:USDC): unica opzione su Deribit, dal 2022.
  - BASE UNICA + RESAMPLE: si scarica UN solo timeframe base (default 5m) e si
    derivano tutti gli altri per aggregazione -> coerenza interna GARANTITA
    (15m == agg dei 5m per costruzione) e si valida UNA serie, non 20.
  - COPERTURA ONESTA a due livelli: major 2018/2019->oggi; alt SOLO dal 2022
    (prima non erano tradabili su Deribit -> backtestarli su Binance pre-2022 =
    validare un edge su un mercato che non potevamo eseguire).

NB: le daily NATIVE Deribit chiudono alle 08:00 UTC (settlement); qui il 1d e' DERIVATO
dal base per resample -> confini 00:00 UTC standard e coerenti con gli altri TF.

    uv run python scripts/analysis/rebuild_history.py --smoke           # prova su slice, NON tocca data/raw
    uv run python scripts/analysis/rebuild_history.py --smoke --asset ETH
    uv run python scripts/analysis/rebuild_history.py --asset BTC ETH    # FULL rebuild (scrive data/raw, con backup)
    uv run python scripts/analysis/rebuild_history.py                    # FULL tutti gli asset
"""
from __future__ import annotations
import sys
import time
import shutil
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))
import numpy as np
import pandas as pd
import ccxt

from src.data.downloader import _parquet_path, DATA_DIR

BACKUP = PROJECT_ROOT / "data" / "_feed_backup"

# asset -> (simbolo ccxt Deribit, start storico reale sul venue)
DERIBIT_INSTR = {
    "BTC":  ("BTC/USD:BTC",   "2018-08-14"),   # inverse, storia lunga ~ lineare entro 3 bps
    "ETH":  ("ETH/USD:ETH",   "2019-03-14"),
    "SOL":  ("SOL/USDC:USDC", "2022-03-15"),   # alt: solo lineare USDC, solo dal 2022
    "ADA":  ("ADA/USDC:USDC", "2022-03-21"),
    "XRP":  ("XRP/USDC:USDC", "2022-03-16"),
    "LTC":  ("LTC/USDC:USDC", "2022-04-28"),
    "DOGE": ("DOGE/USDC:USDC","2022-04-28"),
    "BNB":  ("BNB/USDC:USDC", "2024-10-01"),   # ATTENZIONE: storia cortissima (~1.7y)
}

BASE_TF = "5m"                                  # base unica (5m: supporta tutti i TF standard)
TARGET_TFS = ["5m", "15m", "1h"]                # = set del download_all; 4h/1d il runner li resampla dal 1h
SCHEMA = ["timestamp", "open", "high", "low", "close", "volume"]
COINBASE = ccxt.coinbase({"enableRateLimit": True})


def _deribit():
    return ccxt.deribit({"enableRateLimit": True})


def fetch_base(symbol: str, tf: str, start: str, end: str | None = None) -> pd.DataFrame:
    """OHLCV Deribit mainnet paginato in AVANTI (since esplicito che avanza — evita
    l'ancoraggio-a-now di ccxt deribit con since storico). Ritorna schema canonico."""
    ex = _deribit()
    tf_ms = ex.parse_timeframe(tf) * 1000
    start_ms = int(pd.Timestamp(start, tz="UTC").timestamp() * 1000)
    end_ms = int(pd.Timestamp(end, tz="UTC").timestamp() * 1000) if end else int(time.time() * 1000)
    rows: dict[int, list] = {}
    since, guard = start_ms, 0
    while since <= end_ms and guard < 5000:
        guard += 1
        for attempt in range(3):
            try:
                r = ex.fetch_ohlcv(symbol, tf, since=since, limit=1000)
                break
            except Exception as e:
                if attempt == 2:
                    print(f"      ! {type(e).__name__}: {str(e)[:60]}")
                    r = []
                time.sleep(2 ** attempt)
        r = [x for x in r if int(x[0]) >= since]
        if not r:
            break
        for x in r:
            t = int(x[0])
            if start_ms <= t <= end_ms:
                rows[t] = [t, float(x[1]), float(x[2]), float(x[3]), float(x[4]), float(x[5] or 0)]
        nxt = int(r[-1][0]) + tf_ms
        if nxt <= since:
            break
        since = nxt
    if not rows:
        return pd.DataFrame(columns=SCHEMA)
    df = pd.DataFrame(rows.values(), columns=SCHEMA)
    return df.drop_duplicates("timestamp").sort_values("timestamp").reset_index(drop=True)


def resample(base: pd.DataFrame, tf: str) -> pd.DataFrame:
    """Aggrega il base al timeframe tf (confini 00:00 UTC). Coerente per costruzione."""
    if tf == BASE_TF:
        return base.copy()
    g = base.copy()
    g.index = pd.to_datetime(g["timestamp"], unit="ms", utc=True)
    rule = {"5m": "5min", "15m": "15min", "30m": "30min", "1h": "1h", "2h": "2h",
            "3h": "3h", "4h": "4h", "6h": "6h", "12h": "12h", "1d": "1D"}[tf]
    out = g.resample(rule, label="left", closed="left").agg(
        {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"})
    out = out.dropna(subset=["open"])
    # ms epoch resolution-independent (pandas 2.x indicizza datetime64[ms]: view/asi8
    # cambiano risoluzione e darebbero ts corrotti -> uso una differenza di Timedelta)
    epoch = pd.Timestamp("1970-01-01", tz="UTC")
    ts_ms = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
    out.insert(0, "timestamp", ts_ms)
    return out.reset_index(drop=True)[SCHEMA]


def cross_audit(df1h: pd.DataFrame, asset: str, start: str, end: str) -> str:
    """Certifica il 1h ricostruito contro Coinbase USD (venue indipendente, NON USDT)."""
    sym = {"BTC": "BTC/USD", "ETH": "ETH/USD", "SOL": "SOL/USD", "ADA": "ADA/USD",
           "XRP": "XRP/USD", "LTC": "LTC/USD", "DOGE": "DOGE/USD", "BNB": None}.get(asset)
    if sym is None:
        return "no-coinbase-ref"
    try:
        if not COINBASE.markets:
            COINBASE.load_markets()
        if sym not in COINBASE.markets:
            return f"coinbase: simbolo {sym} assente"
        s_ms = int(pd.Timestamp(start, tz="UTC").timestamp() * 1000)
        e_ms = int(pd.Timestamp(end, tz="UTC").timestamp() * 1000)
        out, since = {}, s_ms
        while since <= e_ms:
            r = COINBASE.fetch_ohlcv(sym, "1h", since=since, limit=300)
            r = [x for x in r if int(x[0]) >= since]
            if not r:
                break
            for x in r:
                if s_ms <= int(x[0]) <= e_ms and x[4]:
                    out[int(x[0])] = float(x[4])
            nxt = int(r[-1][0]) + 3600_000
            if nxt <= since:
                break
            since = nxt
        ref = pd.Series(out)
    except Exception as e:
        return f"coinbase-err {type(e).__name__}"
    a = df1h.set_index("timestamp")["close"]
    m = pd.concat([a.rename("a"), ref.rename("b")], axis=1, join="inner")
    if len(m) == 0:
        return "0 overlap"
    d = (m["a"] - m["b"]).abs() / m["b"] * 1e4
    return f"vs Coinbase USD: barre={len(m)} med {d.median():.1f} bps p95 {d.quantile(.95):.1f} max {d.max():.1f}"


def build(asset: str, write: bool, smoke: bool) -> None:
    symbol, full_start = DERIBIT_INSTR[asset]
    if smoke:
        start, end = "2026-04-15", "2026-05-27"
    else:
        start, end = full_start, None
    print(f"\n  [{asset}] {symbol}  base {BASE_TF}  da {start}{' -> '+end if end else ' -> oggi'}")
    base = fetch_base(symbol, BASE_TF, start, end)
    if base.empty:
        print("      VUOTO — skip"); return
    f0 = pd.to_datetime(base['timestamp'].iloc[0], unit='ms', utc=True).date()
    f1 = pd.to_datetime(base['timestamp'].iloc[-1], unit='ms', utc=True).date()
    print(f"      base: {len(base)} barre {BASE_TF}  [{f0} -> {f1}]")
    built = {tf: resample(base, tf) for tf in TARGET_TFS}
    for tf, d in built.items():
        print(f"      {tf:<4} -> {len(d):>7} barre")
    # certifica il 1h
    aud_start = "2026-04-15" if smoke else "2026-03-01"
    aud_end = "2026-05-27" if smoke else f1.isoformat()
    print(f"      AUDIT 1h {cross_audit(built['1h'], asset, aud_start, aud_end)}")
    if not write:
        print("      (smoke: NON scrivo data/raw)")
        return
    BACKUP.mkdir(parents=True, exist_ok=True)
    for tf, d in built.items():
        path = _parquet_path(asset, tf)
        if path.exists():
            shutil.copy2(path, BACKUP / f"{asset.lower()}_{tf}.parquet.prebuild.bak")
        tmp = path.with_suffix(".parquet.tmp")
        d.to_parquet(tmp, index=False)
        tmp.replace(path)
        print(f"      scritto {path.name}")


def main():
    argv = sys.argv[1:]
    smoke = "--smoke" in argv
    assets = []
    if "--asset" in argv:
        i = argv.index("--asset")
        assets = [a.upper() for a in argv[i + 1:] if not a.startswith("--")]
    assets = assets or list(DERIBIT_INSTR)
    print("=" * 78)
    print(f"  REBUILD STORICO da DERIBIT MAINNET — {'SMOKE (no write)' if smoke else 'FULL (scrive data/raw, backup)'}")
    print(f"  asset: {assets}  base={BASE_TF}  target={TARGET_TFS}")
    print("=" * 78)
    for a in assets:
        build(a, write=not smoke, smoke=smoke)


if __name__ == "__main__":
    main()