"""Client HTTP per Cerbero MCP — Deribit. Ambiente determinato dal TOKEN (vedi API_REFERENCE): TESTNET_TOKEN -> upstream testnet, MAINNET_TOKEN -> upstream live. Default = testnet. Per il micro-test mainnet basta esportare CERBERO_TOKEN (e opzionalmente CERBERO_BOT_TAG) nel `.env` — NESSUNA modifica di codice. Vedi docs/specs/mainnet-microtest-plan.md.""" from __future__ import annotations import os from dataclasses import dataclass, field from typing import Any import requests BASE_URL = os.environ.get("CERBERO_BASE_URL", "https://cerbero-mcp.tielogic.xyz") # Token TESTNET di default; override via env CERBERO_TOKEN per puntare a mainnet. TESTNET_TOKEN = "_hm0FkyC67P9OXJTy7R9SE2lfhGz_Wa6i89KqH_uXrk" TOKEN = os.environ.get("CERBERO_TOKEN", TESTNET_TOKEN) BOT_TAG = os.environ.get("CERBERO_BOT_TAG", "pythagoras-paper") TIMEOUT = 15 def is_mainnet() -> bool: """True se il token attivo NON è quello testnet di default (= upstream live).""" return TOKEN != TESTNET_TOKEN @dataclass class CerberoClient: base_url: str = BASE_URL token: str = TOKEN bot_tag: str = BOT_TAG def _headers(self) -> dict[str, str]: return { "Authorization": f"Bearer {self.token}", "X-Bot-Tag": self.bot_tag, "Content-Type": "application/json", } def _post(self, path: str, payload: dict | None = None) -> dict: resp = requests.post( f"{self.base_url}{path}", headers=self._headers(), json=payload or {}, timeout=TIMEOUT, ) resp.raise_for_status() return resp.json() # --- Market data --- def get_ticker(self, instrument: str = "ETH-PERPETUAL") -> dict: return self._post("/mcp-deribit/tools/get_ticker", {"instrument": instrument}) def get_historical(self, instrument: str, start_date: str, end_date: str, resolution: str = "15") -> list[dict]: data = self._post("/mcp-deribit/tools/get_historical", { "instrument": instrument, "start_date": start_date, "end_date": end_date, "resolution": resolution, }) return data.get("candles", []) def get_historical_v2(self, instrument: str, start_date: str, end_date: str, interval: str = "1h", exchange: str = "deribit") -> list[dict]: """Endpoint unificato v2: /mcp/tools/get_historical (exchange deribit|hyperliquid). Stesso shape candele del legacy: [{timestamp(ms), open, high, low, close, volume}].""" data = self._post("/mcp/tools/get_historical", { "exchange": exchange, "instrument": instrument, "interval": interval, "start_date": start_date, "end_date": end_date, }) return data.get("candles", []) def get_instruments(self, currency: str, kind: str = "future", exchange: str = "deribit", limit: int = 100) -> list[dict]: """Enumera gli strumenti reali (v2). Usato per risolvere il naming senza hardcoding.""" data = self._post("/mcp/tools/get_instruments", { "exchange": exchange, "currency": currency, "kind": kind, "limit": limit, }) return data.get("instruments", data if isinstance(data, list) else []) def get_ticker_batch(self, instruments: list[str]) -> dict: """Prezzi correnti di N strumenti in una sola chiamata (v2, Deribit).""" return self._post("/mcp-deribit/tools/get_ticker_batch", {"instruments": instruments}) # --- Account --- def get_account_summary(self, currency: str = "USDC") -> dict: return self._post("/mcp-deribit/tools/get_account_summary", {"currency": currency}) def get_positions(self, currency: str = "ETH") -> list[dict]: return self._post("/mcp-deribit/tools/get_positions", {"currency": currency}) def get_open_orders(self, currency: str = "USDC", type: str = "all") -> list[dict]: """Ordini APERTI sul conto (limit resting + trigger non scattati). Ogni voce: {order_id, instrument, direction, order_type, order_state, amount, filled_amount, price, trigger_price, reduce_only, label}. NB Deribit puo' omettere i trigger untriggered da type='all' -> per i bracket interrogare anche type='trigger_all' e fare merge per order_id.""" out = self._post("/mcp-deribit/tools/get_open_orders", {"currency": currency, "type": type}) return out if isinstance(out, list) else out.get("orders", []) # --- Trading --- def place_order( self, instrument: str, side: str, amount: float, order_type: str = "market", price: float | None = None, leverage: int | None = None, label: str | None = None, reduce_only: bool = False, ) -> dict: """Piazza un ordine REALE su Deribit. `amount`: per i perp inverse (BTC/ETH-PERPETUAL) e' in USD notional (step BTC $10, ETH $1); per i lineari USDC (BTC_USDC/ETH_USDC-PERPETUAL) e' nel base-coin (step 0.0001/0.001). `reduce_only=True` per chiudere solo la propria quota su uno strumento condiviso (le posizioni si nettano per conto). Ritorna il `result` grezzo Deribit: {"order": {...}, "trades": [{price, amount, fee, ...}]} → le fee REALI sono in trades[].""" payload: dict[str, Any] = { "instrument_name": instrument, "side": side, "amount": amount, "type": order_type, } if price is not None: payload["price"] = price if leverage is not None: payload["leverage"] = leverage if label: payload["label"] = label if reduce_only: payload["reduce_only"] = True return self._post("/mcp-deribit/tools/place_order", payload) def cancel_order(self, order_id: str) -> dict: """Cancella un ordine resting (es. limit reduce-only al TP). Ritorna {order_id, state}; state='error' se l'ordine non e' piu' open (gia' fillato/cancellato) — il chiamante riconcilia via get_trade_history.""" return self._post("/mcp-deribit/tools/cancel_order", {"order_id": order_id}) def close_position(self, instrument: str) -> dict: return self._post("/mcp-deribit/tools/close_position", {"instrument_name": instrument}) def get_trade_history(self, limit: int = 100, instrument_name: str | None = None) -> list[dict]: """Trade ESEGUITI sul conto (fonte autorevole delle fee reali). Ogni voce: {instrument, direction, price, amount, fee, timestamp, order_id}.""" payload: dict[str, Any] = {"limit": limit} if instrument_name: payload["instrument_name"] = instrument_name out = self._post("/mcp-deribit/tools/get_trade_history", payload) return out if isinstance(out, list) else out.get("trades", []) def set_stop_loss(self, order_id: str, stop_price: float) -> dict: return self._post("/mcp-deribit/tools/set_stop_loss", {"order_id": order_id, "stop_price": stop_price}) def set_take_profit(self, order_id: str, tp_price: float) -> dict: return self._post("/mcp-deribit/tools/set_take_profit", {"order_id": order_id, "tp_price": tp_price})