"""Accesso Deribit MAINNET in SOLA LETTURA (via Cerbero MCP) + costruttore ordini deterministico. Serve lo SHADOW MODE di TP01 (`scripts/live/live_trend.py`): legge prezzi / conto / posizioni REALI dal mainnet (token `.env.mainnet`) e costruisce gli ordini di ribilancio **senza inviarli**. Qui NON esiste alcun metodo di trading — by design: l'unica via per piazzare ordini sara' un modulo separato, abilitato esplicitamente, dopo la validazione shadow + micro-test. Disciplina del progetto: **testnet FUORI** (feed farlocco, causa del reset v2.0.0). Solo mainnet reale, e in questa fase solo in lettura. I contratti sono ristretti a BTC/ETH-PERPETUAL (inverse: `amount` in USD notional, step verificato su Deribit: BTC $10, ETH $1). """ from __future__ import annotations import json import os from decimal import Decimal from pathlib import Path import requests PROJECT_ROOT = Path(__file__).resolve().parents[2] BASE_URL = os.environ.get("CERBERO_BASE_URL", "https://cerbero-mcp.tielogic.xyz") TIMEOUT = 15 # Inverse perp: amount = USD notional, step in USD, settle = base-coin (BTC/ETH). # Linear USDC perp: amount = base-coin (BTC/ETH), step in base-coin, settle = USDC (margine USDC). # NB il conto reale e' USDC -> gli strumenti ESEGUIBILI sono i LINEARI _USDC-PERPETUAL. # ⚠️ TABELLA DICHIARATA, ed e' lei l'autorita' per costruire un ordine — non l'API. # L'esecuzione dev'essere deterministica, in git e leggibile offline: un ordine che cambia forma # perche' una GET ha risposto diverso e' un ordine che non si puo' ricostruire dopo. Il venue fa # il CONTROLLORE, non la fonte: `check_specs()` gira ogni ora e allerta quando questa tabella si # scosta dal vero, e allora la si aggiorna qui a mano, di proposito. # # ⚠️ VERIFICATI contro public/get_instrument il 2026-08-19. Deribit ha cambiato le specifiche dei # perpetual lineari USDC il 18/08 alle 09:00 UTC (annuncio del 14/08) e questa tabella non se n'era # accorta: tick BTC 0.5->0.1, tick ETH 0.05->0.01, min/step ETH 0.001->0.0001. Nessun ordine e' # stato rifiutato perche' i cambi erano RIDUZIONI e un valore piu' grosso resta conforme — cioe' # e' andata bene per la direzione del cambiamento, non perche' ce ne fossimo accorti. Il costo # misurato era di granularita': su ETH l'incremento minimo restava $1.92 invece di $0.19 su un # conto da $597 (0.32% contro 0.03%). Il giorno che Deribit ALZA un minimo la stessa cecita' fa # rifiutare gli ordini: per quello adesso c'e' check_specs(). _CONTRACT = { "BTC-PERPETUAL": {"min": 10.0, "step": 10.0, "tick": 0.5, "settle": "BTC"}, "ETH-PERPETUAL": {"min": 1.0, "step": 1.0, "tick": 0.05, "settle": "ETH"}, "BTC_USDC-PERPETUAL": {"min": 0.0001, "step": 0.0001, "tick": 0.1, "settle": "USDC", "linear": True}, "ETH_USDC-PERPETUAL": {"min": 0.0001, "step": 0.0001, "tick": 0.01, "settle": "USDC", "linear": True}, } # nome nostro -> campo di public/get_instrument _SPEC_FIELDS = (("tick", "tick_size"), ("min", "min_trade_amount"), ("step", "contract_size")) def compare_specs(declared: dict, live: dict) -> list[dict]: """PURA, niente rete. Divergenze fra la tabella dichiarata e le specifiche del venue. Ogni voce porta la DIREZIONE, che e' l'informazione che serve per decidere se correre: * `rischio: "granularita'"` — il dichiarato e' piu' GROSSO del vero. L'ordine resta conforme (un multiplo del tick e' un tick valido), si perde solo precisione. Si aggiorna con calma. * `rischio: "rifiuto"` — il dichiarato e' piu' FINE del vero. Il venue RIFIUTA l'ordine. Va corretto prima del prossimo giro. Uno strumento assente dal `live` NON e' una divergenza: puo' essere la rete. Il silenzio non va confuso con l'uguaglianza, e infatti `check_specs()` dichiara a parte cosa non ha letto. """ fuori = [] for strumento, atteso in declared.items(): vero = live.get(strumento) if not vero: continue for nostro, loro in _SPEC_FIELDS: a, v = atteso.get(nostro), vero.get(loro) if a is None or v is None: continue a, v = float(a), float(v) if abs(a - v) <= 1e-12 * max(1.0, abs(v)): continue fuori.append({"strumento": strumento, "campo": nostro, "dichiarato": a, "venue": v, "rischio": "granularita'" if a > v else "rifiuto"}) return fuori def fetch_live_specs(instruments: list[str] | None = None) -> tuple[dict, list[str]]: """public/get_instrument per ogni strumento. RETE. Ritorna (specifiche, non_letti). ⚠️ Non solleva mai: un controllo che non parte non deve poter fermare il cron del book. Cio' che non ha letto lo DICHIARA invece di ometterlo, perche' "non l'ho guardato" e "e' uguale" sono due cose diverse e solo una delle due e' rassicurante. """ import urllib.request fuori, non_letti = {}, [] for strumento in (instruments or list(_CONTRACT)): try: url = ("https://www.deribit.com/api/v2/public/get_instrument" f"?instrument_name={strumento}") with urllib.request.urlopen(url, timeout=15) as r: res = json.loads(r.read()).get("result") or {} if res.get("tick_size") is None: non_letti.append(strumento) else: fuori[strumento] = res except Exception as e: # noqa: BLE001 non_letti.append(f"{strumento} ({type(e).__name__})") return fuori, non_letti def check_specs() -> dict: """Confronta la tabella dichiarata col venue. Sola lettura, mai solleva.""" live, non_letti = fetch_live_specs() return {"divergenze": compare_specs(_CONTRACT, live), "non_letti": non_letti} # Il conto reale e' USDC -> mappiamo gli asset sui perp LINEARI USDC (gli unici eseguibili qui). INSTRUMENT = {"BTC": "BTC_USDC-PERPETUAL", "ETH": "ETH_USDC-PERPETUAL"} DISASTER_LABEL = "tp01-disaster" # label dei bracket disaster-SL (per ritrovarli/gestirli/mostrarli) # ----------------------------- costruzione ordini (pura, testabile, NIENTE rete) ----------------------------- def _quantize_step(value: float, step: float, mn: float) -> float: """Arrotonda al multiplo di `step` (Decimal, niente artefatti float), clampa al minimo.""" n = round(value / step) return float(max(n * Decimal(str(step)), Decimal(str(mn)))) def notional_to_amount(instrument: str, notional_usd: float, price: float | None = None) -> float: """USD notional -> `amount` Deribit, arrotondato allo step e clampato al minimo. Ritorna 0.0 se sotto mezzo step. INVERSE: amount in USD (price ignorato). LINEAR USDC: amount in base-coin (units = notional/price -> serve il `price`; senza price ritorna 0.0).""" spec = _CONTRACT[instrument] step, mn = spec["step"], spec["min"] if spec.get("linear"): if not price: return 0.0 units = abs(notional_usd) / price if units < step / 2: return 0.0 return _quantize_step(units, step, mn) if abs(notional_usd) < step / 2: return 0.0 return _quantize_step(abs(notional_usd), step, mn) def quantize_price(instrument: str, price: float) -> float: """Arrotonda il prezzo al tick dello strumento (per gli ordini stop/limit).""" tick = _CONTRACT[instrument].get("tick") if not tick or price <= 0: return price return float(round(price / tick) * Decimal(str(tick))) def disaster_stop_price(instrument: str, mark: float, sl_pct: float, long: bool = True) -> float: """Prezzo del disaster-stop: ~sl_pct SOTTO il mark per un long (SOPRA per uno short), al tick.""" raw = mark * (1 - sl_pct) if long else mark * (1 + sl_pct) return quantize_price(instrument, raw) def target_notional_usd(target_fraction: float, weight: float, equity_usd: float) -> float: """Notional bersaglio (USD) di un asset = peso nel book * frazione-di-equity TP01 * equity. Coerente col paper trader (esposizione asset = WEIGHT * target * equity).""" return weight * target_fraction * equity_usd def build_rebalance_order(instrument: str, target_fraction: float, weight: float, equity_usd: float, current_pos_usd: float, price: float | None = None) -> dict | None: """COSTRUISCE (non invia) l'ordine di ribilancio verso il target. Ritorna un dict-ordine o None se sotto-soglia. Long-only TP01 -> target_notional >= 0; delta = target - posizione corrente. `price` (mark) serve a convertire il notional in base-coin per gli strumenti LINEARI USDC.""" tgt = target_notional_usd(target_fraction, weight, equity_usd) delta = tgt - current_pos_usd amount = notional_to_amount(instrument, delta, price=price) if amount == 0.0: return None is_exit = abs(tgt) < 1e-9 and abs(current_pos_usd) > 0 return dict( instrument=instrument, side="buy" if delta > 0 else "sell", amount=amount, type="market", reduce_only=is_exit, target_notional=round(tgt, 2), current_notional=round(current_pos_usd, 2), delta_notional=round(delta, 2), ) # ----------------------------- lettura mainnet (Cerbero MCP) — SOLA LETTURA ----------------------------- def _load_mainnet_token() -> tuple[str, str]: """Legge CERBERO_TOKEN (mainnet) + bot-tag da .env.mainnet. Il token NON viene mai stampato.""" env: dict[str, str] = {} for ln in (PROJECT_ROOT / ".env.mainnet").read_text().splitlines(): ln = ln.strip() if ln and not ln.startswith("#") and "=" in ln: k, v = ln.split("=", 1) env[k] = v.strip() if "CERBERO_TOKEN" not in env: raise RuntimeError("CERBERO_TOKEN assente in .env.mainnet") return env["CERBERO_TOKEN"], env.get("CERBERO_BOT_TAG", "pythagoras-shadow") class DeribitRead: """Accesso Deribit mainnet in SOLA LETTURA via Cerbero MCP. Nessun metodo di trading (by design).""" def __init__(self) -> None: self._token, self._tag = _load_mainnet_token() def _post(self, path: str, payload: dict) -> dict | list: r = requests.post( f"{BASE_URL}{path}", headers={"Authorization": f"Bearer {self._token}", "X-Bot-Tag": self._tag, "Content-Type": "application/json"}, json=payload, timeout=TIMEOUT, ) r.raise_for_status() return r.json() @staticmethod def _unwrap(resp: dict | list) -> dict | list: return resp.get("result", resp) if isinstance(resp, dict) else resp def ticker(self, instrument: str) -> dict: return self._unwrap(self._post("/mcp-deribit/tools/get_ticker", {"instrument": instrument})) or {} def mark_price(self, instrument: str) -> float: t = self.ticker(instrument) for k in ("mark_price", "index_price", "last_price", "last"): v = t.get(k) if v: return float(v) raise ValueError(f"prezzo assente nel ticker {instrument}: chiavi={list(t)[:8]}") def account_summary(self, currency: str) -> dict: return self._unwrap(self._post("/mcp-deribit/tools/get_account_summary", {"currency": currency})) or {} def positions(self, currency: str) -> list[dict]: out = self._unwrap(self._post("/mcp-deribit/tools/get_positions", {"currency": currency})) if isinstance(out, list): return out return out.get("positions", []) if isinstance(out, dict) else [] def position_usd(self, instrument: str) -> float: """Size netta (USD notional, segno = direzione) della posizione su `instrument`. 0 se flat.""" cur = _CONTRACT[instrument]["settle"] for p in self.positions(cur): if p.get("instrument_name") == instrument or p.get("instrument") == instrument: return float(p.get("size") or p.get("size_currency") or 0.0) return 0.0 def trade_history(self, instrument: str, limit: int = 20) -> list[dict]: """Trade REALMENTE eseguiti sul conto per `instrument` (fonte autorevole fee/fill).""" out = self._unwrap(self._post("/mcp-deribit/tools/get_trade_history", {"limit": limit, "instrument_name": instrument})) return out if isinstance(out, list) else (out.get("trades", []) if isinstance(out, dict) else []) def open_orders(self, instrument: str) -> list[dict]: """Ordini APERTI su `instrument` (limit resting + trigger). Deribit puo' omettere i trigger untriggered da type='all' -> interroga anche 'trigger_all' e fa merge per order_id.""" cur = _CONTRACT[instrument]["settle"] seen: dict = {} for typ in ("all", "trigger_all"): try: out = self._unwrap(self._post("/mcp-deribit/tools/get_open_orders", {"currency": cur, "type": typ})) lst = out if isinstance(out, list) else (out.get("orders", []) if isinstance(out, dict) else []) for o in lst: if o.get("instrument_name") == instrument or o.get("instrument") == instrument: seen[o.get("order_id")] = o except Exception: pass return list(seen.values())