# -*- coding: utf-8 -*- # PLEASE DO NOT EDIT THIS FILE, IT IS GENERATED AND WILL BE OVERWRITTEN: # https://github.com/ccxt/ccxt/blob/master/CONTRIBUTING.md#how-to-contribute-code import ccxt.async_support from ccxt.async_support.base.ws.cache import ArrayCache, ArrayCacheBySymbolById, ArrayCacheBySymbolBySide, ArrayCacheByTimestamp import asyncio import hashlib from ccxt.base.types import Any, Balances, Bool, Int, Liquidation, Num, Order, OrderBook, OrderSide, OrderType, Position, Str, Strings, Ticker, Tickers, Trade from ccxt.async_support.base.ws.client import Client from typing import List from ccxt.base.errors import ExchangeError from ccxt.base.errors import AuthenticationError from ccxt.base.errors import ArgumentsRequired from ccxt.base.errors import BadRequest from ccxt.base.errors import NotSupported class bybit(ccxt.async_support.bybit): def describe(self) -> Any: return self.deep_extend(super(bybit, self).describe(), { 'has': { 'ws': True, 'createOrderWs': True, 'editOrderWs': True, 'fetchOpenOrdersWs': False, 'fetchOrderWs': False, 'cancelOrderWs': True, 'cancelOrdersWs': False, 'cancelAllOrdersWs': False, 'fetchTradesWs': False, 'fetchBalanceWs': False, 'watchBalance': True, 'watchBidsAsks': True, 'watchLiquidations': True, 'watchLiquidationsForSymbols': False, 'watchMyLiquidations': False, 'watchMyLiquidationsForSymbols': False, 'watchMyTrades': True, 'watchOHLCV': True, 'watchOHLCVForSymbols': True, 'watchOrderBook': True, 'watchOrderBookForSymbols': True, 'watchOrders': True, 'watchTicker': True, 'watchTickers': True, 'watchTrades': True, 'watchPositions': True, 'watchTradesForSymbols': True, 'unWatchTicker': True, 'unWatchTickers': True, 'unWatchOHLCV': True, 'unWatchOHLCVForSymbols': True, 'unWatchOrderBook': True, 'unWatchOrderBookForSymbols': True, 'unWatchTrades': True, 'unWatchTradesForSymbols': True, 'unWatchMyTrades': True, 'unWatchOrders': True, 'unWatchPositions': True, }, 'urls': { 'api': { 'ws': { 'public': { 'spot': 'wss://stream.{hostname}/v5/public/spot', 'inverse': 'wss://stream.{hostname}/v5/public/inverse', 'option': 'wss://stream.{hostname}/v5/public/option', 'linear': 'wss://stream.{hostname}/v5/public/linear', }, 'private': { 'spot': { 'unified': 'wss://stream.{hostname}/v5/private', 'nonUnified': 'wss://stream.{hostname}/spot/private/v3', }, 'contract': 'wss://stream.{hostname}/v5/private', 'usdc': 'wss://stream.{hostname}/trade/option/usdc/private/v1', 'trade': 'wss://stream.bybit.com/v5/trade', }, }, }, 'test': { 'ws': { 'public': { 'spot': 'wss://stream-testnet.{hostname}/v5/public/spot', 'inverse': 'wss://stream-testnet.{hostname}/v5/public/inverse', 'linear': 'wss://stream-testnet.{hostname}/v5/public/linear', 'option': 'wss://stream-testnet.{hostname}/v5/public/option', }, 'private': { 'spot': { 'unified': 'wss://stream-testnet.{hostname}/v5/private', 'nonUnified': 'wss://stream-testnet.{hostname}/spot/private/v3', }, 'contract': 'wss://stream-testnet.{hostname}/v5/private', 'usdc': 'wss://stream-testnet.{hostname}/trade/option/usdc/private/v1', 'trade': 'wss://stream-testnet.bybit.com/v5/trade', }, }, }, 'demotrading': { 'ws': { 'public': { 'spot': 'wss://stream.{hostname}/v5/public/spot', 'inverse': 'wss://stream.{hostname}/v5/public/inverse', 'option': 'wss://stream.{hostname}/v5/public/option', 'linear': 'wss://stream.{hostname}/v5/public/linear', }, 'private': { 'spot': { 'unified': 'wss://stream-demo.{hostname}/v5/private', 'nonUnified': 'wss://stream-demo.{hostname}/spot/private/v3', }, 'contract': 'wss://stream-demo.{hostname}/v5/private', 'usdc': 'wss://stream-demo.{hostname}/trade/option/usdc/private/v1', 'trade': 'wss://stream-demo.bybit.com/v5/trade', }, }, }, }, 'options': { 'watchTicker': { 'name': 'tickers', # 'tickers' for 24hr statistical ticker or 'tickers_lt' for leverage token ticker }, 'watchPositions': { 'fetchPositionsSnapshot': True, # or False 'awaitPositionsSnapshot': True, # whether to wait for the positions snapshot before providing updates }, 'watchMyTrades': { # filter execType: https://bybit-exchange.github.io/docs/api-explorer/v5/position/execution 'filterExecTypes': [ 'Trade', 'AdlTrade', 'BustTrade', 'Settle', ], }, 'spot': { 'timeframes': { '1m': '1m', '3m': '3m', '5m': '5m', '15m': '15m', '30m': '30m', '1h': '1h', '2h': '2h', '4h': '4h', '6h': '6h', '12h': '12h', '1d': '1d', '1w': '1w', '1M': '1M', }, }, 'contract': { 'timeframes': { '1m': '1', '3m': '3', '5m': '5', '15m': '15', '30m': '30', '1h': '60', '2h': '120', '4h': '240', '6h': '360', '12h': '720', '1d': 'D', '1w': 'W', '1M': 'M', }, }, }, 'streaming': { 'ping': self.ping, 'keepAlive': 18000, }, }) def request_id(self): requestId = self.sum(self.safe_integer(self.options, 'requestId', 0), 1) self.options['requestId'] = requestId return requestId async def get_url_by_market_type(self, symbol: Str = None, isPrivate=False, method: Str = None, params={}): accessibility = 'private' if isPrivate else 'public' isUsdcSettled = None isSpot = None type = None market = None url = self.urls['api']['ws'] if symbol is not None: market = self.market(symbol) isUsdcSettled = market['settle'] == 'USDC' type = market['type'] else: type, params = self.handle_market_type_and_params(method, None, params) defaultSettle = self.safe_string(self.options, 'defaultSettle') defaultSettle = self.safe_string_2(params, 'settle', 'defaultSettle', defaultSettle) isUsdcSettled = (defaultSettle == 'USDC') isSpot = (type == 'spot') if isPrivate: unified = await self.isUnifiedEnabled() isUnifiedMargin = self.safe_bool(unified, 0, False) isUnifiedAccount = self.safe_bool(unified, 1, False) if isUsdcSettled and not isUnifiedMargin and not isUnifiedAccount: url = url[accessibility]['usdc'] else: url = url[accessibility]['contract'] else: if isSpot: url = url[accessibility]['spot'] elif (type == 'swap') or (type == 'future'): subType = None subType, params = self.handle_sub_type_and_params(method, market, params, 'linear') url = url[accessibility][subType] else: # option url = url[accessibility]['option'] url = self.implode_hostname(url) return url def clean_params(self, params): params = self.omit(params, ['type', 'subType', 'settle', 'defaultSettle', 'unifiedMargin']) return params async def create_order_ws(self, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}): """ create a trade order https://bybit-exchange.github.io/docs/v5/order/create-order https://bybit-exchange.github.io/docs/v5/websocket/trade/guideline#createamendcancel-order :param str symbol: unified symbol of the market to create an order in :param str type: 'market' or 'limit' :param str side: 'buy' or 'sell' :param float amount: how much of currency you want to trade in units of base currency :param float [price]: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.timeInForce]: "GTC", "IOC", "FOK" :param bool [params.postOnly]: True or False whether the order is post-only :param bool [params.reduceOnly]: True or False whether the order is reduce-only :param str [params.positionIdx]: *contracts only* 0 for one-way mode, 1 buy side of hedged mode, 2 sell side of hedged mode :param boolean [params.isLeverage]: *unified spot only* False then spot trading True then margin trading :param str [params.tpslMode]: *contract only* 'full' or 'partial' :param str [params.mmp]: *option only* market maker protection :param str [params.triggerDirection]: *contract only* the direction for trigger orders, 'above' or 'below' :param float [params.triggerPrice]: The price at which a trigger order is triggered at :param float [params.stopLossPrice]: The price at which a stop loss order is triggered at :param float [params.takeProfitPrice]: The price at which a take profit order is triggered at :param dict [params.takeProfit]: *takeProfit object in params* containing the triggerPrice at which the attached take profit order will be triggered :param float [params.takeProfit.triggerPrice]: take profit trigger price :param dict [params.stopLoss]: *stopLoss object in params* containing the triggerPrice at which the attached stop loss order will be triggered :param float [params.stopLoss.triggerPrice]: stop loss trigger price :param str [params.trailingAmount]: the quote amount to trail away from the current market price :param str [params.trailingTriggerPrice]: the price to trigger a trailing order, default uses the price argument :returns dict: an `order structure ` """ await self.load_markets() orderRequest = self.create_order_request(symbol, type, side, amount, price, params, True) url = self.urls['api']['ws']['private']['trade'] await self.authenticate(url) requestId = str(self.request_id()) request: dict = { 'op': 'order.create', 'reqId': requestId, 'args': [ orderRequest, ], 'header': { 'X-BAPI-TIMESTAMP': str(self.milliseconds()), 'X-BAPI-RECV-WINDOW': str(self.options['recvWindow']), }, } return await self.watch(url, requestId, request, requestId, True) async def edit_order_ws(self, id: str, symbol: str, type: OrderType, side: OrderSide, amount: Num = None, price: Num = None, params={}): """ edit a trade order https://bybit-exchange.github.io/docs/v5/order/amend-order https://bybit-exchange.github.io/docs/v5/websocket/trade/guideline#createamendcancel-order :param str id: cancel order id :param str symbol: unified symbol of the market to create an order in :param str type: 'market' or 'limit' :param str side: 'buy' or 'sell' :param float amount: how much of currency you want to trade in units of base currency :param float price: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders :param dict [params]: extra parameters specific to the exchange API endpoint :param float [params.triggerPrice]: The price that a trigger order is triggered at :param float [params.stopLossPrice]: The price that a stop loss order is triggered at :param float [params.takeProfitPrice]: The price that a take profit order is triggered at :param dict [params.takeProfit]: *takeProfit object in params* containing the triggerPrice that the attached take profit order will be triggered :param float [params.takeProfit.triggerPrice]: take profit trigger price :param dict [params.stopLoss]: *stopLoss object in params* containing the triggerPrice that the attached stop loss order will be triggered :param float [params.stopLoss.triggerPrice]: stop loss trigger price :param str [params.triggerBy]: 'IndexPrice', 'MarkPrice' or 'LastPrice', default is 'LastPrice', required if no initial value for triggerPrice :param str [params.slTriggerBy]: 'IndexPrice', 'MarkPrice' or 'LastPrice', default is 'LastPrice', required if no initial value for stopLoss :param str [params.tpTriggerby]: 'IndexPrice', 'MarkPrice' or 'LastPrice', default is 'LastPrice', required if no initial value for takeProfit :returns dict: an `order structure ` """ await self.load_markets() orderRequest = self.edit_order_request(id, symbol, type, side, amount, price, params) url = self.urls['api']['ws']['private']['trade'] await self.authenticate(url) requestId = str(self.request_id()) request: dict = { 'op': 'order.amend', 'reqId': requestId, 'args': [ orderRequest, ], 'header': { 'X-BAPI-TIMESTAMP': str(self.milliseconds()), 'X-BAPI-RECV-WINDOW': str(self.options['recvWindow']), }, } return await self.watch(url, requestId, request, requestId, True) async def cancel_order_ws(self, id: str, symbol: Str = None, params={}): """ cancels an open order https://bybit-exchange.github.io/docs/v5/order/cancel-order https://bybit-exchange.github.io/docs/v5/websocket/trade/guideline#createamendcancel-order :param str id: order id :param str symbol: unified symbol of the market the order was made in :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.trigger]: *spot only* whether the order is a trigger order :param str [params.orderFilter]: *spot only* 'Order' or 'StopOrder' or 'tpslOrder' :returns dict: An `order structure ` """ await self.load_markets() if symbol is None: raise ArgumentsRequired(self.id + ' cancelOrderWs() requires a symbol argument') orderRequest = self.cancel_order_request(id, symbol, params) url = self.urls['api']['ws']['private']['trade'] await self.authenticate(url) requestId = str(self.request_id()) if 'orderFilter' in orderRequest: del orderRequest['orderFilter'] request: dict = { 'op': 'order.cancel', 'reqId': requestId, 'args': [ orderRequest, ], 'header': { 'X-BAPI-TIMESTAMP': str(self.milliseconds()), 'X-BAPI-RECV-WINDOW': str(self.options['recvWindow']), }, } return await self.watch(url, requestId, request, requestId, True) async def watch_ticker(self, symbol: str, params={}) -> Ticker: """ watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market https://bybit-exchange.github.io/docs/v5/websocket/public/ticker https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] messageHash = 'ticker:' + symbol url = await self.get_url_by_market_type(symbol, False, 'watchTicker', params) params = self.clean_params(params) options = self.safe_value(self.options, 'watchTicker', {}) topic = self.safe_string(options, 'name', 'tickers') if not market['spot'] and topic != 'tickers': raise BadRequest(self.id + ' watchTicker() only supports name tickers for contract markets') topic += '.' + market['id'] topics = [topic] return await self.watch_topics(url, [messageHash], topics, params) async def watch_tickers(self, symbols: Strings = None, params={}) -> Tickers: """ watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for all markets of a specific list https://bybit-exchange.github.io/docs/v5/websocket/public/ticker https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker :param str[] symbols: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False) messageHashes = [] url = await self.get_url_by_market_type(symbols[0], False, 'watchTickers', params) params = self.clean_params(params) options = self.safe_value(self.options, 'watchTickers', {}) topic = self.safe_string(options, 'name', 'tickers') marketIds = self.market_ids(symbols) topics = [] for i in range(0, len(marketIds)): marketId = marketIds[i] topics.append(topic + '.' + marketId) messageHashes.append('ticker:' + symbols[i]) ticker = await self.watch_topics(url, messageHashes, topics, params) if self.newUpdates: result: dict = {} result[ticker['symbol']] = ticker return result return self.filter_by_array(self.tickers, 'symbol', symbols) async def un_watch_tickers(self, symbols: Strings = None, params={}) -> Any: """ unWatches a price ticker https://bybit-exchange.github.io/docs/v5/websocket/public/ticker https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker :param str[] symbols: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False) options = self.safe_value(self.options, 'watchTickers', {}) topic = self.safe_string(options, 'name', 'tickers') messageHashes = [] subMessageHashes = [] marketIds = self.market_ids(symbols) topics = [] for i in range(0, len(marketIds)): marketId = marketIds[i] symbol = symbols[i] topics.append(topic + '.' + marketId) subMessageHashes.append('ticker:' + symbol) messageHashes.append('unsubscribe:ticker:' + symbol) url = await self.get_url_by_market_type(symbols[0], False, 'watchTickers', params) return await self.un_watch_topics(url, 'ticker', symbols, messageHashes, subMessageHashes, topics, params) async def un_watch_ticker(self, symbol: str, params={}) -> Any: """ unWatches a price ticker https://bybit-exchange.github.io/docs/v5/websocket/public/ticker https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker :param str[] symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() return await self.un_watch_tickers([symbol], params) def handle_ticker(self, client: Client, message): # # linear # { # "topic": "tickers.BTCUSDT", # "type": "snapshot", # "data": { # "symbol": "BTCUSDT", # "tickDirection": "PlusTick", # "price24hPcnt": "0.017103", # "lastPrice": "17216.00", # "prevPrice24h": "16926.50", # "highPrice24h": "17281.50", # "lowPrice24h": "16915.00", # "prevPrice1h": "17238.00", # "markPrice": "17217.33", # "indexPrice": "17227.36", # "openInterest": "68744.761", # "openInterestValue": "1183601235.91", # "turnover24h": "1570383121.943499", # "volume24h": "91705.276", # "nextFundingTime": "1673280000000", # "fundingRate": "-0.000212", # "bid1Price": "17215.50", # "bid1Size": "84.489", # "ask1Price": "17216.00", # "ask1Size": "83.020" # }, # "cs": 24987956059, # "ts": 1673272861686 # } # # option # { # "id": "tickers.BTC-6JAN23-17500-C-2480334983-1672917511074", # "topic": "tickers.BTC-6JAN23-17500-C", # "ts": 1672917511074, # "data": { # "symbol": "BTC-6JAN23-17500-C", # "bidPrice": "0", # "bidSize": "0", # "bidIv": "0", # "askPrice": "10", # "askSize": "5.1", # "askIv": "0.514", # "lastPrice": "10", # "highPrice24h": "25", # "lowPrice24h": "5", # "markPrice": "7.86976724", # "indexPrice": "16823.73", # "markPriceIv": "0.4896", # "underlyingPrice": "16815.1", # "openInterest": "49.85", # "turnover24h": "446802.8473", # "volume24h": "26.55", # "totalVolume": "86", # "totalTurnover": "1437431", # "delta": "0.047831", # "gamma": "0.00021453", # "vega": "0.81351067", # "theta": "-19.9115368", # "predictedDeliveryPrice": "0", # "change24h": "-0.33333334" # }, # "type": "snapshot" # } # # spot # { # "topic": "tickers.BTCUSDT", # "ts": 1673853746003, # "type": "snapshot", # "cs": 2588407389, # "data": { # "symbol": "BTCUSDT", # "lastPrice": "21109.77", # "highPrice24h": "21426.99", # "lowPrice24h": "20575", # "prevPrice24h": "20704.93", # "volume24h": "6780.866843", # "turnover24h": "141946527.22907118", # "price24hPcnt": "0.0196", # "usdIndexPrice": "21120.2400136" # } # } # # lt ticker # { # "topic": "tickers_lt.EOS3LUSDT", # "ts": 1672325446847, # "type": "snapshot", # "data": { # "symbol": "EOS3LUSDT", # "lastPrice": "0.41477848043290448", # "highPrice24h": "0.435285472510871305", # "lowPrice24h": "0.394601507960931382", # "prevPrice24h": "0.431502290172376349", # "price24hPcnt": "-0.0388" # } # } # swap delta # { # "topic":"tickers.AAVEUSDT", # "type":"delta", # "data":{ # "symbol":"AAVEUSDT", # "bid1Price":"112.89", # "bid1Size":"2.12", # "ask1Price":"112.90", # "ask1Size":"5.02" # }, # "cs":78039939929, # "ts":1709210212704 # } # topic = self.safe_string(message, 'topic', '') updateType = self.safe_string(message, 'type', '') data = self.safe_dict(message, 'data', {}) isSpot = self.safe_string(data, 'usdIndexPrice') is not None type = 'spot' if isSpot else 'contract' symbol = None parsed = None if (updateType == 'snapshot'): parsed = self.parse_ticker(data) symbol = parsed['symbol'] elif updateType == 'delta': topicParts = topic.split('.') topicLength = len(topicParts) marketId = self.safe_string(topicParts, topicLength - 1) market = self.safe_market(marketId, None, None, type) symbol = market['symbol'] # update the info in place ticker = self.safe_dict(self.tickers, symbol, {}) rawTicker = self.safe_dict(ticker, 'info', {}) merged = self.extend(rawTicker, data) parsed = self.parse_ticker(merged) timestamp = self.safe_integer(message, 'ts') parsed['timestamp'] = timestamp parsed['datetime'] = self.iso8601(timestamp) self.tickers[symbol] = parsed messageHash = 'ticker:' + symbol client.resolve(self.tickers[symbol], messageHash) async def watch_bids_asks(self, symbols: Strings = None, params={}) -> Tickers: """ watches best bid & ask for symbols https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook :param str[] symbols: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False) messageHashes = [] url = await self.get_url_by_market_type(symbols[0], False, 'watchBidsAsks', params) params = self.clean_params(params) marketIds = self.market_ids(symbols) topics = [] for i in range(0, len(marketIds)): marketId = marketIds[i] topic = 'orderbook.1.' + marketId topics.append(topic) messageHashes.append('bidask:' + symbols[i]) ticker = await self.watch_topics(url, messageHashes, topics, params) if self.newUpdates: return ticker return self.filter_by_array(self.bidsasks, 'symbol', symbols) def parse_ws_bid_ask(self, orderbook, market=None): timestamp = self.safe_integer(orderbook, 'timestamp') bids = self.sort_by(self.aggregate(orderbook['bids']), 0) asks = self.sort_by(self.aggregate(orderbook['asks']), 0) bestBid = self.safe_list(bids, 0, []) bestAsk = self.safe_list(asks, 0, []) return self.safe_ticker({ 'symbol': market['symbol'], 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'ask': self.safe_number(bestAsk, 0), 'askVolume': self.safe_number(bestAsk, 1), 'bid': self.safe_number(bestBid, 0), 'bidVolume': self.safe_number(bestBid, 1), 'info': orderbook, }, market) async def watch_ohlcv(self, symbol: str, timeframe: str = '1m', since: Int = None, limit: Int = None, params={}) -> List[list]: """ watches historical candlestick data containing the open, high, low, and close price, and the volume of a market https://bybit-exchange.github.io/docs/v5/websocket/public/kline https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline :param str symbol: unified symbol of the market to fetch OHLCV data for :param str timeframe: the length of time each candle represents :param int [since]: timestamp in ms of the earliest candle to fetch :param int [limit]: the maximum amount of candles to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns int[][]: A list of candles ordered, open, high, low, close, volume """ params['callerMethodName'] = 'watchOHLCV' result = await self.watch_ohlcv_for_symbols([[symbol, timeframe]], since, limit, params) return result[symbol][timeframe] async def watch_ohlcv_for_symbols(self, symbolsAndTimeframes: List[List[str]], since: Int = None, limit: Int = None, params={}): """ watches historical candlestick data containing the open, high, low, and close price, and the volume of a market https://bybit-exchange.github.io/docs/v5/websocket/public/kline https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline :param str[][] symbolsAndTimeframes: array of arrays containing unified symbols and timeframes to fetch OHLCV data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']] :param int [since]: timestamp in ms of the earliest candle to fetch :param int [limit]: the maximum amount of candles to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() symbols = self.get_list_from_object_values(symbolsAndTimeframes, 0) marketSymbols = self.market_symbols(symbols, None, False, True, True) firstSymbol = marketSymbols[0] url = await self.get_url_by_market_type(firstSymbol, False, 'watchOHLCVForSymbols', params) rawHashes = [] messageHashes = [] for i in range(0, len(symbolsAndTimeframes)): data = symbolsAndTimeframes[i] symbolString = self.safe_string(data, 0) market = self.market(symbolString) symbolString = market['symbol'] unfiedTimeframe = self.safe_string(data, 1) timeframeId = self.safe_string(self.timeframes, unfiedTimeframe, unfiedTimeframe) rawHashes.append('kline.' + timeframeId + '.' + market['id']) messageHashes.append('ohlcv::' + symbolString + '::' + unfiedTimeframe) symbol, timeframe, stored = await self.watch_topics(url, messageHashes, rawHashes, params) if self.newUpdates: limit = stored.getLimit(symbol, limit) filtered = self.filter_by_since_limit(stored, since, limit, 0, True) return self.create_ohlcv_object(symbol, timeframe, filtered) async def un_watch_ohlcv_for_symbols(self, symbolsAndTimeframes: List[List[str]], params={}) -> Any: """ unWatches historical candlestick data containing the open, high, low, and close price, and the volume of a market https://bybit-exchange.github.io/docs/v5/websocket/public/kline https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline :param str[][] symbolsAndTimeframes: array of arrays containing unified symbols and timeframes to fetch OHLCV data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']] :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() symbols = self.get_list_from_object_values(symbolsAndTimeframes, 0) marketSymbols = self.market_symbols(symbols, None, False, True, True) firstSymbol = marketSymbols[0] url = await self.get_url_by_market_type(firstSymbol, False, 'watchOHLCVForSymbols', params) rawHashes = [] subMessageHashes = [] messageHashes = [] for i in range(0, len(symbolsAndTimeframes)): data = symbolsAndTimeframes[i] symbolString = self.safe_string(data, 0) market = self.market(symbolString) symbolString = market['symbol'] unfiedTimeframe = self.safe_string(data, 1) timeframeId = self.safe_string(self.timeframes, unfiedTimeframe, unfiedTimeframe) rawHashes.append('kline.' + timeframeId + '.' + market['id']) subMessageHashes.append('ohlcv::' + symbolString + '::' + unfiedTimeframe) messageHashes.append('unsubscribe::ohlcv::' + symbolString + '::' + unfiedTimeframe) subExtension = { 'symbolsAndTimeframes': symbolsAndTimeframes, } return await self.un_watch_topics(url, 'ohlcv', symbols, messageHashes, subMessageHashes, rawHashes, params, subExtension) async def un_watch_ohlcv(self, symbol: str, timeframe: str = '1m', params={}) -> Any: """ unWatches historical candlestick data containing the open, high, low, and close price, and the volume of a market https://bybit-exchange.github.io/docs/v5/websocket/public/kline https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline :param str symbol: unified symbol of the market to fetch OHLCV data for :param str timeframe: the length of time each candle represents :param dict [params]: extra parameters specific to the exchange API endpoint :returns int[][]: A list of candles ordered, open, high, low, close, volume """ params['callerMethodName'] = 'watchOHLCV' return await self.un_watch_ohlcv_for_symbols([[symbol, timeframe]], params) def handle_ohlcv(self, client: Client, message): # # { # "topic": "kline.5.BTCUSDT", # "data": [ # { # "start": 1672324800000, # "end": 1672325099999, # "interval": "5", # "open": "16649.5", # "close": "16677", # "high": "16677", # "low": "16608", # "volume": "2.081", # "turnover": "34666.4005", # "confirm": False, # "timestamp": 1672324988882 # } # ], # "ts": 1672324988882, # "type": "snapshot" # } # data = self.safe_value(message, 'data', {}) topic = self.safe_string(message, 'topic') topicParts = topic.split('.') topicLength = len(topicParts) timeframeId = self.safe_string(topicParts, 1) timeframe = self.find_timeframe(timeframeId) marketId = self.safe_string(topicParts, topicLength - 1) isSpot = client.url.find('spot') > -1 marketType = 'spot' if isSpot else 'contract' market = self.safe_market(marketId, None, None, marketType) symbol = market['symbol'] ohlcvsByTimeframe = self.safe_value(self.ohlcvs, symbol) if ohlcvsByTimeframe is None: self.ohlcvs[symbol] = {} if self.safe_value(ohlcvsByTimeframe, timeframe) is None: limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) self.ohlcvs[symbol][timeframe] = ArrayCacheByTimestamp(limit) stored = self.ohlcvs[symbol][timeframe] for i in range(0, len(data)): parsed = self.parse_ws_ohlcv(data[i], market) stored.append(parsed) messageHash = 'ohlcv::' + symbol + '::' + timeframe resolveData = [symbol, timeframe, stored] client.resolve(resolveData, messageHash) def parse_ws_ohlcv(self, ohlcv, market=None) -> list: # # { # "start": 1670363160000, # "end": 1670363219999, # "interval": "1", # "open": "16987.5", # "close": "16987.5", # "high": "16988", # "low": "16987.5", # "volume": "23.511", # "turnover": "399396.344", # "confirm": False, # "timestamp": 1670363219614 # } # volumeIndex = 'turnover' if (market['inverse']) else 'volume' return [ self.safe_integer(ohlcv, 'start'), self.safe_number(ohlcv, 'open'), self.safe_number(ohlcv, 'high'), self.safe_number(ohlcv, 'low'), self.safe_number(ohlcv, 'close'), self.safe_number(ohlcv, volumeIndex), ] async def watch_order_book(self, symbol: str, limit: Int = None, params={}) -> OrderBook: """ watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook :param str symbol: unified symbol of the market to fetch the order book for :param int [limit]: the maximum amount of order book entries to return. :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A dictionary of `order book structures ` indexed by market symbols """ return await self.watch_order_book_for_symbols([symbol], limit, params) async def watch_order_book_for_symbols(self, symbols: List[str], limit: Int = None, params={}) -> OrderBook: """ watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook :param str[] symbols: unified array of symbols :param int [limit]: the maximum amount of order book entries to return. :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() symbolsLength = len(symbols) if symbolsLength == 0: raise ArgumentsRequired(self.id + ' watchOrderBookForSymbols() requires a non-empty array of symbols') symbols = self.market_symbols(symbols) url = await self.get_url_by_market_type(symbols[0], False, 'watchOrderBook', params) params = self.clean_params(params) market = self.market(symbols[0]) if limit is None: limit = 50 if (market['spot']) else 500 if market['option']: limit = 100 else: limits = { 'spot': [1, 50, 200, 1000], 'option': [25, 100], 'default': [1, 50, 200, 500, 1000], } selectedLimits = self.safe_list_2(limits, market['type'], 'default') if not self.in_array(limit, selectedLimits): raise BadRequest(self.id + ' watchOrderBookForSymbols(): for ' + market['type'] + ' markets limit can be one of: ' + self.json(selectedLimits)) topics = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] marketId = self.market_id(symbol) topic = 'orderbook.' + str(limit) + '.' + marketId topics.append(topic) messageHash = 'orderbook:' + symbol messageHashes.append(messageHash) orderbook = await self.watch_topics(url, messageHashes, topics, params) return orderbook.limit() async def un_watch_order_book_for_symbols(self, symbols: List[str], params={}) -> Any: """ unsubscribe from the orderbook channel https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook :param str[] symbols: unified symbol of the market to unwatch the trades for :param dict [params]: extra parameters specific to the exchange API endpoint :param int [params.limit]: orderbook limit, default is None :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() symbols = self.market_symbols(symbols, None, False) channel = 'orderbook.' limit = self.safe_integer(params, 'limit') if limit is not None: params = self.omit(params, 'limit') else: firstMarket = self.market(symbols[0]) limit = 50 if firstMarket['spot'] else 500 channel += str(limit) subMessageHashes = [] messageHashes = [] topics = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) marketId = market['id'] topic = channel + '.' + marketId messageHashes.append('unsubscribe:orderbook:' + symbol) subMessageHashes.append('orderbook:' + symbol) topics.append(topic) url = await self.get_url_by_market_type(symbols[0], False, 'watchOrderBook', params) return await self.un_watch_topics(url, 'orderbook', symbols, messageHashes, subMessageHashes, topics, params) async def un_watch_order_book(self, symbol: str, params={}) -> Any: """ unsubscribe from the orderbook channel https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook :param str symbol: symbol of the market to unwatch the trades for :param dict [params]: extra parameters specific to the exchange API endpoint :param int [params.limit]: orderbook limit, default is None :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() return await self.un_watch_order_book_for_symbols([symbol], params) def handle_order_book(self, client: Client, message): # # { # "topic": "orderbook.50.BTCUSDT", # "type": "snapshot", # "ts": 1672304484978, # "data": { # "s": "BTCUSDT", # "b": [ # ..., # [ # "16493.50", # "0.006" # ], # [ # "16493.00", # "0.100" # ] # ], # "a": [ # [ # "16611.00", # "0.029" # ], # [ # "16612.00", # "0.213" # ], # ], # "u": 18521288, # "seq": 7961638724 # } # } # topic = self.safe_string(message, 'topic') limit = topic.split('.')[1] isSpot = client.url.find('spot') >= 0 type = self.safe_string(message, 'type') isSnapshot = (type == 'snapshot') data = self.safe_dict(message, 'data', {}) marketId = self.safe_string(data, 's') marketType = 'spot' if isSpot else 'contract' market = self.safe_market(marketId, None, None, marketType) symbol = market['symbol'] timestamp = self.safe_integer(message, 'ts') if not (symbol in self.orderbooks): self.orderbooks[symbol] = self.order_book() orderbook = self.orderbooks[symbol] orderbook['symbol'] = symbol if isSnapshot: snapshot = self.parse_order_book(data, symbol, timestamp, 'b', 'a') orderbook.reset(snapshot) else: asks = self.safe_list(data, 'a', []) bids = self.safe_list(data, 'b', []) self.handle_deltas(orderbook['asks'], asks) self.handle_deltas(orderbook['bids'], bids) orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) messageHash = 'orderbook' + ':' + symbol self.orderbooks[symbol] = orderbook client.resolve(orderbook, messageHash) if limit == '1': bidask = self.parse_ws_bid_ask(self.orderbooks[symbol], market) newBidsAsks: dict = {} newBidsAsks[symbol] = bidask self.bidsasks[symbol] = bidask client.resolve(newBidsAsks, 'bidask:' + symbol) def handle_delta(self, bookside, delta): bidAsk = self.parse_bid_ask(delta, 0, 1) bookside.storeArray(bidAsk) def handle_deltas(self, bookside, deltas): for i in range(0, len(deltas)): self.handle_delta(bookside, deltas[i]) async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ watches information on multiple trades made in a market https://bybit-exchange.github.io/docs/v5/websocket/public/trade :param str symbol: unified market symbol of the market trades were made in :param int [since]: the earliest time in ms to fetch trades for :param int [limit]: the maximum number of trade structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `trade structures ` """ return await self.watch_trades_for_symbols([symbol], since, limit, params) async def watch_trades_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ get the list of most recent trades for a list of symbols https://bybit-exchange.github.io/docs/v5/websocket/public/trade :param str[] symbols: unified symbol of the market to fetch trades for :param int [since]: timestamp in ms of the earliest trade to fetch :param int [limit]: the maximum amount of trades to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `trade structures ` """ await self.load_markets() symbols = self.market_symbols(symbols) symbolsLength = len(symbols) if symbolsLength == 0: raise ArgumentsRequired(self.id + ' watchTradesForSymbols() requires a non-empty array of symbols') params = self.clean_params(params) url = await self.get_url_by_market_type(symbols[0], False, 'watchTrades', params) topics = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) topic = 'publicTrade.' + market['id'] topics.append(topic) messageHash = 'trade:' + symbol messageHashes.append(messageHash) trades = await self.watch_topics(url, messageHashes, topics, params) if self.newUpdates: first = self.safe_value(trades, 0) tradeSymbol = self.safe_string(first, 'symbol') limit = trades.getLimit(tradeSymbol, limit) return self.filter_by_since_limit(trades, since, limit, 'timestamp', True) async def un_watch_trades_for_symbols(self, symbols: List[str], params={}) -> Any: """ unsubscribe from the trades channel https://bybit-exchange.github.io/docs/v5/websocket/public/trade :param str[] symbols: unified symbol of the market to unwatch the trades for :param dict [params]: extra parameters specific to the exchange API endpoint :returns any: status of the unwatch request """ await self.load_markets() symbols = self.market_symbols(symbols, None, False, True) url = await self.get_url_by_market_type(symbols[0], False, 'unWatchTradesForSymbols', params) messageHashes = [] topics = [] subMessageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) topic = 'publicTrade.' + market['id'] topics.append(topic) messageHash = 'unsubscribe:trade:' + symbol messageHashes.append(messageHash) subMessageHashes.append('trade:' + symbol) return await self.un_watch_topics(url, 'trades', symbols, messageHashes, subMessageHashes, topics, params) async def un_watch_trades(self, symbol: str, params={}) -> Any: """ unsubscribe from the trades channel https://bybit-exchange.github.io/docs/v5/websocket/public/trade :param str symbol: unified symbol of the market to unwatch the trades for :param dict [params]: extra parameters specific to the exchange API endpoint :returns any: status of the unwatch request """ await self.load_markets() return await self.un_watch_trades_for_symbols([symbol], params) def handle_trades(self, client: Client, message): # # { # "topic": "publicTrade.BTCUSDT", # "type": "snapshot", # "ts": 1672304486868, # "data": [ # { # "T": 1672304486865, # "s": "BTCUSDT", # "S": "Buy", # "v": "0.001", # "p": "16578.50", # "L": "PlusTick", # "i": "20f43950-d8dd-5b31-9112-a178eb6023af", # "BT": False # } # ] # } # data = self.safe_value(message, 'data', {}) topic = self.safe_string(message, 'topic') trades = data parts = topic.split('.') isSpot = client.url.find('spot') >= 0 marketType = 'spot' if (isSpot) else 'contract' marketId = self.safe_string(parts, 1) market = self.safe_market(marketId, None, None, marketType) symbol = market['symbol'] stored = self.safe_value(self.trades, symbol) if stored is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) stored = ArrayCache(limit) self.trades[symbol] = stored for j in range(0, len(trades)): parsed = self.parse_ws_trade(trades[j], market) stored.append(parsed) messageHash = 'trade' + ':' + symbol client.resolve(stored, messageHash) def parse_ws_trade(self, trade, market=None): # # public # { # "T": 1672304486865, # "s": "BTCUSDT", # "S": "Buy", # "v": "0.001", # "p": "16578.50", # "L": "PlusTick", # "i": "20f43950-d8dd-5b31-9112-a178eb6023af", # "BT": False # } # # spot private # { # "e": "ticketInfo", # "E": "1662348310386", # "s": "BTCUSDT", # "q": "0.001007", # "t": "1662348310373", # "p": "19842.02", # "T": "2100000000002220938", # "o": "1238261807653647872", # "c": "spotx008", # "O": "1238225004531834368", # "a": "533287", # "A": "642908", # "m": False, # "S": "BUY" # } # id = self.safe_string_n(trade, ['i', 'T', 'v']) isContract = ('BT' in trade) marketType = 'contract' if isContract else 'spot' if market is not None: marketType = market['type'] marketId = self.safe_string(trade, 's') market = self.safe_market(marketId, market, None, marketType) symbol = market['symbol'] timestamp = self.safe_integer_2(trade, 't', 'T') side = self.safe_string_lower(trade, 'S') takerOrMaker = None m = self.safe_value(trade, 'm') if side is None: side = 'buy' if m else 'sell' else: # spot private takerOrMaker = m price = self.safe_string(trade, 'p') amount = self.safe_string_2(trade, 'q', 'v') orderId = self.safe_string(trade, 'o') return self.safe_trade({ 'id': id, 'info': trade, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'symbol': symbol, 'order': orderId, 'type': None, 'side': side, 'takerOrMaker': takerOrMaker, 'price': price, 'amount': amount, 'cost': None, 'fee': None, }, market) def get_private_type(self, url): if url.find('spot') >= 0: return 'spot' elif url.find('v5/private') >= 0: return 'unified' else: return 'usdc' async def watch_my_trades(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ watches information on multiple trades made by the user https://bybit-exchange.github.io/docs/v5/websocket/private/execution https://bybit-exchange.github.io/docs/v5/websocket/private/fast-execution :param str symbol: unified market symbol of the market orders were made in :param int [since]: the earliest time in ms to fetch orders for :param int [limit]: the maximum number of order structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.unifiedMargin]: use unified margin account :param boolean [params.executionFast]: use fast execution :returns dict[]: a list of `order structures ` """ method = 'watchMyTrades' messageHash = 'myTrades' await self.load_markets() if symbol is not None: symbol = self.symbol(symbol) messageHash += ':' + symbol url = await self.get_url_by_market_type(symbol, True, method, params) await self.authenticate(url) topicByMarket: dict = { 'spot': 'ticketInfo', 'unified': 'execution', 'usdc': 'user.openapi.perp.trade', } topic = self.safe_value(topicByMarket, self.get_private_type(url)) executionFast = False executionFast, params = self.handle_option_and_params(params, 'watchMyTrades', 'executionFast', False) if executionFast: topic = 'execution.fast' trades = await self.watch_topics(url, [messageHash], [topic], params) if self.newUpdates: limit = trades.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(trades, symbol, since, limit, True) async def un_watch_my_trades(self, symbol: Str = None, params={}) -> Any: """ unWatches information on multiple trades made by the user https://bybit-exchange.github.io/docs/v5/websocket/private/execution https://bybit-exchange.github.io/docs/v5/websocket/private/fast-execution :param str symbol: unified market symbol of the market orders were made in :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.unifiedMargin]: use unified margin account :param boolean [params.executionFast]: use fast execution :returns dict[]: a list of `order structures ` """ method = 'watchMyTrades' messageHash = 'unsubscribe:myTrades' subHash = 'myTrades' await self.load_markets() if symbol is not None: raise NotSupported(self.id + ' unWatchMyTrades() does not support a symbol parameter, you must unwatch all my trades') url = await self.get_url_by_market_type(symbol, True, method, params) await self.authenticate(url) topicByMarket: dict = { 'spot': 'ticketInfo', 'unified': 'execution', 'usdc': 'user.openapi.perp.trade', } topic = self.safe_value(topicByMarket, self.get_private_type(url)) executionFast = False executionFast, params = self.handle_option_and_params(params, 'watchMyTrades', 'executionFast', False) if executionFast: topic = 'execution.fast' return await self.un_watch_topics(url, 'myTrades', [], [messageHash], [subHash], [topic], params) def handle_my_trades(self, client: Client, message): # # spot # { # "type": "snapshot", # "topic": "ticketInfo", # "ts": "1662348310388", # "data": [ # { # "e": "ticketInfo", # "E": "1662348310386", # "s": "BTCUSDT", # "q": "0.001007", # "t": "1662348310373", # "p": "19842.02", # "T": "2100000000002220938", # "o": "1238261807653647872", # "c": "spotx008", # "O": "1238225004531834368", # "a": "533287", # "A": "642908", # "m": False, # "S": "BUY" # } # ] # } # unified # { # "id": "592324803b2785-26fa-4214-9963-bdd4727f07be", # "topic": "execution", # "creationTime": 1672364174455, # "data": [ # { # "category": "linear", # "symbol": "XRPUSDT", # "execFee": "0.005061", # "execId": "7e2ae69c-4edf-5800-a352-893d52b446aa", # "execPrice": "0.3374", # "execQty": "25", # "execType": "Trade", # "execValue": "8.435", # "isMaker": False, # "feeRate": "0.0006", # "tradeIv": "", # "markIv": "", # "blockTradeId": "", # "markPrice": "0.3391", # "indexPrice": "", # "underlyingPrice": "", # "leavesQty": "0", # "orderId": "f6e324ff-99c2-4e89-9739-3086e47f9381", # "orderLinkId": "", # "orderPrice": "0.3207", # "orderQty": "25", # "orderType": "Market", # "stopOrderType": "UNKNOWN", # "side": "Sell", # "execTime": "1672364174443", # "isLeverage": "0" # } # ] # } # # execution.fast # # { # "topic": "execution.fast", # "creationTime": 1757405601981, # "data": [ # { # "category": "linear", # "symbol": "BTCUSDT", # "execId": "ffcac6ac-7571-536d-a28a-847dd7d08a0f", # "execPrice": "112529.6", # "execQty": "0.001", # "orderId": "6e25ab73-7a55-4ae7-adc2-8ea95f167c85", # "isMaker": False, # "orderLinkId": "test-00001", # "side": "Buy", # "execTime": "1757405601977", # "seq": 9515624038 # } # ] # } # topic = self.safe_string(message, 'topic') spot = topic == 'ticketInfo' executionFast = topic == 'execution.fast' data = self.safe_value(message, 'data', []) if not isinstance(data, list): data = self.safe_value(data, 'result', []) if self.myTrades is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) self.myTrades = ArrayCacheBySymbolById(limit) trades = self.myTrades symbols: dict = {} filterExecTypes = self.handle_option('watchMyTrades', 'filterExecTypes', []) for i in range(0, len(data)): rawTrade = data[i] parsed = None if spot and not executionFast: parsed = self.parse_ws_trade(rawTrade) else: # filter unified trades execType = self.safe_string(rawTrade, 'execType', '') if executionFast: execType = 'Trade' if not self.in_array(execType, filterExecTypes): continue parsed = self.parse_trade(rawTrade) symbol = parsed['symbol'] symbols[symbol] = True trades.append(parsed) keys = list(symbols.keys()) for i in range(0, len(keys)): currentMessageHash = 'myTrades:' + keys[i] client.resolve(trades, currentMessageHash) # non-symbol specific messageHash = 'myTrades' client.resolve(trades, messageHash) async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]: """ https://bybit-exchange.github.io/docs/v5/websocket/private/position watch all open positions :param str[] [symbols]: list of unified market symbols :param int [since]: the earliest time in ms to fetch positions for :param int [limit]: the maximum number of positions to retrieve :param dict params: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `position structure ` """ await self.load_markets() method = 'watchPositions' messageHash = '' if not self.is_empty(symbols): symbols = self.market_symbols(symbols) messageHash = '::' + ','.join(symbols) firstSymbol = self.safe_string(symbols, 0) url = await self.get_url_by_market_type(firstSymbol, True, method, params) messageHash = 'positions' + messageHash client = self.client(url) await self.authenticate(url) self.set_positions_cache(client, symbols) cache = self.positions fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', True) awaitPositionsSnapshot = self.handle_option('watchPositions', 'awaitPositionsSnapshot', True) if fetchPositionsSnapshot and awaitPositionsSnapshot and cache is None: snapshot = await client.future('fetchPositionsSnapshot') return self.filter_by_symbols_since_limit(snapshot, symbols, since, limit, True) topics = ['position'] newPositions = await self.watch_topics(url, [messageHash], topics, params) if self.newUpdates: return newPositions return self.filter_by_symbols_since_limit(cache, symbols, since, limit, True) def set_positions_cache(self, client: Client, symbols: Strings = None): if self.positions is not None: return fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', True) if fetchPositionsSnapshot: messageHash = 'fetchPositionsSnapshot' if not (messageHash in client.futures): client.future(messageHash) self.spawn(self.load_positions_snapshot, client, messageHash) else: self.positions = ArrayCacheBySymbolBySide() async def load_positions_snapshot(self, client, messageHash): # one ws channel gives positions for all types, for snapshot must load all positions fetchFunctions = [ self.fetch_positions(None, {'type': 'swap', 'subType': 'linear'}), self.fetch_positions(None, {'type': 'swap', 'subType': 'inverse'}), ] promises = await asyncio.gather(*fetchFunctions) self.positions = ArrayCacheBySymbolBySide() cache = self.positions for i in range(0, len(promises)): positions = promises[i] for ii in range(0, len(positions)): position = positions[ii] cache.append(position) # don't remove the future from the .futures cache future = client.futures[messageHash] future.resolve(cache) client.resolve(cache, 'position') def handle_positions(self, client, message): # # { # topic: 'position', # id: '504b2671629b08e3c4f6960382a59363:3bc4028023786545:0:01', # creationTime: 1694566055295, # data: [{ # bustPrice: '15.00', # category: 'inverse', # createdTime: '1670083436351', # cumRealisedPnl: '0.00011988', # entryPrice: '19358.58553268', # leverage: '10', # liqPrice: '15.00', # markPrice: '25924.00', # positionBalance: '0.0000156', # positionIdx: 0, # positionMM: '0.001', # positionIM: '0.0000015497', # positionStatus: 'Normal', # positionValue: '0.00015497', # riskId: 1, # riskLimitValue: '150', # side: 'Buy', # size: '3', # stopLoss: '0.00', # symbol: 'BTCUSD', # takeProfit: '0.00', # tpslMode: 'Full', # tradeMode: 0, # autoAddMargin: 1, # trailingStop: '0.00', # unrealisedPnl: '0.00003925', # updatedTime: '1694566055293', # adlRankIndicator: 3 # }] # } # # each account is connected to a different endpoint # and has exactly one subscriptionhash which is the account type if self.positions is None: self.positions = ArrayCacheBySymbolBySide() cache = self.positions newPositions = [] rawPositions = self.safe_value(message, 'data', []) for i in range(0, len(rawPositions)): rawPosition = rawPositions[i] position = self.parse_position(rawPosition) side = self.safe_string(position, 'side') # hacky solution to handle closing positions # without crashing, we should handle self properly later newPositions.append(position) if side is None or side == '': # closing update, adding both sides to "reset" both sides # since we don't know which side is being closed position['side'] = 'long' cache.append(position) position['side'] = 'short' cache.append(position) position['side'] = None else: # regular update cache.append(position) messageHashes = self.find_message_hashes(client, 'positions::') for i in range(0, len(messageHashes)): messageHash = messageHashes[i] parts = messageHash.split('::') symbolsString = parts[1] symbols = symbolsString.split(',') positions = self.filter_by_array(newPositions, 'symbol', symbols, False) if not self.is_empty(positions): client.resolve(positions, messageHash) client.resolve(newPositions, 'positions') async def un_watch_positions(self, symbols: Strings = None, params={}) -> Any: """ unWatches all open positions https://bybit-exchange.github.io/docs/v5/websocket/private/position :param str[] [symbols]: list of unified market symbols :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: status of the unwatch request """ await self.load_markets() method = 'watchPositions' messageHash = 'unsubscribe:positions' subHash = 'positions' if not self.is_empty(symbols): raise NotSupported(self.id + ' unWatchPositions() does not support a symbol parameter, you must unwatch all orders') url = await self.get_url_by_market_type(None, True, method, params) await self.authenticate(url) topics = ['position'] return await self.un_watch_topics(url, 'positions', symbols, [messageHash], [subHash], topics, params) async def watch_liquidations(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Liquidation]: """ watch the public liquidations of a trading pair https://bybit-exchange.github.io/docs/v5/websocket/public/liquidation :param str symbol: unified CCXT market symbol :param int [since]: the earliest time in ms to fetch liquidations for :param int [limit]: the maximum number of liquidation structures to retrieve :param dict [params]: exchange specific parameters for the bitmex api endpoint :param str [params.method]: exchange specific method, supported: liquidation, allLiquidation :returns dict: an array of `liquidation structures ` """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] url = await self.get_url_by_market_type(symbol, False, 'watchLiquidations', params) params = self.clean_params(params) method = None method, params = self.handle_option_and_params(params, 'watchLiquidations', 'method', 'liquidation') messageHash = 'liquidations::' + symbol topic = method + '.' + market['id'] newLiquidation = await self.watch_topics(url, [messageHash], [topic], params) if self.newUpdates: return newLiquidation return self.filter_by_symbols_since_limit(self.liquidations, [symbol], since, limit, True) def handle_liquidation(self, client: Client, message): # # { # "data": { # "price": "0.03803", # "side": "Buy", # "size": "1637", # "symbol": "GALAUSDT", # "updatedTime": 1673251091822 # }, # "topic": "liquidation.GALAUSDT", # "ts": 1673251091822, # "type": "snapshot" # } # # { # "topic": "allLiquidation.ROSEUSDT", # "type": "snapshot", # "ts": 1739502303204, # "data": [ # { # "T": 1739502302929, # "s": "ROSEUSDT", # "S": "Sell", # "v": "20000", # "p": "0.04499" # } # ] # } # if isinstance(message['data'], list): rawLiquidations = self.safe_list(message, 'data', []) for i in range(0, len(rawLiquidations)): rawLiquidation = rawLiquidations[i] marketId = self.safe_string(rawLiquidation, 's') market = self.safe_market(marketId, None, '', 'contract') symbol = market['symbol'] liquidation = self.parse_ws_liquidation(rawLiquidation, market) liquidations = self.safe_value(self.liquidations, symbol) if liquidations is None: limit = self.safe_integer(self.options, 'liquidationsLimit', 1000) liquidations = ArrayCache(limit) liquidations.append(liquidation) self.liquidations[symbol] = liquidations client.resolve([liquidation], 'liquidations') client.resolve([liquidation], 'liquidations::' + symbol) else: rawLiquidation = self.safe_dict(message, 'data', {}) marketId = self.safe_string(rawLiquidation, 'symbol') market = self.safe_market(marketId, None, '', 'contract') symbol = market['symbol'] liquidation = self.parse_ws_liquidation(rawLiquidation, market) liquidations = self.safe_value(self.liquidations, symbol) if liquidations is None: limit = self.safe_integer(self.options, 'liquidationsLimit', 1000) liquidations = ArrayCache(limit) liquidations.append(liquidation) self.liquidations[symbol] = liquidations client.resolve([liquidation], 'liquidations') client.resolve([liquidation], 'liquidations::' + symbol) def parse_ws_liquidation(self, liquidation, market=None): # # { # "price": "0.03803", # "side": "Buy", # "size": "1637", # "symbol": "GALAUSDT", # "updatedTime": 1673251091822 # } # # { # "T": 1739502302929, # "s": "ROSEUSDT", # "S": "Sell", # "v": "20000", # "p": "0.04499" # } # marketId = self.safe_string_2(liquidation, 'symbol', 's') market = self.safe_market(marketId, market, '', 'contract') timestamp = self.safe_integer_2(liquidation, 'updatedTime', 'T') return self.safe_liquidation({ 'info': liquidation, 'symbol': market['symbol'], 'contracts': self.safe_number_2(liquidation, 'size', 'v'), 'contractSize': self.safe_number(market, 'contractSize'), 'price': self.safe_number_2(liquidation, 'price', 'p'), 'side': self.safe_string_lower(liquidation, 'side', 'S'), 'baseValue': None, 'quoteValue': None, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), }) async def watch_orders(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ watches information on multiple orders made by the user https://bybit-exchange.github.io/docs/v5/websocket/private/order :param str symbol: unified market symbol of the market orders were made in :param int [since]: the earliest time in ms to fetch orders for :param int [limit]: the maximum number of order structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `order structures ` """ await self.load_markets() method = 'watchOrders' messageHash = 'orders' if symbol is not None: symbol = self.symbol(symbol) messageHash += ':' + symbol url = await self.get_url_by_market_type(symbol, True, method, params) await self.authenticate(url) topicsByMarket: dict = { 'spot': ['order', 'stopOrder'], 'unified': ['order'], 'usdc': ['user.openapi.perp.order'], } topics = self.safe_value(topicsByMarket, self.get_private_type(url)) orders = await self.watch_topics(url, [messageHash], topics, params) if self.newUpdates: limit = orders.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True) async def un_watch_orders(self, symbol: Str = None, params={}) -> Any: """ unWatches information on multiple orders made by the user https://bybit-exchange.github.io/docs/v5/websocket/private/order :param str symbol: unified market symbol of the market orders were made in :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.unifiedMargin]: use unified margin account :returns dict[]: a list of `order structures ` """ await self.load_markets() method = 'watchOrders' messageHash = 'unsubscribe:orders' subHash = 'orders' if symbol is not None: raise NotSupported(self.id + ' unWatchOrders() does not support a symbol parameter, you must unwatch all orders') url = await self.get_url_by_market_type(symbol, True, method, params) await self.authenticate(url) topicsByMarket: dict = { 'spot': ['order', 'stopOrder'], 'unified': ['order'], 'usdc': ['user.openapi.perp.order'], } topics = self.safe_value(topicsByMarket, self.get_private_type(url)) return await self.un_watch_topics(url, 'orders', [], [messageHash], [subHash], topics, params) def handle_order_ws(self, client: Client, message): # # { # "reqId":"1", # "retCode":0, # "retMsg":"OK", # "op":"order.create", # "data":{ # "orderId":"1673523595617593600", # "orderLinkId":"1673523595617593601" # }, # "header":{ # "X-Bapi-Limit":"20", # "X-Bapi-Limit-Status":"19", # "X-Bapi-Limit-Reset-Timestamp":"1714235558880", # "Traceid":"584a06d373f2fdcb3a4dfdd81d27df11", # "Timenow":"1714235558881" # }, # "connId":"cojidqec0hv9fgvhtbt0-40e" # } # messageHash = self.safe_string(message, 'reqId') data = self.safe_dict(message, 'data') order = self.parse_order(data) client.resolve(order, messageHash) def handle_order(self, client: Client, message): # # spot # { # "type": "snapshot", # "topic": "order", # "ts": "1662348310441", # "data": [ # { # "e": "order", # "E": "1662348310441", # "s": "BTCUSDT", # "c": "spotx008", # "S": "BUY", # "o": "MARKET_OF_QUOTE", # "f": "GTC", # "q": "20", # "p": "0", # "X": "CANCELED", # "i": "1238261807653647872", # "M": "1238225004531834368", # "l": "0.001007", # "z": "0.001007", # "L": "19842.02", # "n": "0", # "N": "BTC", # "u": True, # "w": True, # "m": False, # "O": "1662348310368", # "Z": "19.98091414", # "A": "0", # "C": False, # "v": "0", # "d": "NO_LIQ", # "t": "2100000000002220938" # } # ] # } # unified # { # "id": "5923240c6880ab-c59f-420b-9adb-3639adc9dd90", # "topic": "order", # "creationTime": 1672364262474, # "data": [ # { # "symbol": "ETH-30DEC22-1400-C", # "orderId": "5cf98598-39a7-459e-97bf-76ca765ee020", # "side": "Sell", # "orderType": "Market", # "cancelType": "UNKNOWN", # "price": "72.5", # "qty": "1", # "orderIv": "", # "timeInForce": "IOC", # "orderStatus": "Filled", # "orderLinkId": "", # "lastPriceOnCreated": "", # "reduceOnly": False, # "leavesQty": "", # "leavesValue": "", # "cumExecQty": "1", # "cumExecValue": "75", # "avgPrice": "75", # "blockTradeId": "", # "positionIdx": 0, # "cumExecFee": "0.358635", # "createdTime": "1672364262444", # "updatedTime": "1672364262457", # "rejectReason": "EC_NoError", # "stopOrderType": "", # "triggerPrice": "", # "takeProfit": "", # "stopLoss": "", # "tpTriggerBy": "", # "slTriggerBy": "", # "triggerDirection": 0, # "triggerBy": "", # "closeOnTrigger": False, # "category": "option" # } # ] # } # if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) self.orders = ArrayCacheBySymbolById(limit) orders = self.orders rawOrders = self.safe_value(message, 'data', []) first = self.safe_value(rawOrders, 0, {}) category = self.safe_string(first, 'category') isSpot = category == 'spot' if not isSpot: rawOrders = self.safe_value(rawOrders, 'result', rawOrders) symbols: dict = {} for i in range(0, len(rawOrders)): parsed = self.parse_order(rawOrders[i]) # if isSpot: # parsed = self.parseWsSpotOrder(rawOrders[i]) # else: # parsed = self.parse_order(rawOrders[i]) # } symbol = parsed['symbol'] symbols[symbol] = True orders.append(parsed) symbolsArray = list(symbols.keys()) for i in range(0, len(symbolsArray)): currentMessageHash = 'orders:' + symbolsArray[i] client.resolve(orders, currentMessageHash) messageHash = 'orders' client.resolve(orders, messageHash) async def watch_balance(self, params={}) -> Balances: """ watch balance and get the amount of funds available for trading or funds locked in orders https://bybit-exchange.github.io/docs/v5/websocket/private/wallet :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `balance structure ` """ await self.load_markets() method = 'watchBalance' messageHash = 'balances' type = None type, params = self.handle_market_type_and_params('watchBalance', None, params) subType = None subType, params = self.handle_sub_type_and_params('watchBalance', None, params) unified = await self.isUnifiedEnabled() isUnifiedMargin = self.safe_bool(unified, 0, False) isUnifiedAccount = self.safe_bool(unified, 1, False) url = await self.get_url_by_market_type(None, True, method, params) await self.authenticate(url) topicByMarket: dict = { 'spot': 'outboundAccountInfo', 'unified': 'wallet', } if isUnifiedAccount: # unified account if subType == 'inverse': messageHash += ':contract' else: messageHash += ':unified' if not isUnifiedMargin and not isUnifiedAccount: # normal account using v5 if type == 'spot': messageHash += ':spot' else: messageHash += ':contract' if isUnifiedMargin: # unified margin account using v5 if type == 'spot': messageHash += ':spot' else: if subType == 'linear': messageHash += ':unified' else: messageHash += ':contract' topics = [self.safe_value(topicByMarket, self.get_private_type(url))] return await self.watch_topics(url, [messageHash], topics, params) def handle_balance(self, client: Client, message): # # spot # { # "type": "snapshot", # "topic": "outboundAccountInfo", # "ts": "1662107217641", # "data": [ # { # "e": "outboundAccountInfo", # "E": "1662107217640", # "T": True, # "W": True, # "D": True, # "B": [ # { # "a": "USDT", # "f": "176.81254174", # "l": "201.575" # } # ] # } # ] # } # unified # { # "id": "5923242c464be9-25ca-483d-a743-c60101fc656f", # "topic": "wallet", # "creationTime": 1672364262482, # "data": [ # { # "accountIMRate": "0.016", # "accountMMRate": "0.003", # "totalEquity": "12837.78330098", # "totalWalletBalance": "12840.4045924", # "totalMarginBalance": "12837.78330188", # "totalAvailableBalance": "12632.05767702", # "totalPerpUPL": "-2.62129051", # "totalInitialMargin": "205.72562486", # "totalMaintenanceMargin": "39.42876721", # "coin": [ # { # "coin": "USDC", # "equity": "200.62572554", # "usdValue": "200.62572554", # "walletBalance": "201.34882644", # "availableToWithdraw": "0", # "availableToBorrow": "1500000", # "borrowAmount": "0", # "accruedInterest": "0", # "totalOrderIM": "0", # "totalPositionIM": "202.99874213", # "totalPositionMM": "39.14289747", # "unrealisedPnl": "74.2768991", # "cumRealisedPnl": "-209.1544627", # "bonus": "0" # }, # { # "coin": "BTC", # "equity": "0.06488393", # "usdValue": "1023.08402268", # "walletBalance": "0.06488393", # "availableToWithdraw": "0.06488393", # "availableToBorrow": "2.5", # "borrowAmount": "0", # "accruedInterest": "0", # "totalOrderIM": "0", # "totalPositionIM": "0", # "totalPositionMM": "0", # "unrealisedPnl": "0", # "cumRealisedPnl": "0", # "bonus": "0" # }, # { # "coin": "ETH", # "equity": "0", # "usdValue": "0", # "walletBalance": "0", # "availableToWithdraw": "0", # "availableToBorrow": "26", # "borrowAmount": "0", # "accruedInterest": "0", # "totalOrderIM": "0", # "totalPositionIM": "0", # "totalPositionMM": "0", # "unrealisedPnl": "0", # "cumRealisedPnl": "0", # "bonus": "0" # }, # { # "coin": "USDT", # "equity": "11726.64664904", # "usdValue": "11613.58597018", # "walletBalance": "11728.54414904", # "availableToWithdraw": "11723.92075829", # "availableToBorrow": "2500000", # "borrowAmount": "0", # "accruedInterest": "0", # "totalOrderIM": "0", # "totalPositionIM": "2.72589075", # "totalPositionMM": "0.28576575", # "unrealisedPnl": "-1.8975", # "cumRealisedPnl": "0.64782276", # "bonus": "0" # }, # { # "coin": "EOS3L", # "equity": "215.0570412", # "usdValue": "0", # "walletBalance": "215.0570412", # "availableToWithdraw": "215.0570412", # "availableToBorrow": "0", # "borrowAmount": "0", # "accruedInterest": "", # "totalOrderIM": "0", # "totalPositionIM": "0", # "totalPositionMM": "0", # "unrealisedPnl": "0", # "cumRealisedPnl": "0", # "bonus": "0" # }, # { # "coin": "BIT", # "equity": "1.82", # "usdValue": "0.48758257", # "walletBalance": "1.82", # "availableToWithdraw": "1.82", # "availableToBorrow": "0", # "borrowAmount": "0", # "accruedInterest": "", # "totalOrderIM": "0", # "totalPositionIM": "0", # "totalPositionMM": "0", # "unrealisedPnl": "0", # "cumRealisedPnl": "0", # "bonus": "0" # } # ], # "accountType": "UNIFIED" # } # ] # } # if self.balance is None: self.balance = {} messageHash = 'balance' topic = self.safe_value(message, 'topic') info = None rawBalances = [] account = None if topic == 'outboundAccountInfo': account = 'spot' data = self.safe_value(message, 'data', []) for i in range(0, len(data)): B = self.safe_value(data[i], 'B', []) rawBalances = self.array_concat(rawBalances, B) info = rawBalances if topic == 'wallet': data = self.safe_value(message, 'data', {}) for i in range(0, len(data)): result = self.safe_value(data, 0, {}) account = self.safe_string_lower(result, 'accountType') rawBalances = self.array_concat(rawBalances, self.safe_value(result, 'coin', [])) info = data for i in range(0, len(rawBalances)): self.parse_ws_balance(rawBalances[i], account) if account is not None: if self.safe_value(self.balance, account) is None: self.balance[account] = {} self.balance[account]['info'] = info timestamp = self.safe_integer(message, 'ts') self.balance[account]['timestamp'] = timestamp self.balance[account]['datetime'] = self.iso8601(timestamp) self.balance[account] = self.safe_balance(self.balance[account]) messageHash = 'balances:' + account client.resolve(self.balance[account], messageHash) else: self.balance['info'] = info timestamp = self.safe_integer(message, 'ts') self.balance['timestamp'] = timestamp self.balance['datetime'] = self.iso8601(timestamp) self.balance = self.safe_balance(self.balance) messageHash = 'balances' client.resolve(self.balance, messageHash) def parse_ws_balance(self, balance, accountType=None): # # spot # { # "a": "USDT", # "f": "176.81254174", # "l": "201.575" # } # unified # { # "coin": "BTC", # "equity": "0.06488393", # "usdValue": "1023.08402268", # "walletBalance": "0.06488393", # "availableToWithdraw": "0.06488393", # "availableToBorrow": "2.5", # "borrowAmount": "0", # "accruedInterest": "0", # "totalOrderIM": "0", # "totalPositionIM": "0", # "totalPositionMM": "0", # "unrealisedPnl": "0", # "cumRealisedPnl": "0", # "bonus": "0" # } # account = self.account() currencyId = self.safe_string_2(balance, 'a', 'coin') code = self.safe_currency_code(currencyId) account['free'] = self.safe_string_n(balance, ['availableToWithdraw', 'f', 'free', 'availableToWithdraw']) account['used'] = self.safe_string_2(balance, 'l', 'locked') account['total'] = self.safe_string(balance, 'walletBalance') if accountType is not None: if self.safe_value(self.balance, accountType) is None: self.balance[accountType] = {} self.balance[accountType][code] = account else: self.balance[code] = account async def watch_topics(self, url, messageHashes, topics, params={}): request: dict = { 'op': 'subscribe', 'req_id': self.request_id(), 'args': topics, } message = self.extend(request, params) return await self.watch_multiple(url, messageHashes, message, messageHashes) async def un_watch_topics(self, url: str, topic: str, symbols: Strings, messageHashes: List[str], subMessageHashes: List[str], topics, params={}, subExtension={}): reqId = self.request_id() request: dict = { 'op': 'unsubscribe', 'req_id': reqId, 'args': topics, } subscription = { 'id': reqId, 'topic': topic, 'messageHashes': messageHashes, 'subMessageHashes': subMessageHashes, 'symbols': symbols, } message = self.extend(request, params) return await self.watch_multiple(url, messageHashes, message, messageHashes, self.extend(subscription, subExtension)) async def authenticate(self, url, params={}): self.check_required_credentials() messageHash = 'authenticated' client = self.client(url) future = client.reusableFuture(messageHash) authenticated = self.safe_value(client.subscriptions, messageHash) if authenticated is None: expiresInt = self.milliseconds() + 10000 expires = self.number_to_string(expiresInt) path = 'GET/realtime' auth = path + expires signature = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256, 'hex') request: dict = { 'op': 'auth', 'args': [ self.apiKey, expires, signature, ], } message = self.extend(request, params) self.watch(url, messageHash, message, messageHash) return await future def handle_error_message(self, client: Client, message) -> Bool: # # { # "success": False, # "ret_msg": "error:invalid op", # "conn_id": "5e079fdd-9c7f-404d-9dbf-969d650838b5", # "request": {op: '', args: null} # } # # auth error # # { # "success": False, # "ret_msg": "error:USVC1111", # "conn_id": "e73770fb-a0dc-45bd-8028-140e20958090", # "request": { # "op": "auth", # "args": [ # "9rFT6uR4uz9Imkw4Wx", # "1653405853543", # "542e71bd85597b4db0290f0ce2d13ed1fd4bb5df3188716c1e9cc69a879f7889" # ] # } # # {code: '-10009', desc: "Invalid period!"} # # { # "reqId":"1", # "retCode":170131, # "retMsg":"Insufficient balance.", # "op":"order.create", # "data":{ # # }, # "header":{ # "X-Bapi-Limit":"20", # "X-Bapi-Limit-Status":"19", # "X-Bapi-Limit-Reset-Timestamp":"1714236608944", # "Traceid":"3d7168a137bf32a947b7e5e6a575ac7f", # "Timenow":"1714236608946" # }, # "connId":"cojifin88smerbj9t560-406" # } # code = self.safe_string_n(message, ['code', 'ret_code', 'retCode']) try: if code is not None and code != '0': feedback = self.id + ' ' + self.json(message) self.throw_exactly_matched_exception(self.exceptions['exact'], code, feedback) msg = self.safe_string_2(message, 'retMsg', 'ret_msg') self.throw_broadly_matched_exception(self.exceptions['broad'], msg, feedback) raise ExchangeError(feedback) success = self.safe_value(message, 'success') if success is not None and not success: ret_msg = self.safe_string(message, 'ret_msg') request = self.safe_value(message, 'request', {}) op = self.safe_string(request, 'op') if op == 'auth': raise AuthenticationError('Authentication failed: ' + ret_msg) else: raise ExchangeError(self.id + ' ' + ret_msg) return False except Exception as error: if isinstance(error, AuthenticationError): messageHash = 'authenticated' client.reject(error, messageHash) if messageHash in client.subscriptions: del client.subscriptions[messageHash] else: messageHash = self.safe_string(message, 'reqId') client.reject(error, messageHash) return True def handle_message(self, client: Client, message): topic = self.safe_string_2(message, 'topic', 'op', '') if self.handle_error_message(client, message): return # contract pong ret_msg = self.safe_string(message, 'ret_msg') if (ret_msg == 'pong') or (topic == 'pong'): self.handle_pong(client, message) return # spot pong pong = self.safe_integer(message, 'pong') if pong is not None: self.handle_pong(client, message) return # pong event = self.safe_string(message, 'event') if event == 'sub' or (topic == 'subscribe'): self.handle_subscription_status(client, message) return methods: dict = { 'orderbook': self.handle_order_book, 'kline': self.handle_ohlcv, 'order': self.handle_order, 'stopOrder': self.handle_order, 'ticker': self.handle_ticker, 'trade': self.handle_trades, 'publicTrade': self.handle_trades, 'depth': self.handle_order_book, 'wallet': self.handle_balance, 'outboundAccountInfo': self.handle_balance, 'execution': self.handle_my_trades, 'execution.fast': self.handle_my_trades, 'ticketInfo': self.handle_my_trades, 'user.openapi.perp.trade': self.handle_my_trades, 'position': self.handle_positions, 'liquidation': self.handle_liquidation, 'allLiquidation': self.handle_liquidation, 'pong': self.handle_pong, 'order.create': self.handle_order_ws, 'order.amend': self.handle_order_ws, 'order.cancel': self.handle_order_ws, 'auth': self.handle_authenticate, 'unsubscribe': self.handle_un_subscribe, } exacMethod = self.safe_value(methods, topic) if exacMethod is not None: exacMethod(client, message) return keys = list(methods.keys()) for i in range(0, len(keys)): key = keys[i] if topic.find(keys[i]) >= 0: method = methods[key] method(client, message) return # unified auth acknowledgement type = self.safe_string(message, 'type') if type == 'AUTH_RESP': self.handle_authenticate(client, message) def ping(self, client: Client): return { 'req_id': self.request_id(), 'op': 'ping', } def handle_pong(self, client: Client, message): # # { # "success": True, # "ret_msg": "pong", # "conn_id": "db3158a0-8960-44b9-a9de-ac350ee13158", # "request": {op: "ping", args: null} # } # # {pong: 1653296711335} # # # { # "req_id": "2", # "op": "pong", # "args": ["1757405570352"], # "conn_id": "d266o6hqo29sqmnq4vk0-1yus1" # } # client.lastPong = self.safe_integer(message, 'pong') return message def handle_authenticate(self, client: Client, message): # # { # "success": True, # "ret_msg": '', # "op": "auth", # "conn_id": "ce3dpomvha7dha97tvp0-2xh" # } # # { # "retCode":0, # "retMsg":"OK", # "op":"auth", # "connId":"cojifin88smerbj9t560-404" # } # # { # "success": True, # "ret_msg": "", # "op": "auth", # "conn_id": "d266o6hqo29sqmnq4vk0-1yus1" # } # success = self.safe_value(message, 'success') code = self.safe_integer(message, 'retCode') messageHash = 'authenticated' if success or code == 0: future = self.safe_value(client.futures, messageHash) future.resolve(True) else: error = AuthenticationError(self.id + ' ' + self.json(message)) client.reject(error, messageHash) if messageHash in client.subscriptions: del client.subscriptions[messageHash] return message def handle_subscription_status(self, client: Client, message): # # { # "topic": "kline", # "event": "sub", # "params": { # "symbol": "LTCUSDT", # "binary": "false", # "klineType": "1m", # "symbolName": "LTCUSDT" # }, # "code": "0", # "msg": "Success" # } # return message def handle_un_subscribe(self, client: Client, message): # # {"success":true,"ret_msg":"","conn_id":"7188110e-6908-41e9-b863-6365127e92ad","req_id":"3","op":"unsubscribe"} # # client.subscription will be something like: # { # "publicTrade.LTCUSDT":true, # "publicTrade.ADAUSDT":true, # "unsubscribe:trade:LTC/USDT:USDT": { # "id":4, # "subHash": "trade:LTC/USDT" # }, # } reqId = self.safe_string(message, 'req_id') keys = list(client.subscriptions.keys()) for i in range(0, len(keys)): messageHash = keys[i] if not (messageHash in client.subscriptions): continue # the previous iteration can have deleted the messageHash from the subscriptions if messageHash.startswith('unsubscribe'): subscription = client.subscriptions[messageHash] subId = self.safe_string(subscription, 'id') if reqId != subId: continue messageHashes = self.safe_list(subscription, 'messageHashes', []) subMessageHashes = self.safe_list(subscription, 'subMessageHashes', []) for j in range(0, len(messageHashes)): unsubHash = messageHashes[j] subHash = subMessageHashes[j] usePrefix = (subHash == 'orders') or (subHash == 'myTrades') or (subHash == 'positions') self.clean_unsubscription(client, subHash, unsubHash, usePrefix) self.clean_cache(subscription) return message