feat(data): discovery + validazione strumenti con gate nel downloader
src/data/instruments.py: enumera i perpetui per exchange (Deribit, Hyperliquid;
esclusi Alpaca/stocks e Bybit per feed testnet farlocco) e valida ogni strumento
sui DATI STORICI realmente raccoglibili:
- esistenza, congruenza OHLC, not-flat (scarta contratti morti)
- liquidita' (volume daily) e congruenza prezzo cross-exchange via mediana
del base-coin (scarta outlier come Deribit SOL-PERPETUAL=9.6 vs SOL ~82)
Produce data/instruments_registry.json con strumenti validi, timeframe e start-date.
Gate: _download_cerbero_range rifiuta strumenti non validati (override esplicito
allow_unvalidated). La raccolta dati e' possibile solo per strumenti validati.
Registry attuale (testnet): Deribit 18/106 validi (BTC dal 2018, alt dal 2022),
Hyperliquid 66/74. I major liquidi (BTC,ETH,SOL,LTC,ADA,XRP,DOGE,AVAX,BNB,...) passano.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
+12
-1
@@ -69,8 +69,19 @@ def _fetch_binance(symbol: str, tf: str, since_ms: int, limit: int = 1000) -> li
|
||||
|
||||
|
||||
def _download_cerbero_range(
|
||||
instrument: str, resolution: str, tf: str, start_date: str, end_date: str
|
||||
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]
|
||||
current = datetime.fromisoformat(start_date)
|
||||
|
||||
@@ -0,0 +1,278 @@
|
||||
"""Discovery + validazione strumenti per gli exchange implementati (via Cerbero MCP).
|
||||
|
||||
Per ogni exchange (Deribit, Hyperliquid — esclusi Alpaca/stocks e Bybit, il cui
|
||||
feed testnet e' farlocco) enumera i perpetui, ne verifica i dati e produce un
|
||||
registry di strumenti VALIDATI.
|
||||
Solo gli strumenti nel registry possono essere usati per la raccolta dati
|
||||
(vedi gate in src/data/downloader.py).
|
||||
|
||||
Controlli di validazione (uno strumento e' valido solo se TUTTI passano):
|
||||
- exists : la storia daily ritorna candele
|
||||
- ohlc_sane : high>=low, high>=max(o,c), low<=min(o,c), prezzi>0
|
||||
- not_flat : non e' un contratto morto (quasi tutte le barre O==H==L==C)
|
||||
- liquid : volume_24h>0 dal ticker
|
||||
- congruent : il prezzo concorda (entro tolleranza) con la MEDIANA dello
|
||||
stesso base-coin su tutti gli exchange. Scarta i feed testnet
|
||||
farlocchi (es. Bybit BTC=300k) e i contratti sbagliati
|
||||
(es. Deribit SOL-PERPETUAL=9.6 vs SOL reale ~82).
|
||||
|
||||
NB: il token Cerbero punta a TESTNET; la congruenza cross-exchange e' il filtro
|
||||
che distingue i feed realistici (Deribit, Hyperliquid) da quelli farlocchi.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import statistics
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from src.live.cerbero_client import CerberoClient
|
||||
|
||||
REGISTRY_PATH = Path(__file__).resolve().parents[2] / "data" / "instruments_registry.json"
|
||||
|
||||
# I nostri timestep -> codice risoluzione per ciascun exchange
|
||||
TF_CODES = {
|
||||
"deribit": {"1m": "1", "5m": "5", "15m": "15", "1h": "60", "1d": "1D"},
|
||||
"hyperliquid": {"1m": "1m", "5m": "5m", "15m": "15m", "1h": "1h", "1d": "1d"},
|
||||
}
|
||||
CONGRUENCE_TOL = 0.05 # 5% di scostamento dalla mediana del base-coin
|
||||
|
||||
|
||||
def _today() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Quote:
|
||||
base: str
|
||||
symbol: str
|
||||
last: float | None = None
|
||||
volume_24h: float | None = None
|
||||
open_interest: float | None = None
|
||||
|
||||
|
||||
# --------------------------- adapters ---------------------------
|
||||
class ExchangeAdapter:
|
||||
name = "base"
|
||||
|
||||
def __init__(self, client: CerberoClient):
|
||||
self.c = client
|
||||
|
||||
def _post(self, tool: str, payload: dict) -> dict:
|
||||
return self.c._post(f"/mcp-{self.name}/tools/{tool}", payload)
|
||||
|
||||
def list_symbols(self) -> list[Quote]:
|
||||
"""Lista perpetui (economica). I prezzi possono essere None (vedi ticker)."""
|
||||
raise NotImplementedError
|
||||
|
||||
def ticker(self, q: Quote) -> None:
|
||||
"""Riempie last/volume/OI sul Quote (per-simbolo). No-op se gia' pieni."""
|
||||
|
||||
def candles(self, symbol: str, tf: str, start: str, end: str) -> pd.DataFrame:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class DeribitAdapter(ExchangeAdapter):
|
||||
name = "deribit"
|
||||
|
||||
def list_symbols(self) -> list[Quote]:
|
||||
perps, offset = [], 0
|
||||
while True:
|
||||
r = self._post("get_instruments", {"currency": "any", "kind": "future",
|
||||
"offset": offset, "limit": 100})
|
||||
insts = r.get("instruments", [])
|
||||
perps += [i["name"] for i in insts if i.get("name", "").endswith("-PERPETUAL")]
|
||||
if not r.get("has_more") or not insts:
|
||||
break
|
||||
offset += len(insts)
|
||||
if offset > 2000:
|
||||
break
|
||||
out = []
|
||||
for name in perps:
|
||||
base = name.split("-PERPETUAL")[0].replace("_USDC", "").replace("_USD", "")
|
||||
out.append(Quote(base, name))
|
||||
return out
|
||||
|
||||
def ticker(self, q: Quote) -> None:
|
||||
t = self._post("get_ticker", {"instrument": q.symbol})
|
||||
q.last, q.volume_24h, q.open_interest = t.get("last_price"), t.get("volume_24h"), t.get("open_interest")
|
||||
|
||||
def candles(self, symbol, tf, start, end) -> pd.DataFrame:
|
||||
r = self._post("get_historical", {"instrument": symbol, "start_date": start,
|
||||
"end_date": end, "resolution": TF_CODES["deribit"][tf]})
|
||||
return _to_df(r.get("candles", []))
|
||||
|
||||
|
||||
class HyperliquidAdapter(ExchangeAdapter):
|
||||
name = "hyperliquid"
|
||||
|
||||
def list_symbols(self) -> list[Quote]:
|
||||
r = self._post("get_markets", {})
|
||||
markets = r if isinstance(r, list) else r.get("markets", [])
|
||||
return [Quote(m["asset"], m["asset"], m.get("mark_price"),
|
||||
m.get("volume_24h"), m.get("open_interest")) for m in markets]
|
||||
|
||||
# prezzi gia' presenti da get_markets -> ticker no-op
|
||||
|
||||
def candles(self, symbol, tf, start, end) -> pd.DataFrame:
|
||||
r = self._post("get_historical", {"asset": symbol, "start_date": start, "end_date": end,
|
||||
"resolution": TF_CODES["hyperliquid"][tf], "limit": 5000})
|
||||
return _to_df(r.get("candles", []))
|
||||
|
||||
|
||||
ADAPTERS = {"deribit": DeribitAdapter, "hyperliquid": HyperliquidAdapter}
|
||||
|
||||
|
||||
def _to_df(candles: list[dict]) -> pd.DataFrame:
|
||||
if not candles:
|
||||
return pd.DataFrame()
|
||||
df = pd.DataFrame(candles)
|
||||
df["timestamp"] = df["timestamp"].astype("int64")
|
||||
return df.sort_values("timestamp").reset_index(drop=True)
|
||||
|
||||
|
||||
# --------------------------- validazione ---------------------------
|
||||
def _ohlc_sane(df: pd.DataFrame) -> bool:
|
||||
if df.empty:
|
||||
return False
|
||||
o, h, l, c = df["open"], df["high"], df["low"], df["close"]
|
||||
ok = (h >= l) & (h >= o) & (h >= c) & (l <= o) & (l <= c) & (c > 0) & (l > 0)
|
||||
return bool(ok.mean() > 0.99)
|
||||
|
||||
|
||||
def _not_flat(df: pd.DataFrame) -> bool:
|
||||
if df.empty:
|
||||
return False
|
||||
flat = (df["open"] == df["high"]) & (df["high"] == df["low"]) & (df["low"] == df["close"])
|
||||
return bool(flat.mean() < 0.90)
|
||||
|
||||
|
||||
def build_registry(exchanges: list[str] | None = None,
|
||||
tf_check: tuple[str, ...] = ("1m", "5m", "15m", "1h"),
|
||||
start_scan_from: str = "2017-01-01",
|
||||
save: bool = True) -> dict:
|
||||
exchanges = exchanges or ["deribit", "hyperliquid"] # NO alpaca, NO bybit (testnet farlocco)
|
||||
client = CerberoClient()
|
||||
adapters = {ex: ADAPTERS[ex](client) for ex in exchanges}
|
||||
|
||||
# 1) lista economica per ogni exchange
|
||||
listed: dict[str, list[Quote]] = {}
|
||||
for ex, ad in adapters.items():
|
||||
try:
|
||||
listed[ex] = ad.list_symbols()
|
||||
print(f" [{ex}] {len(listed[ex])} strumenti elencati")
|
||||
except Exception as e:
|
||||
print(f" [{ex}] discovery FALLITA: {type(e).__name__}: {e}")
|
||||
listed[ex] = []
|
||||
|
||||
# 2) universo = base-coin presenti su Deribit (il nostro venue). Bybit/HL
|
||||
# vengono validati solo sull'overlap (cross-check), non su 500+ simboli.
|
||||
deribit_bases = {q.base for q in listed.get("deribit", [])}
|
||||
selected: dict[str, list[Quote]] = {}
|
||||
for ex, qs in listed.items():
|
||||
selected[ex] = qs if ex == "deribit" else [q for q in qs if q.base in deribit_bases]
|
||||
|
||||
# 3) timeframe disponibili per exchange (testati su BTC recente)
|
||||
ref = {"deribit": "BTC-PERPETUAL", "hyperliquid": "BTC"}
|
||||
tf_by_ex: dict[str, list[str]] = {}
|
||||
for ex, ad in adapters.items():
|
||||
oks = []
|
||||
for tf in tf_check:
|
||||
try:
|
||||
if not ad.candles(ref[ex], tf, _today(), _today()).empty:
|
||||
oks.append(tf)
|
||||
except Exception:
|
||||
pass
|
||||
tf_by_ex[ex] = oks
|
||||
print(f" [{ex}] timeframe ok: {oks}")
|
||||
|
||||
# 4) UNA fetch daily per strumento: e' il dato che davvero raccoglieremmo.
|
||||
# Da qui ricaviamo esistenza, OHLC, not-flat, start-date, prezzo-per-congruenza
|
||||
# (ultima close STORICA, non il ticker) e liquidita' (volume daily recente).
|
||||
scan: dict[tuple[str, str], dict] = {}
|
||||
for ex, ad in adapters.items():
|
||||
for q in selected[ex]:
|
||||
rec = {"reasons": [], "last_close": None, "start_date": None, "vol": 0.0}
|
||||
try:
|
||||
d = ad.candles(q.symbol, "1d", start_scan_from, _today())
|
||||
if d.empty:
|
||||
rec["reasons"].append("no_history")
|
||||
else:
|
||||
if not _ohlc_sane(d):
|
||||
rec["reasons"].append("ohlc_insane")
|
||||
if not _not_flat(d):
|
||||
rec["reasons"].append("flat_dead")
|
||||
rec["last_close"] = float(d["close"].iloc[-1])
|
||||
rec["vol"] = float(d["volume"].tail(7).mean())
|
||||
rec["start_date"] = str(pd.to_datetime(d["timestamp"].iloc[0], unit="ms", utc=True).date())
|
||||
except Exception as e:
|
||||
rec["reasons"].append(f"history_err:{type(e).__name__}")
|
||||
scan[(ex, q.symbol)] = rec
|
||||
|
||||
# 5) mediana per base-coin dall'ULTIMA CLOSE STORICA (riferimento congruenza)
|
||||
by_base: dict[str, list[float]] = {}
|
||||
for (ex, sym), rec in scan.items():
|
||||
base = next(q.base for q in selected[ex] if q.symbol == sym)
|
||||
if rec["last_close"] and rec["last_close"] > 0:
|
||||
by_base.setdefault(base, []).append(rec["last_close"])
|
||||
median_px = {b: statistics.median(v) for b, v in by_base.items()}
|
||||
|
||||
# 6) finalizza validazione
|
||||
registry: dict = {"generated_at": datetime.now(timezone.utc).isoformat(),
|
||||
"congruence_tol": CONGRUENCE_TOL, "testnet": True, "exchanges": {}}
|
||||
for ex, ad in adapters.items():
|
||||
registry["exchanges"][ex] = {"timeframes": tf_by_ex[ex], "instruments": {}}
|
||||
for q in selected[ex]:
|
||||
rec = scan[(ex, q.symbol)]
|
||||
reasons = list(rec["reasons"])
|
||||
px, med, n_src = rec["last_close"], median_px.get(q.base), len(by_base.get(q.base, []))
|
||||
if not (rec["vol"] and rec["vol"] > 0):
|
||||
reasons.append("no_volume")
|
||||
if px is None or px <= 0:
|
||||
if "no_history" not in reasons:
|
||||
reasons.append("no_price")
|
||||
elif med and n_src >= 2 and abs(px - med) / med > CONGRUENCE_TOL:
|
||||
reasons.append(f"incongruent(px={px:.4g},med={med:.4g})")
|
||||
valid = len(reasons) == 0
|
||||
registry["exchanges"][ex]["instruments"][q.symbol] = {
|
||||
"base": q.base, "valid": valid, "reasons": reasons,
|
||||
"last_price": px, "start_date": rec["start_date"],
|
||||
"timeframes": tf_by_ex[ex] if valid else [],
|
||||
}
|
||||
if save:
|
||||
REGISTRY_PATH.write_text(json.dumps(registry, indent=2))
|
||||
print(f" registry salvato in {REGISTRY_PATH}")
|
||||
return registry
|
||||
|
||||
|
||||
# --------------------------- gate per il downloader ---------------------------
|
||||
def load_registry() -> dict:
|
||||
return json.loads(REGISTRY_PATH.read_text()) if REGISTRY_PATH.exists() else {}
|
||||
|
||||
|
||||
def is_validated(symbol: str, tf: str, exchange: str = "deribit") -> bool:
|
||||
"""True solo se lo strumento e' nel registry come valido per quel timeframe."""
|
||||
inst = load_registry().get("exchanges", {}).get(exchange, {}).get("instruments", {}).get(symbol)
|
||||
return bool(inst and inst.get("valid") and tf in inst.get("timeframes", []))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
reg = build_registry()
|
||||
print("\n" + "=" * 96)
|
||||
print(" REGISTRY STRUMENTI VALIDATI")
|
||||
print("=" * 96)
|
||||
for ex, exd in reg["exchanges"].items():
|
||||
insts = exd["instruments"]
|
||||
valid = {s: i for s, i in insts.items() if i["valid"]}
|
||||
print(f"\n {ex.upper()} | tf={exd['timeframes']} | validi {len(valid)}/{len(insts)}")
|
||||
for s, i in sorted(valid.items(), key=lambda kv: kv[1]["base"]):
|
||||
print(f" {s:30s} {i['base']:10s} px={i['last_price']:<12.6g} dal {i['start_date']}")
|
||||
bad = {s: i for s, i in insts.items() if not i["valid"]}
|
||||
if bad:
|
||||
shown = list(bad.items())[:6]
|
||||
print(f" -- scartati {len(bad)} (primi {len(shown)}):")
|
||||
for s, i in shown:
|
||||
print(f" {s:30s} {','.join(i['reasons'])[:64]}")
|
||||
Reference in New Issue
Block a user