"""Official Bybit V5 REST & WebSocket Client.""" from __future__ import annotations import hashlib import hmac import json import time import urllib.error import urllib.parse import urllib.request from decimal import Decimal from typing import Any, Optional from backend.app.audit.logger import audit_logger, redact_sensitive_data from backend.app.config.settings import TradingMode, settings from backend.app.schemas.market import Candle, OrderBook, OrderBookLevel, Ticker from backend.app.schemas.risk import InstrumentFilters class BybitV5Client: """Official Bybit V5 REST API Connector supporting Demo and Live trading with strict isolation.""" def __init__( self, api_key: Optional[str] = None, api_secret: Optional[str] = None, base_url: Optional[str] = None, mode: TradingMode = TradingMode.PAPER, ) -> None: self.mode = mode if mode == TradingMode.BYBIT_DEMO: self.base_url = base_url or settings.BYBIT_DEMO_REST_URL self.api_key = api_key or settings.BYBIT_DEMO_API_KEY self.api_secret = api_secret or settings.BYBIT_DEMO_API_SECRET elif mode == TradingMode.LIVE: if not settings.ENABLE_LIVE_TRADING: raise PermissionError("Attempted to initialize Live Bybit client while ENABLE_LIVE_TRADING is False.") self.base_url = base_url or settings.BYBIT_LIVE_REST_URL self.api_key = api_key or settings.BYBIT_LIVE_API_KEY self.api_secret = api_secret or settings.BYBIT_LIVE_API_SECRET else: # Backtest or Paper: Read public market data only self.base_url = "https://api.bybit.com" self.api_key = "" self.api_secret = "" def _generate_signature(self, timestamp: str, recv_window: str, query_or_body: str) -> str: param_str = f"{timestamp}{self.api_key}{recv_window}{query_or_body}" return hmac.new( self.api_secret.encode("utf-8"), param_str.encode("utf-8"), hashlib.sha256, ).hexdigest() def _request( self, method: str, path: str, params: Optional[dict[str, Any]] = None, data: Optional[dict[str, Any]] = None, auth_required: bool = False, ) -> dict[str, Any]: params = params or {} data_str = json.dumps(data) if data else "" query_str = urllib.parse.urlencode(params) if params else "" full_url = f"{self.base_url}{path}" if query_str: full_url += f"?{query_str}" headers = { "Content-Type": "application/json", "User-Agent": "AegisTradeAI/1.0", } if auth_required: if not self.api_key or not self.api_secret: raise ValueError("Authentication credentials missing for private Bybit request.") timestamp = str(int(time.time() * 1000)) recv_window = "5000" payload_to_sign = data_str if method.upper() == "POST" else query_str signature = self._generate_signature(timestamp, recv_window, payload_to_sign) headers.update({ "X-BAPI-API-KEY": self.api_key, "X-BAPI-TIMESTAMP": timestamp, "X-BAPI-SIGN": signature, "X-BAPI-RECV-WINDOW": recv_window, }) req = urllib.request.Request( full_url, data=data_str.encode("utf-8") if data_str else None, headers=headers, method=method.upper(), ) try: with urllib.request.urlopen(req, timeout=10) as resp: body = resp.read().decode("utf-8") res = json.loads(body) if res.get("retCode") != 0: audit_logger.warning("Bybit API response warning: %s", redact_sensitive_data(str(res))) return res except urllib.error.HTTPError as e: err_body = e.read().decode("utf-8") if e.fp else "" audit_logger.error( "Bybit HTTP Error %s: %s", e.code, redact_sensitive_data(err_body) ) raise ConnectionError(f"Bybit HTTP {e.code}: {err_body}") from e except Exception as e: audit_logger.error("Bybit network error: %s", redact_sensitive_data(str(e))) raise ConnectionError(f"Failed connection to Bybit API: {str(e)}") from e def get_server_time(self) -> int: """Fetch exchange server time in milliseconds.""" res = self._request("GET", "/v5/market/time") if "result" in res and "timeSecond" in res["result"]: return int(res["result"]["timeSecond"]) * 1000 return int(time.time() * 1000) def get_klines(self, symbol: str, interval: str = "15", limit: int = 200) -> list[Candle]: """Fetch historical candlesticks from Bybit V5 API.""" # Convert standard interval notation to Bybit interval interval_map = {"1m": "1", "3m": "3", "5m": "5", "15m": "15", "30m": "30", "1h": "60", "4h": "240", "1d": "D"} bybit_interval = interval_map.get(interval, interval) params = { "category": "linear", "symbol": symbol, "interval": bybit_interval, "limit": limit, } res = self._request("GET", "/v5/market/kline", params=params) candles: list[Candle] = [] if res.get("retCode") == 0 and "result" in res and "list" in res["result"]: # Bybit returns klines newest first: [startTime, open, high, low, close, volume, turnover] for row in reversed(res["result"]["list"]): candles.append( Candle( symbol=symbol, interval=interval, timestamp=int(row[0]), open=Decimal(row[1]), high=Decimal(row[2]), low=Decimal(row[3]), close=Decimal(row[4]), volume=Decimal(row[5]), turnover=Decimal(row[6]) if len(row) > 6 else None, is_closed=True, ) ) return candles def get_ticker(self, symbol: str) -> Ticker: """Fetch latest market ticker.""" res = self._request("GET", "/v5/market/tickers", params={"category": "linear", "symbol": symbol}) if res.get("retCode") == 0 and res.get("result", {}).get("list"): row = res["result"]["list"][0] return Ticker( symbol=symbol, bid_price=Decimal(row.get("bid1Price", row.get("lastPrice", "0"))), ask_price=Decimal(row.get("ask1Price", row.get("lastPrice", "0"))), last_price=Decimal(row.get("lastPrice", "0")), volume_24h=Decimal(row.get("volume24h", "0")), price_24h_pcnt=Decimal(row.get("price24hPcnt", "0")), high_24h=Decimal(row.get("highPrice24h", "0")), low_24h=Decimal(row.get("lowPrice24h", "0")), timestamp=int(time.time() * 1000), ) raise ValueError(f"Failed to fetch ticker for {symbol}: {res}") def get_orderbook(self, symbol: str, limit: int = 50) -> OrderBook: """Fetch orderbook depth.""" res = self._request("GET", "/v5/market/orderbook", params={"category": "linear", "symbol": symbol, "limit": limit}) bids: list[OrderBookLevel] = [] asks: list[OrderBookLevel] = [] if res.get("retCode") == 0 and "result" in res: res_obj = res["result"] for b in res_obj.get("b", []): bids.append(OrderBookLevel(price=Decimal(b[0]), size=Decimal(b[1]))) for a in res_obj.get("a", []): asks.append(OrderBookLevel(price=Decimal(a[0]), size=Decimal(a[1]))) return OrderBook(symbol=symbol, bids=bids, asks=asks, timestamp=int(time.time() * 1000)) def get_instrument_filters(self, symbol: str) -> InstrumentFilters: """Fetch precision, lot size, and tick size filters from Bybit.""" res = self._request("GET", "/v5/market/instruments-info", params={"category": "linear", "symbol": symbol}) if res.get("retCode") == 0 and res.get("result", {}).get("list"): info = res["result"]["list"][0] lot_filter = info.get("lotSizeFilter", {}) price_filter = info.get("priceFilter", {}) return InstrumentFilters( symbol=symbol, min_order_qty=Decimal(lot_filter.get("minOrderQty", "0.001")), qty_step=Decimal(lot_filter.get("qtyStep", "0.001")), min_notional=Decimal(lot_filter.get("minNotionalValue", "5.0")), tick_size=Decimal(price_filter.get("tickSize", "0.1")), max_leverage=Decimal(info.get("leverageFilter", {}).get("maxLeverage", "10")), ) # Safe default fallback for BTCUSDT if offline return InstrumentFilters( symbol=symbol, min_order_qty=Decimal("0.001"), qty_step=Decimal("0.001"), min_notional=Decimal("5.0"), tick_size=Decimal("0.1"), max_leverage=Decimal("10"), ) def get_wallet_balance(self, account_type: str = "UNIFIED") -> dict[str, Any]: """Fetch account wallet balance.""" return self._request( "GET", "/v5/account/wallet-balance", params={"accountType": account_type}, auth_required=True, ) def place_order( self, symbol: str, side: str, order_type: str, qty: str, price: Optional[str] = None, stop_loss: Optional[str] = None, take_profit: Optional[str] = None, client_order_id: Optional[str] = None, reduce_only: bool = False, ) -> dict[str, Any]: """Place new order with exchange-side protective stop loss.""" data: dict[str, Any] = { "category": "linear", "symbol": symbol, "side": side.capitalize(), "orderType": order_type.capitalize(), "qty": qty, "reduceOnly": reduce_only, "timeInForce": "GTC", } if price and order_type.upper() == "LIMIT": data["price"] = price if stop_loss: data["stopLoss"] = stop_loss if take_profit: data["takeProfit"] = take_profit if client_order_id: data["orderLinkId"] = client_order_id return self._request("POST", "/v5/order/create", data=data, auth_required=True) def cancel_order(self, symbol: str, client_order_id: str) -> dict[str, Any]: """Cancel order by client order ID.""" data = { "category": "linear", "symbol": symbol, "orderLinkId": client_order_id, } return self._request("POST", "/v5/order/cancel", data=data, auth_required=True)