# -*- 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, 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 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, }, '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() 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, symbols: 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[] 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() return await self.un_watch_tickers([symbols], 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='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='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: if not market['spot']: if market['option']: if (limit != 25) and (limit != 100): raise BadRequest(self.id + ' watchOrderBookForSymbols() can only use limit 25 and 100 for option markets.') elif (limit != 1) and (limit != 50) and (limit != 200) and (limit != 500): # bybit only support limit 1, 50, 200, 500 for contract raise BadRequest(self.id + ' watchOrderBookForSymbols() can only use limit 1, 50, 200 and 500 for swap and future markets.') else: if (limit != 1) and (limit != 50) and (limit != 200): raise BadRequest(self.id + ' watchOrderBookForSymbols() can only use limit 1,50, and 200 for spot markets.') 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: Strings, 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] 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: Strings, 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 :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 :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)) 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 :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 ` """ method = 'watchMyTrades' messageHash = 'unsubscribe:myTrades' subHash = 'myTrades' await self.load_markets() if symbol is not None: symbol = self.symbol(symbol) subHash += ':' + 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)) 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" # } # ] # } # topic = self.safe_string(message, 'topic') spot = topic == 'ticketInfo' 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: parsed = self.parse_ws_trade(rawTrade) else: # filter unified trades execType = self.safe_string(rawTrade, 'execType', '') 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 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'), '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: symbol = self.symbol(symbol) subHash += ':' + 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)) 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 = None if isSpot: parsed = self.parse_ws_spot_order(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) def parse_ws_spot_order(self, order, market=None): # # { # "e": "executionReport", # "E": "1653297251061", # timestamp # "s": "LTCUSDT", # symbol # "c": "1653297250740", # user id # "S": "SELL", # side # "o": "MARKET_OF_BASE", # order type # "f": "GTC", # time in force # "q": "0.16233", # quantity # "p": "0", # price # "X": "NEW", # status # "i": "1162336018974750208", # order id # "M": "0", # "l": "0", # last filled # "z": "0", # total filled # "L": "0", # last traded price # "n": "0", # trading fee # "N": '', # fee asset # "u": True, # "w": True, # "m": False, # is limit_maker # "O": "1653297251042", # order creation # "Z": "0", # total filled # "A": "0", # account id # "C": False, # is close # "v": "0", # leverage # "d": "NO_LIQ" # } # v5 # { # "category":"spot", # "symbol":"LTCUSDT", # "orderId":"1474764674982492160", # "orderLinkId":"1690541649154749", # "blockTradeId":"", # "side":"Buy", # "positionIdx":0, # "orderStatus":"Cancelled", # "cancelType":"UNKNOWN", # "rejectReason":"EC_NoError", # "timeInForce":"GTC", # "isLeverage":"0", # "price":"0", # "qty":"5.00000", # "avgPrice":"0", # "leavesQty":"0.00000", # "leavesValue":"5.0000000", # "cumExecQty":"0.00000", # "cumExecValue":"0.0000000", # "cumExecFee":"", # "orderType":"Market", # "stopOrderType":"", # "orderIv":"", # "triggerPrice":"0.000", # "takeProfit":"", # "stopLoss":"", # "triggerBy":"", # "tpTriggerBy":"", # "slTriggerBy":"", # "triggerDirection":0, # "placeType":"", # "lastPriceOnCreated":"0.000", # "closeOnTrigger":false, # "reduceOnly":false, # "smpGroup":0, # "smpType":"None", # "smpOrderId":"", # "createdTime":"1690541649160", # "updatedTime":"1690541649168" # } # id = self.safe_string_2(order, 'i', 'orderId') marketId = self.safe_string_2(order, 's', 'symbol') symbol = self.safe_symbol(marketId, market, None, 'spot') timestamp = self.safe_integer_2(order, 'O', 'createdTime') price = self.safe_string_2(order, 'p', 'price') if price == '0': price = None # market orders filled = self.safe_string_2(order, 'z', 'cumExecQty') status = self.parse_order_status(self.safe_string_2(order, 'X', 'orderStatus')) side = self.safe_string_lower_2(order, 'S', 'side') lastTradeTimestamp = self.safe_string_2(order, 'E', 'updatedTime') timeInForce = self.safe_string_2(order, 'f', 'timeInForce') amount = None cost = self.safe_string_2(order, 'Z', 'cumExecValue') type = self.safe_string_lower_2(order, 'o', 'orderType') if (type is not None) and (type.find('market') >= 0): type = 'market' if type == 'market' and side == 'buy': amount = filled else: amount = self.safe_string_2(order, 'orderQty', 'qty') fee = None feeCost = self.safe_string_2(order, 'n', 'cumExecFee') if feeCost is not None and feeCost != '0': feeCurrencyId = self.safe_string(order, 'N') feeCurrencyCode = self.safe_currency_code(feeCurrencyId) fee = { 'cost': feeCost, 'currency': feeCurrencyCode, } triggerPrice = self.omit_zero(self.safe_string(order, 'triggerPrice')) return self.safe_order({ 'info': order, 'id': id, 'clientOrderId': self.safe_string_2(order, 'c', 'orderLinkId'), 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': lastTradeTimestamp, 'symbol': symbol, 'type': type, 'timeInForce': timeInForce, 'postOnly': None, 'side': side, 'price': price, 'stopPrice': triggerPrice, 'triggerPrice': triggerPrice, 'takeProfitPrice': self.safe_string(order, 'takeProfit'), 'stopLossPrice': self.safe_string(order, 'stopLoss'), 'reduceOnly': self.safe_value(order, 'reduceOnly'), 'amount': amount, 'cost': cost, 'average': self.safe_string(order, 'avgPrice'), 'filled': filled, 'remaining': None, 'status': status, 'fee': fee, }, market) 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: List[str], 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.future(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): # # { # "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): if self.handle_error_message(client, message): return # contract pong ret_msg = self.safe_string(message, 'ret_msg') if ret_msg == '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': self.handle_subscription_status(client, message) return topic = self.safe_string_2(message, 'topic', 'op', '') 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, '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} # 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 = 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] self.clean_unsubscription(client, subHash, unsubHash) self.clean_cache(subscription) return message