"""Deribit REST API client."""
import hashlib
import hmac
import logging
import time
from typing import Any, Dict, List, Optional
import aiohttp
from .config import settings
logger = logging.getLogger(__name__)
class DeribitRestClient:
"""REST API client for Deribit exchange."""
def __init__(self):
self.base_url = settings.deribit_rest_url
self.api_key = settings.deribit_api_key
self.api_secret = settings.deribit_api_secret
self.session: Optional[aiohttp.ClientSession] = None
self.access_token: Optional[str] = None
self.token_expiry: Optional[float] = None
async def __aenter__(self):
"""Context manager entry."""
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
await self.disconnect()
async def connect(self) -> None:
"""Initialize HTTP session."""
if not self.session:
self.session = aiohttp.ClientSession()
logger.info("Initialized Deribit REST client")
# Authenticate if credentials provided
if self.api_key and self.api_secret:
await self._authenticate()
async def disconnect(self) -> None:
"""Close HTTP session."""
if self.session:
await self.session.close()
self.session = None
logger.info("Closed Deribit REST client")
async def _authenticate(self) -> None:
"""Authenticate with Deribit API."""
try:
data = {
"grant_type": "client_credentials",
"client_id": self.api_key,
"client_secret": self.api_secret,
}
result = await self._request("public/auth", data)
if "access_token" in result:
self.access_token = result["access_token"]
self.token_expiry = time.time() + result.get("expires_in", 3600)
logger.info("Successfully authenticated with Deribit REST API")
else:
logger.warning(f"Authentication response: {result}")
except Exception as e:
logger.error(f"Failed to authenticate: {e}")
raise
async def _ensure_authenticated(self) -> None:
"""Ensure we have a valid access token."""
if not self.access_token or (
self.token_expiry and time.time() >= self.token_expiry - 60
):
await self._authenticate()
async def _request(
self, method: str, params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Make a request to Deribit API."""
if not self.session:
await self.connect()
url = f"{self.base_url}/{method}"
# Add access token to params for private methods
if method.startswith("private/") and self.access_token:
await self._ensure_authenticated()
if params is None:
params = {}
params["access_token"] = self.access_token
try:
async with self.session.get(url, params=params) as response:
data = await response.json()
if "error" in data:
error_msg = data["error"].get("message", "Unknown error")
logger.error(f"API error: {error_msg}")
raise Exception(f"Deribit API error: {error_msg}")
return data.get("result", {})
except aiohttp.ClientError as e:
logger.error(f"HTTP request failed: {e}")
raise
async def get_instruments(self, currency: str = "BTC", kind: str = "future") -> List[Dict]:
"""Get available instruments."""
params = {"currency": currency, "kind": kind}
return await self._request("public/get_instruments", params)
async def get_ticker(self, instrument: str) -> Dict[str, Any]:
"""Get ticker information for an instrument."""
params = {"instrument_name": instrument}
return await self._request("public/ticker", params)
async def get_account_summary(self, currency: str = "BTC") -> Dict[str, Any]:
"""Get account summary."""
params = {"currency": currency, "extended": True}
return await self._request("private/get_account_summary", params)
async def get_positions(self, currency: str = "BTC") -> List[Dict]:
"""Get current positions."""
params = {"currency": currency}
result = await self._request("private/get_positions", params)
return result if isinstance(result, list) else []
async def get_order_book(
self, instrument: str, depth: int = 10
) -> Dict[str, Any]:
"""Get order book for an instrument."""
params = {"instrument_name": instrument, "depth": depth}
return await self._request("public/get_order_book", params)
async def buy(
self,
instrument: str,
amount: float,
order_type: str = "market",
price: Optional[float] = None,
) -> Dict[str, Any]:
"""Place a buy order."""
params = {
"instrument_name": instrument,
"amount": amount,
"type": order_type,
}
if price and order_type == "limit":
params["price"] = price
return await self._request("private/buy", params)
async def sell(
self,
instrument: str,
amount: float,
order_type: str = "market",
price: Optional[float] = None,
) -> Dict[str, Any]:
"""Place a sell order."""
params = {
"instrument_name": instrument,
"amount": amount,
"type": order_type,
}
if price and order_type == "limit":
params["price"] = price
return await self._request("private/sell", params)
async def cancel_order(self, order_id: str) -> Dict[str, Any]:
"""Cancel an order."""
params = {"order_id": order_id}
return await self._request("private/cancel", params)
async def get_open_orders(
self, instrument: Optional[str] = None, currency: str = "BTC"
) -> List[Dict]:
"""Get open orders."""
params = {"currency": currency}
if instrument:
params["instrument_name"] = instrument
result = await self._request("private/get_open_orders", params)
return result if isinstance(result, list) else []
