feat(V2): unified /mcp interface; retire Bybit/Alpaca to old/

Add a common cross-exchange interface (/mcp) over the integrated venues
(deribit, hyperliquid):

- get_instruments: uniform schema where each row carries its own
  `exchange`, `fees` (maker/taker, live from Deribit, null where the
  venue has no per-instrument schedule) and `history_start` (listing
  date, live from Deribit creation_timestamp), plus type/tick_size and a
  lossless `native` blob. Optional `exchange` filter; fan-out otherwise.
- get_historical: generalized to {exchange, instrument, interval,
  start_date, end_date}, returning a single chosen venue's candles.
  Consensus merge stays available on /mcp-cross.

New: routers/unified.py, exchanges/cross/instruments.py (normalizers),
UnifiedClient in cross/client.py, schemas in cross/tools.py. Deribit
get_instruments now also surfaces maker/taker_commission and
creation_timestamp (additive).

Retire Bybit and Alpaca from the API surface: move clients, routers,
settings classes and their tests under old/ (history preserved via
git mv); drop them from the builder, /mcp-cross dispatch and symbol_map.
Bybit remains a public funding/OI data source in sentiment (not the
trading client). IBKR is intentionally excluded from /mcp for now.

Docs: rewrite API_REFERENCE.md (remove Bybit/Alpaca, document /mcp,
clarify that data_timestamp is injected globally by middleware).

Tests: add unified-interface coverage; update cross/settings/builder/boot
tests for the reduced venue set. Fix a pre-existing flaky assertion in
the Hyperliquid signing test (r/s use eth_utils.to_hex like the official
SDK, so a leading zero byte yields <66 chars ~1/256 of the time).

323 passed, ruff clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano
2026-05-29 09:09:22 +00:00
parent eb5f0148ad
commit bc75d3980a
36 changed files with 644 additions and 382 deletions
+519
View File
@@ -0,0 +1,519 @@
"""Alpaca client su httpx puro (V2.0.0).
Riscrittura full-REST del client `alpaca-py` originale: 4 endpoint base
(trading, stock data, crypto data, options data), auth via header
APCA-API-KEY-ID / APCA-API-SECRET-KEY, parità completa con la versione V1
(stesse firme, stessa shape dei dict ritornati).
- `base_url` parametro override applica SOLO al trading endpoint
(coerente con `url_override` di alpaca-py.TradingClient). Gli endpoint
data restano hardcoded su `https://data.alpaca.markets`.
- I metodi ritornano `dict` / `list[dict]` direttamente dal JSON REST
(al posto dei modelli pydantic alpaca-py serializzati). Le chiavi sono
quelle restituite dall'API Alpaca; equivalgono al `model_dump()` dei
modelli SDK precedenti.
"""
from __future__ import annotations
import datetime as _dt
from typing import Any
import httpx
from cerbero_mcp.common.candles import validate_candles
from cerbero_mcp.common.http import async_client
# ── Endpoint base ────────────────────────────────────────────────
_TRADING_LIVE = "https://api.alpaca.markets"
_TRADING_PAPER = "https://paper-api.alpaca.markets"
_DATA = "https://data.alpaca.markets"
# ── Mappa timeframe → query param Alpaca ─────────────────────────
# Alpaca v2 bars: timeframe = "1Min" / "5Min" / "15Min" / "30Min" / "1Hour" / "1Day" / "1Week"
_TF_MAP = {
"1min": "1Min",
"5min": "5Min",
"15min": "15Min",
"30min": "30Min",
"1h": "1Hour",
"1d": "1Day",
"1w": "1Week",
}
_ASSET_CLASS_MAP = {
"stocks": "us_equity",
"crypto": "crypto",
"options": "us_option",
}
def _tf(interval: str) -> str:
if interval in _TF_MAP:
return _TF_MAP[interval]
raise ValueError(f"unsupported timeframe: {interval}")
def _asset_class_param(ac: str) -> str:
ac = ac.lower()
if ac in _ASSET_CLASS_MAP:
return _ASSET_CLASS_MAP[ac]
raise ValueError(f"invalid asset_class: {ac}")
def _iso(value: _dt.datetime | _dt.date | None) -> str | None:
if value is None:
return None
return value.isoformat()
class AlpacaClient:
"""Client httpx-based per Alpaca REST API v2.
Auth via header `APCA-API-KEY-ID` / `APCA-API-SECRET-KEY`.
"""
def __init__(
self,
api_key: str,
secret_key: str,
paper: bool = True,
base_url: str | None = None,
http: httpx.AsyncClient | None = None,
) -> None:
self.api_key = api_key
self.secret_key = secret_key
self.paper = paper
# `base_url` mantenuto come attributo pubblico (test/build_client lo
# leggono). Override del solo endpoint trading; data endpoints sono
# sempre `data.alpaca.markets` (Alpaca non offre paper data feed).
self.base_url = base_url
if base_url:
self._trading_base = base_url
else:
self._trading_base = _TRADING_PAPER if paper else _TRADING_LIVE
self._data_base = _DATA
# Single long-lived AsyncClient → reuse connection pool.
self._http = http or async_client(timeout=30.0)
async def aclose(self) -> None:
"""Chiudi connessioni HTTP. Idempotente."""
if not self._http.is_closed:
await self._http.aclose()
async def health(self) -> dict[str, Any]:
"""Probe minimo per /health/ready: nessuna chiamata di rete."""
return {"status": "ok", "paper": self.paper}
# ── Helpers ──────────────────────────────────────────────────
@property
def _headers(self) -> dict[str, str]:
return {
"APCA-API-KEY-ID": self.api_key,
"APCA-API-SECRET-KEY": self.secret_key,
"Accept": "application/json",
}
async def _request(
self,
method: str,
base: str,
path: str,
*,
params: dict[str, Any] | None = None,
json_body: dict[str, Any] | None = None,
) -> Any:
"""Esegue una richiesta HTTP autenticata e ritorna il JSON parsato.
Per response body vuoto (es. DELETE 204) ritorna `{}`.
Solleva `httpx.HTTPStatusError` su 4xx/5xx tramite raise_for_status.
"""
url = f"{base}{path}"
# httpx scarta i query params con valore None automaticamente solo se
# passati come list of tuples; con dict dobbiamo filtrare a monte.
clean_params: dict[str, Any] | None = None
if params is not None:
clean_params = {k: v for k, v in params.items() if v is not None}
if not clean_params:
clean_params = None
resp = await self._http.request(
method,
url,
params=clean_params,
json=json_body,
headers=self._headers,
)
resp.raise_for_status()
if not resp.content:
return {}
return resp.json()
# ── Account / positions ──────────────────────────────────────
async def get_account(self) -> dict:
data = await self._request("GET", self._trading_base, "/v2/account")
return dict(data) if data else {}
async def get_positions(self) -> list[dict]:
data = await self._request("GET", self._trading_base, "/v2/positions")
return list(data) if data else []
async def get_activities(self, limit: int = 50) -> list[dict]:
data = await self._request(
"GET",
self._trading_base,
"/v2/account/activities",
params={"page_size": limit},
)
items = list(data) if data else []
return items[:limit]
# ── Assets ──────────────────────────────────────────────────
async def get_assets(
self, asset_class: str = "stocks", status: str = "active"
) -> list[dict]:
data = await self._request(
"GET",
self._trading_base,
"/v2/assets",
params={
"status": status,
"asset_class": _asset_class_param(asset_class),
},
)
items = list(data) if data else []
return items[:500]
# ── Market data ─────────────────────────────────────────────
async def get_ticker(self, symbol: str, asset_class: str = "stocks") -> dict:
ac = asset_class.lower()
if ac == "stocks":
trade_resp = await self._request(
"GET",
self._data_base,
f"/v2/stocks/{symbol}/trades/latest",
)
quote_resp = await self._request(
"GET",
self._data_base,
f"/v2/stocks/{symbol}/quotes/latest",
)
trade = (trade_resp or {}).get("trade") or {}
quote = (quote_resp or {}).get("quote") or {}
return {
"symbol": symbol,
"asset_class": "stocks",
"last_price": trade.get("p"),
"bid": quote.get("bp"),
"ask": quote.get("ap"),
"bid_size": quote.get("bs"),
"ask_size": quote.get("as"),
"timestamp": trade.get("t"),
}
if ac == "crypto":
trade_resp = await self._request(
"GET",
self._data_base,
"/v1beta3/crypto/us/latest/trades",
params={"symbols": symbol},
)
quote_resp = await self._request(
"GET",
self._data_base,
"/v1beta3/crypto/us/latest/quotes",
params={"symbols": symbol},
)
trade = ((trade_resp or {}).get("trades") or {}).get(symbol) or {}
quote = ((quote_resp or {}).get("quotes") or {}).get(symbol) or {}
return {
"symbol": symbol,
"asset_class": "crypto",
"last_price": trade.get("p"),
"bid": quote.get("bp"),
"ask": quote.get("ap"),
"timestamp": trade.get("t"),
}
if ac == "options":
quote_resp = await self._request(
"GET",
self._data_base,
f"/v1beta1/options/{symbol}/quotes/latest",
)
quote = (quote_resp or {}).get("quote") or {}
return {
"symbol": symbol,
"asset_class": "options",
"bid": quote.get("bp"),
"ask": quote.get("ap"),
"timestamp": quote.get("t"),
}
raise ValueError(f"invalid asset_class: {asset_class}")
async def get_bars(
self,
symbol: str,
asset_class: str = "stocks",
interval: str = "1d",
start: str | None = None,
end: str | None = None,
limit: int = 1000,
) -> dict:
tf = _tf(interval)
start_dt = (
_dt.datetime.fromisoformat(start)
if start
else (_dt.datetime.now(_dt.UTC) - _dt.timedelta(days=30))
)
end_dt = _dt.datetime.fromisoformat(end) if end else _dt.datetime.now(_dt.UTC)
ac = asset_class.lower()
params: dict[str, Any] = {
"symbols": symbol,
"timeframe": tf,
"start": _iso(start_dt),
"end": _iso(end_dt),
"limit": limit,
}
if ac == "stocks":
# IEX feed di default — coerente con default alpaca-py free tier.
params["feed"] = "iex"
data = await self._request(
"GET", self._data_base, "/v2/stocks/bars", params=params
)
elif ac == "crypto":
data = await self._request(
"GET",
self._data_base,
"/v1beta3/crypto/us/bars",
params=params,
)
elif ac == "options":
data = await self._request(
"GET",
self._data_base,
"/v1beta1/options/bars",
params=params,
)
else:
raise ValueError(f"invalid asset_class: {asset_class}")
bars_dict = (data or {}).get("bars") or {}
rows = bars_dict.get(symbol) or []
def _iso_to_ms(ts: str | int | None) -> int | None:
if ts is None or isinstance(ts, int):
return ts
return int(_dt.datetime.fromisoformat(
ts.replace("Z", "+00:00")
).timestamp() * 1000)
candles = validate_candles([
{
"timestamp": _iso_to_ms(b.get("t")),
"open": b.get("o"),
"high": b.get("h"),
"low": b.get("l"),
"close": b.get("c"),
"volume": b.get("v"),
}
for b in rows
])
return {
"symbol": symbol,
"asset_class": ac,
"interval": interval,
"candles": candles,
}
async def get_snapshot(self, symbol: str) -> dict:
data = await self._request(
"GET",
self._data_base,
"/v2/stocks/snapshots",
params={"symbols": symbol},
)
# API ritorna {"AAPL": {snapshot}} o {"snapshots": {...}} — gestiamo
# entrambi i formati; v2/stocks/snapshots ritorna dict top-level
# symbol→snapshot.
if data is None:
return {}
if symbol in data:
return data[symbol] or {}
snaps = data.get("snapshots") or {}
return snaps.get(symbol) or {}
async def get_option_chain(
self,
underlying: str,
expiry: str | None = None,
) -> dict:
params: dict[str, Any] = {}
if expiry:
# Validazione date (solleva ValueError su input invalido,
# parità con V1 che usava _dt.date.fromisoformat).
_dt.date.fromisoformat(expiry)
params["expiration_date_gte"] = expiry
params["expiration_date_lte"] = expiry
data = await self._request(
"GET",
self._data_base,
f"/v1beta1/options/snapshots/{underlying}",
params=params or None,
)
contracts = (data or {}).get("snapshots") if data else None
return {
"underlying": underlying,
"expiry": expiry,
"contracts": contracts if contracts is not None else (data or {}),
}
# ── Orders ──────────────────────────────────────────────────
async def get_open_orders(self, limit: int = 50) -> list[dict]:
data = await self._request(
"GET",
self._trading_base,
"/v2/orders",
params={"status": "open", "limit": limit},
)
return list(data) if data else []
async def place_order(
self,
symbol: str,
side: str,
qty: float | None = None,
notional: float | None = None,
order_type: str = "market",
limit_price: float | None = None,
stop_price: float | None = None,
tif: str = "day",
asset_class: str = "stocks",
) -> dict:
ot = order_type.lower()
body: dict[str, Any] = {
"symbol": symbol,
"side": side.lower(),
"type": ot,
"time_in_force": tif.lower(),
}
if qty is not None:
body["qty"] = str(qty)
if notional is not None:
body["notional"] = str(notional)
if ot == "market":
pass
elif ot == "limit":
if limit_price is None:
raise ValueError("limit_price required for limit order")
body["limit_price"] = str(limit_price)
elif ot == "stop":
if stop_price is None:
raise ValueError("stop_price required for stop order")
body["stop_price"] = str(stop_price)
else:
raise ValueError(f"unsupported order_type: {order_type}")
# `asset_class` non è un parametro REST; mantenuto in firma per parità
# con V1 (era usato solo da SDK per scegliere il request model).
_ = asset_class
data = await self._request(
"POST",
self._trading_base,
"/v2/orders",
json_body=body,
)
return dict(data) if data else {}
async def amend_order(
self,
order_id: str,
qty: float | None = None,
limit_price: float | None = None,
stop_price: float | None = None,
tif: str | None = None,
) -> dict:
body: dict[str, Any] = {}
if qty is not None:
body["qty"] = str(qty)
if limit_price is not None:
body["limit_price"] = str(limit_price)
if stop_price is not None:
body["stop_price"] = str(stop_price)
if tif is not None:
body["time_in_force"] = tif.lower()
data = await self._request(
"PATCH",
self._trading_base,
f"/v2/orders/{order_id}",
json_body=body,
)
return dict(data) if data else {}
async def cancel_order(self, order_id: str) -> dict:
# DELETE /v2/orders/{id} → 204 No Content su success.
await self._request(
"DELETE", self._trading_base, f"/v2/orders/{order_id}"
)
return {"order_id": order_id, "canceled": True}
async def cancel_all_orders(self) -> list[dict]:
# DELETE /v2/orders → 207 Multi-Status con array di {id, status}
data = await self._request(
"DELETE", self._trading_base, "/v2/orders"
)
return list(data) if data else []
# ── Position close ──────────────────────────────────────────
async def close_position(
self, symbol: str, qty: float | None = None, percentage: float | None = None
) -> dict:
# DELETE /v2/positions/{symbol}?qty=... oppure ?percentage=...
params: dict[str, Any] = {}
if qty is not None:
params["qty"] = str(qty)
if percentage is not None:
params["percentage"] = str(percentage)
data = await self._request(
"DELETE",
self._trading_base,
f"/v2/positions/{symbol}",
params=params or None,
)
return dict(data) if data else {}
async def close_all_positions(self, cancel_orders: bool = True) -> list[dict]:
data = await self._request(
"DELETE",
self._trading_base,
"/v2/positions",
params={"cancel_orders": "true" if cancel_orders else "false"},
)
return list(data) if data else []
# ── Clock / calendar ────────────────────────────────────────
async def get_clock(self) -> dict:
data = await self._request("GET", self._trading_base, "/v2/clock")
return dict(data) if data else {}
async def get_calendar(
self, start: str | None = None, end: str | None = None
) -> list[dict]:
params: dict[str, Any] = {}
if start:
_dt.date.fromisoformat(start) # validazione, parità V1
params["start"] = start
if end:
_dt.date.fromisoformat(end)
params["end"] = end
data = await self._request(
"GET",
self._trading_base,
"/v2/calendar",
params=params or None,
)
return list(data) if data else []
+56
View File
@@ -0,0 +1,56 @@
"""Leverage cap server-side per place_order.
Cap letto dal secret JSON via campo `max_leverage`. Default 1 (cash) se assente.
"""
from __future__ import annotations
from fastapi import HTTPException
def get_max_leverage(creds: dict) -> int:
"""Legge max_leverage dal secret. Default 1 se mancante."""
raw = creds.get("max_leverage", 1)
try:
value = int(raw)
except (TypeError, ValueError):
value = 1
return max(1, value)
def enforce_leverage(
requested: int | float | None,
*,
creds: dict,
exchange: str,
) -> int:
"""Verifica e applica leverage cap. Ritorna leverage applicabile.
Solleva HTTPException(403, LEVERAGE_CAP_EXCEEDED) se requested > cap.
Se requested is None, applica il cap come default.
"""
cap = get_max_leverage(creds)
if requested is None:
return cap
lev = int(requested)
if lev < 1:
raise HTTPException(
status_code=403,
detail={
"error": "LEVERAGE_CAP_EXCEEDED",
"exchange": exchange,
"requested": lev,
"max": cap,
"reason": "leverage must be >= 1",
},
)
if lev > cap:
raise HTTPException(
status_code=403,
detail={
"error": "LEVERAGE_CAP_EXCEEDED",
"exchange": exchange,
"requested": lev,
"max": cap,
},
)
return lev
+279
View File
@@ -0,0 +1,279 @@
"""Tool alpaca V2: pydantic schemas + async functions.
Ogni funzione prende (client: AlpacaClient, params: <Req>) e restituisce
un dict (o list[dict]). Pure logica, no FastAPI dependency, no ACL.
L'autenticazione bearer è gestita dal middleware in cerbero_mcp.auth;
l'audit verrà cablato dal router via request.state.environment.
"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel
from cerbero_mcp.exchanges.alpaca.client import AlpacaClient
from cerbero_mcp.exchanges.alpaca.leverage_cap import get_max_leverage
# === Schemas: reads ===
class GetAccountReq(BaseModel):
pass
class GetPositionsReq(BaseModel):
pass
class GetActivitiesReq(BaseModel):
limit: int = 50
class GetAssetsReq(BaseModel):
asset_class: str = "stocks"
status: str = "active"
class GetTickerReq(BaseModel):
symbol: str
asset_class: str = "stocks"
class GetBarsReq(BaseModel):
symbol: str
asset_class: str = "stocks"
interval: str = "1d"
start: str | None = None
end: str | None = None
limit: int = 1000
class GetSnapshotReq(BaseModel):
symbol: str
class GetOptionChainReq(BaseModel):
underlying: str
expiry: str | None = None
class GetOpenOrdersReq(BaseModel):
limit: int = 50
class GetClockReq(BaseModel):
pass
class GetCalendarReq(BaseModel):
start: str | None = None
end: str | None = None
# === Schemas: writes ===
class PlaceOrderReq(BaseModel):
symbol: str
side: str # "buy" | "sell"
qty: float | None = None
notional: float | None = None
order_type: str = "market"
limit_price: float | None = None
stop_price: float | None = None
tif: str = "day"
asset_class: str = "stocks"
model_config = {
"json_schema_extra": {
"examples": [
{
"summary": "Market buy 1 share AAPL",
"value": {
"symbol": "AAPL",
"side": "buy",
"qty": 1,
"order_type": "market",
"asset_class": "stocks",
},
}
]
}
}
class AmendOrderReq(BaseModel):
order_id: str
qty: float | None = None
limit_price: float | None = None
stop_price: float | None = None
tif: str | None = None
class CancelOrderReq(BaseModel):
order_id: str
class CancelAllOrdersReq(BaseModel):
pass
class ClosePositionReq(BaseModel):
symbol: str
qty: float | None = None
percentage: float | None = None
class CloseAllPositionsReq(BaseModel):
cancel_orders: bool = True
# === Tools (reads) ===
async def environment_info(
client: AlpacaClient, *, creds: dict, env_info: Any | None = None
) -> dict:
if env_info is None:
return {
"exchange": "alpaca",
"environment": "testnet" if getattr(client, "paper", True) else "mainnet",
"source": "credentials",
"env_value": None,
"base_url": getattr(client, "base_url", None),
"max_leverage": get_max_leverage(creds),
}
return {
"exchange": env_info.exchange,
"environment": env_info.environment,
"source": env_info.source,
"env_value": env_info.env_value,
"base_url": env_info.base_url,
"max_leverage": get_max_leverage(creds),
}
async def get_account(client: AlpacaClient, params: GetAccountReq) -> dict:
return await client.get_account()
async def get_positions(
client: AlpacaClient, params: GetPositionsReq
) -> dict:
return {"positions": await client.get_positions()}
async def get_activities(
client: AlpacaClient, params: GetActivitiesReq
) -> dict:
return {"activities": await client.get_activities(params.limit)}
async def get_assets(client: AlpacaClient, params: GetAssetsReq) -> dict:
return {
"assets": await client.get_assets(params.asset_class, params.status)
}
async def get_ticker(client: AlpacaClient, params: GetTickerReq) -> dict:
return await client.get_ticker(params.symbol, params.asset_class)
async def get_bars(client: AlpacaClient, params: GetBarsReq) -> dict:
return await client.get_bars(
params.symbol,
params.asset_class,
params.interval,
params.start,
params.end,
params.limit,
)
async def get_snapshot(
client: AlpacaClient, params: GetSnapshotReq
) -> dict:
return await client.get_snapshot(params.symbol)
async def get_option_chain(
client: AlpacaClient, params: GetOptionChainReq
) -> dict:
return await client.get_option_chain(params.underlying, params.expiry)
async def get_open_orders(
client: AlpacaClient, params: GetOpenOrdersReq
) -> dict:
return {"orders": await client.get_open_orders(params.limit)}
async def get_clock(client: AlpacaClient, params: GetClockReq) -> dict:
return await client.get_clock()
async def get_calendar(
client: AlpacaClient, params: GetCalendarReq
) -> dict:
return {"calendar": await client.get_calendar(params.start, params.end)}
# === Tools (writes) ===
async def place_order(
client: AlpacaClient, params: PlaceOrderReq, *, creds: dict
) -> dict:
# Alpaca: cap default 1 (cash account). Niente leverage parametro;
# cap presente per coerenza con altri exchange e per audit.
return await client.place_order(
symbol=params.symbol,
side=params.side,
qty=params.qty,
notional=params.notional,
order_type=params.order_type,
limit_price=params.limit_price,
stop_price=params.stop_price,
tif=params.tif,
asset_class=params.asset_class,
)
async def amend_order(
client: AlpacaClient, params: AmendOrderReq
) -> dict:
return await client.amend_order(
params.order_id,
params.qty,
params.limit_price,
params.stop_price,
params.tif,
)
async def cancel_order(
client: AlpacaClient, params: CancelOrderReq
) -> dict:
return await client.cancel_order(params.order_id)
async def cancel_all_orders(
client: AlpacaClient, params: CancelAllOrdersReq
) -> dict:
return {"canceled": await client.cancel_all_orders()}
async def close_position(
client: AlpacaClient, params: ClosePositionReq
) -> dict:
return await client.close_position(
params.symbol, params.qty, params.percentage
)
async def close_all_positions(
client: AlpacaClient, params: CloseAllPositionsReq
) -> dict:
return {
"closed": await client.close_all_positions(params.cancel_orders)
}
View File
+904
View File
@@ -0,0 +1,904 @@
"""Bybit V5 REST API client (httpx puro, no SDK).
Implementazione diretta su `httpx.AsyncClient` per i tool Cerbero MCP V2.
Mantiene parità di interfaccia con la versione precedente basata su
`pybit.unified_trading.HTTP` per non rompere `tools.py` né i router.
Auth Bybit V5:
Header X-BAPI-SIGN = HMAC_SHA256(secret,
timestamp + api_key + recv_window + (body_json | querystring))
"""
from __future__ import annotations
import hashlib
import hmac
import json
import time
import uuid
from typing import Any
from urllib.parse import urlencode
import httpx
from cerbero_mcp.common import indicators as ind
from cerbero_mcp.common import microstructure as micro
from cerbero_mcp.common.candles import validate_candles
BASE_MAINNET = "https://api.bybit.com"
BASE_TESTNET = "https://api-testnet.bybit.com"
DEFAULT_RECV_WINDOW = "5000"
DEFAULT_TIMEOUT = 15.0
def _f(v: Any) -> float | None:
try:
return float(v)
except (TypeError, ValueError):
return None
def _i(v: Any) -> int | None:
try:
return int(v)
except (TypeError, ValueError):
return None
class BybitAPIError(RuntimeError):
"""Errore di trasporto Bybit V5 (non gestito a livello envelope)."""
class BybitClient:
"""Async REST client per Bybit V5 (linear/inverse/spot/option)."""
def __init__(
self,
api_key: str,
api_secret: str,
testnet: bool = True,
http: httpx.AsyncClient | None = None,
base_url: str | None = None,
) -> None:
self.api_key = api_key
self.api_secret = api_secret
self.testnet = testnet
self.base_url = base_url or (BASE_TESTNET if testnet else BASE_MAINNET)
self.recv_window = DEFAULT_RECV_WINDOW
# `http` injection è usato dai test per montare un AsyncClient con
# `httpx.MockTransport`. In produzione creiamo un client dedicato.
self._owns_http = http is None
self._http: httpx.AsyncClient = http or httpx.AsyncClient(
timeout=DEFAULT_TIMEOUT
)
async def aclose(self) -> None:
"""Chiude l'AsyncClient httpx se di nostra proprietà."""
if self._owns_http:
await self._http.aclose()
async def health(self) -> dict[str, Any]:
"""Probe minimo per /health/ready: nessuna chiamata di rete."""
return {"status": "ok", "testnet": self.testnet}
# ── auth helpers ───────────────────────────────────────────
def _timestamp_ms(self) -> str:
return str(int(time.time() * 1000))
def _sign(self, timestamp: str, payload: str) -> str:
msg = timestamp + self.api_key + self.recv_window + payload
return hmac.new(
self.api_secret.encode("utf-8"),
msg.encode("utf-8"),
hashlib.sha256,
).hexdigest()
def _signed_headers(self, payload: str) -> dict[str, str]:
ts = self._timestamp_ms()
sig = self._sign(ts, payload)
return {
"X-BAPI-API-KEY": self.api_key,
"X-BAPI-TIMESTAMP": ts,
"X-BAPI-RECV-WINDOW": self.recv_window,
"X-BAPI-SIGN": sig,
"Content-Type": "application/json",
}
@staticmethod
def _clean_params(params: dict[str, Any] | None) -> dict[str, Any]:
if not params:
return {}
return {k: v for k, v in params.items() if v is not None}
@staticmethod
def _querystring(params: dict[str, Any]) -> str:
# Bybit accetta querystring nell'ordine in cui viene serializzata la
# request. Per la signature usiamo lo stesso urlencode (ordine
# inserzione dict). In Python 3.7+ dict mantiene insertion order:
# mantenere coerenza tra signature payload e URL effettivo.
return urlencode(params)
# ── request primitives ─────────────────────────────────────
async def _request_public(
self,
method: str,
path: str,
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
clean = self._clean_params(params)
url = self.base_url + path
resp = await self._http.request(
method, url, params=clean if clean else None
)
return self._parse_response(resp)
async def _request_signed(
self,
method: str,
path: str,
params: dict[str, Any] | None = None,
body: dict[str, Any] | None = None,
) -> dict[str, Any]:
url = self.base_url + path
method = method.upper()
if method == "GET":
clean = self._clean_params(params)
qs = self._querystring(clean)
headers = self._signed_headers(qs)
resp = await self._http.request(
method, url, params=clean if clean else None, headers=headers
)
else:
payload_body = body or {}
body_json = json.dumps(payload_body, separators=(",", ":"))
headers = self._signed_headers(body_json)
resp = await self._http.request(
method, url, content=body_json, headers=headers
)
return self._parse_response(resp)
@staticmethod
def _parse_response(resp: httpx.Response) -> dict[str, Any]:
try:
data = resp.json()
except Exception as e: # pragma: no cover - difficilmente raggiungibile
raise BybitAPIError(
f"invalid JSON from Bybit (status={resp.status_code}): {resp.text[:200]}"
) from e
if resp.status_code >= 500:
raise BybitAPIError(
f"bybit server error {resp.status_code}: "
f"{data.get('retMsg', resp.text[:200])}"
)
if not isinstance(data, dict):
raise BybitAPIError(f"unexpected payload type: {type(data).__name__}")
return data
def _envelope(self, resp: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
code = resp.get("retCode", 0)
if code != 0:
return {"error": resp.get("retMsg", "bybit_error"), "code": code}
return payload
# ── parsers shared ─────────────────────────────────────────
@staticmethod
def _parse_ticker(row: dict[str, Any]) -> dict[str, Any]:
return {
"symbol": row.get("symbol"),
"last_price": _f(row.get("lastPrice")),
"mark_price": _f(row.get("markPrice")),
"bid": _f(row.get("bid1Price")),
"ask": _f(row.get("ask1Price")),
"volume_24h": _f(row.get("volume24h")),
"turnover_24h": _f(row.get("turnover24h")),
"funding_rate": _f(row.get("fundingRate")),
"open_interest": _f(row.get("openInterest")),
}
# ── market data (public) ───────────────────────────────────
async def get_ticker(self, symbol: str, category: str = "linear") -> dict:
resp = await self._request_public(
"GET",
"/v5/market/tickers",
params={"category": category, "symbol": symbol},
)
rows = (resp.get("result") or {}).get("list") or []
if not rows:
return {"symbol": symbol, "error": "not_found"}
return self._parse_ticker(rows[0])
async def get_ticker_batch(
self, symbols: list[str], category: str = "linear"
) -> dict[str, dict]:
out: dict[str, dict] = {}
for sym in symbols:
out[sym] = await self.get_ticker(sym, category=category)
return out
async def get_orderbook(
self, symbol: str, category: str = "linear", limit: int = 50
) -> dict:
resp = await self._request_public(
"GET",
"/v5/market/orderbook",
params={"category": category, "symbol": symbol, "limit": limit},
)
r = resp.get("result") or {}
return {
"symbol": r.get("s"),
"bids": [[float(p), float(q)] for p, q in (r.get("b") or [])],
"asks": [[float(p), float(q)] for p, q in (r.get("a") or [])],
"timestamp": r.get("ts"),
}
async def get_historical(
self,
symbol: str,
category: str = "linear",
interval: str = "60",
start: int | None = None,
end: int | None = None,
limit: int = 1000,
) -> dict:
params: dict[str, Any] = {
"category": category,
"symbol": symbol,
"interval": interval,
"limit": limit,
}
if start is not None:
params["start"] = start
if end is not None:
params["end"] = end
resp = await self._request_public("GET", "/v5/market/kline", params=params)
rows = (resp.get("result") or {}).get("list") or []
candles = validate_candles([
{
"timestamp": int(r[0]),
"open": r[1],
"high": r[2],
"low": r[3],
"close": r[4],
"volume": r[5],
}
for r in rows
])
return {"symbol": symbol, "candles": candles}
async def get_indicators(
self,
symbol: str,
category: str = "linear",
indicators: list[str] | None = None,
interval: str = "60",
start: int | None = None,
end: int | None = None,
) -> dict:
indicators = indicators or ["rsi", "atr", "macd", "adx"]
historical = await self.get_historical(
symbol, category=category, interval=interval, start=start, end=end
)
candles = historical.get("candles", [])
closes = [c["close"] for c in candles]
highs = [c["high"] for c in candles]
lows = [c["low"] for c in candles]
out: dict[str, Any] = {"symbol": symbol, "category": category}
for name in indicators:
n = name.lower()
if n == "sma":
out["sma"] = ind.sma(closes, 20)
elif n == "rsi":
out["rsi"] = ind.rsi(closes)
elif n == "atr":
out["atr"] = ind.atr(highs, lows, closes)
elif n == "macd":
out["macd"] = ind.macd(closes)
elif n == "adx":
out["adx"] = ind.adx(highs, lows, closes)
else:
out[n] = None
return out
async def get_funding_rate(self, symbol: str, category: str = "linear") -> dict:
resp = await self._request_public(
"GET",
"/v5/market/tickers",
params={"category": category, "symbol": symbol},
)
rows = (resp.get("result") or {}).get("list") or []
if not rows:
return {"symbol": symbol, "error": "not_found"}
row = rows[0]
return {
"symbol": row.get("symbol"),
"funding_rate": _f(row.get("fundingRate")),
"next_funding_time": _i(row.get("nextFundingTime")),
}
async def get_funding_history(
self, symbol: str, category: str = "linear", limit: int = 100
) -> dict:
resp = await self._request_public(
"GET",
"/v5/market/funding/history",
params={"category": category, "symbol": symbol, "limit": limit},
)
rows = (resp.get("result") or {}).get("list") or []
hist = [
{
"timestamp": int(r.get("fundingRateTimestamp", 0)),
"rate": float(r.get("fundingRate", 0)),
}
for r in rows
]
return {"symbol": symbol, "history": hist}
async def get_open_interest(
self,
symbol: str,
category: str = "linear",
interval: str = "5min",
limit: int = 288,
) -> dict:
resp = await self._request_public(
"GET",
"/v5/market/open-interest",
params={
"category": category,
"symbol": symbol,
"intervalTime": interval,
"limit": limit,
},
)
rows = (resp.get("result") or {}).get("list") or []
points = [
{
"timestamp": int(r.get("timestamp", 0)),
"oi": float(r.get("openInterest", 0)),
}
for r in rows
]
current_oi = points[0]["oi"] if points else None
return {
"symbol": symbol,
"category": category,
"interval": interval,
"current_oi": current_oi,
"points": points,
}
async def get_instruments(
self, category: str = "linear", symbol: str | None = None
) -> dict:
params: dict[str, Any] = {"category": category}
if symbol:
params["symbol"] = symbol
resp = await self._request_public(
"GET", "/v5/market/instruments-info", params=params
)
rows = (resp.get("result") or {}).get("list") or []
instruments = []
for r in rows:
pf = r.get("priceFilter") or {}
lf = r.get("lotSizeFilter") or {}
instruments.append(
{
"symbol": r.get("symbol"),
"status": r.get("status"),
"base_coin": r.get("baseCoin"),
"quote_coin": r.get("quoteCoin"),
"tick_size": _f(pf.get("tickSize")),
"qty_step": _f(lf.get("qtyStep")),
"min_qty": _f(lf.get("minOrderQty")),
}
)
return {"category": category, "instruments": instruments}
async def get_option_chain(self, base_coin: str, expiry: str | None = None) -> dict:
resp = await self._request_public(
"GET",
"/v5/market/instruments-info",
params={"category": "option", "baseCoin": base_coin.upper()},
)
rows = (resp.get("result") or {}).get("list") or []
options = []
for r in rows:
delivery = r.get("deliveryTime")
if expiry and expiry not in r.get("symbol", ""):
continue
options.append(
{
"symbol": r.get("symbol"),
"base_coin": r.get("baseCoin"),
"settle_coin": r.get("settleCoin"),
"type": r.get("optionsType"),
"launch_time": int(r.get("launchTime", 0)),
"delivery_time": int(delivery) if delivery else None,
}
)
return {"base_coin": base_coin.upper(), "options": options}
# ── account / positions / orders (signed) ─────────────────
async def get_positions(
self, category: str = "linear", settle_coin: str = "USDT"
) -> list[dict]:
params: dict[str, Any] = {"category": category}
if category in ("linear", "inverse"):
params["settleCoin"] = settle_coin
resp = await self._request_signed("GET", "/v5/position/list", params=params)
rows = (resp.get("result") or {}).get("list") or []
out = []
for r in rows:
out.append(
{
"symbol": r.get("symbol"),
"side": r.get("side"),
"size": _f(r.get("size")),
"entry_price": _f(r.get("avgPrice")),
"unrealized_pnl": _f(r.get("unrealisedPnl")),
"leverage": _f(r.get("leverage")),
"liquidation_price": _f(r.get("liqPrice")),
"position_value": _f(r.get("positionValue")),
}
)
return out
async def get_account_summary(self, account_type: str = "UNIFIED") -> dict:
resp = await self._request_signed(
"GET",
"/v5/account/wallet-balance",
params={"accountType": account_type},
)
rows = (resp.get("result") or {}).get("list") or []
if not rows:
return {"error": "no_account"}
a = rows[0]
coins = []
for c in a.get("coin") or []:
coins.append(
{
"coin": c.get("coin"),
"wallet_balance": _f(c.get("walletBalance")),
"equity": _f(c.get("equity")),
}
)
return {
"account_type": a.get("accountType"),
"equity": _f(a.get("totalEquity")),
"wallet_balance": _f(a.get("totalWalletBalance")),
"margin_balance": _f(a.get("totalMarginBalance")),
"available_balance": _f(a.get("totalAvailableBalance")),
"unrealized_pnl": _f(a.get("totalPerpUPL")),
"coins": coins,
}
async def get_trade_history(
self, category: str = "linear", limit: int = 50
) -> list[dict]:
resp = await self._request_signed(
"GET",
"/v5/execution/list",
params={"category": category, "limit": limit},
)
rows = (resp.get("result") or {}).get("list") or []
return [
{
"symbol": r.get("symbol"),
"side": r.get("side"),
"size": _f(r.get("execQty")),
"price": _f(r.get("execPrice")),
"fee": _f(r.get("execFee")),
"timestamp": _i(r.get("execTime")),
"order_id": r.get("orderId"),
}
for r in rows
]
async def get_open_orders(
self,
category: str = "linear",
symbol: str | None = None,
settle_coin: str = "USDT",
) -> list[dict]:
params: dict[str, Any] = {"category": category}
if category in ("linear", "inverse") and not symbol:
params["settleCoin"] = settle_coin
if symbol:
params["symbol"] = symbol
resp = await self._request_signed(
"GET", "/v5/order/realtime", params=params
)
rows = (resp.get("result") or {}).get("list") or []
return [
{
"order_id": r.get("orderId"),
"symbol": r.get("symbol"),
"side": r.get("side"),
"qty": _f(r.get("qty")),
"price": _f(r.get("price")),
"type": r.get("orderType"),
"status": r.get("orderStatus"),
"reduce_only": bool(r.get("reduceOnly")),
}
for r in rows
]
# ── microstructure / basis ─────────────────────────────────
async def get_orderbook_imbalance(
self,
symbol: str,
category: str = "linear",
depth: int = 10,
) -> dict:
ob = await self.get_orderbook(
symbol=symbol, category=category, limit=max(depth, 50)
)
result = micro.orderbook_imbalance(
ob.get("bids") or [], ob.get("asks") or [], depth=depth
)
return {
"symbol": symbol,
"category": category,
"depth": depth,
**result,
"timestamp": ob.get("timestamp"),
}
async def get_basis_term_structure(self, asset: str) -> dict:
import datetime as _dt
asset = asset.upper()
spot = await self.get_ticker(f"{asset}USDT", category="spot")
perp = await self.get_ticker(f"{asset}USDT", category="linear")
sp = spot.get("last_price")
pp = perp.get("last_price")
instr = await self.get_instruments(category="linear")
items = instr.get("instruments") or []
futures = [
x
for x in items
if x.get("symbol", "").startswith(f"{asset}-")
or x.get("symbol", "").startswith(f"{asset}USDT-")
]
rows: list[dict[str, Any]] = []
if sp:
now_ms = int(_dt.datetime.now(_dt.UTC).timestamp() * 1000)
for f in futures[:10]:
tk = await self.get_ticker(f["symbol"], category="linear")
fp = tk.get("last_price")
expiry_ms = f.get("delivery_time")
if not fp or not expiry_ms:
continue
days = max((int(expiry_ms) - now_ms) / 86_400_000, 1)
basis_pct = 100.0 * (fp - sp) / sp
annualized = basis_pct * 365.0 / days
rows.append(
{
"symbol": f["symbol"],
"expiry_ms": int(expiry_ms),
"days_to_expiry": round(days, 2),
"future_price": fp,
"basis_pct": round(basis_pct, 4),
"annualized_basis_pct": round(annualized, 4),
}
)
rows.sort(key=lambda r: r["days_to_expiry"])
return {
"asset": asset,
"spot_price": sp,
"perp_price": pp,
"perp_basis_pct": round(100.0 * (pp - sp) / sp, 4)
if (sp and pp)
else None,
"term_structure": rows,
"data_timestamp": _dt.datetime.now(_dt.UTC).isoformat(),
}
async def get_basis_spot_perp(self, asset: str) -> dict:
asset = asset.upper()
symbol = f"{asset}USDT"
spot = await self.get_ticker(symbol, category="spot")
perp = await self.get_ticker(symbol, category="linear")
sp = spot.get("last_price")
pp = perp.get("last_price")
basis_abs = basis_pct = None
if sp and pp:
basis_abs = pp - sp
basis_pct = 100.0 * basis_abs / sp
return {
"asset": asset,
"symbol": symbol,
"spot_price": sp,
"perp_price": pp,
"basis_abs": basis_abs,
"basis_pct": basis_pct,
"funding_rate": perp.get("funding_rate"),
}
# ── trading (signed, write) ────────────────────────────────
async def place_order(
self,
category: str,
symbol: str,
side: str,
qty: float,
order_type: str = "Limit",
price: float | None = None,
tif: str = "GTC",
reduce_only: bool = False,
position_idx: int | None = None,
) -> dict:
body: dict[str, Any] = {
"category": category,
"symbol": symbol,
"side": side,
"qty": str(qty),
"orderType": order_type,
"timeInForce": tif,
"reduceOnly": reduce_only,
}
if price is not None:
body["price"] = str(price)
if position_idx is not None:
body["positionIdx"] = position_idx
if category == "option":
body["orderLinkId"] = f"cerbero-{uuid.uuid4().hex[:16]}"
resp = await self._request_signed("POST", "/v5/order/create", body=body)
r = resp.get("result") or {}
return self._envelope(
resp,
{
"order_id": r.get("orderId"),
"order_link_id": r.get("orderLinkId"),
"status": "submitted",
},
)
async def place_combo_order(
self,
category: str,
legs: list[dict[str, Any]],
) -> dict:
if category != "option":
raise ValueError(
"place_combo_order: Bybit batch_order è disponibile solo su category='option'"
)
if len(legs) < 2:
raise ValueError("combo requires at least 2 legs")
request: list[dict[str, Any]] = []
for leg in legs:
entry: dict[str, Any] = {
"symbol": leg["symbol"],
"side": leg["side"],
"qty": str(leg["qty"]),
"orderType": leg.get("order_type", "Limit"),
"timeInForce": leg.get("tif", "GTC"),
"reduceOnly": leg.get("reduce_only", False),
"orderLinkId": f"cerbero-{uuid.uuid4().hex[:16]}",
}
if leg.get("price") is not None:
entry["price"] = str(leg["price"])
request.append(entry)
body = {"category": category, "request": request}
resp = await self._request_signed(
"POST", "/v5/order/create-batch", body=body
)
result_list = (resp.get("result") or {}).get("list") or []
orders = [
{
"order_id": r.get("orderId"),
"order_link_id": r.get("orderLinkId"),
"status": "submitted",
}
for r in result_list
]
return self._envelope(resp, {"orders": orders})
async def amend_order(
self,
category: str,
symbol: str,
order_id: str,
new_qty: float | None = None,
new_price: float | None = None,
) -> dict:
body: dict[str, Any] = {
"category": category,
"symbol": symbol,
"orderId": order_id,
}
if new_qty is not None:
body["qty"] = str(new_qty)
if new_price is not None:
body["price"] = str(new_price)
resp = await self._request_signed("POST", "/v5/order/amend", body=body)
r = resp.get("result") or {}
return self._envelope(
resp,
{
"order_id": r.get("orderId", order_id),
"status": "amended",
},
)
async def cancel_order(self, category: str, symbol: str, order_id: str) -> dict:
body = {"category": category, "symbol": symbol, "orderId": order_id}
resp = await self._request_signed("POST", "/v5/order/cancel", body=body)
r = resp.get("result") or {}
return self._envelope(
resp,
{
"order_id": r.get("orderId", order_id),
"status": "cancelled",
},
)
async def cancel_all_orders(
self, category: str, symbol: str | None = None
) -> dict:
body: dict[str, Any] = {"category": category}
if symbol:
body["symbol"] = symbol
resp = await self._request_signed(
"POST", "/v5/order/cancel-all", body=body
)
r = resp.get("result") or {}
ids = [x.get("orderId") for x in (r.get("list") or [])]
return self._envelope(
resp,
{
"cancelled_ids": ids,
"count": len(ids),
},
)
async def set_stop_loss(
self,
category: str,
symbol: str,
stop_loss: float,
position_idx: int = 0,
) -> dict:
body = {
"category": category,
"symbol": symbol,
"stopLoss": str(stop_loss),
"positionIdx": position_idx,
}
resp = await self._request_signed(
"POST", "/v5/position/trading-stop", body=body
)
return self._envelope(
resp,
{
"symbol": symbol,
"stop_loss": stop_loss,
"status": "stop_loss_set",
},
)
async def set_take_profit(
self,
category: str,
symbol: str,
take_profit: float,
position_idx: int = 0,
) -> dict:
body = {
"category": category,
"symbol": symbol,
"takeProfit": str(take_profit),
"positionIdx": position_idx,
}
resp = await self._request_signed(
"POST", "/v5/position/trading-stop", body=body
)
return self._envelope(
resp,
{
"symbol": symbol,
"take_profit": take_profit,
"status": "take_profit_set",
},
)
async def close_position(self, category: str, symbol: str) -> dict:
positions = await self.get_positions(category=category)
target = next(
(p for p in positions if p["symbol"] == symbol and (p["size"] or 0) > 0),
None,
)
if not target:
return {"error": "no_open_position", "symbol": symbol}
close_side = "Sell" if target["side"] == "Buy" else "Buy"
return await self.place_order(
category=category,
symbol=symbol,
side=close_side,
qty=target["size"],
order_type="Market",
reduce_only=True,
tif="IOC",
)
async def set_leverage(
self, category: str, symbol: str, leverage: int
) -> dict:
body = {
"category": category,
"symbol": symbol,
"buyLeverage": str(leverage),
"sellLeverage": str(leverage),
}
resp = await self._request_signed(
"POST", "/v5/position/set-leverage", body=body
)
return self._envelope(
resp,
{
"symbol": symbol,
"leverage": leverage,
"status": "leverage_set",
},
)
async def switch_position_mode(
self, category: str, symbol: str, mode: str
) -> dict:
mode_code = 3 if mode.lower() == "hedge" else 0
body = {
"category": category,
"symbol": symbol,
"mode": mode_code,
}
resp = await self._request_signed(
"POST", "/v5/position/switch-mode", body=body
)
return self._envelope(
resp,
{
"symbol": symbol,
"mode": mode,
"status": "mode_switched",
},
)
async def transfer_asset(
self,
coin: str,
amount: float,
from_type: str,
to_type: str,
) -> dict:
body = {
"transferId": str(uuid.uuid4()),
"coin": coin,
"amount": str(amount),
"fromAccountType": from_type,
"toAccountType": to_type,
}
resp = await self._request_signed(
"POST", "/v5/asset/transfer/inter-transfer", body=body
)
r = resp.get("result") or {}
return self._envelope(
resp,
{
"transfer_id": r.get("transferId"),
"coin": coin,
"amount": amount,
"status": "submitted",
},
)
+56
View File
@@ -0,0 +1,56 @@
"""Leverage cap server-side per place_order.
Cap letto dal secret JSON via campo `max_leverage`. Default 1 (cash) se assente.
"""
from __future__ import annotations
from fastapi import HTTPException
def get_max_leverage(creds: dict) -> int:
"""Legge max_leverage dal secret. Default 1 se mancante."""
raw = creds.get("max_leverage", 1)
try:
value = int(raw)
except (TypeError, ValueError):
value = 1
return max(1, value)
def enforce_leverage(
requested: int | float | None,
*,
creds: dict,
exchange: str,
) -> int:
"""Verifica e applica leverage cap. Ritorna leverage applicabile.
Solleva HTTPException(403, LEVERAGE_CAP_EXCEEDED) se requested > cap.
Se requested is None, applica il cap come default.
"""
cap = get_max_leverage(creds)
if requested is None:
return cap
lev = int(requested)
if lev < 1:
raise HTTPException(
status_code=403,
detail={
"error": "LEVERAGE_CAP_EXCEEDED",
"exchange": exchange,
"requested": lev,
"max": cap,
"reason": "leverage must be >= 1",
},
)
if lev > cap:
raise HTTPException(
status_code=403,
detail={
"error": "LEVERAGE_CAP_EXCEEDED",
"exchange": exchange,
"requested": lev,
"max": cap,
},
)
return lev
+440
View File
@@ -0,0 +1,440 @@
"""Tool bybit V2: pydantic schemas + async functions.
Ogni funzione prende (client: BybitClient, params: <Req>) e restituisce
un dict (o un model Pydantic). Pure logica, no FastAPI dependency, no ACL.
L'autenticazione bearer è gestita dal middleware in cerbero_mcp.auth;
l'audit verrà cablato dal router via request.state.environment.
"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, Field
from cerbero_mcp.exchanges.bybit.client import BybitClient
from cerbero_mcp.exchanges.bybit.leverage_cap import (
enforce_leverage as _enforce_leverage,
)
from cerbero_mcp.exchanges.bybit.leverage_cap import get_max_leverage
# === Schemas: reads ===
class TickerReq(BaseModel):
symbol: str
category: str = "linear"
class TickerBatchReq(BaseModel):
symbols: list[str]
category: str = "linear"
class OrderbookReq(BaseModel):
symbol: str
category: str = "linear"
limit: int = 50
class HistoricalReq(BaseModel):
symbol: str
category: str = "linear"
interval: str = "60"
start: int | None = None
end: int | None = None
limit: int = 1000
class IndicatorsReq(BaseModel):
symbol: str
category: str = "linear"
indicators: list[str] = ["rsi", "atr", "macd", "adx"]
interval: str = "60"
start: int | None = None
end: int | None = None
class FundingRateReq(BaseModel):
symbol: str
category: str = "linear"
class FundingHistoryReq(BaseModel):
symbol: str
category: str = "linear"
limit: int = 100
class OpenInterestReq(BaseModel):
symbol: str
category: str = "linear"
interval: str = "5min"
limit: int = 288
class InstrumentsReq(BaseModel):
category: str = "linear"
symbol: str | None = None
class OptionChainReq(BaseModel):
base_coin: str
expiry: str | None = None
class PositionsReq(BaseModel):
category: str = "linear"
class AccountSummaryReq(BaseModel):
pass
class TradeHistoryReq(BaseModel):
category: str = "linear"
limit: int = 50
class OpenOrdersReq(BaseModel):
category: str = "linear"
symbol: str | None = None
class BasisSpotPerpReq(BaseModel):
asset: str
class OrderbookImbalanceReq(BaseModel):
symbol: str
category: str = "linear"
depth: int = 10
class BasisTermStructureReq(BaseModel):
asset: str
# === Schemas: writes ===
class PlaceOrderReq(BaseModel):
category: str
symbol: str
side: str
qty: float
order_type: str = "Limit"
price: float | None = None
tif: str = "GTC"
reduce_only: bool = False
position_idx: int | None = None
model_config = {
"json_schema_extra": {
"examples": [
{
"summary": "Market buy 0.01 BTCUSDT linear perp",
"value": {
"category": "linear",
"symbol": "BTCUSDT",
"side": "Buy",
"qty": 0.01,
"order_type": "Market",
},
}
]
}
}
class ComboLegReq(BaseModel):
symbol: str
side: str
qty: float
order_type: str = "Limit"
price: float | None = None
tif: str = "GTC"
reduce_only: bool = False
class PlaceComboOrderReq(BaseModel):
category: str = "option"
legs: list[ComboLegReq] = Field(..., min_length=2)
class AmendOrderReq(BaseModel):
category: str
symbol: str
order_id: str
new_qty: float | None = None
new_price: float | None = None
class CancelOrderReq(BaseModel):
category: str
symbol: str
order_id: str
class CancelAllReq(BaseModel):
category: str
symbol: str | None = None
class SetStopLossReq(BaseModel):
category: str
symbol: str
stop_loss: float
position_idx: int = 0
class SetTakeProfitReq(BaseModel):
category: str
symbol: str
take_profit: float
position_idx: int = 0
class ClosePositionReq(BaseModel):
category: str
symbol: str
class SetLeverageReq(BaseModel):
category: str
symbol: str
leverage: int
class SwitchModeReq(BaseModel):
category: str
symbol: str
mode: str
class TransferReq(BaseModel):
coin: str
amount: float
from_type: str
to_type: str
# === Tools (reads) ===
async def environment_info(
client: BybitClient, *, creds: dict, env_info: Any | None = None
) -> dict:
if env_info is None:
return {
"exchange": "bybit",
"environment": "testnet" if client.testnet else "mainnet",
"source": "credentials",
"env_value": None,
"base_url": getattr(client, "base_url", None),
"max_leverage": get_max_leverage(creds),
}
return {
"exchange": env_info.exchange,
"environment": env_info.environment,
"source": env_info.source,
"env_value": env_info.env_value,
"base_url": env_info.base_url,
"max_leverage": get_max_leverage(creds),
}
async def get_ticker(client: BybitClient, params: TickerReq) -> dict:
return await client.get_ticker(params.symbol, params.category)
async def get_ticker_batch(client: BybitClient, params: TickerBatchReq) -> dict:
return await client.get_ticker_batch(params.symbols, params.category)
async def get_orderbook(client: BybitClient, params: OrderbookReq) -> dict:
return await client.get_orderbook(params.symbol, params.category, params.limit)
async def get_historical(client: BybitClient, params: HistoricalReq) -> dict:
return await client.get_historical(
params.symbol,
params.category,
params.interval,
params.start,
params.end,
params.limit,
)
async def get_indicators(client: BybitClient, params: IndicatorsReq) -> dict:
return await client.get_indicators(
params.symbol,
params.category,
params.indicators,
params.interval,
params.start,
params.end,
)
async def get_funding_rate(client: BybitClient, params: FundingRateReq) -> dict:
return await client.get_funding_rate(params.symbol, params.category)
async def get_funding_history(client: BybitClient, params: FundingHistoryReq) -> dict:
return await client.get_funding_history(
params.symbol, params.category, params.limit
)
async def get_open_interest(client: BybitClient, params: OpenInterestReq) -> dict:
return await client.get_open_interest(
params.symbol, params.category, params.interval, params.limit
)
async def get_instruments(client: BybitClient, params: InstrumentsReq) -> dict:
return await client.get_instruments(params.category, params.symbol)
async def get_option_chain(client: BybitClient, params: OptionChainReq) -> dict:
return await client.get_option_chain(params.base_coin, params.expiry)
async def get_positions(client: BybitClient, params: PositionsReq) -> dict:
return {"positions": await client.get_positions(params.category)}
async def get_account_summary(
client: BybitClient, params: AccountSummaryReq
) -> dict:
return await client.get_account_summary()
async def get_trade_history(client: BybitClient, params: TradeHistoryReq) -> dict:
return {
"trades": await client.get_trade_history(params.category, params.limit)
}
async def get_open_orders(client: BybitClient, params: OpenOrdersReq) -> dict:
return {
"orders": await client.get_open_orders(params.category, params.symbol)
}
async def get_basis_spot_perp(client: BybitClient, params: BasisSpotPerpReq) -> dict:
return await client.get_basis_spot_perp(params.asset)
async def get_orderbook_imbalance(
client: BybitClient, params: OrderbookImbalanceReq
) -> dict:
return await client.get_orderbook_imbalance(
params.symbol, params.category, params.depth
)
async def get_basis_term_structure(
client: BybitClient, params: BasisTermStructureReq
) -> dict:
return await client.get_basis_term_structure(params.asset)
# === Tools (writes) ===
async def place_order(
client: BybitClient, params: PlaceOrderReq, *, creds: dict
) -> dict:
# Bybit non ha leverage_cap parametro per place_order; cap applicato a set_leverage.
result = await client.place_order(
category=params.category,
symbol=params.symbol,
side=params.side,
qty=params.qty,
order_type=params.order_type,
price=params.price,
tif=params.tif,
reduce_only=params.reduce_only,
position_idx=params.position_idx,
)
return result
async def place_combo_order(
client: BybitClient, params: PlaceComboOrderReq, *, creds: dict
) -> dict:
result = await client.place_combo_order(
category=params.category,
legs=[leg.model_dump() for leg in params.legs],
)
return result
async def amend_order(client: BybitClient, params: AmendOrderReq) -> dict:
result = await client.amend_order(
params.category,
params.symbol,
params.order_id,
params.new_qty,
params.new_price,
)
return result
async def cancel_order(client: BybitClient, params: CancelOrderReq) -> dict:
result = await client.cancel_order(
params.category, params.symbol, params.order_id
)
return result
async def cancel_all_orders(client: BybitClient, params: CancelAllReq) -> dict:
result = await client.cancel_all_orders(params.category, params.symbol)
return result
async def set_stop_loss(client: BybitClient, params: SetStopLossReq) -> dict:
result = await client.set_stop_loss(
params.category, params.symbol, params.stop_loss, params.position_idx
)
return result
async def set_take_profit(client: BybitClient, params: SetTakeProfitReq) -> dict:
result = await client.set_take_profit(
params.category, params.symbol, params.take_profit, params.position_idx
)
return result
async def close_position(client: BybitClient, params: ClosePositionReq) -> dict:
result = await client.close_position(params.category, params.symbol)
return result
async def set_leverage(
client: BybitClient, params: SetLeverageReq, *, creds: dict
) -> dict:
_enforce_leverage(params.leverage, creds=creds, exchange="bybit")
result = await client.set_leverage(
params.category, params.symbol, params.leverage
)
return result
async def switch_position_mode(
client: BybitClient, params: SwitchModeReq
) -> dict:
result = await client.switch_position_mode(
params.category, params.symbol, params.mode
)
return result
async def transfer_asset(client: BybitClient, params: TransferReq) -> dict:
result = await client.transfer_asset(
params.coin, params.amount, params.from_type, params.to_type
)
return result
+222
View File
@@ -0,0 +1,222 @@
"""Router /mcp-alpaca/* — DI per env, client e (write) creds.
Mappa 1:1 i tool di `cerbero_mcp.exchanges.alpaca.tools` a endpoint
`POST /mcp-alpaca/tools/{tool_name}`. L'autenticazione bearer è gestita
dal middleware in `cerbero_mcp.auth`; qui leggiamo solo `request.state.environment`.
"""
from __future__ import annotations
from typing import Literal, cast
from fastapi import APIRouter, Depends, Request
from cerbero_mcp.client_registry import ClientRegistry
from cerbero_mcp.common.audit_helpers import audit_call
from cerbero_mcp.exchanges.alpaca import tools as t
from cerbero_mcp.exchanges.alpaca.client import AlpacaClient
Environment = Literal["testnet", "mainnet"]
def get_environment(request: Request) -> Environment:
return cast(Environment, request.state.environment)
async def get_alpaca_client(
request: Request, env: Environment = Depends(get_environment)
) -> AlpacaClient:
registry: ClientRegistry = request.app.state.registry
return cast(AlpacaClient, await registry.get("alpaca", env))
def _build_creds(request: Request) -> dict:
"""Costruisce dict `creds` minimale per leverage cap / metadata."""
settings = request.app.state.settings
return {
"max_leverage": settings.alpaca.max_leverage,
"api_key_id": settings.alpaca.api_key_id,
}
def make_router() -> APIRouter:
r = APIRouter(prefix="/mcp-alpaca", tags=["alpaca"])
# === READ tools ===
@r.post("/tools/environment_info")
async def _environment_info(
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
creds = _build_creds(request)
return await t.environment_info(client, creds=creds)
@r.post("/tools/get_account")
async def _get_account(
params: t.GetAccountReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_account(client, params)
@r.post("/tools/get_positions")
async def _get_positions(
params: t.GetPositionsReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_positions(client, params)
@r.post("/tools/get_activities")
async def _get_activities(
params: t.GetActivitiesReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_activities(client, params)
@r.post("/tools/get_assets")
async def _get_assets(
params: t.GetAssetsReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_assets(client, params)
@r.post("/tools/get_ticker")
async def _get_ticker(
params: t.GetTickerReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_ticker(client, params)
@r.post("/tools/get_bars")
async def _get_bars(
params: t.GetBarsReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_bars(client, params)
@r.post("/tools/get_snapshot")
async def _get_snapshot(
params: t.GetSnapshotReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_snapshot(client, params)
@r.post("/tools/get_option_chain")
async def _get_option_chain(
params: t.GetOptionChainReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_option_chain(client, params)
@r.post("/tools/get_open_orders")
async def _get_open_orders(
params: t.GetOpenOrdersReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_open_orders(client, params)
@r.post("/tools/get_clock")
async def _get_clock(
params: t.GetClockReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_clock(client, params)
@r.post("/tools/get_calendar")
async def _get_calendar(
params: t.GetCalendarReq,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await t.get_calendar(client, params)
# === WRITE tools ===
@r.post("/tools/place_order")
async def _place_order(
params: t.PlaceOrderReq,
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
creds = _build_creds(request)
return await audit_call(
request=request,
exchange="alpaca",
action="place_order",
target_field="symbol",
params=params,
tool_fn=lambda: t.place_order(client, params, creds=creds),
)
@r.post("/tools/amend_order")
async def _amend_order(
params: t.AmendOrderReq,
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await audit_call(
request=request,
exchange="alpaca",
action="amend_order",
target_field="order_id",
params=params,
tool_fn=lambda: t.amend_order(client, params),
)
@r.post("/tools/cancel_order")
async def _cancel_order(
params: t.CancelOrderReq,
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await audit_call(
request=request,
exchange="alpaca",
action="cancel_order",
target_field="order_id",
params=params,
tool_fn=lambda: t.cancel_order(client, params),
)
@r.post("/tools/cancel_all_orders")
async def _cancel_all_orders(
params: t.CancelAllOrdersReq,
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await audit_call(
request=request,
exchange="alpaca",
action="cancel_all_orders",
params=params,
tool_fn=lambda: t.cancel_all_orders(client, params),
)
@r.post("/tools/close_position")
async def _close_position(
params: t.ClosePositionReq,
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await audit_call(
request=request,
exchange="alpaca",
action="close_position",
target_field="symbol",
params=params,
tool_fn=lambda: t.close_position(client, params),
)
@r.post("/tools/close_all_positions")
async def _close_all_positions(
params: t.CloseAllPositionsReq,
request: Request,
client: AlpacaClient = Depends(get_alpaca_client),
):
return await audit_call(
request=request,
exchange="alpaca",
action="close_all_positions",
params=params,
tool_fn=lambda: t.close_all_positions(client, params),
)
return r
+346
View File
@@ -0,0 +1,346 @@
"""Router /mcp-bybit/* — DI per env, client e (write) creds.
Mappa 1:1 i tool di `cerbero_mcp.exchanges.bybit.tools` a endpoint
`POST /mcp-bybit/tools/{tool_name}`. L'autenticazione bearer è gestita
dal middleware in `cerbero_mcp.auth`; qui leggiamo solo `request.state.environment`.
"""
from __future__ import annotations
from typing import Literal, cast
from fastapi import APIRouter, Depends, Request
from cerbero_mcp.client_registry import ClientRegistry
from cerbero_mcp.common.audit_helpers import audit_call
from cerbero_mcp.exchanges.bybit import tools as t
from cerbero_mcp.exchanges.bybit.client import BybitClient
Environment = Literal["testnet", "mainnet"]
def get_environment(request: Request) -> Environment:
return cast(Environment, request.state.environment)
async def get_bybit_client(
request: Request, env: Environment = Depends(get_environment)
) -> BybitClient:
registry: ClientRegistry = request.app.state.registry
return cast(BybitClient, await registry.get("bybit", env))
def _build_creds(request: Request) -> dict:
"""Costruisce dict `creds` minimale per leverage cap / metadata.
Le credenziali vere sono già iniettate nel client da ClientRegistry;
qui passiamo solo il cap di leverage e l'api_key (metadata audit).
"""
settings = request.app.state.settings
return {
"max_leverage": settings.bybit.max_leverage,
"api_key": settings.bybit.api_key,
}
def make_router() -> APIRouter:
r = APIRouter(prefix="/mcp-bybit", tags=["bybit"])
# === READ tools ===
@r.post("/tools/environment_info")
async def _environment_info(
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
creds = _build_creds(request)
return await t.environment_info(client, creds=creds)
@r.post("/tools/get_ticker")
async def _get_ticker(
params: t.TickerReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_ticker(client, params)
@r.post("/tools/get_ticker_batch")
async def _get_ticker_batch(
params: t.TickerBatchReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_ticker_batch(client, params)
@r.post("/tools/get_orderbook")
async def _get_orderbook(
params: t.OrderbookReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_orderbook(client, params)
@r.post("/tools/get_historical")
async def _get_historical(
params: t.HistoricalReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_historical(client, params)
@r.post("/tools/get_indicators")
async def _get_indicators(
params: t.IndicatorsReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_indicators(client, params)
@r.post("/tools/get_funding_rate")
async def _get_funding_rate(
params: t.FundingRateReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_funding_rate(client, params)
@r.post("/tools/get_funding_history")
async def _get_funding_history(
params: t.FundingHistoryReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_funding_history(client, params)
@r.post("/tools/get_open_interest")
async def _get_open_interest(
params: t.OpenInterestReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_open_interest(client, params)
@r.post("/tools/get_instruments")
async def _get_instruments(
params: t.InstrumentsReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_instruments(client, params)
@r.post("/tools/get_option_chain")
async def _get_option_chain(
params: t.OptionChainReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_option_chain(client, params)
@r.post("/tools/get_positions")
async def _get_positions(
params: t.PositionsReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_positions(client, params)
@r.post("/tools/get_account_summary")
async def _get_account_summary(
params: t.AccountSummaryReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_account_summary(client, params)
@r.post("/tools/get_trade_history")
async def _get_trade_history(
params: t.TradeHistoryReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_trade_history(client, params)
@r.post("/tools/get_open_orders")
async def _get_open_orders(
params: t.OpenOrdersReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_open_orders(client, params)
@r.post("/tools/get_basis_spot_perp")
async def _get_basis_spot_perp(
params: t.BasisSpotPerpReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_basis_spot_perp(client, params)
@r.post("/tools/get_orderbook_imbalance")
async def _get_orderbook_imbalance(
params: t.OrderbookImbalanceReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_orderbook_imbalance(client, params)
@r.post("/tools/get_basis_term_structure")
async def _get_basis_term_structure(
params: t.BasisTermStructureReq,
client: BybitClient = Depends(get_bybit_client),
):
return await t.get_basis_term_structure(client, params)
# === WRITE tools (richiedono creds per leverage cap / audit) ===
@r.post("/tools/place_order")
async def _place_order(
params: t.PlaceOrderReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
creds = _build_creds(request)
return await audit_call(
request=request,
exchange="bybit",
action="place_order",
target_field="symbol",
params=params,
tool_fn=lambda: t.place_order(client, params, creds=creds),
)
@r.post("/tools/place_combo_order")
async def _place_combo_order(
params: t.PlaceComboOrderReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
creds = _build_creds(request)
return await audit_call(
request=request,
exchange="bybit",
action="place_combo_order",
params=params,
tool_fn=lambda: t.place_combo_order(client, params, creds=creds),
)
@r.post("/tools/amend_order")
async def _amend_order(
params: t.AmendOrderReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="amend_order",
target_field="symbol",
params=params,
tool_fn=lambda: t.amend_order(client, params),
)
@r.post("/tools/cancel_order")
async def _cancel_order(
params: t.CancelOrderReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="cancel_order",
target_field="order_id",
params=params,
tool_fn=lambda: t.cancel_order(client, params),
)
@r.post("/tools/cancel_all_orders")
async def _cancel_all_orders(
params: t.CancelAllReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="cancel_all_orders",
target_field="symbol",
params=params,
tool_fn=lambda: t.cancel_all_orders(client, params),
)
@r.post("/tools/set_stop_loss")
async def _set_stop_loss(
params: t.SetStopLossReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="set_stop_loss",
target_field="symbol",
params=params,
tool_fn=lambda: t.set_stop_loss(client, params),
)
@r.post("/tools/set_take_profit")
async def _set_take_profit(
params: t.SetTakeProfitReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="set_take_profit",
target_field="symbol",
params=params,
tool_fn=lambda: t.set_take_profit(client, params),
)
@r.post("/tools/close_position")
async def _close_position(
params: t.ClosePositionReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="close_position",
target_field="symbol",
params=params,
tool_fn=lambda: t.close_position(client, params),
)
@r.post("/tools/set_leverage")
async def _set_leverage(
params: t.SetLeverageReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
creds = _build_creds(request)
return await audit_call(
request=request,
exchange="bybit",
action="set_leverage",
target_field="symbol",
params=params,
tool_fn=lambda: t.set_leverage(client, params, creds=creds),
)
@r.post("/tools/switch_position_mode")
async def _switch_position_mode(
params: t.SwitchModeReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="switch_position_mode",
target_field="symbol",
params=params,
tool_fn=lambda: t.switch_position_mode(client, params),
)
@r.post("/tools/transfer_asset")
async def _transfer_asset(
params: t.TransferReq,
request: Request,
client: BybitClient = Depends(get_bybit_client),
):
return await audit_call(
request=request,
exchange="bybit",
action="transfer_asset",
target_field="coin",
params=params,
tool_fn=lambda: t.transfer_asset(client, params),
)
return r
View File
+23
View File
@@ -0,0 +1,23 @@
from __future__ import annotations
import pytest
from cerbero_mcp.exchanges.alpaca.client import AlpacaClient
@pytest.fixture
async def client():
"""AlpacaClient paper su httpx mock (gestito da pytest-httpx)."""
c = AlpacaClient(api_key="test_key", secret_key="test_secret", paper=True)
try:
yield c
finally:
await c.aclose()
@pytest.fixture
async def client_live():
c = AlpacaClient(api_key="test_key", secret_key="test_secret", paper=False)
try:
yield c
finally:
await c.aclose()
+417
View File
@@ -0,0 +1,417 @@
"""Test AlpacaClient httpx-based (V2.0.0).
Mockano gli endpoint REST tramite pytest-httpx. Coprono account/positions,
ordini (place/cancel/limit-error), close position, clock, asset class
invalida → ValueError.
"""
from __future__ import annotations
import re
import pytest
from cerbero_mcp.exchanges.alpaca.client import AlpacaClient
from pytest_httpx import HTTPXMock
PAPER = "https://paper-api.alpaca.markets"
DATA = "https://data.alpaca.markets"
@pytest.mark.asyncio
async def test_init_paper_mode(client: AlpacaClient):
assert client.paper is True
assert client.base_url is None
assert client._trading_base == PAPER
@pytest.mark.asyncio
async def test_init_live_mode(client_live: AlpacaClient):
assert client_live.paper is False
assert client_live._trading_base == "https://api.alpaca.markets"
@pytest.mark.asyncio
async def test_init_base_url_override():
c = AlpacaClient(
api_key="k",
secret_key="s",
paper=True,
base_url="https://alpaca-custom.example.com",
)
try:
assert c.base_url == "https://alpaca-custom.example.com"
assert c._trading_base == "https://alpaca-custom.example.com"
# Data endpoint NON viene overridato
assert c._data_base == DATA
finally:
await c.aclose()
@pytest.mark.asyncio
async def test_get_account(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=f"{PAPER}/v2/account",
json={"id": "abc", "equity": "100000.00", "buying_power": "200000.00"},
)
result = await client.get_account()
assert result["id"] == "abc"
assert result["equity"] == "100000.00"
@pytest.mark.asyncio
async def test_get_account_sends_auth_headers(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(url=f"{PAPER}/v2/account", json={"id": "x"})
await client.get_account()
req = httpx_mock.get_requests()[0]
assert req.headers["APCA-API-KEY-ID"] == "test_key"
assert req.headers["APCA-API-SECRET-KEY"] == "test_secret"
assert req.headers["Accept"] == "application/json"
@pytest.mark.asyncio
async def test_get_positions_returns_list(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(
url=f"{PAPER}/v2/positions",
json=[{"symbol": "AAPL", "qty": "10", "side": "long"}],
)
result = await client.get_positions()
assert len(result) == 1
assert result[0]["symbol"] == "AAPL"
@pytest.mark.asyncio
async def test_place_market_order_stocks(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(
method="POST",
url=f"{PAPER}/v2/orders",
json={"id": "o123", "symbol": "AAPL", "status": "accepted"},
)
result = await client.place_order(
symbol="AAPL", side="buy", qty=1, order_type="market", asset_class="stocks"
)
assert result["id"] == "o123"
# body POST corretto
req = httpx_mock.get_requests()[0]
import json as _j
body = _j.loads(req.content)
assert body["symbol"] == "AAPL"
assert body["side"] == "buy"
assert body["type"] == "market"
assert body["qty"] == "1"
assert body["time_in_force"] == "day"
@pytest.mark.asyncio
async def test_place_limit_order_requires_price(client: AlpacaClient):
with pytest.raises(ValueError, match="limit_price"):
await client.place_order(
symbol="AAPL", side="buy", qty=1, order_type="limit"
)
@pytest.mark.asyncio
async def test_place_stop_order_requires_price(client: AlpacaClient):
with pytest.raises(ValueError, match="stop_price"):
await client.place_order(
symbol="AAPL", side="buy", qty=1, order_type="stop"
)
@pytest.mark.asyncio
async def test_place_unsupported_order_type(client: AlpacaClient):
with pytest.raises(ValueError, match="unsupported order_type"):
await client.place_order(
symbol="AAPL", side="buy", qty=1, order_type="trailing_stop"
)
@pytest.mark.asyncio
async def test_cancel_order(httpx_mock: HTTPXMock, client: AlpacaClient):
# 204 No Content su success
httpx_mock.add_response(
method="DELETE",
url=f"{PAPER}/v2/orders/o1",
status_code=204,
)
result = await client.cancel_order("o1")
assert result == {"order_id": "o1", "canceled": True}
@pytest.mark.asyncio
async def test_cancel_all_orders(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
method="DELETE",
url=f"{PAPER}/v2/orders",
json=[
{"id": "a", "status": 200},
{"id": "b", "status": 200},
],
)
result = await client.cancel_all_orders()
assert len(result) == 2
@pytest.mark.asyncio
async def test_close_position_no_options(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(
method="DELETE",
url=f"{PAPER}/v2/positions/AAPL",
json={"id": "close-1", "symbol": "AAPL"},
)
result = await client.close_position("AAPL")
assert result["id"] == "close-1"
@pytest.mark.asyncio
async def test_close_position_with_qty(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(
method="DELETE",
url=f"{PAPER}/v2/positions/AAPL?qty=5.0",
json={"id": "close-2"},
)
result = await client.close_position("AAPL", qty=5.0)
assert result["id"] == "close-2"
@pytest.mark.asyncio
async def test_close_all_positions(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(
method="DELETE",
url=f"{PAPER}/v2/positions?cancel_orders=true",
json=[{"symbol": "AAPL", "status": 200}],
)
result = await client.close_all_positions(cancel_orders=True)
assert len(result) == 1
assert result[0]["symbol"] == "AAPL"
@pytest.mark.asyncio
async def test_get_clock(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=f"{PAPER}/v2/clock",
json={"is_open": True, "next_close": "2026-04-21T20:00:00Z"},
)
result = await client.get_clock()
assert result["is_open"] is True
@pytest.mark.asyncio
async def test_invalid_asset_class(client: AlpacaClient):
with pytest.raises(ValueError, match="invalid asset_class"):
await client.get_ticker("AAPL", asset_class="forex")
@pytest.mark.asyncio
async def test_get_ticker_stocks(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=f"{DATA}/v2/stocks/AAPL/trades/latest",
json={
"symbol": "AAPL",
"trade": {"p": 175.50, "s": 100, "t": "2026-04-18T15:30:00Z"},
},
)
httpx_mock.add_response(
url=f"{DATA}/v2/stocks/AAPL/quotes/latest",
json={
"symbol": "AAPL",
"quote": {
"bp": 175.40,
"ap": 175.55,
"bs": 50,
"as": 25,
"t": "2026-04-18T15:30:00Z",
},
},
)
result = await client.get_ticker("AAPL", asset_class="stocks")
assert result["asset_class"] == "stocks"
assert result["last_price"] == 175.50
assert result["bid"] == 175.40
assert result["ask"] == 175.55
@pytest.mark.asyncio
async def test_get_bars_stocks(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{DATA}/v2/stocks/bars\?.*"),
json={
"bars": {
"AAPL": [
{
"t": "2026-04-17T00:00:00Z",
"o": 170.0,
"h": 176.0,
"l": 169.5,
"c": 175.0,
"v": 1000000,
}
]
}
},
)
result = await client.get_bars(
symbol="AAPL",
asset_class="stocks",
interval="1d",
start="2026-04-17T00:00:00",
end="2026-04-18T00:00:00",
limit=10,
)
assert result["symbol"] == "AAPL"
assert result["interval"] == "1d"
assert len(result["candles"]) == 1
assert result["candles"][0]["close"] == 175.0
@pytest.mark.asyncio
async def test_get_bars_unsupported_timeframe(client: AlpacaClient):
with pytest.raises(ValueError, match="unsupported timeframe"):
await client.get_bars(
symbol="AAPL",
asset_class="stocks",
interval="3min",
)
@pytest.mark.asyncio
async def test_get_bars_invalid_asset_class(client: AlpacaClient):
with pytest.raises(ValueError, match="invalid asset_class"):
await client.get_bars(symbol="AAPL", asset_class="forex")
@pytest.mark.asyncio
async def test_get_assets(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{PAPER}/v2/assets\?.*"),
json=[
{"symbol": "AAPL", "tradable": True, "class": "us_equity"},
{"symbol": "GOOG", "tradable": True, "class": "us_equity"},
],
)
result = await client.get_assets(asset_class="stocks", status="active")
assert len(result) == 2
assert result[0]["symbol"] == "AAPL"
@pytest.mark.asyncio
async def test_get_assets_invalid_class(client: AlpacaClient):
with pytest.raises(ValueError, match="invalid asset_class"):
await client.get_assets(asset_class="forex")
@pytest.mark.asyncio
async def test_get_open_orders(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{PAPER}/v2/orders\?.*"),
json=[{"id": "o1", "status": "open", "symbol": "AAPL"}],
)
result = await client.get_open_orders(limit=10)
assert len(result) == 1
assert result[0]["id"] == "o1"
@pytest.mark.asyncio
async def test_amend_order(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
method="PATCH",
url=f"{PAPER}/v2/orders/o1",
json={"id": "o1", "qty": "5", "limit_price": "180.0"},
)
result = await client.amend_order(
"o1", qty=5, limit_price=180.0, tif="gtc"
)
assert result["id"] == "o1"
req = httpx_mock.get_requests()[0]
import json as _j
body = _j.loads(req.content)
assert body["qty"] == "5"
assert body["limit_price"] == "180.0"
assert body["time_in_force"] == "gtc"
@pytest.mark.asyncio
async def test_get_calendar(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{PAPER}/v2/calendar.*"),
json=[{"date": "2026-04-20", "open": "09:30", "close": "16:00"}],
)
result = await client.get_calendar(start="2026-04-20", end="2026-04-20")
assert len(result) == 1
assert result[0]["date"] == "2026-04-20"
@pytest.mark.asyncio
async def test_get_calendar_no_filters(
httpx_mock: HTTPXMock, client: AlpacaClient
):
httpx_mock.add_response(
url=f"{PAPER}/v2/calendar",
json=[{"date": "2026-04-20"}],
)
result = await client.get_calendar()
assert len(result) == 1
@pytest.mark.asyncio
async def test_get_snapshot(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{DATA}/v2/stocks/snapshots\?.*"),
json={
"AAPL": {
"latestTrade": {"p": 175.0},
"latestQuote": {"bp": 174.9, "ap": 175.1},
}
},
)
result = await client.get_snapshot("AAPL")
assert result["latestTrade"]["p"] == 175.0
@pytest.mark.asyncio
async def test_get_option_chain(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{DATA}/v1beta1/options/snapshots/AAPL.*"),
json={
"snapshots": {
"AAPL250620C00200000": {
"latestQuote": {"bp": 1.20, "ap": 1.30}
}
}
},
)
result = await client.get_option_chain("AAPL", expiry="2026-06-20")
assert result["underlying"] == "AAPL"
assert result["expiry"] == "2026-06-20"
assert "AAPL250620C00200000" in result["contracts"]
@pytest.mark.asyncio
async def test_get_activities(httpx_mock: HTTPXMock, client: AlpacaClient):
httpx_mock.add_response(
url=re.compile(rf"^{PAPER}/v2/account/activities.*"),
json=[
{"id": "1", "activity_type": "FILL"},
{"id": "2", "activity_type": "TRANS"},
],
)
result = await client.get_activities(limit=10)
assert len(result) == 2
@pytest.mark.asyncio
async def test_aclose_idempotent(client: AlpacaClient):
await client.aclose()
await client.aclose() # nessun raise
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
import pytest
from cerbero_mcp.exchanges.alpaca.leverage_cap import enforce_leverage, get_max_leverage
from fastapi import HTTPException
def test_get_max_leverage_returns_creds_value():
creds = {"max_leverage": 4}
assert get_max_leverage(creds) == 4
def test_get_max_leverage_default_when_missing():
"""Default 1 (cash) se il secret non ha max_leverage."""
assert get_max_leverage({}) == 1
def test_enforce_leverage_pass_at_cap_one():
"""Alpaca cash account: cap 1, leverage 1 OK."""
creds = {"max_leverage": 1}
enforce_leverage(1, creds=creds, exchange="alpaca") # no raise
def test_enforce_leverage_reject_over_cap_one():
creds = {"max_leverage": 1}
with pytest.raises(HTTPException) as exc:
enforce_leverage(2, creds=creds, exchange="alpaca")
assert exc.value.status_code == 403
assert exc.value.detail["error"] == "LEVERAGE_CAP_EXCEEDED"
assert exc.value.detail["exchange"] == "alpaca"
assert exc.value.detail["requested"] == 2
assert exc.value.detail["max"] == 1
def test_enforce_leverage_reject_when_below_one():
creds = {"max_leverage": 1}
with pytest.raises(HTTPException) as exc:
enforce_leverage(0, creds=creds, exchange="alpaca")
assert exc.value.status_code == 403
def test_enforce_leverage_default_when_none():
"""Se requested è None, applica il cap come default."""
creds = {"max_leverage": 1}
result = enforce_leverage(None, creds=creds, exchange="alpaca")
assert result == 1
View File
+18
View File
@@ -0,0 +1,18 @@
from __future__ import annotations
import pytest
from cerbero_mcp.exchanges.bybit.client import BybitClient
@pytest.fixture
def client():
"""BybitClient con base_url testnet e AsyncClient interno.
pytest-httpx intercetta le chiamate dell'AsyncClient httpx creato dal
costruttore (auto-mock), quindi non serve injection esplicita.
"""
return BybitClient(
api_key="test_key",
api_secret="test_secret",
testnet=True,
)
+914
View File
@@ -0,0 +1,914 @@
"""Test del client Bybit V5 (httpx puro) con pytest-httpx.
Tutte le chiamate HTTP vengono intercettate via `httpx_mock`. Le request
non hanno bisogno di un AsyncClient iniettato: BybitClient costruisce
internamente `httpx.AsyncClient`, e pytest-httpx applica un MockTransport
globale.
"""
from __future__ import annotations
import json
import httpx
import pytest
from cerbero_mcp.exchanges.bybit.client import (
BASE_TESTNET,
BybitClient,
_f,
_i,
)
from pytest_httpx import HTTPXMock
# ── helpers / costanti ────────────────────────────────────────
def _url(path: str) -> str:
return BASE_TESTNET + path
def _last_request_json(httpx_mock: HTTPXMock) -> dict:
req = httpx_mock.get_requests()[-1]
return json.loads(req.content.decode("utf-8"))
# ── init / helpers basics ─────────────────────────────────────
def test_client_init_stores_attrs():
c = BybitClient(api_key="k", api_secret="s", testnet=True)
assert c.testnet is True
assert c.base_url == BASE_TESTNET
assert c.api_key == "k"
def test_client_init_mainnet_base_url():
c = BybitClient(api_key="k", api_secret="s", testnet=False)
assert c.testnet is False
assert c.base_url == "https://api.bybit.com"
def test_client_init_custom_base_url():
c = BybitClient(
api_key="k", api_secret="s", testnet=True, base_url="https://example.test"
)
assert c.base_url == "https://example.test"
def test_client_init_injected_http_not_owned():
http = httpx.AsyncClient()
c = BybitClient(api_key="k", api_secret="s", http=http)
assert c._http is http
assert c._owns_http is False
def test_parse_helpers():
assert _f("1.5") == 1.5
assert _f("") is None
assert _f(None) is None
assert _i("42") == 42
assert _i("") is None
assert _i(None) is None
async def test_aclose_owned_client_closes():
c = BybitClient(api_key="k", api_secret="s", testnet=True)
await c.aclose()
# An aclosed client cannot dispatch requests
assert c._http.is_closed is True
# ── market data (public) ──────────────────────────────────────
async def test_get_ticker(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=linear&symbol=BTCUSDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"lastPrice": "60000",
"markPrice": "60010",
"bid1Price": "59995",
"ask1Price": "60005",
"volume24h": "1500.5",
"turnover24h": "90000000",
"fundingRate": "0.0001",
"openInterest": "50000",
}
]
},
},
)
t = await client.get_ticker("BTCUSDT", category="linear")
assert t["symbol"] == "BTCUSDT"
assert t["last_price"] == 60000.0
assert t["mark_price"] == 60010.0
assert t["bid"] == 59995.0
assert t["ask"] == 60005.0
assert t["volume_24h"] == 1500.5
assert t["funding_rate"] == 0.0001
assert t["open_interest"] == 50000.0
async def test_get_ticker_batch(client, httpx_mock: HTTPXMock):
def _row(sym: str) -> dict:
return {
"symbol": sym,
"lastPrice": "1",
"markPrice": "1",
"bid1Price": "1",
"ask1Price": "1",
"volume24h": "0",
"turnover24h": "0",
"fundingRate": "0",
"openInterest": "0",
}
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=linear&symbol=BTCUSDT"),
json={"retCode": 0, "result": {"list": [_row("BTCUSDT")]}},
)
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=linear&symbol=ETHUSDT"),
json={"retCode": 0, "result": {"list": [_row("ETHUSDT")]}},
)
out = await client.get_ticker_batch(["BTCUSDT", "ETHUSDT"], category="linear")
assert set(out.keys()) == {"BTCUSDT", "ETHUSDT"}
async def test_get_ticker_not_found(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=linear&symbol=UNKNOWNUSDT"),
json={"retCode": 0, "result": {"list": []}},
)
t = await client.get_ticker("UNKNOWNUSDT", category="linear")
assert t == {"symbol": "UNKNOWNUSDT", "error": "not_found"}
async def test_get_orderbook(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/orderbook?category=linear&symbol=BTCUSDT&limit=25"),
json={
"retCode": 0,
"result": {
"s": "BTCUSDT",
"b": [["59990", "0.5"], ["59980", "1.0"]],
"a": [["60010", "0.3"], ["60020", "0.7"]],
"ts": 1700000000000,
},
},
)
ob = await client.get_orderbook("BTCUSDT", category="linear", limit=25)
assert ob["symbol"] == "BTCUSDT"
assert ob["bids"] == [[59990.0, 0.5], [59980.0, 1.0]]
assert ob["asks"] == [[60010.0, 0.3], [60020.0, 0.7]]
assert ob["timestamp"] == 1700000000000
async def test_get_historical(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url(
"/v5/market/kline?category=linear&symbol=BTCUSDT&interval=60&limit=1000"
"&start=1700000000000&end=1700003600000"
),
json={
"retCode": 0,
"result": {
"list": [
[
"1700000000000",
"60000",
"60500",
"59500",
"60200",
"100",
"6020000",
],
[
"1700003600000",
"60200",
"60700",
"60000",
"60400",
"80",
"4832000",
],
]
},
},
)
out = await client.get_historical(
"BTCUSDT",
category="linear",
interval="60",
start=1700000000000,
end=1700003600000,
)
assert len(out["candles"]) == 2
c0 = out["candles"][0]
assert c0["timestamp"] == 1700000000000
assert c0["open"] == 60000.0
assert c0["high"] == 60500.0
assert c0["low"] == 59500.0
assert c0["close"] == 60200.0
assert c0["volume"] == 100.0
async def test_get_indicators(client, httpx_mock: HTTPXMock):
rows = [
[
str(1700000000000 + i * 3600_000),
str(60000 + i * 10),
str(60000 + i * 10 + 5),
str(60000 + i * 10 - 5),
str(60000 + i * 10 + 2),
"100",
"6000000",
]
for i in range(35)
]
httpx_mock.add_response(
url=_url(
"/v5/market/kline?category=linear&symbol=BTCUSDT&interval=60&limit=1000"
),
json={"retCode": 0, "result": {"list": rows}},
)
out = await client.get_indicators(
"BTCUSDT",
category="linear",
indicators=["rsi", "atr", "macd", "adx"],
interval="60",
)
assert "rsi" in out and out["rsi"] is not None
assert "atr" in out and out["atr"] is not None
assert "macd" in out and out["macd"]["macd"] is not None
assert "adx" in out and out["adx"]["adx"] is not None
async def test_get_funding_rate(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=linear&symbol=BTCUSDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"fundingRate": "0.0001",
"nextFundingTime": "1700003600000",
"lastPrice": "60000",
"markPrice": "60000",
"bid1Price": "0",
"ask1Price": "0",
"volume24h": "0",
"turnover24h": "0",
"openInterest": "0",
}
]
},
},
)
out = await client.get_funding_rate("BTCUSDT", category="linear")
assert out["symbol"] == "BTCUSDT"
assert out["funding_rate"] == 0.0001
assert out["next_funding_time"] == 1700003600000
async def test_get_funding_history(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url(
"/v5/market/funding/history?category=linear&symbol=BTCUSDT&limit=50"
),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"fundingRate": "0.0001",
"fundingRateTimestamp": "1700000000000",
},
{
"symbol": "BTCUSDT",
"fundingRate": "0.00008",
"fundingRateTimestamp": "1699996400000",
},
]
},
},
)
out = await client.get_funding_history("BTCUSDT", category="linear", limit=50)
assert len(out["history"]) == 2
assert out["history"][0]["rate"] == 0.0001
async def test_get_open_interest(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url(
"/v5/market/open-interest?category=linear&symbol=BTCUSDT"
"&intervalTime=5min&limit=100"
),
json={
"retCode": 0,
"result": {
"list": [
{"openInterest": "50000", "timestamp": "1700000000000"},
{"openInterest": "49000", "timestamp": "1699996400000"},
]
},
},
)
out = await client.get_open_interest(
"BTCUSDT", category="linear", interval="5min", limit=100
)
assert len(out["points"]) == 2
assert out["current_oi"] == 50000.0
async def test_get_instruments(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/instruments-info?category=linear"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"status": "Trading",
"baseCoin": "BTC",
"quoteCoin": "USDT",
"priceFilter": {"tickSize": "0.1"},
"lotSizeFilter": {
"qtyStep": "0.001",
"minOrderQty": "0.001",
},
}
]
},
},
)
out = await client.get_instruments(category="linear")
assert len(out["instruments"]) == 1
inst = out["instruments"][0]
assert inst["symbol"] == "BTCUSDT"
assert inst["tick_size"] == 0.1
assert inst["qty_step"] == 0.001
async def test_get_option_chain(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/instruments-info?category=option&baseCoin=BTC"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTC-30JUN25-50000-C",
"baseCoin": "BTC",
"settleCoin": "USDC",
"optionsType": "Call",
"launchTime": "1700000000000",
"deliveryTime": "1719734400000",
},
{
"symbol": "BTC-30JUN25-50000-P",
"baseCoin": "BTC",
"settleCoin": "USDC",
"optionsType": "Put",
"launchTime": "1700000000000",
"deliveryTime": "1719734400000",
},
]
},
},
)
out = await client.get_option_chain(base_coin="BTC")
assert len(out["options"]) == 2
assert out["options"][0]["type"] == "Call"
# ── account / positions / orders (signed reads) ───────────────
async def test_get_positions(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/position/list?category=linear&settleCoin=USDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"side": "Buy",
"size": "0.1",
"avgPrice": "60000",
"unrealisedPnl": "50",
"leverage": "10",
"liqPrice": "50000",
"positionValue": "6000",
}
]
},
},
)
out = await client.get_positions(category="linear")
assert len(out) == 1
p = out[0]
assert p["symbol"] == "BTCUSDT"
assert p["side"] == "Buy"
assert p["size"] == 0.1
assert p["entry_price"] == 60000.0
assert p["liquidation_price"] == 50000.0
# signed: header X-BAPI-API-KEY presente
req = httpx_mock.get_requests()[-1]
assert req.headers.get("X-BAPI-API-KEY") == "test_key"
assert "X-BAPI-SIGN" in req.headers
async def test_get_account_summary(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/account/wallet-balance?accountType=UNIFIED"),
json={
"retCode": 0,
"result": {
"list": [
{
"accountType": "UNIFIED",
"totalEquity": "10000",
"totalWalletBalance": "9500",
"totalMarginBalance": "9800",
"totalAvailableBalance": "9000",
"totalPerpUPL": "200",
"coin": [
{
"coin": "USDT",
"walletBalance": "9500",
"equity": "9700",
}
],
}
]
},
},
)
out = await client.get_account_summary()
assert out["equity"] == 10000.0
assert out["available_balance"] == 9000.0
assert out["unrealized_pnl"] == 200.0
assert len(out["coins"]) == 1
assert out["coins"][0]["coin"] == "USDT"
async def test_get_trade_history(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/execution/list?category=linear&limit=50"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"side": "Buy",
"execQty": "0.01",
"execPrice": "60000",
"execFee": "0.1",
"execTime": "1700000000000",
"orderId": "abc",
}
]
},
},
)
out = await client.get_trade_history(category="linear", limit=50)
assert len(out) == 1
assert out[0]["symbol"] == "BTCUSDT"
assert out[0]["size"] == 0.01
assert out[0]["price"] == 60000.0
async def test_get_open_orders(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/realtime?category=linear&settleCoin=USDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"orderId": "o1",
"side": "Buy",
"qty": "0.1",
"price": "59000",
"orderType": "Limit",
"orderStatus": "New",
"reduceOnly": False,
}
]
},
},
)
out = await client.get_open_orders(category="linear")
assert len(out) == 1
assert out[0]["order_id"] == "o1"
assert out[0]["price"] == 59000.0
# ── basis / spreads ────────────────────────────────────────────
async def test_get_basis_spot_perp(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=spot&symbol=BTCUSDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"lastPrice": "60000",
"markPrice": "60000",
"bid1Price": "59995",
"ask1Price": "60005",
"volume24h": "0",
"turnover24h": "0",
"fundingRate": "0",
"openInterest": "0",
}
]
},
},
)
httpx_mock.add_response(
url=_url("/v5/market/tickers?category=linear&symbol=BTCUSDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"lastPrice": "60120",
"markPrice": "60120",
"bid1Price": "60115",
"ask1Price": "60125",
"volume24h": "0",
"turnover24h": "0",
"fundingRate": "0.0001",
"openInterest": "0",
}
]
},
},
)
out = await client.get_basis_spot_perp("BTC")
assert out["asset"] == "BTC"
assert out["spot_price"] == 60000.0
assert out["perp_price"] == 60120.0
assert out["basis_abs"] == 120.0
assert round(out["basis_pct"], 3) == 0.2
# ── trading writes ────────────────────────────────────────────
async def test_place_order_limit(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/create"),
method="POST",
json={
"retCode": 0,
"result": {"orderId": "ord123", "orderLinkId": ""},
},
)
out = await client.place_order(
category="linear",
symbol="BTCUSDT",
side="Buy",
qty=0.01,
order_type="Limit",
price=60000.0,
tif="GTC",
)
assert out["order_id"] == "ord123"
body = _last_request_json(httpx_mock)
assert body["category"] == "linear"
assert body["symbol"] == "BTCUSDT"
assert body["side"] == "Buy"
assert body["qty"] == "0.01"
assert body["orderType"] == "Limit"
assert body["price"] == "60000.0"
assert body["timeInForce"] == "GTC"
async def test_place_order_error(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/create"),
method="POST",
json={"retCode": 10001, "retMsg": "insufficient balance"},
)
out = await client.place_order(
category="linear",
symbol="BTCUSDT",
side="Buy",
qty=0.01,
order_type="Market",
)
assert out.get("error") == "insufficient balance"
assert out.get("code") == 10001
async def test_amend_order(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/amend"),
method="POST",
json={"retCode": 0, "result": {"orderId": "ord1"}},
)
out = await client.amend_order(
category="linear", symbol="BTCUSDT", order_id="ord1", new_qty=0.02
)
assert out["order_id"] == "ord1"
body = _last_request_json(httpx_mock)
assert body["orderId"] == "ord1"
assert body["qty"] == "0.02"
assert "price" not in body
async def test_place_order_option_adds_link_id(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/create"),
method="POST",
json={
"retCode": 0,
"result": {"orderId": "opt1", "orderLinkId": "cerbero-abc"},
},
)
await client.place_order(
category="option",
symbol="BTC-24APR26-96000-C-USDT",
side="Buy",
qty=0.01,
order_type="Limit",
price=5.0,
)
body = _last_request_json(httpx_mock)
assert "orderLinkId" in body
assert body["orderLinkId"].startswith("cerbero-")
async def test_place_order_linear_no_link_id(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/create"),
method="POST",
json={"retCode": 0, "result": {"orderId": "x"}},
)
await client.place_order(
category="linear",
symbol="BTCUSDT",
side="Buy",
qty=0.01,
order_type="Market",
)
body = _last_request_json(httpx_mock)
assert "orderLinkId" not in body
async def test_place_combo_order_batch_option(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/create-batch"),
method="POST",
json={
"retCode": 0,
"result": {
"list": [
{"orderId": "ord-1", "orderLinkId": "cerbero-leg1"},
{"orderId": "ord-2", "orderLinkId": "cerbero-leg2"},
]
},
},
)
legs = [
{
"symbol": "BTC-30APR26-75000-C-USDT",
"side": "Buy",
"qty": 0.01,
"order_type": "Limit",
"price": 5.0,
},
{
"symbol": "BTC-30APR26-80000-C-USDT",
"side": "Sell",
"qty": 0.01,
"order_type": "Limit",
"price": 3.0,
},
]
out = await client.place_combo_order(category="option", legs=legs)
assert len(out["orders"]) == 2
assert out["orders"][0]["order_id"] == "ord-1"
body = _last_request_json(httpx_mock)
assert body["category"] == "option"
request = body["request"]
assert len(request) == 2
assert request[0]["symbol"] == "BTC-30APR26-75000-C-USDT"
assert request[0]["qty"] == "0.01"
assert request[0]["orderType"] == "Limit"
assert "orderLinkId" in request[0]
async def test_place_combo_order_error(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/create-batch"),
method="POST",
json={"retCode": 10001, "retMsg": "invalid leg"},
)
out = await client.place_combo_order(
category="option",
legs=[
{
"symbol": "X",
"side": "Buy",
"qty": 1,
"order_type": "Limit",
"price": 1.0,
},
{
"symbol": "Y",
"side": "Sell",
"qty": 1,
"order_type": "Limit",
"price": 1.0,
},
],
)
assert out["error"] == "invalid leg"
assert out["code"] == 10001
async def test_place_combo_order_rejects_non_option(client):
with pytest.raises(ValueError, match="option"):
await client.place_combo_order(
category="linear",
legs=[
{
"symbol": "BTCUSDT",
"side": "Buy",
"qty": 0.01,
"order_type": "Market",
},
{
"symbol": "ETHUSDT",
"side": "Sell",
"qty": 0.01,
"order_type": "Market",
},
],
)
async def test_cancel_order(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/cancel"),
method="POST",
json={"retCode": 0, "result": {"orderId": "ord1"}},
)
out = await client.cancel_order(category="linear", symbol="BTCUSDT", order_id="ord1")
assert out["order_id"] == "ord1"
assert out["status"] == "cancelled"
body = _last_request_json(httpx_mock)
assert body["category"] == "linear"
assert body["symbol"] == "BTCUSDT"
assert body["orderId"] == "ord1"
async def test_cancel_all_orders(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/order/cancel-all"),
method="POST",
json={
"retCode": 0,
"result": {"list": [{"orderId": "o1"}, {"orderId": "o2"}]},
},
)
out = await client.cancel_all_orders(category="linear", symbol="BTCUSDT")
assert out["cancelled_ids"] == ["o1", "o2"]
body = _last_request_json(httpx_mock)
assert body["category"] == "linear"
assert body["symbol"] == "BTCUSDT"
async def test_set_stop_loss(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/position/trading-stop"),
method="POST",
json={"retCode": 0, "result": {}},
)
out = await client.set_stop_loss(
category="linear", symbol="BTCUSDT", stop_loss=55000.0
)
body = _last_request_json(httpx_mock)
assert body["category"] == "linear"
assert body["symbol"] == "BTCUSDT"
assert body["stopLoss"] == "55000.0"
assert body.get("positionIdx", 0) == 0
assert out["status"] == "stop_loss_set"
async def test_set_take_profit(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/position/trading-stop"),
method="POST",
json={"retCode": 0, "result": {}},
)
out = await client.set_take_profit(
category="linear", symbol="BTCUSDT", take_profit=65000.0
)
body = _last_request_json(httpx_mock)
assert body["takeProfit"] == "65000.0"
assert out["status"] == "take_profit_set"
async def test_close_position(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/position/list?category=linear&settleCoin=USDT"),
json={
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"side": "Buy",
"size": "0.1",
"avgPrice": "60000",
"unrealisedPnl": "0",
"leverage": "10",
"liqPrice": "0",
"positionValue": "6000",
}
]
},
},
)
httpx_mock.add_response(
url=_url("/v5/order/create"),
method="POST",
json={
"retCode": 0,
"result": {"orderId": "closeord", "orderLinkId": ""},
},
)
out = await client.close_position(category="linear", symbol="BTCUSDT")
assert out["status"] == "submitted"
# the LAST request is the place_order body
body = _last_request_json(httpx_mock)
assert body["side"] == "Sell"
assert body["qty"] == "0.1"
assert body["reduceOnly"] is True
assert body["orderType"] == "Market"
async def test_set_leverage(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/position/set-leverage"),
method="POST",
json={"retCode": 0, "result": {}},
)
out = await client.set_leverage(category="linear", symbol="BTCUSDT", leverage=5)
body = _last_request_json(httpx_mock)
assert body["category"] == "linear"
assert body["symbol"] == "BTCUSDT"
assert body["buyLeverage"] == "5"
assert body["sellLeverage"] == "5"
assert out["status"] == "leverage_set"
async def test_switch_position_mode(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/position/switch-mode"),
method="POST",
json={"retCode": 0, "result": {}},
)
out = await client.switch_position_mode(
category="linear", symbol="BTCUSDT", mode="hedge"
)
body = _last_request_json(httpx_mock)
assert body["mode"] == 3
assert out["status"] == "mode_switched"
async def test_transfer_asset(client, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_url("/v5/asset/transfer/inter-transfer"),
method="POST",
json={"retCode": 0, "result": {"transferId": "tx123"}},
)
out = await client.transfer_asset(
coin="USDT", amount=100.0, from_type="UNIFIED", to_type="FUND"
)
body = _last_request_json(httpx_mock)
assert body["coin"] == "USDT"
assert body["amount"] == "100.0"
assert body["fromAccountType"] == "UNIFIED"
assert body["toAccountType"] == "FUND"
assert out["transfer_id"] == "tx123"
+50
View File
@@ -0,0 +1,50 @@
from __future__ import annotations
import pytest
from cerbero_mcp.exchanges.bybit.leverage_cap import enforce_leverage, get_max_leverage
from fastapi import HTTPException
def test_get_max_leverage_returns_creds_value():
creds = {"max_leverage": 5}
assert get_max_leverage(creds) == 5
def test_get_max_leverage_default_when_missing():
"""Default 1 (cash) se il secret non ha max_leverage."""
assert get_max_leverage({}) == 1
def test_enforce_leverage_pass_under_cap():
creds = {"max_leverage": 3}
enforce_leverage(2, creds=creds, exchange="bybit") # no raise
def test_enforce_leverage_pass_at_cap():
creds = {"max_leverage": 3}
enforce_leverage(3, creds=creds, exchange="bybit") # no raise
def test_enforce_leverage_reject_over_cap():
creds = {"max_leverage": 3}
with pytest.raises(HTTPException) as exc:
enforce_leverage(10, creds=creds, exchange="bybit")
assert exc.value.status_code == 403
assert exc.value.detail["error"] == "LEVERAGE_CAP_EXCEEDED"
assert exc.value.detail["exchange"] == "bybit"
assert exc.value.detail["requested"] == 10
assert exc.value.detail["max"] == 3
def test_enforce_leverage_reject_when_below_one():
creds = {"max_leverage": 3}
with pytest.raises(HTTPException) as exc:
enforce_leverage(0, creds=creds, exchange="bybit")
assert exc.value.status_code == 403
def test_enforce_leverage_default_when_none():
"""Se requested è None, applica il cap come default."""
creds = {"max_leverage": 3}
result = enforce_leverage(None, creds=creds, exchange="bybit")
assert result == 3