Adriano Dal Pastro
9fa3e6a0be
Il feed BTC/ETH del book era congelato 7 giorni (2026-07-08 -> 07-15): rebuild_history.py usava shutil.copy2 per il .prebuild.bak; copy2=copyfile+copystat e copystat (os.utime/chmod con valori espliciti) richiede la PROPRIETA' del file, non basta il group-write. I .bak sono root:adriano, il cron gira come adriano -> EPERM ogni notte, abortendo PRIMA della scrittura del feed (data/raw/*.parquet fermo a Jul 8). Il book ha ri-girato ogni ora su barra vecchia (TP01 1d non ri-valutato); nessuna perdita (segnale fresco == tenuto) ma cieco 7g. Fix: backup best-effort = shutil.copyfile (solo contenuto, no copystat) in try/except OSError. Un .bak difensivo non deve mai poter bloccare la scrittura del feed live. Verificato: rebuild rigirato -> feed a 2026-07-15, audit cross-venue BTC 1.9/ETH 2.0 bps, file ora adriano:adriano (auto-sana), book dry-run legge ultima barra 2026-07-15 -> HOLD. Diario: docs/diary/2026-07-15-feed-freeze-rebuild-copystat.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
222 lines
10 KiB
Python
222 lines
10 KiB
Python
"""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():
|
|
# Backup best-effort: copyfile (solo contenuto, NO copystat).
|
|
# copy2 copiava anche i metadati -> os.utime/chmod su un .bak di proprieta'
|
|
# altrui (es. root) fallisce con EPERM per un membro-gruppo e ABORTIVA il
|
|
# rebuild del feed live (incidente 2026-07-15: feed fermo 7g). Un backup e'
|
|
# difensivo: un suo errore non deve MAI bloccare la scrittura del feed.
|
|
try:
|
|
shutil.copyfile(path, BACKUP / f"{asset.lower()}_{tf}.parquet.prebuild.bak")
|
|
except OSError as e:
|
|
print(f" WARN backup {path.name} saltato (non fatale): {e}")
|
|
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()
|