116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""Client HTTP per Cerbero MCP — Deribit testnet."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
BASE_URL = "https://cerbero-mcp.tielogic.xyz"
|
|
TOKEN = "_hm0FkyC67P9OXJTy7R9SE2lfhGz_Wa6i89KqH_uXrk"
|
|
BOT_TAG = "pythagoras-paper"
|
|
TIMEOUT = 15
|
|
|
|
|
|
@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})
|
|
|
|
# --- Trading ---
|
|
|
|
def place_order(
|
|
self,
|
|
instrument: str,
|
|
side: str,
|
|
amount: float,
|
|
order_type: str = "market",
|
|
price: float | None = None,
|
|
leverage: int | None = 3,
|
|
label: str | None = None,
|
|
) -> dict:
|
|
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
|
|
return self._post("/mcp-deribit/tools/place_order", payload)
|
|
|
|
def close_position(self, instrument: str) -> dict:
|
|
return self._post("/mcp-deribit/tools/close_position", {"instrument_name": instrument})
|
|
|
|
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})
|