# -*- 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 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 ArgumentsRequired from ccxt.base.errors import BadRequest from ccxt.base.errors import NotSupported from ccxt.base.errors import ChecksumError from ccxt.base.precise import Precise class binance(ccxt.async_support.binance): def describe(self) -> Any: superDescribe = super(binance, self).describe() return self.deep_extend(superDescribe, self.describe_data()) def describe_data(self): return { 'has': { 'ws': True, 'watchBalance': True, 'watchLiquidations': True, 'watchLiquidationsForSymbols': True, 'watchMyLiquidations': True, 'watchMyLiquidationsForSymbols': True, 'watchBidsAsks': True, 'watchMyTrades': True, 'watchOHLCV': True, 'watchOHLCVForSymbols': True, 'watchOrderBook': True, 'watchOrderBookForSymbols': True, 'watchOrders': True, 'watchOrdersForSymbols': True, 'watchPositions': True, 'watchTicker': True, 'watchTickers': True, 'watchMarkPrices': True, 'watchMarkPrice': True, 'watchTrades': True, 'watchTradesForSymbols': True, 'createOrderWs': True, 'editOrderWs': True, 'cancelOrderWs': True, 'cancelOrdersWs': False, 'cancelAllOrdersWs': True, 'fetchBalanceWs': True, 'fetchDepositsWs': False, 'fetchMarketsWs': False, 'fetchMyTradesWs': True, 'fetchOHLCVWs': True, 'fetchOrderBookWs': True, 'fetchOpenOrdersWs': True, 'fetchOrderWs': True, 'fetchOrdersWs': True, 'fetchPositionWs': True, 'fetchPositionForSymbolWs': True, 'fetchPositionsWs': True, 'fetchTickerWs': True, 'fetchTradesWs': True, 'fetchTradingFeesWs': False, 'fetchWithdrawalsWs': False, }, 'urls': { 'test': { 'ws': { 'spot': 'wss://stream.testnet.binance.vision/ws', 'margin': 'wss://stream.testnet.binance.vision/ws', 'future': 'wss://fstream.binancefuture.com/ws', 'delivery': 'wss://dstream.binancefuture.com/ws', 'ws-api': { 'spot': 'wss://ws-api.testnet.binance.vision/ws-api/v3', 'future': 'wss://testnet.binancefuture.com/ws-fapi/v1', 'delivery': 'wss://testnet.binancefuture.com/ws-dapi/v1', }, }, }, 'api': { 'ws': { 'spot': 'wss://stream.binance.com:9443/ws', 'margin': 'wss://stream.binance.com:9443/ws', 'future': 'wss://fstream.binance.com/ws', 'delivery': 'wss://dstream.binance.com/ws', 'ws-api': { 'spot': 'wss://ws-api.binance.com:443/ws-api/v3', 'future': 'wss://ws-fapi.binance.com/ws-fapi/v1', 'delivery': 'wss://ws-dapi.binance.com/ws-dapi/v1', }, 'papi': 'wss://fstream.binance.com/pm/ws', }, }, 'doc': 'https://developers.binance.com/en', }, 'streaming': { 'keepAlive': 180000, }, 'options': { 'returnRateLimits': False, 'streamLimits': { 'spot': 50, # max 1024 'margin': 50, # max 1024 'future': 50, # max 200 'delivery': 50, # max 200 }, 'subscriptionLimitByStream': { 'spot': 200, 'margin': 200, 'future': 200, 'delivery': 200, }, 'streamBySubscriptionsHash': self.create_safe_dictionary(), 'streamIndex': -1, # get updates every 1000ms or 100ms # or every 0ms in real-time for futures 'watchOrderBookRate': 100, 'liquidationsLimit': 1000, 'myLiquidationsLimit': 1000, 'tradesLimit': 1000, 'ordersLimit': 1000, 'OHLCVLimit': 1000, 'requestId': self.create_safe_dictionary(), 'watchOrderBookLimit': 1000, # default limit 'watchTrades': { 'name': 'trade', # 'trade' or 'aggTrade' }, 'watchTicker': { 'name': 'ticker', # ticker or miniTicker or ticker_ }, 'watchTickers': { 'name': 'ticker', # ticker or miniTicker or ticker_ }, 'watchOHLCV': { 'name': 'kline', # or indexPriceKline or markPriceKline(coin-m futures) }, 'watchOrderBook': { 'maxRetries': 3, 'checksum': True, }, 'watchBalance': { 'fetchBalanceSnapshot': False, # or True 'awaitBalanceSnapshot': True, # whether to wait for the balance snapshot before providing updates }, 'watchLiquidationsForSymbols': { 'defaultType': 'swap', }, 'watchPositions': { 'fetchPositionsSnapshot': True, # or False 'awaitPositionsSnapshot': True, # whether to wait for the positions snapshot before providing updates }, 'wallet': 'wb', # wb = wallet balance, cw = cross balance 'listenKeyRefreshRate': 1200000, # 20 mins 'ws': { 'cost': 5, }, 'tickerChannelsMap': { '24hrTicker': 'ticker', '24hrMiniTicker': 'miniTicker', 'markPriceUpdate': 'markPrice', # rolling window tickers '1hTicker': 'ticker_1h', '4hTicker': 'ticker_4h', '1dTicker': 'ticker_1d', 'bookTicker': 'bookTicker', }, }, } def request_id(self, url): options = self.safe_dict(self.options, 'requestId', self.create_safe_dictionary()) previousValue = self.safe_integer(options, url, 0) newValue = self.sum(previousValue, 1) self.options['requestId'][url] = newValue return newValue def stream(self, type: Str, subscriptionHash: Str, numSubscriptions=1): streamBySubscriptionsHash = self.safe_dict(self.options, 'streamBySubscriptionsHash', self.create_safe_dictionary()) stream = self.safe_string(streamBySubscriptionsHash, subscriptionHash) if stream is None: streamIndex = self.safe_integer(self.options, 'streamIndex', -1) streamLimits = self.safe_value(self.options, 'streamLimits') streamLimit = self.safe_integer(streamLimits, type) streamIndex = streamIndex + 1 normalizedIndex = streamIndex % streamLimit self.options['streamIndex'] = streamIndex stream = self.number_to_string(normalizedIndex) self.options['streamBySubscriptionsHash'][subscriptionHash] = stream subscriptionsByStreams = self.safe_value(self.options, 'numSubscriptionsByStream') if subscriptionsByStreams is None: self.options['numSubscriptionsByStream'] = self.create_safe_dictionary() subscriptionsByStream = self.safe_integer(self.options['numSubscriptionsByStream'], stream, 0) newNumSubscriptions = subscriptionsByStream + numSubscriptions subscriptionLimitByStream = self.safe_integer(self.options['subscriptionLimitByStream'], type, 200) if newNumSubscriptions > subscriptionLimitByStream: raise BadRequest(self.id + ' reached the limit of subscriptions by stream. Increase the number of streams, or increase the stream limit or subscription limit by stream if the exchange allows.') self.options['numSubscriptionsByStream'][stream] = subscriptionsByStream + numSubscriptions return stream 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://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Liquidation-Order-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Liquidation-Order-Streams :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 :returns dict: an array of `liquidation structures ` """ return await self.watch_liquidations_for_symbols([symbol], since, limit, params) async def watch_liquidations_for_symbols(self, symbols: List[str] = None, since: Int = None, limit: Int = None, params={}) -> List[Liquidation]: """ watch the public liquidations of a trading pair https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Market-Liquidation-Order-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/All-Market-Liquidation-Order-Streams :param str[] symbols: list of unified market symbols :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 :returns dict: an array of `liquidation structures ` """ await self.load_markets() subscriptionHashes = [] messageHashes = [] streamHash = 'liquidations' symbols = self.market_symbols(symbols, None, True, True) if self.is_empty(symbols): subscriptionHashes.append('!' + 'forceOrder@arr') messageHashes.append('liquidations') else: for i in range(0, len(symbols)): market = self.market(symbols[i]) subscriptionHashes.append(market['lowercaseId'] + '@forceOrder') messageHashes.append('liquidations::' + symbols[i]) streamHash += '::' + ','.join(symbols) firstMarket = self.get_market_from_symbols(symbols) type = None type, params = self.handle_market_type_and_params('watchLiquidationsForSymbols', firstMarket, params) if type == 'spot': raise BadRequest(self.id + ' watchLiquidationsForSymbols is not supported for spot symbols') subType = None subType, params = self.handle_sub_type_and_params('watchLiquidationsForSymbols', firstMarket, params) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' numSubscriptions = len(subscriptionHashes) url = self.urls['api']['ws'][type] + '/' + self.stream(type, streamHash, numSubscriptions) requestId = self.request_id(url) request = { 'method': 'SUBSCRIBE', 'params': subscriptionHashes, 'id': requestId, } subscribe = { 'id': requestId, } newLiquidations = await self.watch_multiple(url, messageHashes, self.extend(request, params), subscriptionHashes, subscribe) if self.newUpdates: return newLiquidations return self.filter_by_symbols_since_limit(self.liquidations, symbols, since, limit, True) def handle_liquidation(self, client: Client, message): # # future # { # "e":"forceOrder", # "E":1698871323061, # "o":{ # "s":"BTCUSDT", # "S":"BUY", # "o":"LIMIT", # "f":"IOC", # "q":"1.437", # "p":"35100.81", # "ap":"34959.70", # "X":"FILLED", # "l":"1.437", # "z":"1.437", # "T":1698871323059 # } # } # delivery # { # "e":"forceOrder", # Event Type # "E": 1591154240950, # Event Time # "o":{ # "s":"BTCUSD_200925", # Symbol # "ps": "BTCUSD", # Pair # "S":"SELL", # Side # "o":"LIMIT", # Order Type # "f":"IOC", # Time in Force # "q":"1", # Original Quantity # "p":"9425.5", # Price # "ap":"9496.5", # Average Price # "X":"FILLED", # Order Status # "l":"1", # Order Last Filled Quantity # "z":"1", # Order Filled Accumulated Quantity # "T": 1591154240949, # Order Trade Time # } # } # rawLiquidation = self.safe_value(message, 'o', {}) 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) def parse_ws_liquidation(self, liquidation, market=None): # # future # { # "s":"BTCUSDT", # "S":"BUY", # "o":"LIMIT", # "f":"IOC", # "q":"1.437", # "p":"35100.81", # "ap":"34959.70", # "X":"FILLED", # "l":"1.437", # "z":"1.437", # "T":1698871323059 # } # delivery # { # "s":"BTCUSD_200925", # Symbol # "ps": "BTCUSD", # Pair # "S":"SELL", # Side # "o":"LIMIT", # Order Type # "f":"IOC", # Time in Force # "q":"1", # Original Quantity # "p":"9425.5", # Price # "ap":"9496.5", # Average Price # "X":"FILLED", # Order Status # "l":"1", # Order Last Filled Quantity # "z":"1", # Order Filled Accumulated Quantity # "T": 1591154240949, # Order Trade Time # } # myLiquidation # { # "s":"BTCUSDT", # Symbol # "c":"TEST", # Client Order Id # # special client order id: # # starts with "autoclose-": liquidation order # # "adl_autoclose": ADL auto close order # # "settlement_autoclose-": settlement order for delisting or delivery # "S":"SELL", # Side # "o":"TRAILING_STOP_MARKET", # Order Type # "f":"GTC", # Time in Force # "q":"0.001", # Original Quantity # "p":"0", # Original Price # "ap":"0", # Average Price # "sp":"7103.04", # Stop Price. Please ignore with TRAILING_STOP_MARKET order # "x":"NEW", # Execution Type # "X":"NEW", # Order Status # "i":8886774, # Order Id # "l":"0", # Order Last Filled Quantity # "z":"0", # Order Filled Accumulated Quantity # "L":"0", # Last Filled Price # "N":"USDT", # Commission Asset, will not push if no commission # "n":"0", # Commission, will not push if no commission # "T":1568879465650, # Order Trade Time # "t":0, # Trade Id # "b":"0", # Bids Notional # "a":"9.91", # Ask Notional # "m":false, # Is self trade the maker side? # "R":false, # Is self reduce only # "wt":"CONTRACT_PRICE", # Stop Price Working Type # "ot":"TRAILING_STOP_MARKET",// Original Order Type # "ps":"LONG", # Position Side # "cp":false, # If Close-All, pushed with conditional order # "AP":"7476.89", # Activation Price, only puhed with TRAILING_STOP_MARKET order # "cr":"5.0", # Callback Rate, only puhed with TRAILING_STOP_MARKET order # "pP": False, # If price protection is turned on # "si": 0, # ignore # "ss": 0, # ignore # "rp":"0", # Realized Profit of the trade # "V":"EXPIRE_TAKER", # STP mode # "pm":"OPPONENT", # Price match mode # "gtd":0 # TIF GTD order auto cancel time # } # marketId = self.safe_string(liquidation, 's') market = self.safe_market(marketId, market) timestamp = self.safe_integer(liquidation, 'T') return self.safe_liquidation({ 'info': liquidation, 'symbol': self.safe_symbol(marketId, market), 'contracts': self.safe_number(liquidation, 'l'), 'contractSize': self.safe_number(market, 'contractSize'), 'price': self.safe_number(liquidation, 'ap'), 'baseValue': None, 'quoteValue': None, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), }) async def watch_my_liquidations(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Liquidation]: """ watch the private liquidations of a trading pair https://developers.binance.com/docs/derivatives/usds-margined-futures/user-data-streams/Event-Order-Update https://developers.binance.com/docs/derivatives/coin-margined-futures/user-data-streams/Event-Order-Update :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 :returns dict: an array of `liquidation structures ` """ return self.watch_my_liquidations_for_symbols([symbol], since, limit, params) async def watch_my_liquidations_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Liquidation]: """ watch the private liquidations of a trading pair https://developers.binance.com/docs/derivatives/usds-margined-futures/user-data-streams/Event-Order-Update https://developers.binance.com/docs/derivatives/coin-margined-futures/user-data-streams/Event-Order-Update :param str[] symbols: list of unified market symbols :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 :returns dict: an array of `liquidation structures ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, True, True, True) market = self.get_market_from_symbols(symbols) messageHashes = ['myLiquidations'] if not self.is_empty(symbols): for i in range(0, len(symbols)): symbol = symbols[i] messageHashes.append('myLiquidations::' + symbol) type = None type, params = self.handle_market_type_and_params('watchMyLiquidationsForSymbols', market, params) subType = None subType, params = self.handle_sub_type_and_params('watchMyLiquidationsForSymbols', market, params) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' await self.authenticate(params) url = self.urls['api']['ws'][type] + '/' + self.options[type]['listenKey'] message = None newLiquidations = await self.watch_multiple(url, messageHashes, message, [type]) if self.newUpdates: return newLiquidations return self.filter_by_symbols_since_limit(self.liquidations, symbols, since, limit) def handle_my_liquidation(self, client: Client, message): # # { # "s":"BTCUSDT", # Symbol # "c":"TEST", # Client Order Id # # special client order id: # # starts with "autoclose-": liquidation order # # "adl_autoclose": ADL auto close order # # "settlement_autoclose-": settlement order for delisting or delivery # "S":"SELL", # Side # "o":"TRAILING_STOP_MARKET", # Order Type # "f":"GTC", # Time in Force # "q":"0.001", # Original Quantity # "p":"0", # Original Price # "ap":"0", # Average Price # "sp":"7103.04", # Stop Price. Please ignore with TRAILING_STOP_MARKET order # "x":"NEW", # Execution Type # "X":"NEW", # Order Status # "i":8886774, # Order Id # "l":"0", # Order Last Filled Quantity # "z":"0", # Order Filled Accumulated Quantity # "L":"0", # Last Filled Price # "N":"USDT", # Commission Asset, will not push if no commission # "n":"0", # Commission, will not push if no commission # "T":1568879465650, # Order Trade Time # "t":0, # Trade Id # "b":"0", # Bids Notional # "a":"9.91", # Ask Notional # "m":false, # Is self trade the maker side? # "R":false, # Is self reduce only # "wt":"CONTRACT_PRICE", # Stop Price Working Type # "ot":"TRAILING_STOP_MARKET",// Original Order Type # "ps":"LONG", # Position Side # "cp":false, # If Close-All, pushed with conditional order # "AP":"7476.89", # Activation Price, only puhed with TRAILING_STOP_MARKET order # "cr":"5.0", # Callback Rate, only puhed with TRAILING_STOP_MARKET order # "pP": False, # If price protection is turned on # "si": 0, # ignore # "ss": 0, # ignore # "rp":"0", # Realized Profit of the trade # "V":"EXPIRE_TAKER", # STP mode # "pm":"OPPONENT", # Price match mode # "gtd":0 # TIF GTD order auto cancel time # } # orderType = self.safe_string(message, 'o') if orderType != 'LIQUIDATION': return marketId = self.safe_string(message, 's') market = self.safe_market(marketId) symbol = self.safe_symbol(marketId) liquidation = self.parse_ws_liquidation(message, market) myLiquidations = self.safe_value(self.myLiquidations, symbol) if myLiquidations is None: limit = self.safe_integer(self.options, 'myLiquidationsLimit', 1000) myLiquidations = ArrayCache(limit) myLiquidations.append(liquidation) self.myLiquidations[symbol] = myLiquidations client.resolve([liquidation], 'myLiquidations') client.resolve([liquidation], 'myLiquidations::' + symbol) 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://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#partial-book-depth-streams https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#diff-depth-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams :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 """ # # todo add support for -snapshots(depth) # https://github.com/binance-exchange/binance-official-api-docs/blob/master/web-socket-streams.md#partial-book-depth-streams # @depth@100ms or @depth(1000ms) # valid are 5, 10, or 20 # # default 100, max 1000, valid limits 5, 10, 20, 50, 100, 500, 1000 # # notice the differences between trading futures and spot trading # the algorithms use different urls in step 1 # delta caching and merging also differs in steps 4, 5, 6 # # spot/margin # https://binance-docs.github.io/apidocs/spot/en/#how-to-manage-a-local-order-book-correctly # # 1. Open a stream to wss://stream.binance.com:9443/ws/bnbbtc@depth. # 2. Buffer the events you receive from the stream. # 3. Get a depth snapshot from https://www.binance.com/api/v1/depth?symbol=BNBBTC&limit=1000 . # 4. Drop any event where u is <= lastUpdateId in the snapshot. # 5. The first processed event should have U <= lastUpdateId+1 AND u >= lastUpdateId+1. # 6. While listening to the stream, each new event's U should be equal to the previous event's u+1. # 7. The data in each event is the absolute quantity for a price level. # 8. If the quantity is 0, remove the price level. # 9. Receiving an event that removes a price level that is not in your local order book can happen and is normal. # # futures # https://binance-docs.github.io/apidocs/futures/en/#how-to-manage-a-local-order-book-correctly # # 1. Open a stream to wss://fstream.binance.com/stream?streams=btcusdt@depth. # 2. Buffer the events you receive from the stream. For same price, latest received update covers the previous one. # 3. Get a depth snapshot from https://fapi.binance.com/fapi/v1/depth?symbol=BTCUSDT&limit=1000 . # 4. Drop any event where u is < lastUpdateId in the snapshot. # 5. The first processed event should have U <= lastUpdateId AND u >= lastUpdateId # 6. While listening to the stream, each new event's pu should be equal to the previous event's u, otherwise initialize the process from step 3. # 7. The data in each event is the absolute quantity for a price level. # 8. If the quantity is 0, remove the price level. # 9. Receiving an event that removes a price level that is not in your local order book can happen and is normal. # 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://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#partial-book-depth-streams https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#diff-depth-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams :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() symbols = self.market_symbols(symbols, None, False, True, True) firstMarket = self.market(symbols[0]) type = firstMarket['type'] if firstMarket['contract']: type = 'future' if firstMarket['linear'] else 'delivery' name = 'depth' streamHash = 'multipleOrderbook' if symbols is not None: symbolsLength = len(symbols) if symbolsLength > 200: raise BadRequest(self.id + ' watchOrderBookForSymbols() accepts 200 symbols at most. To watch more symbols call watchOrderBookForSymbols() multiple times') streamHash += '::' + ','.join(symbols) watchOrderBookRate = self.safe_string(self.options, 'watchOrderBookRate', '100') subParams = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) messageHashes.append('orderbook::' + symbol) subscriptionHash = market['lowercaseId'] + '@' + name symbolHash = subscriptionHash + '@' + watchOrderBookRate + 'ms' subParams.append(symbolHash) messageHashesLength = len(messageHashes) url = self.urls['api']['ws'][type] + '/' + self.stream(type, streamHash, messageHashesLength) requestId = self.request_id(url) request: dict = { 'method': 'SUBSCRIBE', 'params': subParams, 'id': requestId, } subscription: dict = { 'id': str(requestId), 'name': name, 'symbols': symbols, 'method': self.handle_order_book_subscription, 'limit': limit, 'type': type, 'params': params, } orderbook = await self.watch_multiple(url, messageHashes, self.extend(request, params), messageHashes, subscription) return orderbook.limit() async def un_watch_order_book_for_symbols(self, symbols: List[str], params={}) -> Any: """ unWatches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#partial-book-depth-streams https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#diff-depth-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams :param str[] symbols: unified array of symbols :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() symbols = self.market_symbols(symbols, None, False, True, True) firstMarket = self.market(symbols[0]) type = firstMarket['type'] if firstMarket['contract']: type = 'future' if firstMarket['linear'] else 'delivery' name = 'depth' streamHash = 'multipleOrderbook' if symbols is not None: streamHash += '::' + ','.join(symbols) watchOrderBookRate = self.safe_string(self.options, 'watchOrderBookRate', '100') subParams = [] subMessageHashes = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) subMessageHashes.append('orderbook::' + symbol) messageHashes.append('unsubscribe:orderbook:' + symbol) subscriptionHash = market['lowercaseId'] + '@' + name symbolHash = subscriptionHash + '@' + watchOrderBookRate + 'ms' subParams.append(symbolHash) messageHashesLength = len(subMessageHashes) url = self.urls['api']['ws'][type] + '/' + self.stream(type, streamHash, messageHashesLength) requestId = self.request_id(url) request: dict = { 'method': 'UNSUBSCRIBE', 'params': subParams, 'id': requestId, } subscription: dict = { 'unsubscribe': True, 'id': str(requestId), 'symbols': symbols, 'subMessageHashes': subMessageHashes, 'messageHashes': messageHashes, 'topic': 'orderbook', } return await self.watch_multiple(url, messageHashes, self.extend(request, params), messageHashes, subscription) async def un_watch_order_book(self, symbol: str, params={}) -> Any: """ unWatches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#partial-book-depth-streams https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#diff-depth-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Partial-Book-Depth-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Diff-Book-Depth-Streams :param str symbol: unified array of symbols :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.un_watch_order_book_for_symbols([symbol], params) async def fetch_order_book_ws(self, symbol: str, limit: Int = None, params={}) -> OrderBook: """ fetches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#order-book https://developers.binance.com/docs/derivatives/usds-margined-futures/market-data/websocket-api/Order-Book :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 """ await self.load_markets() market = self.market(symbol) payload: dict = { 'symbol': market['id'], } if limit is not None: payload['limit'] = limit marketType = self.get_market_type('fetchOrderBookWs', market, params) if marketType != 'future': raise BadRequest(self.id + ' fetchOrderBookWs only supports swap markets') url = self.urls['api']['ws']['ws-api'][marketType] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'createOrderWs', 'returnRateLimits', False) payload['returnRateLimits'] = returnRateLimits params = self.omit(params, 'test') message: dict = { 'id': messageHash, 'method': 'depth', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_fetch_order_book, } orderbook = await self.watch(url, messageHash, message, messageHash, subscription) orderbook['symbol'] = market['symbol'] return orderbook def handle_fetch_order_book(self, client: Client, message): # # { # "id":"51e2affb-0aba-4821-ba75-f2625006eb43", # "status":200, # "result":{ # "lastUpdateId":1027024, # "E":1589436922972, # "T":1589436922959, # "bids":[ # [ # "4.00000000", # "431.00000000" # ] # ], # "asks":[ # [ # "4.00000200", # "12.00000000" # ] # ] # } # } # messageHash = self.safe_string(message, 'id') result = self.safe_dict(message, 'result') timestamp = self.safe_integer(result, 'T') orderbook = self.parse_order_book(result, None, timestamp) orderbook['nonce'] = self.safe_integer_2(result, 'lastUpdateId', 'u') client.resolve(orderbook, messageHash) async def fetch_order_book_snapshot(self, client, message, subscription): symbol = self.safe_string(subscription, 'symbol') messageHash = 'orderbook::' + symbol try: defaultLimit = self.safe_integer(self.options, 'watchOrderBookLimit', 1000) type = self.safe_value(subscription, 'type') limit = self.safe_integer(subscription, 'limit', defaultLimit) params = self.safe_value(subscription, 'params') # 3. Get a depth snapshot from https://www.binance.com/api/v1/depth?symbol=BNBBTC&limit=1000 . # todo: self is a synch blocking call - make it async # default 100, max 1000, valid limits 5, 10, 20, 50, 100, 500, 1000 snapshot = await self.fetch_rest_order_book_safe(symbol, limit, params) if self.safe_value(self.orderbooks, symbol) is None: # if the orderbook is dropped before the snapshot is received return orderbook = self.orderbooks[symbol] orderbook.reset(snapshot) # unroll the accumulated deltas messages = orderbook.cache orderbook.cache = [] for i in range(0, len(messages)): messageItem = messages[i] U = self.safe_integer(messageItem, 'U') u = self.safe_integer(messageItem, 'u') pu = self.safe_integer(messageItem, 'pu') if type == 'future': # 4. Drop any event where u is < lastUpdateId in the snapshot if u < orderbook['nonce']: continue # 5. The first processed event should have U <= lastUpdateId AND u >= lastUpdateId if (U <= orderbook['nonce']) and (u >= orderbook['nonce']) or (pu == orderbook['nonce']): self.handle_order_book_message(client, messageItem, orderbook) else: # 4. Drop any event where u is <= lastUpdateId in the snapshot if u <= orderbook['nonce']: continue # 5. The first processed event should have U <= lastUpdateId+1 AND u >= lastUpdateId+1 if ((U - 1) <= orderbook['nonce']) and ((u - 1) >= orderbook['nonce']): self.handle_order_book_message(client, messageItem, orderbook) self.orderbooks[symbol] = orderbook client.resolve(orderbook, messageHash) except Exception as e: del client.subscriptions[messageHash] client.reject(e, messageHash) def handle_delta(self, bookside, delta): price = self.safe_float(delta, 0) amount = self.safe_float(delta, 1) bookside.store(price, amount) def handle_deltas(self, bookside, deltas): for i in range(0, len(deltas)): self.handle_delta(bookside, deltas[i]) def handle_order_book_message(self, client: Client, message, orderbook): u = self.safe_integer(message, 'u') self.handle_deltas(orderbook['asks'], self.safe_value(message, 'a', [])) self.handle_deltas(orderbook['bids'], self.safe_value(message, 'b', [])) orderbook['nonce'] = u timestamp = self.safe_integer(message, 'E') orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) return orderbook def handle_order_book(self, client: Client, message): # # initial snapshot is fetched with ccxt's fetchOrderBook # the feed does not include a snapshot, just the deltas # # { # "e": "depthUpdate", # Event type # "E": 1577554482280, # Event time # "s": "BNBBTC", # Symbol # "U": 157, # First update ID in event # "u": 160, # Final update ID in event # "b": [ # bids # ["0.0024", "10"], # price, size # ], # "a": [ # asks # ["0.0026", "100"], # price, size # ] # } # isSpot = (client.url.find('/stream') > -1) marketType = 'spot' if (isSpot) else 'contract' marketId = self.safe_string(message, 's') market = self.safe_market(marketId, None, None, marketType) symbol = market['symbol'] messageHash = 'orderbook::' + symbol if not (symbol in self.orderbooks): # # https://github.com/ccxt/ccxt/issues/6672 # # Sometimes Binance sends the first delta before the subscription # confirmation arrives. At that point the orderbook is not # initialized yet and the snapshot has not been requested yet # therefore it is safe to drop these premature messages. # return orderbook = self.orderbooks[symbol] nonce = self.safe_integer(orderbook, 'nonce') if nonce is None: # 2. Buffer the events you receive from the stream. orderbook.cache.append(message) else: try: U = self.safe_integer(message, 'U') u = self.safe_integer(message, 'u') pu = self.safe_integer(message, 'pu') if pu is None: # spot # 4. Drop any event where u is <= lastUpdateId in the snapshot if u > orderbook['nonce']: timestamp = self.safe_integer(orderbook, 'timestamp') conditional = None if timestamp is None: # 5. The first processed event should have U <= lastUpdateId+1 AND u >= lastUpdateId+1 conditional = ((U - 1) <= orderbook['nonce']) and ((u - 1) >= orderbook['nonce']) else: # 6. While listening to the stream, each new event's U should be equal to the previous event's u+1. conditional = ((U - 1) == orderbook['nonce']) if conditional: self.handle_order_book_message(client, message, orderbook) if nonce < orderbook['nonce']: client.resolve(orderbook, messageHash) else: checksum = self.handle_option('watchOrderBook', 'checksum', True) if checksum: # todo: client.reject from handleOrderBookMessage properly raise ChecksumError(self.id + ' ' + self.orderbook_checksum_message(symbol)) else: # future # 4. Drop any event where u is < lastUpdateId in the snapshot if u >= orderbook['nonce']: # 5. The first processed event should have U <= lastUpdateId AND u >= lastUpdateId # 6. While listening to the stream, each new event's pu should be equal to the previous event's u, otherwise initialize the process from step 3 if (U <= orderbook['nonce']) or (pu == orderbook['nonce']): self.handle_order_book_message(client, message, orderbook) if nonce <= orderbook['nonce']: client.resolve(orderbook, messageHash) else: checksum = self.handle_option('watchOrderBook', 'checksum', True) if checksum: # todo: client.reject from handleOrderBookMessage properly raise ChecksumError(self.id + ' ' + self.orderbook_checksum_message(symbol)) except Exception as e: del self.orderbooks[symbol] del client.subscriptions[messageHash] client.reject(e, messageHash) def handle_order_book_subscription(self, client: Client, message, subscription): defaultLimit = self.safe_integer(self.options, 'watchOrderBookLimit', 1000) # messageHash = self.safe_string(subscription, 'messageHash') symbolOfSubscription = self.safe_string(subscription, 'symbol') # watchOrderBook symbols = self.safe_value(subscription, 'symbols', [symbolOfSubscription]) # watchOrderBookForSymbols limit = self.safe_integer(subscription, 'limit', defaultLimit) # handle list of symbols for i in range(0, len(symbols)): symbol = symbols[i] if symbol in self.orderbooks: del self.orderbooks[symbol] self.orderbooks[symbol] = self.order_book({}, limit) subscription = self.extend(subscription, {'symbol': symbol}) # fetch the snapshot in a separate async call self.spawn(self.fetch_order_book_snapshot, client, message, subscription) def handle_subscription_status(self, client: Client, message): # # { # "result": null, # "id": 1574649734450 # } # id = self.safe_string(message, 'id') subscriptionsById = self.index_by(client.subscriptions, 'id') subscription = self.safe_value(subscriptionsById, id, {}) method = self.safe_value(subscription, 'method') if method is not None: method(client, message, subscription) isUnSubMessage = self.safe_bool(subscription, 'unsubscribe', False) if isUnSubMessage: self.handle_un_subscription(client, subscription) return message def handle_un_subscription(self, client: Client, subscription: dict): 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) 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://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#aggregate-trades https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#recent-trades https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Aggregate-Trade-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Aggregate-Trade-Streams :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 :param str [params.name]: the name of the method to call, 'trade' or 'aggTrade', default is 'trade' :returns dict[]: a list of `trade structures ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False, True, True) streamHash = 'multipleTrades' if symbols is not None: symbolsLength = len(symbols) if symbolsLength > 200: raise BadRequest(self.id + ' watchTradesForSymbols() accepts 200 symbols at most. To watch more symbols call watchTradesForSymbols() multiple times') streamHash += '::' + ','.join(symbols) name = None name, params = self.handle_option_and_params(params, 'watchTradesForSymbols', 'name', 'trade') params = self.omit(params, 'callerMethodName') firstMarket = self.market(symbols[0]) type = firstMarket['type'] if firstMarket['contract']: type = 'future' if firstMarket['linear'] else 'delivery' messageHashes = [] subParams = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) messageHashes.append('trade::' + symbol) rawHash = market['lowercaseId'] + '@' + name subParams.append(rawHash) query = self.omit(params, 'type') subParamsLength = len(subParams) url = self.urls['api']['ws'][type] + '/' + self.stream(type, streamHash, subParamsLength) requestId = self.request_id(url) request: dict = { 'method': 'SUBSCRIBE', 'params': subParams, 'id': requestId, } subscribe: dict = { 'id': requestId, } trades = await self.watch_multiple(url, messageHashes, self.extend(request, query), messageHashes, subscribe) 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: """ unsubscribes from the trades channel https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#aggregate-trades https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#recent-trades https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Aggregate-Trade-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Aggregate-Trade-Streams :param str[] symbols: unified symbol of the market to fetch trades for :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.name]: the name of the method to call, 'trade' or 'aggTrade', default is 'trade' :returns dict[]: a list of `trade structures ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False, True, True) streamHash = 'multipleTrades' if symbols is not None: symbolsLength = len(symbols) if symbolsLength > 200: raise BadRequest(self.id + ' watchTradesForSymbols() accepts 200 symbols at most. To watch more symbols call watchTradesForSymbols() multiple times') streamHash += '::' + ','.join(symbols) name = None name, params = self.handle_option_and_params(params, 'watchTradesForSymbols', 'name', 'trade') params = self.omit(params, 'callerMethodName') firstMarket = self.market(symbols[0]) type = firstMarket['type'] if firstMarket['contract']: type = 'future' if firstMarket['linear'] else 'delivery' subMessageHashes = [] subParams = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) subMessageHashes.append('trade::' + symbol) messageHashes.append('unsubscribe:trade:' + symbol) rawHash = market['lowercaseId'] + '@' + name subParams.append(rawHash) query = self.omit(params, 'type') subParamsLength = len(subParams) url = self.urls['api']['ws'][type] + '/' + self.stream(type, streamHash, subParamsLength) requestId = self.request_id(url) request: dict = { 'method': 'UNSUBSCRIBE', 'params': subParams, 'id': requestId, } subscription: dict = { 'unsubscribe': True, 'id': str(requestId), 'subMessageHashes': subMessageHashes, 'messageHashes': messageHashes, 'symbols': symbols, 'topic': 'trades', } return await self.watch_multiple(url, messageHashes, self.extend(request, query), messageHashes, subscription) async def un_watch_trades(self, symbol: str, params={}) -> Any: """ unsubscribes from the trades channel https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#aggregate-trades https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#recent-trades https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Aggregate-Trade-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Aggregate-Trade-Streams :param str symbol: unified symbol of the market to fetch trades for :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.name]: the name of the method to call, 'trade' or 'aggTrade', default is 'trade' :returns dict[]: a list of `trade structures ` """ await self.load_markets() return await self.un_watch_trades_for_symbols([symbol], params) async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ get the list of most recent trades for a particular symbol https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#aggregate-trades https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#recent-trades https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Aggregate-Trade-Streams https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Aggregate-Trade-Streams :param str symbol: 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 :param str [params.name]: the name of the method to call, 'trade' or 'aggTrade', default is 'trade' :returns dict[]: a list of `trade structures ` """ params['callerMethodName'] = 'watchTrades' return await self.watch_trades_for_symbols([symbol], since, limit, params) def parse_ws_trade(self, trade, market=None) -> Trade: # # public watchTrades # # { # "e": "trade", # event type # "E": 1579481530911, # event time # "s": "ETHBTC", # symbol # "t": 158410082, # trade id # "p": "0.01914100", # price # "q": "0.00700000", # quantity # "b": 586187049, # buyer order id # "a": 586186710, # seller order id # "T": 1579481530910, # trade time # "m": False, # is the buyer the market maker # "M": True # binance docs say it should be ignored # } # # { # "e": "aggTrade", # Event type # "E": 123456789, # Event time # "s": "BNBBTC", # Symbol # "a": 12345, # Aggregate trade ID # "p": "0.001", # Price # "q": "100", # Quantity # "f": 100, # First trade ID # "l": 105, # Last trade ID # "T": 123456785, # Trade time # "m": True, # Is the buyer the market maker? # "M": True # Ignore # } # # private watchMyTrades spot # # { # "e": "executionReport", # "E": 1611063861489, # "s": "BNBUSDT", # "c": "m4M6AD5MF3b1ERe65l4SPq", # "S": "BUY", # "o": "MARKET", # "f": "GTC", # "q": "2.00000000", # "p": "0.00000000", # "P": "0.00000000", # "F": "0.00000000", # "g": -1, # "C": '', # "x": "TRADE", # "X": "PARTIALLY_FILLED", # "r": "NONE", # "i": 1296882607, # "l": "0.33200000", # "z": "0.33200000", # "L": "46.86600000", # "n": "0.00033200", # "N": "BNB", # "T": 1611063861488, # "t": 109747654, # "I": 2696953381, # "w": False, # "m": False, # "M": True, # "O": 1611063861488, # "Z": "15.55951200", # "Y": "15.55951200", # "Q": "0.00000000" # } # # private watchMyTrades future/delivery # # { # "s": "BTCUSDT", # "c": "pb2jD6ZQHpfzSdUac8VqMK", # "S": "SELL", # "o": "MARKET", # "f": "GTC", # "q": "0.001", # "p": "0", # "ap": "33468.46000", # "sp": "0", # "x": "TRADE", # "X": "FILLED", # "i": 13351197194, # "l": "0.001", # "z": "0.001", # "L": "33468.46", # "n": "0.00027086", # "N": "BNB", # "T": 1612095165362, # "t": 458032604, # "b": "0", # "a": "0", # "m": False, # "R": False, # "wt": "CONTRACT_PRICE", # "ot": "MARKET", # "ps": "BOTH", # "cp": False, # "rp": "0.00335000", # "pP": False, # "si": 0, # "ss": 0 # } # executionType = self.safe_string(trade, 'x') isTradeExecution = (executionType == 'TRADE') if not isTradeExecution: return self.parse_trade(trade, market) id = self.safe_string_2(trade, 't', 'a') timestamp = self.safe_integer(trade, 'T') price = self.safe_string_2(trade, 'L', 'p') amount = self.safe_string(trade, 'q') if isTradeExecution: amount = self.safe_string(trade, 'l', amount) cost = self.safe_string(trade, 'Y') if cost is None: if (price is not None) and (amount is not None): cost = Precise.string_mul(price, amount) marketId = self.safe_string(trade, 's') marketType = 'contract' if ('ps' in trade) else 'spot' symbol = self.safe_symbol(marketId, None, None, marketType) side = self.safe_string_lower(trade, 'S') takerOrMaker = None orderId = self.safe_string(trade, 'i') if 'm' in trade: if side is None: side = 'sell' if trade['m'] else 'buy' # self is reversed intentionally takerOrMaker = 'maker' if trade['m'] else 'taker' fee = None feeCost = self.safe_string(trade, 'n') if feeCost is not None: feeCurrencyId = self.safe_string(trade, 'N') feeCurrencyCode = self.safe_currency_code(feeCurrencyId) fee = { 'cost': feeCost, 'currency': feeCurrencyCode, } type = self.safe_string_lower(trade, 'o') return self.safe_trade({ 'info': trade, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'symbol': symbol, 'id': id, 'order': orderId, 'type': type, 'takerOrMaker': takerOrMaker, 'side': side, 'price': price, 'amount': amount, 'cost': cost, 'fee': fee, }) def handle_trade(self, client: Client, message): # the trade streams push raw trade information in real-time # each trade has a unique buyer and seller isSpot = (client.url.find('/stream') > -1) marketType = 'spot' if (isSpot) else 'contract' marketId = self.safe_string(message, 's') market = self.safe_market(marketId, None, None, marketType) symbol = market['symbol'] messageHash = 'trade::' + symbol trade = self.parse_ws_trade(message, market) tradesArray = self.safe_value(self.trades, symbol) if tradesArray is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) tradesArray = ArrayCache(limit) tradesArray.append(trade) self.trades[symbol] = tradesArray client.resolve(tradesArray, messageHash) 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://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#klines https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Kline-Candlestick-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Kline-Candlestick-Streams :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 :param dict [params.timezone]: if provided, kline intervals are interpreted in that timezone instead of UTC, example '+08:00' :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] 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://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#klines https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Kline-Candlestick-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Kline-Candlestick-Streams :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 :param dict [params.timezone]: if provided, kline intervals are interpreted in that timezone instead of UTC, example '+08:00' :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() klineType = None klineType, params = self.handle_param_string_2(params, 'channel', 'name', 'kline') symbols = self.get_list_from_object_values(symbolsAndTimeframes, 0) marketSymbols = self.market_symbols(symbols, None, False, False, True) firstMarket = self.market(marketSymbols[0]) type = firstMarket['type'] if firstMarket['contract']: type = 'future' if firstMarket['linear'] else 'delivery' isSpot = (type == 'spot') timezone = None timezone, params = self.handle_param_string(params, 'timezone', None) isUtc8 = (timezone is not None) and ((timezone == '+08:00') or Precise.string_eq(timezone, '8')) rawHashes = [] messageHashes = [] for i in range(0, len(symbolsAndTimeframes)): symAndTf = symbolsAndTimeframes[i] symbolString = symAndTf[0] timeframeString = symAndTf[1] interval = self.safe_string(self.timeframes, timeframeString, timeframeString) market = self.market(symbolString) marketId = market['lowercaseId'] if klineType == 'indexPriceKline': # weird behavior for index price kline we can't use the perp suffix marketId = marketId.replace('_perp', '') shouldUseUTC8 = (isUtc8 and isSpot) suffix = '@+08:00' utcSuffix = suffix if shouldUseUTC8 else '' rawHashes.append(marketId + '@' + klineType + '_' + interval + utcSuffix) messageHashes.append('ohlcv::' + market['symbol'] + '::' + timeframeString) url = self.urls['api']['ws'][type] + '/' + self.stream(type, 'multipleOHLCV') requestId = self.request_id(url) request = { 'method': 'SUBSCRIBE', 'params': rawHashes, 'id': requestId, } subscribe = { 'id': requestId, } params = self.omit(params, 'callerMethodName') symbol, timeframe, candles = await self.watch_multiple(url, messageHashes, self.extend(request, params), messageHashes, subscribe) if self.newUpdates: limit = candles.getLimit(symbol, limit) filtered = self.filter_by_since_limit(candles, 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://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#klines https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Kline-Candlestick-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Kline-Candlestick-Streams :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 :param dict [params.timezone]: if provided, kline intervals are interpreted in that timezone instead of UTC, example '+08:00' :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() klineType = None klineType, params = self.handle_param_string_2(params, 'channel', 'name', 'kline') symbols = self.get_list_from_object_values(symbolsAndTimeframes, 0) marketSymbols = self.market_symbols(symbols, None, False, False, True) firstMarket = self.market(marketSymbols[0]) type = firstMarket['type'] if firstMarket['contract']: type = 'future' if firstMarket['linear'] else 'delivery' isSpot = (type == 'spot') timezone = None timezone, params = self.handle_param_string(params, 'timezone', None) isUtc8 = (timezone is not None) and ((timezone == '+08:00') or Precise.string_eq(timezone, '8')) rawHashes = [] subMessageHashes = [] messageHashes = [] for i in range(0, len(symbolsAndTimeframes)): symAndTf = symbolsAndTimeframes[i] symbolString = symAndTf[0] timeframeString = symAndTf[1] interval = self.safe_string(self.timeframes, timeframeString, timeframeString) market = self.market(symbolString) marketId = market['lowercaseId'] if klineType == 'indexPriceKline': # weird behavior for index price kline we can't use the perp suffix marketId = marketId.replace('_perp', '') shouldUseUTC8 = (isUtc8 and isSpot) suffix = '@+08:00' utcSuffix = suffix if shouldUseUTC8 else '' rawHashes.append(marketId + '@' + klineType + '_' + interval + utcSuffix) subMessageHashes.append('ohlcv::' + market['symbol'] + '::' + timeframeString) messageHashes.append('unsubscribe::ohlcv::' + market['symbol'] + '::' + timeframeString) url = self.urls['api']['ws'][type] + '/' + self.stream(type, 'multipleOHLCV') requestId = self.request_id(url) request = { 'method': 'UNSUBSCRIBE', 'params': rawHashes, 'id': requestId, } subscribe = { 'unsubscribe': True, 'id': str(requestId), 'symbols': symbols, 'symbolsAndTimeframes': symbolsAndTimeframes, 'subMessageHashes': subMessageHashes, 'messageHashes': messageHashes, 'topic': 'ohlcv', } params = self.omit(params, 'callerMethodName') return await self.watch_multiple(url, messageHashes, self.extend(request, params), messageHashes, subscribe) 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://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#klines https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Kline-Candlestick-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Kline-Candlestick-Streams :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 :param dict [params.timezone]: if provided, kline intervals are interpreted in that timezone instead of UTC, example '+08:00' :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] params['callerMethodName'] = 'watchOHLCV' return await self.un_watch_ohlcv_for_symbols([[symbol, timeframe]], params) def handle_ohlcv(self, client: Client, message): # # { # "e": "kline", # "E": 1579482921215, # "s": "ETHBTC", # "k": { # "t": 1579482900000, # "T": 1579482959999, # "s": "ETHBTC", # "i": "1m", # "f": 158411535, # "L": 158411550, # "o": "0.01913200", # "c": "0.01913500", # "h": "0.01913700", # "l": "0.01913200", # "v": "5.08400000", # "n": 16, # "x": False, # "q": "0.09728060", # "V": "3.30200000", # "Q": "0.06318500", # "B": "0" # } # } # event = self.safe_string(message, 'e') eventMap: dict = { 'indexPrice_kline': 'indexPriceKline', 'markPrice_kline': 'markPriceKline', } event = self.safe_string(eventMap, event, event) kline = self.safe_value(message, 'k') marketId = self.safe_string_2(kline, 's', 'ps') if event == 'indexPriceKline': # indexPriceKline doesn't have the _PERP suffix marketId = self.safe_string(message, 'ps') interval = self.safe_string(kline, 'i') # use a reverse lookup in a static map instead unifiedTimeframe = self.find_timeframe(interval) parsed = [ self.safe_integer(kline, 't'), self.safe_float(kline, 'o'), self.safe_float(kline, 'h'), self.safe_float(kline, 'l'), self.safe_float(kline, 'c'), self.safe_float(kline, 'v'), ] isSpot = (client.url.find('/stream') > -1) marketType = 'spot' if (isSpot) else 'contract' symbol = self.safe_symbol(marketId, None, None, marketType) messageHash = 'ohlcv::' + symbol + '::' + unifiedTimeframe self.ohlcvs[symbol] = self.safe_value(self.ohlcvs, symbol, {}) stored = self.safe_value(self.ohlcvs[symbol], unifiedTimeframe) if stored is None: limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) stored = ArrayCacheByTimestamp(limit) self.ohlcvs[symbol][unifiedTimeframe] = stored stored.append(parsed) resolveData = [symbol, unifiedTimeframe, stored] client.resolve(resolveData, messageHash) async def fetch_ticker_ws(self, symbol: str, params={}) -> Ticker: """ fetches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.method]: method to use can be ticker.price or ticker.book :param boolean [params.returnRateLimits]: return the rate limits for the exchange :returns dict: a `ticker structure ` """ await self.load_markets() market = self.market(symbol) payload: dict = { 'symbol': market['id'], } type = self.get_market_type('fetchTickerWs', market, params) if type != 'future': raise BadRequest(self.id + ' fetchTickerWs only supports swap markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) subscription: dict = { 'method': self.handle_ticker_ws, } returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchTickerWs', 'returnRateLimits', False) payload['returnRateLimits'] = returnRateLimits params = self.omit(params, 'test') method = None method, params = self.handle_option_and_params(params, 'fetchTickerWs', 'method', 'ticker.book') message: dict = { 'id': messageHash, 'method': method, 'params': self.sign_params(self.extend(payload, params)), } ticker = await self.watch(url, messageHash, message, messageHash, subscription) return ticker async def fetch_ohlcv_ws(self, symbol: str, timeframe='1m', since: Int = None, limit: Int = None, params={}) -> List[list]: """ query historical candlestick data containing the open, high, low, and close price, and the volume of a market https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#klines :param str symbol: unified symbol of the market to query 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 :param int params['until']: timestamp in ms of the earliest candle to fetch EXCHANGE SPECIFIC PARAMETERS :param str params['timeZone']: default=0(UTC) :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() market = self.market(symbol) marketType = self.get_market_type('fetchOHLCVWs', market, params) if marketType != 'spot' and marketType != 'future': raise BadRequest(self.id + ' fetchOHLCVWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][marketType] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchOHLCVWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, 'interval': self.timeframes[timeframe], } until = self.safe_integer(params, 'until') params = self.omit(params, 'until') if since is not None: payload['startTime'] = since if limit is not None: payload['limit'] = limit if until is not None: payload['endTime'] = until message: dict = { 'id': messageHash, 'method': 'klines', 'params': self.extend(payload, params), } subscription: dict = { 'method': self.handle_fetch_ohlcv, } return await self.watch(url, messageHash, message, messageHash, subscription) def handle_fetch_ohlcv(self, client: Client, message): # # { # "id": "1dbbeb56-8eea-466a-8f6e-86bdcfa2fc0b", # "status": 200, # "result": [ # [ # 1655971200000, # Kline open time # "0.01086000", # Open price # "0.01086600", # High price # "0.01083600", # Low price # "0.01083800", # Close price # "2290.53800000", # Volume # 1655974799999, # Kline close time # "24.85074442", # Quote asset volume # 2283, # Number of trades # "1171.64000000", # Taker buy base asset volume # "12.71225884", # Taker buy quote asset volume # "0" # Unused field, ignore # ] # ], # "rateLimits": [ # { # "rateLimitType": "REQUEST_WEIGHT", # "interval": "MINUTE", # "intervalNum": 1, # "limit": 6000, # "count": 2 # } # ] # } # result = self.safe_list(message, 'result') parsed = self.parse_ohlcvs(result) # use a reverse lookup in a static map instead messageHash = self.safe_string(message, 'id') client.resolve(parsed, messageHash) 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://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-mini-ticker-stream https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#all-market-mini-tickers-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.name]: stream to use can be ticker or miniTicker :returns dict: a `ticker structure ` """ await self.load_markets() symbol = self.symbol(symbol) tickers = await self.watch_tickers([symbol], self.extend(params, {'callerMethodName': 'watchTicker'})) return tickers[symbol] async def watch_mark_price(self, symbol: str, params={}) -> Ticker: """ watches a mark price for a specific market https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Mark-Price-Stream :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.use1sFreq]: *default is True* if set to True, the mark price will be updated every second, otherwise every 3 seconds :returns dict: a `ticker structure ` """ await self.load_markets() symbol = self.symbol(symbol) tickers = await self.watch_mark_prices([symbol], self.extend(params, {'callerMethodName': 'watchMarkPrice'})) return tickers[symbol] async def watch_mark_prices(self, symbols: Strings = None, params={}) -> Tickers: """ watches the mark price for all markets https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Mark-Price-Stream-for-All-market :param str[] symbols: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.use1sFreq]: *default is True* if set to True, the mark price will be updated every second, otherwise every 3 seconds :returns dict: a `ticker structure ` """ channelName = None # for now watchmarkPrice uses the same messageHash # so it's impossible to watch both at the same time # refactor self to use different messageHashes channelName, params = self.handle_option_and_params(params, 'watchMarkPrices', 'name', 'markPrice') newTickers = await self.watch_multi_ticker_helper('watchMarkPrices', channelName, symbols, params) if self.newUpdates: return newTickers return self.filter_by_array(self.tickers, 'symbol', symbols) 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://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-mini-ticker-stream https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#all-market-mini-tickers-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams :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 ` """ channelName = None channelName, params = self.handle_option_and_params(params, 'watchTickers', 'name', 'ticker') if channelName == 'bookTicker': raise BadRequest(self.id + ' deprecation notice - to subscribe for bids-asks, use watch_bids_asks() method instead') newTickers = await self.watch_multi_ticker_helper('watchTickers', channelName, symbols, params) if self.newUpdates: return newTickers return self.filter_by_array(self.tickers, 'symbol', symbols) async def un_watch_tickers(self, symbols: Strings = None, params={}) -> Any: """ unWatches a price ticker, a statistical calculation with the information calculated over the past 24 hours for all markets of a specific list https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-mini-ticker-stream https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#all-market-mini-tickers-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams :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 ` """ channelName = None channelName, params = self.handle_option_and_params(params, 'watchTickers', 'name', 'ticker') if channelName == 'bookTicker': raise BadRequest(self.id + ' deprecation notice - to subscribe for bids-asks, use watch_bids_asks() method instead') await self.load_markets() methodName = 'watchTickers' symbols = self.market_symbols(symbols, None, True, False, True) firstMarket = None marketType = None symbolsDefined = (symbols is not None) if symbolsDefined: firstMarket = self.market(symbols[0]) marketType, params = self.handle_market_type_and_params(methodName, firstMarket, params) subType = None subType, params = self.handle_sub_type_and_params(methodName, firstMarket, params) rawMarketType = None if self.isLinear(marketType, subType): rawMarketType = 'future' elif self.isInverse(marketType, subType): rawMarketType = 'delivery' elif marketType == 'spot': rawMarketType = marketType else: raise NotSupported(self.id + ' ' + methodName + '() does not support options markets') isBidAsk = (channelName == 'bookTicker') subscriptionArgs = [] subMessageHashes = [] messageHashes = [] if symbolsDefined: for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) subscriptionArgs.append(market['lowercaseId'] + '@' + channelName) subMessageHashes.append(self.get_message_hash(channelName, market['symbol'], isBidAsk)) messageHashes.append('unsubscribe:ticker:' + symbol) else: if isBidAsk: if marketType == 'spot': raise ArgumentsRequired(self.id + ' ' + methodName + '() requires symbols for self channel for spot markets') subscriptionArgs.append('!' + channelName) else: subscriptionArgs.append('!' + channelName + '@arr') subMessageHashes.append(self.get_message_hash(channelName, None, isBidAsk)) messageHashes.append('unsubscribe:ticker') streamHash = channelName if symbolsDefined: streamHash = channelName + '::' + ','.join(symbols) url = self.urls['api']['ws'][rawMarketType] + '/' + self.stream(rawMarketType, streamHash) requestId = self.request_id(url) request: dict = { 'method': 'UNSUBSCRIBE', 'params': subscriptionArgs, 'id': requestId, } subscription: dict = { 'unsubscribe': True, 'id': str(requestId), 'subMessageHashes': subMessageHashes, 'messageHashes': subMessageHashes, 'symbols': symbols, 'topic': 'ticker', } return await self.watch_multiple(url, subMessageHashes, self.extend(request, params), subMessageHashes, subscription) async def un_watch_ticker(self, symbol: str, params={}) -> Any: """ unWatches a price ticker, a statistical calculation with the information calculated over the past 24 hours for all markets of a specific list https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-mini-ticker-stream https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#all-market-mini-tickers-stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/Individual-Symbol-Ticker-Streams :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 ` """ return await self.un_watch_tickers([symbol], params) async def watch_bids_asks(self, symbols: Strings = None, params={}) -> Tickers: """ watches best bid & ask for symbols https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#symbol-order-book-ticker https://developers.binance.com/docs/derivatives/coin-margined-futures/websocket-market-streams/All-Book-Tickers-Stream https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Book-Tickers-Stream :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, True, False, True) result = await self.watch_multi_ticker_helper('watchBidsAsks', 'bookTicker', symbols, params) if self.newUpdates: return result return self.filter_by_array(self.bidsasks, 'symbol', symbols) async def watch_multi_ticker_helper(self, methodName, channelName: str, symbols: Strings = None, params={}): await self.load_markets() symbols = self.market_symbols(symbols, None, True, False, True) isBidAsk = (channelName == 'bookTicker') isMarkPrice = (channelName == 'markPrice') use1sFreq = self.safe_bool(params, 'use1sFreq', True) firstMarket = None marketType = None symbolsDefined = (symbols is not None) if symbolsDefined: firstMarket = self.market(symbols[0]) defaultMarket = 'swap' if (isMarkPrice) else None marketType, params = self.handle_market_type_and_params(methodName, firstMarket, params, defaultMarket) subType = None subType, params = self.handle_sub_type_and_params(methodName, firstMarket, params) rawMarketType = None if self.isLinear(marketType, subType): rawMarketType = 'future' elif self.isInverse(marketType, subType): rawMarketType = 'delivery' elif marketType == 'spot': rawMarketType = marketType else: raise NotSupported(self.id + ' ' + methodName + '() does not support options markets') subscriptionArgs = [] messageHashes = [] suffix = '' if isMarkPrice: suffix = '@1s' if (use1sFreq) else '' if symbolsDefined: for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) subscriptionArgs.append(market['lowercaseId'] + '@' + channelName + suffix) messageHashes.append(self.get_message_hash(channelName, market['symbol'], isBidAsk)) else: if isBidAsk: if marketType == 'spot': raise ArgumentsRequired(self.id + ' ' + methodName + '() requires symbols for self channel for spot markets') subscriptionArgs.append('!' + channelName) elif isMarkPrice: subscriptionArgs.append('!' + channelName + '@arr' + suffix) else: subscriptionArgs.append('!' + channelName + '@arr') messageHashes.append(self.get_message_hash(channelName, None, isBidAsk)) streamHash = channelName if symbolsDefined: streamHash = channelName + '::' + ','.join(symbols) url = self.urls['api']['ws'][rawMarketType] + '/' + self.stream(rawMarketType, streamHash) requestId = self.request_id(url) request: dict = { 'method': 'SUBSCRIBE', 'params': subscriptionArgs, 'id': requestId, } subscribe: dict = { 'id': requestId, } result = await self.watch_multiple(url, messageHashes, self.deep_extend(request, params), subscriptionArgs, subscribe) # for efficiency, we have two type of returned structure here - if symbols array was provided, then individual # ticker dict comes in, otherwise all-tickers dict comes in if not symbolsDefined: return result else: newDict: dict = {} newDict[result['symbol']] = result return newDict def parse_ws_ticker(self, message, marketType): # markPrice # { # "e": "markPriceUpdate", # Event type # "E": 1562305380000, # Event time # "s": "BTCUSDT", # Symbol # "p": "11794.15000000", # Mark price # "i": "11784.62659091", # Index price # "P": "11784.25641265", # Estimated Settle Price, only useful in the last hour before the settlement starts # "r": "0.00038167", # Funding rate # "T": 1562306400000 # Next funding time # } # # ticker # { # "e": "24hrTicker", # event type # "E": 1579485598569, # event time # "s": "ETHBTC", # symbol # "p": "-0.00004000", # price change # "P": "-0.209", # price change percent # "w": "0.01920495", # weighted average price # "x": "0.01916500", # the price of the first trade before the 24hr rolling window # "c": "0.01912500", # last(closing) price # "Q": "0.10400000", # last quantity # "b": "0.01912200", # best bid # "B": "4.10400000", # best bid quantity # "a": "0.01912500", # best ask # "A": "0.00100000", # best ask quantity # "o": "0.01916500", # open price # "h": "0.01956500", # high price # "l": "0.01887700", # low price # "v": "173518.11900000", # base volume # "q": "3332.40703994", # quote volume # "O": 1579399197842, # open time # "C": 1579485597842, # close time # "F": 158251292, # first trade id # "L": 158414513, # last trade id # "n": 163222, # total number of trades # } # # miniTicker # { # "e": "24hrMiniTicker", # "E": 1671617114585, # "s": "MOBBUSD", # "c": "0.95900000", # "o": "0.91200000", # "h": "1.04000000", # "l": "0.89400000", # "v": "2109995.32000000", # "q": "2019254.05788000" # } # fetchTickerWs # { # "symbol":"BTCUSDT", # "price":"72606.70", # "time":1712526204284 # } # fetchTickerWs - ticker.book # { # "lastUpdateId":1027024, # "symbol":"BTCUSDT", # "bidPrice":"4.00000000", # "bidQty":"431.00000000", # "askPrice":"4.00000200", # "askQty":"9.00000000", # "time":1589437530011, # } # marketId = self.safe_string_2(message, 's', 'symbol') symbol = self.safe_symbol(marketId, None, None, marketType) event = self.safe_string(message, 'e', 'bookTicker') if event == '24hrTicker': event = 'ticker' if event == 'markPriceUpdate': # handle self separately because some fields clash with the ticker fields return self.safe_ticker({ 'symbol': symbol, 'timestamp': self.safe_integer(message, 'E'), 'datetime': self.iso8601(self.safe_integer(message, 'E')), 'info': message, 'markPrice': self.safe_string(message, 'p'), 'indexPrice': self.safe_string(message, 'i'), }) timestamp = None if event == 'bookTicker': # take the event timestamp, if available, for spot tickers it is not timestamp = self.safe_integer_2(message, 'E', 'time') else: # take the timestamp of the closing price for candlestick streams timestamp = self.safe_integer_n(message, ['C', 'E', 'time']) market = self.safe_market(marketId, None, None, marketType) last = self.safe_string_2(message, 'c', 'price') return self.safe_ticker({ 'symbol': symbol, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'high': self.safe_string(message, 'h'), 'low': self.safe_string(message, 'l'), 'bid': self.safe_string_2(message, 'b', 'bidPrice'), 'bidVolume': self.safe_string_2(message, 'B', 'bidQty'), 'ask': self.safe_string_2(message, 'a', 'askPrice'), 'askVolume': self.safe_string_2(message, 'A', 'askQty'), 'vwap': self.safe_string(message, 'w'), 'open': self.safe_string(message, 'o'), 'close': last, 'last': last, 'previousClose': self.safe_string(message, 'x'), # previous day close 'change': self.safe_string(message, 'p'), 'percentage': self.safe_string(message, 'P'), 'average': None, 'baseVolume': self.safe_string(message, 'v'), 'quoteVolume': self.safe_string(message, 'q'), 'info': message, }, market) def handle_ticker_ws(self, client: Client, message): # # ticker.price # { # "id":"1", # "status":200, # "result":{ # "symbol":"BTCUSDT", # "price":"73178.50", # "time":1712527052374 # } # } # ticker.book # { # "id":"9d32157c-a556-4d27-9866-66760a174b57", # "status":200, # "result":{ # "lastUpdateId":1027024, # "symbol":"BTCUSDT", # "bidPrice":"4.00000000", # "bidQty":"431.00000000", # "askPrice":"4.00000200", # "askQty":"9.00000000", # "time":1589437530011 # Transaction time # } # } # messageHash = self.safe_string(message, 'id') result = self.safe_value(message, 'result', {}) ticker = self.parse_ws_ticker(result, 'future') client.resolve(ticker, messageHash) def handle_bids_asks(self, client: Client, message): # # arrives one symbol dict or array of symbol dicts # # { # "u": 7488717758, # "s": "BTCUSDT", # "b": "28621.74000000", # "B": "1.43278800", # "a": "28621.75000000", # "A": "2.52500800" # } # self.handle_tickers_and_bids_asks(client, message, 'bidasks') def handle_tickers(self, client: Client, message): # # arrives one symbol dict or array of symbol dicts # # { # "e": "24hrTicker", # event type # "E": 1579485598569, # event time # "s": "ETHBTC", # symbol # "p": "-0.00004000", # price change # "P": "-0.209", # price change percent # "w": "0.01920495", # weighted average price # "x": "0.01916500", # the price of the first trade before the 24hr rolling window # "c": "0.01912500", # last(closing) price # "Q": "0.10400000", # last quantity # "b": "0.01912200", # best bid # "B": "4.10400000", # best bid quantity # "a": "0.01912500", # best ask # "A": "0.00100000", # best ask quantity # "o": "0.01916500", # open price # "h": "0.01956500", # high price # "l": "0.01887700", # low price # "v": "173518.11900000", # base volume # "q": "3332.40703994", # quote volume # "O": 1579399197842, # open time # "C": 1579485597842, # close time # "F": 158251292, # first trade id # "L": 158414513, # last trade id # "n": 163222, # total number of trades # } # self.handle_tickers_and_bids_asks(client, message, 'tickers') def handle_tickers_and_bids_asks(self, client: Client, message, methodType): isSpot = (client.url.find('/stream') > -1) marketType = 'spot' if (isSpot) else 'contract' isBidAsk = (methodType == 'bidasks') channelName = None resolvedMessageHashes = [] rawTickers = [] newTickers: dict = {} if isinstance(message, list): rawTickers = message else: rawTickers.append(message) for i in range(0, len(rawTickers)): ticker = rawTickers[i] event = self.safe_string(ticker, 'e') if isBidAsk: event = 'bookTicker' # in `handleMessage`, bookTicker doesn't have identifier, so manually set here channelName = self.safe_string(self.options['tickerChannelsMap'], event, event) if channelName is None: continue parsedTicker = self.parse_ws_ticker(ticker, marketType) symbol = parsedTicker['symbol'] newTickers[symbol] = parsedTicker if isBidAsk: self.bidsasks[symbol] = parsedTicker else: self.tickers[symbol] = parsedTicker messageHash = self.get_message_hash(channelName, symbol, isBidAsk) resolvedMessageHashes.append(messageHash) client.resolve(parsedTicker, messageHash) # resolve batch endpoint length = len(resolvedMessageHashes) if length > 0: batchMessageHash = self.get_message_hash(channelName, None, isBidAsk) client.resolve(newTickers, batchMessageHash) def get_message_hash(self, channelName: str, symbol: Str, isBidAsk: bool): prefix = 'bidask' if isBidAsk else 'ticker' if symbol is not None: return prefix + ':' + channelName + '@' + symbol else: return prefix + 's' + ':' + channelName def sign_params(self, params={}): self.check_required_credentials() extendedParams = self.extend({ 'timestamp': self.nonce(), 'apiKey': self.apiKey, }, params) defaultRecvWindow = self.safe_integer(self.options, 'recvWindow') if defaultRecvWindow is not None: params['recvWindow'] = defaultRecvWindow recvWindow = self.safe_integer(params, 'recvWindow') if recvWindow is not None: params['recvWindow'] = recvWindow extendedParams = self.keysort(extendedParams) query = self.urlencode(extendedParams) signature = None if self.secret.find('PRIVATE KEY') > -1: if len(self.secret) > 120: signature = self.rsa(query, self.secret, 'sha256') else: signature = self.eddsa(self.encode(query), self.secret, 'ed25519') else: signature = self.hmac(self.encode(query), self.encode(self.secret), hashlib.sha256) extendedParams['signature'] = signature return extendedParams async def authenticate(self, params={}): time = self.milliseconds() type = None type, params = self.handle_market_type_and_params('authenticate', None, params) subType = None subType, params = self.handle_sub_type_and_params('authenticate', None, params) isPortfolioMargin = None isPortfolioMargin, params = self.handle_option_and_params_2(params, 'authenticate', 'papi', 'portfolioMargin', False) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' marginMode = None marginMode, params = self.handle_margin_mode_and_params('authenticate', params) isIsolatedMargin = (marginMode == 'isolated') isCrossMargin = (marginMode == 'cross') or (marginMode is None) symbol = self.safe_string(params, 'symbol') params = self.omit(params, 'symbol') options = self.safe_value(self.options, type, {}) lastAuthenticatedTime = self.safe_integer(options, 'lastAuthenticatedTime', 0) listenKeyRefreshRate = self.safe_integer(self.options, 'listenKeyRefreshRate', 1200000) delay = self.sum(listenKeyRefreshRate, 10000) if time - lastAuthenticatedTime > delay: response = None if isPortfolioMargin: response = await self.papiPostListenKey(params) params = self.extend(params, {'portfolioMargin': True}) elif type == 'future': response = await self.fapiPrivatePostListenKey(params) elif type == 'delivery': response = await self.dapiPrivatePostListenKey(params) elif type == 'margin' and isCrossMargin: response = await self.sapiPostUserDataStream(params) elif isIsolatedMargin: if symbol is None: raise ArgumentsRequired(self.id + ' authenticate() requires a symbol argument for isolated margin mode') marketId = self.market_id(symbol) params = self.extend(params, {'symbol': marketId}) response = await self.sapiPostUserDataStreamIsolated(params) else: response = await self.publicPostUserDataStream(params) self.options[type] = self.extend(options, { 'listenKey': self.safe_string(response, 'listenKey'), 'lastAuthenticatedTime': time, }) self.delay(listenKeyRefreshRate, self.keep_alive_listen_key, params) async def keep_alive_listen_key(self, params={}): # https://binance-docs.github.io/apidocs/spot/en/#listen-key-spot type = self.safe_string_2(self.options, 'defaultType', 'authenticate', 'spot') type = self.safe_string(params, 'type', type) isPortfolioMargin = None isPortfolioMargin, params = self.handle_option_and_params_2(params, 'keepAliveListenKey', 'papi', 'portfolioMargin', False) subTypeInfo = self.handle_sub_type_and_params('keepAliveListenKey', None, params) subType = subTypeInfo[0] if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' options = self.safe_value(self.options, type, {}) listenKey = self.safe_string(options, 'listenKey') if listenKey is None: # A network error happened: we can't renew a listen key that does not exist. return request: dict = {} symbol = self.safe_string(params, 'symbol') params = self.omit(params, ['type', 'symbol']) time = self.milliseconds() try: if isPortfolioMargin: await self.papiPutListenKey(self.extend(request, params)) params = self.extend(params, {'portfolioMargin': True}) elif type == 'future': await self.fapiPrivatePutListenKey(self.extend(request, params)) elif type == 'delivery': await self.dapiPrivatePutListenKey(self.extend(request, params)) else: request['listenKey'] = listenKey if type == 'margin': request['symbol'] = symbol await self.sapiPutUserDataStream(self.extend(request, params)) else: await self.publicPutUserDataStream(self.extend(request, params)) except Exception as error: urlType = type if isPortfolioMargin: urlType = 'papi' url = self.urls['api']['ws'][urlType] + '/' + self.options[type]['listenKey'] client = self.client(url) messageHashes = list(client.futures.keys()) for i in range(0, len(messageHashes)): messageHash = messageHashes[i] client.reject(error, messageHash) self.options[type] = self.extend(options, { 'listenKey': None, 'lastAuthenticatedTime': 0, }) return self.options[type] = self.extend(options, { 'listenKey': listenKey, 'lastAuthenticatedTime': time, }) # whether or not to schedule another listenKey keepAlive request clients = list(self.clients.values()) listenKeyRefreshRate = self.safe_integer(self.options, 'listenKeyRefreshRate', 1200000) for i in range(0, len(clients)): client = clients[i] subscriptionKeys = list(client.subscriptions.keys()) for j in range(0, len(subscriptionKeys)): subscribeType = subscriptionKeys[j] if subscribeType == type: self.delay(listenKeyRefreshRate, self.keep_alive_listen_key, params) return def set_balance_cache(self, client: Client, type, isPortfolioMargin=False): if type in client.subscriptions: return options = self.safe_value(self.options, 'watchBalance') fetchBalanceSnapshot = self.safe_bool(options, 'fetchBalanceSnapshot', False) if fetchBalanceSnapshot: messageHash = type + ':fetchBalanceSnapshot' if not (messageHash in client.futures): client.future(messageHash) self.spawn(self.load_balance_snapshot, client, messageHash, type, isPortfolioMargin) else: self.balance[type] = {} async def load_balance_snapshot(self, client, messageHash, type, isPortfolioMargin): params: dict = { 'type': type, } if isPortfolioMargin: params['portfolioMargin'] = True response = await self.fetch_balance(params) self.balance[type] = self.extend(response, self.safe_value(self.balance, type, {})) # don't remove the future from the .futures cache future = client.futures[messageHash] future.resolve() client.resolve(self.balance[type], type + ':balance') async def fetch_balance_ws(self, params={}) -> Balances: """ fetch balance and get the amount of funds available for trading or funds locked in orders https://developers.binance.com/docs/derivatives/usds-margined-futures/account/websocket-api/Futures-Account-Balance https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/account-requests#account-information-user_data https://developers.binance.com/docs/derivatives/coin-margined-futures/account/websocket-api :param dict [params]: extra parameters specific to the exchange API endpoint :param str|None [params.type]: 'future', 'delivery', 'savings', 'funding', or 'spot' :param str|None [params.marginMode]: 'cross' or 'isolated', for margin trading, uses self.options.defaultMarginMode if not passed, defaults to None/None/None :param str[]|None [params.symbols]: unified market symbols, only used in isolated margin mode :param str|None [params.method]: method to use. Can be account.balance, account.status, v2/account.balance or v2/account.status :returns dict: a `balance structure ` """ await self.load_markets() type = self.get_market_type('fetchBalanceWs', None, params) if type != 'spot' and type != 'future' and type != 'delivery': raise BadRequest(self.id + ' fetchBalanceWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchBalanceWs', 'returnRateLimits', False) payload: dict = { 'returnRateLimits': returnRateLimits, } method = None method, params = self.handle_option_and_params(params, 'fetchBalanceWs', 'method', 'account.status') message: dict = { 'id': messageHash, 'method': method, 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_account_status_ws if (method == 'account.status') else self.handle_balance_ws, } return await self.watch(url, messageHash, message, messageHash, subscription) def handle_balance_ws(self, client: Client, message): # # messageHash = self.safe_string(message, 'id') rawBalance = None if isinstance(message['result'], list): # account.balance rawBalance = self.safe_list(message, 'result', []) else: # account.status result = self.safe_dict(message, 'result', {}) rawBalance = self.safe_list(result, 'assets', []) parsedBalances = self.parseBalanceCustom(rawBalance) client.resolve(parsedBalances, messageHash) def handle_account_status_ws(self, client: Client, message): # # spot # { # "id": "605a6d20-6588-4cb9-afa0-b0ab087507ba", # "status": 200, # "result": { # "makerCommission": 15, # "takerCommission": 15, # "buyerCommission": 0, # "sellerCommission": 0, # "canTrade": True, # "canWithdraw": True, # "canDeposit": True, # "commissionRates": { # "maker": "0.00150000", # "taker": "0.00150000", # "buyer": "0.00000000", # "seller": "0.00000000" # }, # "brokered": False, # "requireSelfTradePrevention": False, # "updateTime": 1660801833000, # "accountType": "SPOT", # "balances": [{ # "asset": "BNB", # "free": "0.00000000", # "locked": "0.00000000" # }, # { # "asset": "BTC", # "free": "1.3447112", # "locked": "0.08600000" # }, # { # "asset": "USDT", # "free": "1021.21000000", # "locked": "0.00000000" # } # ], # "permissions": [ # "SPOT" # ] # } # } # swap # messageHash = self.safe_string(message, 'id') result = self.safe_dict(message, 'result', {}) parsedBalances = self.parseBalanceCustom(result) client.resolve(parsedBalances, messageHash) async def fetch_position_ws(self, symbol: str, params={}) -> List[Position]: """ fetch data on an open position https://developers.binance.com/docs/derivatives/usds-margined-futures/trade/websocket-api/Position-Information :param str symbol: unified market symbol of the market the position is held in :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `position structure ` """ return await self.fetch_positions_ws([symbol], params) async def fetch_positions_ws(self, symbols: Strings = None, params={}) -> List[Position]: """ fetch all open positions https://developers.binance.com/docs/derivatives/usds-margined-futures/trade/websocket-api/Position-Information https://developers.binance.com/docs/derivatives/coin-margined-futures/trade/websocket-api/Position-Information :param str[] [symbols]: list of unified market symbols :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.returnRateLimits]: set to True to return rate limit informations, defaults to False. :param str|None [params.method]: method to use. Can be account.position or v2/account.position :returns dict[]: a list of `position structure ` """ await self.load_markets() payload: dict = {} market = None symbols = self.market_symbols(symbols, 'swap', True, True, True) if symbols is not None: symbolsLength = len(symbols) if symbolsLength == 1: market = self.market(symbols[0]) payload['symbol'] = market['id'] type = self.get_market_type('fetchPositionsWs', market, params) if type != 'future' and type != 'delivery': raise BadRequest(self.id + ' fetchPositionsWs only supports swap markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchPositionsWs', 'returnRateLimits', False) payload['returnRateLimits'] = returnRateLimits method = None method, params = self.handle_option_and_params(params, 'fetchPositionsWs', 'method', 'account.position') message: dict = { 'id': messageHash, 'method': method, 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_positions_ws, } result = await self.watch(url, messageHash, message, messageHash, subscription) return self.filter_by_array_positions(result, 'symbol', symbols, False) def handle_positions_ws(self, client: Client, message): # # { # id: '1', # status: 200, # result: [ # { # symbol: 'BTCUSDT', # positionAmt: '-0.014', # entryPrice: '42901.1', # breakEvenPrice: '30138.83333142', # markPrice: '71055.98470333', # unRealizedProfit: '-394.16838584', # liquidationPrice: '137032.02272908', # leverage: '123', # maxNotionalValue: '50000', # marginType: 'cross', # isolatedMargin: '0.00000000', # isAutoAddMargin: 'false', # positionSide: 'BOTH', # notional: '-994.78378584', # isolatedWallet: '0', # updateTime: 1708906343111, # isolated: False, # adlQuantile: 2 # }, # ... # ] # } # # messageHash = self.safe_string(message, 'id') result = self.safe_list(message, 'result', []) positions = [] for i in range(0, len(result)): parsed = self.parse_position_risk(result[i]) entryPrice = self.safe_string(parsed, 'entryPrice') if (entryPrice != '0') and (entryPrice != '0.0') and (entryPrice != '0.00000000'): positions.append(parsed) client.resolve(positions, messageHash) async def watch_balance(self, params={}) -> Balances: """ watch balance and get the amount of funds available for trading or funds locked in orders :param dict [params]: extra parameters specific to the exchange API endpoint :param boolean [params.portfolioMargin]: set to True if you would like to watch the balance of a portfolio margin account :returns dict: a `balance structure ` """ await self.load_markets() await self.authenticate(params) defaultType = self.safe_string(self.options, 'defaultType', 'spot') type = self.safe_string(params, 'type', defaultType) subType = None subType, params = self.handle_sub_type_and_params('watchBalance', None, params) isPortfolioMargin = None isPortfolioMargin, params = self.handle_option_and_params_2(params, 'watchBalance', 'papi', 'portfolioMargin', False) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' urlType = type if isPortfolioMargin: urlType = 'papi' url = self.urls['api']['ws'][urlType] + '/' + self.options[type]['listenKey'] client = self.client(url) self.set_balance_cache(client, type, isPortfolioMargin) self.set_positions_cache(client, type, None, isPortfolioMargin) options = self.safe_dict(self.options, 'watchBalance') fetchBalanceSnapshot = self.safe_bool(options, 'fetchBalanceSnapshot', False) awaitBalanceSnapshot = self.safe_bool(options, 'awaitBalanceSnapshot', True) if fetchBalanceSnapshot and awaitBalanceSnapshot: await client.future(type + ':fetchBalanceSnapshot') messageHash = type + ':balance' message = None return await self.watch(url, messageHash, message, type) def handle_balance(self, client: Client, message): # # sent upon a balance update not related to orders # # { # "e": "balanceUpdate", # "E": 1629352505586, # "a": "IOTX", # "d": "0.43750000", # "T": 1629352505585 # } # # sent upon creating or filling an order # # { # "e": "outboundAccountPosition", # Event type # "E": 1564034571105, # Event Time # "u": 1564034571073, # Time of last account update # "B": [ # Balances Array # { # "a": "ETH", # Asset # "f": "10000.000000", # Free # "l": "0.000000" # Locked # } # ] # } # # future/delivery # # { # "e": "ACCOUNT_UPDATE", # Event Type # "E": 1564745798939, # Event Time # "T": 1564745798938 , # Transaction # "i": "SfsR", # Account Alias # "a": { # Update Data # "m":"ORDER", # Event reason type # "B":[ # Balances # { # "a":"BTC", # Asset # "wb":"122624.12345678", # Wallet Balance # "cw":"100.12345678" # Cross Wallet Balance # }, # ], # "P":[ # { # "s":"BTCUSD_200925", # Symbol # "pa":"0", # Position Amount # "ep":"0.0", # Entry Price # "cr":"200", #(Pre-fee) Accumulated Realized # "up":"0", # Unrealized PnL # "mt":"isolated", # Margin Type # "iw":"0.00000000", # Isolated Wallet(if isolated position) # "ps":"BOTH" # Position Side # }, # ] # } # } # wallet = self.safe_string(self.options, 'wallet', 'wb') # cw for cross wallet # each account is connected to a different endpoint # and has exactly one subscriptionhash which is the account type subscriptions = list(client.subscriptions.keys()) accountType = subscriptions[0] messageHash = accountType + ':balance' if self.balance[accountType] is None: self.balance[accountType] = {} self.balance[accountType]['info'] = message event = self.safe_string(message, 'e') if event == 'balanceUpdate': currencyId = self.safe_string(message, 'a') code = self.safe_currency_code(currencyId) account = self.account() delta = self.safe_string(message, 'd') if code in self.balance[accountType]: previousValue = self.balance[accountType][code]['free'] if not isinstance(previousValue, str): previousValue = self.number_to_string(previousValue) account['free'] = Precise.string_add(previousValue, delta) else: account['free'] = delta self.balance[accountType][code] = account else: message = self.safe_dict(message, 'a', message) B = self.safe_list(message, 'B') for i in range(0, len(B)): entry = B[i] currencyId = self.safe_string(entry, 'a') code = self.safe_currency_code(currencyId) account = self.account() account['free'] = self.safe_string(entry, 'f') account['used'] = self.safe_string(entry, 'l') account['total'] = self.safe_string(entry, wallet) self.balance[accountType][code] = account timestamp = self.safe_integer(message, 'E') self.balance[accountType]['timestamp'] = timestamp self.balance[accountType]['datetime'] = self.iso8601(timestamp) self.balance[accountType] = self.safe_balance(self.balance[accountType]) client.resolve(self.balance[accountType], messageHash) def get_market_type(self, method, market, params={}): type = None type, params = self.handle_market_type_and_params(method, market, params) subType = None subType, params = self.handle_sub_type_and_params(method, market, params) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' return type async def create_order_ws(self, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}) -> Order: """ create a trade order https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#place-new-order-trade https://developers.binance.com/docs/derivatives/usds-margined-futures/trade/websocket-api/New-Order https://developers.binance.com/docs/derivatives/coin-margined-futures/trade/websocket-api :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|None [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 boolean params['test']: test order, default False :param boolean params['returnRateLimits']: set to True to return rate limit information, default False :returns dict: an `order structure ` """ await self.load_markets() market = self.market(symbol) marketType = self.get_market_type('createOrderWs', market, params) if marketType != 'spot' and marketType != 'future' and marketType != 'delivery': raise BadRequest(self.id + ' createOrderWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][marketType] requestId = self.request_id(url) messageHash = str(requestId) sor = self.safe_bool_2(params, 'sor', 'SOR', False) params = self.omit(params, 'sor', 'SOR') payload = self.create_order_request(symbol, type, side, amount, price, params) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'createOrderWs', 'returnRateLimits', False) payload['returnRateLimits'] = returnRateLimits test = self.safe_bool(params, 'test', False) params = self.omit(params, 'test') message: dict = { 'id': messageHash, 'method': 'order.place', 'params': self.sign_params(self.extend(payload, params)), } if test: if sor: message['method'] = 'sor.order.test' else: message['method'] = 'order.test' subscription: dict = { 'method': self.handle_order_ws, } return await self.watch(url, messageHash, message, messageHash, subscription) def handle_order_ws(self, client: Client, message): # # { # "id": 1, # "status": 200, # "result": { # "symbol": "BTCUSDT", # "orderId": 7663053, # "orderListId": -1, # "clientOrderId": "x-R4BD3S82d8959d0f5114499487a614", # "transactTime": 1687642291434, # "price": "25000.00000000", # "origQty": "0.00100000", # "executedQty": "0.00000000", # "cummulativeQuoteQty": "0.00000000", # "status": "NEW", # "timeInForce": "GTC", # "type": "LIMIT", # "side": "BUY", # "workingTime": 1687642291434, # "fills": [], # "selfTradePreventionMode": "NONE" # }, # "rateLimits": [ # { # "rateLimitType": "ORDERS", # "interval": "SECOND", # "intervalNum": 10, # "limit": 50, # "count": 1 # }, # { # "rateLimitType": "ORDERS", # "interval": "DAY", # "intervalNum": 1, # "limit": 160000, # "count": 1 # }, # { # "rateLimitType": "REQUEST_WEIGHT", # "interval": "MINUTE", # "intervalNum": 1, # "limit": 1200, # "count": 12 # } # ] # } # messageHash = self.safe_string(message, 'id') result = self.safe_dict(message, 'result', {}) order = self.parse_order(result) client.resolve(order, messageHash) def handle_orders_ws(self, client: Client, message): # # { # "id": 1, # "status": 200, # "result": [{ # "symbol": "BTCUSDT", # "orderId": 7665584, # "orderListId": -1, # "clientOrderId": "x-R4BD3S82b54769abdd3e4b57874c52", # "price": "26000.00000000", # "origQty": "0.00100000", # "executedQty": "0.00000000", # "cummulativeQuoteQty": "0.00000000", # "status": "NEW", # "timeInForce": "GTC", # "type": "LIMIT", # "side": "BUY", # "stopPrice": "0.00000000", # "icebergQty": "0.00000000", # "time": 1687642884646, # "updateTime": 1687642884646, # "isWorking": True, # "workingTime": 1687642884646, # "origQuoteOrderQty": "0.00000000", # "selfTradePreventionMode": "NONE" # }, # ... # ], # "rateLimits": [{ # "rateLimitType": "REQUEST_WEIGHT", # "interval": "MINUTE", # "intervalNum": 1, # "limit": 1200, # "count": 14 # }] # } # messageHash = self.safe_string(message, 'id') result = self.safe_list(message, 'result', []) orders = self.parse_orders(result) client.resolve(orders, messageHash) async def edit_order_ws(self, id: str, symbol: str, type: OrderType, side: OrderSide, amount: Num = None, price: Num = None, params={}) -> Order: """ edit a trade order https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#cancel-and-replace-order-trade https://developers.binance.com/docs/derivatives/usds-margined-futures/trade/websocket-api/Modify-Order https://developers.binance.com/docs/derivatives/coin-margined-futures/trade/websocket-api/Modify-Order :param str id: 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 the currency you want to trade in units of the base currency :param float|None [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 :returns dict: an `order structure ` """ await self.load_markets() market = self.market(symbol) marketType = self.get_market_type('editOrderWs', market, params) if marketType != 'spot' and marketType != 'future' and marketType != 'delivery': raise BadRequest(self.id + ' editOrderWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][marketType] requestId = self.request_id(url) messageHash = str(requestId) isSwap = (marketType == 'future' or marketType == 'delivery') payload = None if marketType == 'spot': payload = self.editSpotOrderRequest(id, symbol, type, side, amount, price, params) elif isSwap: payload = self.editContractOrderRequest(id, symbol, type, side, amount, price, params) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'editOrderWs', 'returnRateLimits', False) payload['returnRateLimits'] = returnRateLimits message: dict = { 'id': messageHash, 'method': 'order.modify' if (isSwap) else 'order.cancelReplace', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_edit_order_ws, } return await self.watch(url, messageHash, message, messageHash, subscription) def handle_edit_order_ws(self, client: Client, message): # # spot # { # "id": 1, # "status": 200, # "result": { # "cancelResult": "SUCCESS", # "newOrderResult": "SUCCESS", # "cancelResponse": { # "symbol": "BTCUSDT", # "origClientOrderId": "x-R4BD3S82813c5d7ffa594104917de2", # "orderId": 7665177, # "orderListId": -1, # "clientOrderId": "mbrnbQsQhtCXCLY45d5q7S", # "price": "26000.00000000", # "origQty": "0.00100000", # "executedQty": "0.00000000", # "cummulativeQuoteQty": "0.00000000", # "status": "CANCELED", # "timeInForce": "GTC", # "type": "LIMIT", # "side": "BUY", # "selfTradePreventionMode": "NONE" # }, # "newOrderResponse": { # "symbol": "BTCUSDT", # "orderId": 7665584, # "orderListId": -1, # "clientOrderId": "x-R4BD3S82b54769abdd3e4b57874c52", # "transactTime": 1687642884646, # "price": "26000.00000000", # "origQty": "0.00100000", # "executedQty": "0.00000000", # "cummulativeQuoteQty": "0.00000000", # "status": "NEW", # "timeInForce": "GTC", # "type": "LIMIT", # "side": "BUY", # "workingTime": 1687642884646, # "fills": [], # "selfTradePreventionMode": "NONE" # } # }, # "rateLimits": [{ # "rateLimitType": "ORDERS", # "interval": "SECOND", # "intervalNum": 10, # "limit": 50, # "count": 1 # }, # { # "rateLimitType": "ORDERS", # "interval": "DAY", # "intervalNum": 1, # "limit": 160000, # "count": 3 # }, # { # "rateLimitType": "REQUEST_WEIGHT", # "interval": "MINUTE", # "intervalNum": 1, # "limit": 1200, # "count": 12 # } # ] # } # swap # { # "id":"1", # "status":200, # "result":{ # "orderId":667061487, # "symbol":"LTCUSDT", # "status":"NEW", # "clientOrderId":"x-xcKtGhcu91a74c818749ee42c0f70", # "price":"82.00", # "avgPrice":"0.00", # "origQty":"1.000", # "executedQty":"0.000", # "cumQty":"0.000", # "cumQuote":"0.00000", # "timeInForce":"GTC", # "type":"LIMIT", # "reduceOnly":false, # "closePosition":false, # "side":"BUY", # "positionSide":"BOTH", # "stopPrice":"0.00", # "workingType":"CONTRACT_PRICE", # "priceProtect":false, # "origType":"LIMIT", # "priceMatch":"NONE", # "selfTradePreventionMode":"NONE", # "goodTillDate":0, # "updateTime":1712918927511 # } # } # messageHash = self.safe_string(message, 'id') result = self.safe_dict(message, 'result', {}) newSpotOrder = self.safe_dict(result, 'newOrderResponse') order = None if newSpotOrder is not None: order = self.parse_order(newSpotOrder) else: order = self.parse_order(result) client.resolve(order, messageHash) async def cancel_order_ws(self, id: str, symbol: Str = None, params={}) -> Order: """ cancel multiple orders https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#cancel-order-trade https://developers.binance.com/docs/derivatives/usds-margined-futures/trade/websocket-api/Cancel-Order https://developers.binance.com/docs/derivatives/coin-margined-futures/trade/websocket-api/Cancel-Order :param str id: order id :param str [symbol]: unified market symbol, default is None :param dict [params]: extra parameters specific to the exchange API endpoint :param str|None [params.cancelRestrictions]: Supported values: ONLY_NEW - Cancel will succeed if the order status is NEW. ONLY_PARTIALLY_FILLED - Cancel will succeed if order status is PARTIALLY_FILLED. :returns dict: an list of `order structures ` """ await self.load_markets() if symbol is None: raise BadRequest(self.id + ' cancelOrderWs requires a symbol') market = self.market(symbol) type = self.get_market_type('cancelOrderWs', market, params) url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'cancelOrderWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, } clientOrderId = self.safe_string_2(params, 'origClientOrderId', 'clientOrderId') if clientOrderId is not None: payload['origClientOrderId'] = clientOrderId else: payload['orderId'] = self.parse_to_int(id) params = self.omit(params, ['origClientOrderId', 'clientOrderId']) message: dict = { 'id': messageHash, 'method': 'order.cancel', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_order_ws, } return await self.watch(url, messageHash, message, messageHash, subscription) async def cancel_all_orders_ws(self, symbol: Str = None, params={}): """ cancel all open orders in a market https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#cancel-open-orders-trade :param str [symbol]: unified market symbol of the market to cancel orders in :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `order structures ` """ await self.load_markets() market = self.market(symbol) type = self.get_market_type('cancelAllOrdersWs', market, params) if type != 'spot' and type != 'future': raise BadRequest(self.id + ' cancelAllOrdersWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'cancelAllOrdersWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, } message: dict = { 'id': messageHash, 'method': 'order.cancel', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_orders_ws, } return await self.watch(url, messageHash, message, messageHash, subscription) async def fetch_order_ws(self, id: str, symbol: Str = None, params={}) -> Order: """ fetches information on an order made by the user https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#query-order-user_data https://developers.binance.com/docs/derivatives/usds-margined-futures/trade/websocket-api/Query-Order https://developers.binance.com/docs/derivatives/coin-margined-futures/trade/websocket-api/Query-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 :returns dict: An `order structure ` """ await self.load_markets() if symbol is None: raise BadRequest(self.id + ' cancelOrderWs requires a symbol') market = self.market(symbol) type = self.get_market_type('fetchOrderWs', market, params) if type != 'spot' and type != 'future' and type != 'delivery': raise BadRequest(self.id + ' fetchOrderWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchOrderWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, } clientOrderId = self.safe_string_2(params, 'origClientOrderId', 'clientOrderId') if clientOrderId is not None: payload['origClientOrderId'] = clientOrderId else: payload['orderId'] = self.parse_to_int(id) message: dict = { 'id': messageHash, 'method': 'order.status', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_order_ws, } return await self.watch(url, messageHash, message, messageHash, subscription) async def fetch_orders_ws(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ fetches information on multiple orders made by the user https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#order-lists :param str symbol: unified market symbol of the market orders were made in :param int|None [since]: the earliest time in ms to fetch orders for :param int|None [limit]: the maximum number of order structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :param int [params.orderId]: order id to begin at :param int [params.startTime]: earliest time in ms to retrieve orders for :param int [params.endTime]: latest time in ms to retrieve orders for :param int [params.limit]: the maximum number of order structures to retrieve :returns dict[]: a list of `order structures ` """ await self.load_markets() if symbol is None: raise BadRequest(self.id + ' fetchOrdersWs requires a symbol') market = self.market(symbol) type = self.get_market_type('fetchOrdersWs', market, params) if type != 'spot': raise BadRequest(self.id + ' fetchOrdersWs only supports spot markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchOrderWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, } message: dict = { 'id': messageHash, 'method': 'allOrders', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_orders_ws, } orders = await self.watch(url, messageHash, message, messageHash, subscription) return self.filter_by_symbol_since_limit(orders, symbol, since, limit) async def fetch_closed_orders_ws(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ fetch closed orders https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#order-lists :param str symbol: unified market symbol :param int [since]: the earliest time in ms to fetch open orders for :param int [limit]: the maximum number of open orders structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `order structures ` """ orders = await self.fetch_orders_ws(symbol, since, limit, params) closedOrders = [] for i in range(0, len(orders)): order = orders[i] if order['status'] == 'closed': closedOrders.append(order) return closedOrders async def fetch_open_orders_ws(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ fetch all unfilled currently open orders https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/trading-requests#current-open-orders-user_data :param str symbol: unified market symbol :param int|None [since]: the earliest time in ms to fetch open orders for :param int|None [limit]: the maximum number of open orders 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() market = self.market(symbol) type = self.get_market_type('fetchOpenOrdersWs', market, params) if type != 'spot' and type != 'future': raise BadRequest(self.id + ' fetchOpenOrdersWs only supports spot or swap markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchOrderWs', 'returnRateLimits', False) payload: dict = { 'returnRateLimits': returnRateLimits, } if symbol is not None: payload['symbol'] = self.market_id(symbol) message: dict = { 'id': messageHash, 'method': 'openOrders.status', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_orders_ws, } orders = await self.watch(url, messageHash, message, messageHash, subscription) return self.filter_by_symbol_since_limit(orders, symbol, since, limit) 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://developers.binance.com/docs/binance-spot-api-docs/user-data-stream#order-update https://developers.binance.com/docs/margin_trading/trade-data-stream/Event-Order-Update https://developers.binance.com/docs/derivatives/usds-margined-futures/user-data-streams/Event-Order-Update :param str symbol: unified market symbol of the market the 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 str|None [params.marginMode]: 'cross' or 'isolated', for spot margin :param boolean [params.portfolioMargin]: set to True if you would like to watch portfolio margin account orders :returns dict[]: a list of `order structures ` """ await self.load_markets() messageHash = 'orders' market = None if symbol is not None: market = self.market(symbol) symbol = market['symbol'] messageHash += ':' + symbol type = None type, params = self.handle_market_type_and_params('watchOrders', market, params) subType = None subType, params = self.handle_sub_type_and_params('watchOrders', market, params) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' params = self.extend(params, {'type': type, 'symbol': symbol}) # needed inside authenticate for isolated margin await self.authenticate(params) marginMode = None marginMode, params = self.handle_margin_mode_and_params('watchOrders', params) urlType = type if (type == 'margin') or ((type == 'spot') and (marginMode is not None)): urlType = 'spot' # spot-margin shares the same stream spot isPortfolioMargin = None isPortfolioMargin, params = self.handle_option_and_params_2(params, 'watchOrders', 'papi', 'portfolioMargin', False) if isPortfolioMargin: urlType = 'papi' url = self.urls['api']['ws'][urlType] + '/' + self.options[type]['listenKey'] client = self.client(url) self.set_balance_cache(client, type, isPortfolioMargin) self.set_positions_cache(client, type, None, isPortfolioMargin) message = None orders = await self.watch(url, messageHash, message, type) if self.newUpdates: limit = orders.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True) def parse_ws_order(self, order, market=None): # # spot # # { # "e": "executionReport", # Event type # "E": 1499405658658, # Event time # "s": "ETHBTC", # Symbol # "c": "mUvoqJxFIILMdfAW5iGSOW", # Client order ID # "S": "BUY", # Side # "o": "LIMIT", # Order type # "f": "GTC", # Time in force # "q": "1.00000000", # Order quantity # "p": "0.10264410", # Order price # "P": "0.00000000", # Stop price # "F": "0.00000000", # Iceberg quantity # "g": -1, # OrderListId # "C": null, # Original client order ID; This is the ID of the order being canceled # "x": "NEW", # Current execution type # "X": "NEW", # Current order status # "r": "NONE", # Order reject reason; will be an error code. # "i": 4293153, # Order ID # "l": "0.00000000", # Last executed quantity # "z": "0.00000000", # Cumulative filled quantity # "L": "0.00000000", # Last executed price # "n": "0", # Commission amount # "N": null, # Commission asset # "T": 1499405658657, # Transaction time # "t": -1, # Trade ID # "I": 8641984, # Ignore # "w": True, # Is the order on the book? # "m": False, # Is self trade the maker side? # "M": False, # Ignore # "O": 1499405658657, # Order creation time # "Z": "0.00000000", # Cumulative quote asset transacted quantity # "Y": "0.00000000" # Last quote asset transacted quantity(i.e. lastPrice * lastQty), # "Q": "0.00000000" # Quote Order Qty # } # # future # # { # "s":"BTCUSDT", # Symbol # "c":"TEST", # Client Order Id # # special client order id: # # starts with "autoclose-": liquidation order # # "adl_autoclose": ADL auto close order # "S":"SELL", # Side # "o":"TRAILING_STOP_MARKET", # Order Type # "f":"GTC", # Time in Force # "q":"0.001", # Original Quantity # "p":"0", # Original Price # "ap":"0", # Average Price # "sp":"7103.04", # Stop Price. Please ignore with TRAILING_STOP_MARKET order # "x":"NEW", # Execution Type # "X":"NEW", # Order Status # "i":8886774, # Order Id # "l":"0", # Order Last Filled Quantity # "z":"0", # Order Filled Accumulated Quantity # "L":"0", # Last Filled Price # "N":"USDT", # Commission Asset, will not push if no commission # "n":"0", # Commission, will not push if no commission # "T":1568879465651, # Order Trade Time # "t":0, # Trade Id # "b":"0", # Bids Notional # "a":"9.91", # Ask Notional # "m":false, # Is self trade the maker side? # "R":false, # Is self reduce only # "wt":"CONTRACT_PRICE", # Stop Price Working Type # "ot":"TRAILING_STOP_MARKET", # Original Order Type # "ps":"LONG", # Position Side # "cp":false, # If Close-All, pushed with conditional order # "AP":"7476.89", # Activation Price, only puhed with TRAILING_STOP_MARKET order # "cr":"5.0", # Callback Rate, only puhed with TRAILING_STOP_MARKET order # "rp":"0" # Realized Profit of the trade # } # executionType = self.safe_string(order, 'x') orderId = self.safe_string(order, 'i') marketId = self.safe_string(order, 's') marketType = 'contract' if ('ps' in order) else 'spot' symbol = self.safe_symbol(marketId, None, None, marketType) timestamp = self.safe_integer(order, 'O') T = self.safe_integer(order, 'T') lastTradeTimestamp = None if executionType == 'NEW' or executionType == 'AMENDMENT' or executionType == 'CANCELED': if timestamp is None: timestamp = T elif executionType == 'TRADE': lastTradeTimestamp = T lastUpdateTimestamp = T fee = None feeCost = self.safe_string(order, 'n') if (feeCost is not None) and (Precise.string_gt(feeCost, '0')): feeCurrencyId = self.safe_string(order, 'N') feeCurrency = self.safe_currency_code(feeCurrencyId) fee = { 'cost': feeCost, 'currency': feeCurrency, } price = self.safe_string(order, 'p') amount = self.safe_string(order, 'q') side = self.safe_string_lower(order, 'S') type = self.safe_string_lower(order, 'o') filled = self.safe_string(order, 'z') cost = self.safe_string(order, 'Z') average = self.safe_string(order, 'ap') rawStatus = self.safe_string(order, 'X') status = self.parse_order_status(rawStatus) trades = None clientOrderId = self.safe_string(order, 'C') if (clientOrderId is None) or (len(clientOrderId) == 0): clientOrderId = self.safe_string(order, 'c') stopPrice = self.safe_string_2(order, 'P', 'sp') timeInForce = self.safe_string(order, 'f') if timeInForce == 'GTX': # GTX means "Good Till Crossing" and is an equivalent way of saying Post Only timeInForce = 'PO' return self.safe_order({ 'info': order, 'symbol': symbol, 'id': orderId, 'clientOrderId': clientOrderId, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': lastTradeTimestamp, 'lastUpdateTimestamp': lastUpdateTimestamp, 'type': type, 'timeInForce': timeInForce, 'postOnly': None, 'reduceOnly': self.safe_bool(order, 'R'), 'side': side, 'price': price, 'stopPrice': stopPrice, 'triggerPrice': stopPrice, 'amount': amount, 'cost': cost, 'average': average, 'filled': filled, 'remaining': None, 'status': status, 'fee': fee, 'trades': trades, }) def handle_order_update(self, client: Client, message): # # spot # # { # "e": "executionReport", # Event type # "E": 1499405658658, # Event time # "s": "ETHBTC", # Symbol # "c": "mUvoqJxFIILMdfAW5iGSOW", # Client order ID # "S": "BUY", # Side # "o": "LIMIT", # Order type # "f": "GTC", # Time in force # "q": "1.00000000", # Order quantity # "p": "0.10264410", # Order price # "P": "0.00000000", # Stop price # "F": "0.00000000", # Iceberg quantity # "g": -1, # OrderListId # "C": null, # Original client order ID; This is the ID of the order being canceled # "x": "NEW", # Current execution type # "X": "NEW", # Current order status # "r": "NONE", # Order reject reason; will be an error code. # "i": 4293153, # Order ID # "l": "0.00000000", # Last executed quantity # "z": "0.00000000", # Cumulative filled quantity # "L": "0.00000000", # Last executed price # "n": "0", # Commission amount # "N": null, # Commission asset # "T": 1499405658657, # Transaction time # "t": -1, # Trade ID # "I": 8641984, # Ignore # "w": True, # Is the order on the book? # "m": False, # Is self trade the maker side? # "M": False, # Ignore # "O": 1499405658657, # Order creation time # "Z": "0.00000000", # Cumulative quote asset transacted quantity # "Y": "0.00000000" # Last quote asset transacted quantity(i.e. lastPrice * lastQty), # "Q": "0.00000000" # Quote Order Qty # } # # future # # { # "e":"ORDER_TRADE_UPDATE", # Event Type # "E":1568879465651, # Event Time # "T":1568879465650, # Trasaction Time # "o": { # "s":"BTCUSDT", # Symbol # "c":"TEST", # Client Order Id # # special client order id: # # starts with "autoclose-": liquidation order # # "adl_autoclose": ADL auto close order # "S":"SELL", # Side # "o":"TRAILING_STOP_MARKET", # Order Type # "f":"GTC", # Time in Force # "q":"0.001", # Original Quantity # "p":"0", # Original Price # "ap":"0", # Average Price # "sp":"7103.04", # Stop Price. Please ignore with TRAILING_STOP_MARKET order # "x":"NEW", # Execution Type # "X":"NEW", # Order Status # "i":8886774, # Order Id # "l":"0", # Order Last Filled Quantity # "z":"0", # Order Filled Accumulated Quantity # "L":"0", # Last Filled Price # "N":"USDT", # Commission Asset, will not push if no commission # "n":"0", # Commission, will not push if no commission # "T":1568879465651, # Order Trade Time # "t":0, # Trade Id # "b":"0", # Bids Notional # "a":"9.91", # Ask Notional # "m":false, # Is self trade the maker side? # "R":false, # Is self reduce only # "wt":"CONTRACT_PRICE", # Stop Price Working Type # "ot":"TRAILING_STOP_MARKET", # Original Order Type # "ps":"LONG", # Position Side # "cp":false, # If Close-All, pushed with conditional order # "AP":"7476.89", # Activation Price, only puhed with TRAILING_STOP_MARKET order # "cr":"5.0", # Callback Rate, only puhed with TRAILING_STOP_MARKET order # "rp":"0" # Realized Profit of the trade # } # } # e = self.safe_string(message, 'e') if e == 'ORDER_TRADE_UPDATE': message = self.safe_dict(message, 'o', message) self.handle_my_trade(client, message) self.handle_order(client, message) self.handle_my_liquidation(client, message) async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]: """ watch all open positions :param str[]|None symbols: list of unified market symbols :param number [since]: since timestamp :param number [limit]: limit :param dict params: extra parameters specific to the exchange API endpoint :param boolean [params.portfolioMargin]: set to True if you would like to watch positions in a portfolio margin account :returns dict[]: a list of `position structure ` """ await self.load_markets() market = None messageHash = '' symbols = self.market_symbols(symbols) if not self.is_empty(symbols): market = self.get_market_from_symbols(symbols) messageHash = '::' + ','.join(symbols) type = None type, params = self.handle_market_type_and_params('watchPositions', market, params) if type == 'spot' or type == 'margin': type = 'future' subType = None subType, params = self.handle_sub_type_and_params('watchPositions', market, params) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' marketTypeObject: dict = {} marketTypeObject['type'] = type marketTypeObject['subType'] = subType await self.authenticate(self.extend(marketTypeObject, params)) messageHash = type + ':positions' + messageHash isPortfolioMargin = None isPortfolioMargin, params = self.handle_option_and_params_2(params, 'watchPositions', 'papi', 'portfolioMargin', False) urlType = type if isPortfolioMargin: urlType = 'papi' url = self.urls['api']['ws'][urlType] + '/' + self.options[type]['listenKey'] client = self.client(url) self.set_balance_cache(client, type, isPortfolioMargin) self.set_positions_cache(client, type, symbols, isPortfolioMargin) fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', True) awaitPositionsSnapshot = self.handle_option('watchPositions', 'awaitPositionsSnapshot', True) cache = self.safe_value(self.positions, type) if fetchPositionsSnapshot and awaitPositionsSnapshot and cache is None: snapshot = await client.future(type + ':fetchPositionsSnapshot') return self.filter_by_symbols_since_limit(snapshot, symbols, since, limit, True) newPositions = await self.watch(url, messageHash, None, type) if self.newUpdates: return newPositions return self.filter_by_symbols_since_limit(cache, symbols, since, limit, True) def set_positions_cache(self, client: Client, type, symbols: Strings = None, isPortfolioMargin=False): if type == 'spot': return if self.positions is None: self.positions = {} if type in self.positions: return fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', False) if fetchPositionsSnapshot: messageHash = type + ':fetchPositionsSnapshot' if not (messageHash in client.futures): client.future(messageHash) self.spawn(self.load_positions_snapshot, client, messageHash, type, isPortfolioMargin) else: self.positions[type] = ArrayCacheBySymbolBySide() async def load_positions_snapshot(self, client, messageHash, type, isPortfolioMargin): params: dict = { 'type': type, } if isPortfolioMargin: params['portfolioMargin'] = True positions = await self.fetch_positions(None, params) self.positions[type] = ArrayCacheBySymbolBySide() cache = self.positions[type] for i in range(0, len(positions)): position = positions[i] contracts = self.safe_number(position, 'contracts', 0) if contracts > 0: cache.append(position) # don't remove the future from the .futures cache future = client.futures[messageHash] future.resolve(cache) client.resolve(cache, type + ':position') def handle_positions(self, client, message): # # { # e: 'ACCOUNT_UPDATE', # T: 1667881353112, # E: 1667881353115, # a: { # B: [{ # a: 'USDT', # wb: '1127.95750089', # cw: '1040.82091149', # bc: '0' # }], # P: [{ # s: 'BTCUSDT', # pa: '-0.089', # ep: '19700.03933', # cr: '-1260.24809979', # up: '1.53058860', # mt: 'isolated', # iw: '87.13658940', # ps: 'BOTH', # ma: 'USDT' # }], # m: 'ORDER' # } # } # # each account is connected to a different endpoint # and has exactly one subscriptionhash which is the account type subscriptions = list(client.subscriptions.keys()) accountType = subscriptions[0] if self.positions is None: self.positions = {} if not (accountType in self.positions): self.positions[accountType] = ArrayCacheBySymbolBySide() cache = self.positions[accountType] data = self.safe_dict(message, 'a', {}) rawPositions = self.safe_list(data, 'P', []) newPositions = [] for i in range(0, len(rawPositions)): rawPosition = rawPositions[i] position = self.parse_ws_position(rawPosition) timestamp = self.safe_integer(message, 'E') position['timestamp'] = timestamp position['datetime'] = self.iso8601(timestamp) newPositions.append(position) cache.append(position) messageHashes = self.find_message_hashes(client, accountType + ':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, accountType + ':positions') def parse_ws_position(self, position, market=None): # # { # "s": "BTCUSDT", # Symbol # "pa": "0", # Position Amount # "ep": "0.00000", # Entry Price # "cr": "200", #(Pre-fee) Accumulated Realized # "up": "0", # Unrealized PnL # "mt": "isolated", # Margin Type # "iw": "0.00000000", # Isolated Wallet(if isolated position) # "ps": "BOTH" # Position Side # } # marketId = self.safe_string(position, 's') contracts = self.safe_string(position, 'pa') contractsAbs = Precise.string_abs(self.safe_string(position, 'pa')) positionSide = self.safe_string_lower(position, 'ps') hedged = True if positionSide == 'both': hedged = False if not Precise.string_eq(contracts, '0'): if Precise.string_lt(contracts, '0'): positionSide = 'short' else: positionSide = 'long' return self.safe_position({ 'info': position, 'id': None, 'symbol': self.safe_symbol(marketId, None, None, 'contract'), 'notional': None, 'marginMode': self.safe_string(position, 'mt'), 'liquidationPrice': None, 'entryPrice': self.safe_number(position, 'ep'), 'unrealizedPnl': self.safe_number(position, 'up'), 'percentage': None, 'contracts': self.parse_number(contractsAbs), 'contractSize': None, 'markPrice': None, 'side': positionSide, 'hedged': hedged, 'timestamp': None, 'datetime': None, 'maintenanceMargin': None, 'maintenanceMarginPercentage': None, 'collateral': None, 'initialMargin': None, 'initialMarginPercentage': None, 'leverage': None, 'marginRatio': None, }) async def fetch_my_trades_ws(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ fetch all trades made by the user https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/account-requests#account-trade-history-user_data :param str symbol: unified market symbol :param int|None [since]: the earliest time in ms to fetch trades for :param int|None [limit]: the maximum number of trades structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :param int [params.endTime]: the latest time in ms to fetch trades for :param int [params.fromId]: first trade Id to fetch :returns dict[]: a list of `trade structures ` """ await self.load_markets() if symbol is None: raise BadRequest(self.id + ' fetchMyTradesWs requires a symbol') market = self.market(symbol) type = self.get_market_type('fetchMyTradesWs', market, params) if type != 'spot' and type != 'future': raise BadRequest(self.id + ' fetchMyTradesWs does not support ' + type + ' markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchMyTradesWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, } if since is not None: payload['startTime'] = since if limit is not None: payload['limit'] = limit fromId = self.safe_integer(params, 'fromId') if fromId is not None and since is not None: raise BadRequest(self.id + ' fetchMyTradesWs does not support fetching by both fromId and since parameters at the same time') message: dict = { 'id': messageHash, 'method': 'myTrades', 'params': self.sign_params(self.extend(payload, params)), } subscription: dict = { 'method': self.handle_trades_ws, } trades = await self.watch(url, messageHash, message, messageHash, subscription) return self.filter_by_symbol_since_limit(trades, symbol, since, limit) async def fetch_trades_ws(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ fetch all trades made by the user https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/market-data-requests#recent-trades :param str symbol: unified market symbol :param int [since]: the earliest time in ms to fetch trades for :param int [limit]: the maximum number of trades structures to retrieve, default=500, max=1000 :param dict [params]: extra parameters specific to the exchange API endpoint EXCHANGE SPECIFIC PARAMETERS :param int [params.fromId]: trade ID to begin at :returns dict[]: a list of `trade structures ` """ await self.load_markets() if symbol is None: raise BadRequest(self.id + ' fetchTradesWs() requires a symbol argument') market = self.market(symbol) type = self.get_market_type('fetchTradesWs', market, params) if type != 'spot' and type != 'future': raise BadRequest(self.id + ' fetchTradesWs does not support ' + type + ' markets') url = self.urls['api']['ws']['ws-api'][type] requestId = self.request_id(url) messageHash = str(requestId) returnRateLimits = False returnRateLimits, params = self.handle_option_and_params(params, 'fetchTradesWs', 'returnRateLimits', False) payload: dict = { 'symbol': self.market_id(symbol), 'returnRateLimits': returnRateLimits, } if limit is not None: payload['limit'] = limit message: dict = { 'id': messageHash, 'method': 'trades.historical', 'params': self.extend(payload, params), } subscription: dict = { 'method': self.handle_trades_ws, } trades = await self.watch(url, messageHash, message, messageHash, subscription) return self.filter_by_since_limit(trades, since, limit) def handle_trades_ws(self, client: Client, message): # # fetchMyTradesWs # # { # "id": "f4ce6a53-a29d-4f70-823b-4ab59391d6e8", # "status": 200, # "result": [ # { # "symbol": "BTCUSDT", # "id": 1650422481, # "orderId": 12569099453, # "orderListId": -1, # "price": "23416.10000000", # "qty": "0.00635000", # "quoteQty": "148.69223500", # "commission": "0.00000000", # "commissionAsset": "BNB", # "time": 1660801715793, # "isBuyer": False, # "isMaker": True, # "isBestMatch": True # }, # ... # ], # } # # fetchTradesWs # # { # "id": "f4ce6a53-a29d-4f70-823b-4ab59391d6e8", # "status": 200, # "result": [ # { # "id": 0, # "price": "0.00005000", # "qty": "40.00000000", # "quoteQty": "0.00200000", # "time": 1500004800376, # "isBuyerMaker": True, # "isBestMatch": True # } # ... # ], # } # messageHash = self.safe_string(message, 'id') result = self.safe_list(message, 'result', []) trades = self.parse_trades(result) client.resolve(trades, messageHash) 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 :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.portfolioMargin]: set to True if you would like to watch trades in a portfolio margin account :returns dict[]: a list of `trade structures ` """ await self.load_markets() type = None market = None if symbol is not None: market = self.market(symbol) symbol = market['symbol'] type, params = self.handle_market_type_and_params('watchMyTrades', market, params) subType = None subType, params = self.handle_sub_type_and_params('watchMyTrades', market, params) if self.isLinear(type, subType): type = 'future' elif self.isInverse(type, subType): type = 'delivery' messageHash = 'myTrades' if symbol is not None: symbol = self.symbol(symbol) messageHash += ':' + symbol params = self.extend(params, {'type': market['type'], 'symbol': symbol}) await self.authenticate(params) urlType = type # we don't change type because the listening key is different if type == 'margin': urlType = 'spot' # spot-margin shares the same stream spot isPortfolioMargin = None isPortfolioMargin, params = self.handle_option_and_params_2(params, 'watchMyTrades', 'papi', 'portfolioMargin', False) if isPortfolioMargin: urlType = 'papi' url = self.urls['api']['ws'][urlType] + '/' + self.options[type]['listenKey'] client = self.client(url) self.set_balance_cache(client, type, isPortfolioMargin) self.set_positions_cache(client, type, None, isPortfolioMargin) message = None trades = await self.watch(url, messageHash, message, type) if self.newUpdates: limit = trades.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(trades, symbol, since, limit, True) def handle_my_trade(self, client: Client, message): messageHash = 'myTrades' executionType = self.safe_string(message, 'x') if executionType == 'TRADE': trade = self.parse_ws_trade(message) orderId = self.safe_string(trade, 'order') tradeFee = self.safe_dict(trade, 'fee', {}) tradeFee = self.extend({}, tradeFee) symbol = self.safe_string(trade, 'symbol') if orderId is not None and tradeFee is not None and symbol is not None: cachedOrders = self.orders if cachedOrders is not None: orders = self.safe_value(cachedOrders.hashmap, symbol, {}) order = self.safe_value(orders, orderId) if order is not None: # accumulate order fees fees = self.safe_value(order, 'fees') fee = self.safe_value(order, 'fee') if not self.is_empty(fees): insertNewFeeCurrency = True for i in range(0, len(fees)): orderFee = fees[i] if orderFee['currency'] == tradeFee['currency']: feeCost = self.sum(tradeFee['cost'], orderFee['cost']) order['fees'][i]['cost'] = float(self.currency_to_precision(tradeFee['currency'], feeCost)) insertNewFeeCurrency = False break if insertNewFeeCurrency: order['fees'].append(tradeFee) elif fee is not None: if fee['currency'] == tradeFee['currency']: feeCost = self.sum(fee['cost'], tradeFee['cost']) order['fee']['cost'] = float(self.currency_to_precision(tradeFee['currency'], feeCost)) elif fee['currency'] is None: order['fee'] = tradeFee else: order['fees'] = [fee, tradeFee] order['fee'] = None else: order['fee'] = tradeFee # save self trade in the order orderTrades = self.safe_list(order, 'trades', []) orderTrades.append(trade) order['trades'] = orderTrades # don't append twice cause it breaks newUpdates mode # self order already exists in the cache if self.myTrades is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) self.myTrades = ArrayCacheBySymbolById(limit) myTrades = self.myTrades myTrades.append(trade) client.resolve(self.myTrades, messageHash) messageHashSymbol = messageHash + ':' + symbol client.resolve(self.myTrades, messageHashSymbol) def handle_order(self, client: Client, message): parsed = self.parse_ws_order(message) symbol = self.safe_string(parsed, 'symbol') orderId = self.safe_string(parsed, 'id') if symbol is not None: if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) self.orders = ArrayCacheBySymbolById(limit) cachedOrders = self.orders orders = self.safe_value(cachedOrders.hashmap, symbol, {}) order = self.safe_value(orders, orderId) if order is not None: fee = self.safe_value(order, 'fee') if fee is not None: parsed['fee'] = fee fees = self.safe_value(order, 'fees') if fees is not None: parsed['fees'] = fees parsed['trades'] = self.safe_value(order, 'trades') timestamp = self.safe_integer(parsed, 'timestamp') if timestamp is None: parsed['timestamp'] = self.safe_integer(order, 'timestamp') parsed['datetime'] = self.safe_string(order, 'datetime') cachedOrders.append(parsed) messageHash = 'orders' symbolSpecificMessageHash = 'orders:' + symbol client.resolve(cachedOrders, messageHash) client.resolve(cachedOrders, symbolSpecificMessageHash) def handle_acount_update(self, client, message): self.handle_balance(client, message) self.handle_positions(client, message) def handle_ws_error(self, client: Client, message): # # { # "error": { # "code": 2, # "msg": "Invalid request: invalid stream" # }, # "id": 1 # } # id = self.safe_string(message, 'id') rejected = False error = self.safe_dict(message, 'error', {}) code = self.safe_integer(error, 'code') msg = self.safe_string(error, 'msg') try: self.handle_errors(code, msg, client.url, None, None, self.json(error), error, None, None) except Exception as e: rejected = True # private endpoint uses id client.reject(e, id) # public endpoint stores messageHash in subscriptions subscriptionKeys = list(client.subscriptions.keys()) for i in range(0, len(subscriptionKeys)): subscriptionHash = subscriptionKeys[i] subscriptionId = self.safe_string(client.subscriptions[subscriptionHash], 'id') if id == subscriptionId: client.reject(e, subscriptionHash) if not rejected: client.reject(message, id) # reset connection if 5xx error codeString = self.safe_string(error, 'code') if (codeString is not None) and (codeString[0] == '5'): client.reset(message) def handle_message(self, client: Client, message): # handle WebSocketAPI status = self.safe_string(message, 'status') error = self.safe_value(message, 'error') if (error is not None) or (status is not None and status != '200'): self.handle_ws_error(client, message) return id = self.safe_string(message, 'id') subscriptions = self.safe_value(client.subscriptions, id) method = self.safe_value(subscriptions, 'method') if method is not None: method(client, message) return # handle other APIs methods: dict = { 'depthUpdate': self.handle_order_book, 'trade': self.handle_trade, 'aggTrade': self.handle_trade, 'kline': self.handle_ohlcv, 'markPrice_kline': self.handle_ohlcv, 'indexPrice_kline': self.handle_ohlcv, '1hTicker@arr': self.handle_tickers, '4hTicker@arr': self.handle_tickers, '1dTicker@arr': self.handle_tickers, '24hrTicker@arr': self.handle_tickers, '24hrMiniTicker@arr': self.handle_tickers, '1hTicker': self.handle_tickers, '4hTicker': self.handle_tickers, '1dTicker': self.handle_tickers, '24hrTicker': self.handle_tickers, '24hrMiniTicker': self.handle_tickers, 'markPriceUpdate': self.handle_tickers, 'markPriceUpdate@arr': self.handle_tickers, 'bookTicker': self.handle_bids_asks, # there is no "bookTicker@arr" endpoint 'outboundAccountPosition': self.handle_balance, 'balanceUpdate': self.handle_balance, 'ACCOUNT_UPDATE': self.handle_acount_update, 'executionReport': self.handle_order_update, 'ORDER_TRADE_UPDATE': self.handle_order_update, 'forceOrder': self.handle_liquidation, } event = self.safe_string(message, 'e') if isinstance(message, list): data = message[0] event = self.safe_string(data, 'e') + '@arr' method = self.safe_value(methods, event) if method is None: requestId = self.safe_string(message, 'id') if requestId is not None: self.handle_subscription_status(client, message) return # special case for the real-time bookTicker, since it comes without an event identifier # # { # "u": 7488717758, # "s": "BTCUSDT", # "b": "28621.74000000", # "B": "1.43278800", # "a": "28621.75000000", # "A": "2.52500800" # } # if event is None and ('a' in message) and ('b' in message): self.handle_bids_asks(client, message) else: method(client, message)