"""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]}")