# -*- 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 from ccxt.async_support.base.ws.order_book_side import Asks, Bids import hashlib from ccxt.base.types import Any, Balances, Int, Market, Order, OrderBook, Position, Str, Strings, Ticker, Tickers, Trade from ccxt.async_support.base.ws.client import Client from typing import List from ccxt.base.errors import ExchangeError from ccxt.base.errors import AuthenticationError from ccxt.base.errors import NotSupported class bitmart(ccxt.async_support.bitmart): def describe(self) -> Any: return self.deep_extend(super(bitmart, self).describe(), { 'has': { 'createOrderWs': False, 'editOrderWs': False, 'fetchOpenOrdersWs': False, 'fetchOrderWs': False, 'cancelOrderWs': False, 'cancelOrdersWs': False, 'cancelAllOrdersWs': False, 'ws': True, 'watchBalance': True, 'watchTicker': True, 'watchTickers': True, 'watchBidsAsks': True, 'watchOrderBook': True, 'watchOrderBookForSymbols': True, 'watchOrders': True, 'watchTrades': True, 'watchTradesForSymbols': True, 'watchOHLCV': True, 'watchPosition': 'emulated', 'watchPositions': True, }, 'urls': { 'api': { 'ws': { 'spot': { 'public': 'wss://ws-manager-compress.{hostname}/api?protocol=1.1', 'private': 'wss://ws-manager-compress.{hostname}/user?protocol=1.1', }, 'swap': { 'public': 'wss://openapi-ws-v2.{hostname}/api?protocol=1.1', 'private': 'wss://openapi-ws-v2.{hostname}/user?protocol=1.1', }, }, }, }, 'options': { 'defaultType': 'spot', 'watchBalance': { 'fetchBalanceSnapshot': True, # or False 'awaitBalanceSnapshot': False, # whether to wait for the balance snapshot before providing updates }, # # orderbook channels can have: # - 'depth5', 'depth20', 'depth50' # these endpoints emit full Orderbooks once in every 500ms # - 'depth/increase100' # self endpoint is preferred, because it emits once in 100ms. however, when self value is chosen, it only affects spot-market, but contracts markets automatically `depth50` will be being used 'watchOrderBook': { 'depth': 'depth/increase100', }, 'watchOrderBookForSymbols': { 'depth': 'depth/increase100', }, 'watchTrades': { 'ignoreDuplicates': True, }, 'ws': { 'inflate': True, }, 'timeframes': { '1m': '1m', '3m': '3m', '5m': '5m', '15m': '15m', '30m': '30m', '45m': '45m', '1h': '1H', '2h': '2H', '3h': '3H', '4h': '4H', '1d': '1D', '1w': '1W', '1M': '1M', }, }, 'streaming': { 'keepAlive': 15000, }, }) async def subscribe(self, channel, symbol, type, params={}): market = self.market(symbol) url = self.implode_hostname(self.urls['api']['ws'][type]['public']) request = {} messageHash = None if type == 'spot': messageHash = 'spot/' + channel + ':' + market['id'] request = { 'op': 'subscribe', 'args': [messageHash], } else: messageHash = 'futures/' + channel + ':' + market['id'] speed = self.safe_string(params, 'speed') if speed is not None: params = self.omit(params, 'speed') messageHash += ':' + speed request = { 'action': 'subscribe', 'args': [messageHash], } return await self.watch(url, messageHash, self.deep_extend(request, params), messageHash) async def subscribe_multiple(self, channel: str, type: str, symbols: Strings = None, params={}): symbols = self.market_symbols(symbols, type, False, True) url = self.implode_hostname(self.urls['api']['ws'][type]['public']) channelType = 'spot' if (type == 'spot') else 'futures' actionType = 'op' if (type == 'spot') else 'action' rawSubscriptions = [] messageHashes = [] for i in range(0, len(symbols)): market = self.market(symbols[i]) message = channelType + '/' + channel + ':' + market['id'] rawSubscriptions.append(message) messageHashes.append(channel + ':' + market['symbol']) # exclusion, futures "tickers" need one generic request for all symbols # if (type != 'spot') and (channel == 'ticker'): # rawSubscriptions = [channelType + '/' + channel] # } # Exchange update from 2025-02-11 supports subscription by trading pair for swap request: dict = { 'args': rawSubscriptions, } request[actionType] = 'subscribe' return await self.watch_multiple(url, messageHashes, self.deep_extend(request, params), rawSubscriptions) async def watch_balance(self, params={}) -> Balances: """ https://developer-pro.bitmart.com/en/spot/#private-balance-change https://developer-pro.bitmart.com/en/futuresv2/#private-assets-channel 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 :returns dict: a `balance structure ` """ await self.load_markets() type = 'spot' type, params = self.handle_market_type_and_params('watchBalance', None, params) await self.authenticate(type, params) request = {} if type == 'spot': request = { 'op': 'subscribe', 'args': ['spot/user/balance:BALANCE_UPDATE'], } else: request = { 'action': 'subscribe', 'args': ['futures/asset:USDT', 'futures/asset:BTC', 'futures/asset:ETH'], } messageHash = 'balance:' + type url = self.implode_hostname(self.urls['api']['ws'][type]['private']) client = self.client(url) self.set_balance_cache(client, type, messageHash) fetchBalanceSnapshot = None awaitBalanceSnapshot = None fetchBalanceSnapshot, params = self.handle_option_and_params(self.options, 'watchBalance', 'fetchBalanceSnapshot', True) awaitBalanceSnapshot, params = self.handle_option_and_params(self.options, 'watchBalance', 'awaitBalanceSnapshot', False) if fetchBalanceSnapshot and awaitBalanceSnapshot: await client.future(type + ':fetchBalanceSnapshot') return await self.watch(url, messageHash, self.deep_extend(request, params), messageHash) def set_balance_cache(self, client: Client, type, subscribeHash): if subscribeHash in client.subscriptions: return options = self.safe_value(self.options, 'watchBalance') snapshot = self.safe_bool(options, 'fetchBalanceSnapshot', True) if snapshot: messageHash = type + ':' + 'fetchBalanceSnapshot' if not (messageHash in client.futures): client.future(messageHash) self.spawn(self.load_balance_snapshot, client, messageHash, type) self.balance[type] = {} # without self comment, transpilation breaks for some reason... async def load_balance_snapshot(self, client, messageHash, type): response = await self.fetch_balance({'type': type}) 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], 'balance:' + type) def handle_balance(self, client: Client, message): # # spot # { # "data":[ # { # "balance_details":[ # { # "av_bal":"0.206000000000000000000000000000", # "ccy":"LTC", # "fz_bal":"0.100000000000000000000000000000" # } # ], # "event_time":"1701632345415", # "event_type":"TRANSACTION_COMPLETED" # } # ], # "table":"spot/user/balance" # } # swap # { # group: 'futures/asset:USDT', # data: { # currency: 'USDT', # available_balance: '37.19688649135', # position_deposit: '0.788687546', # frozen_balance: '0' # } # } # channel = self.safe_string_2(message, 'table', 'group') data = self.safe_value(message, 'data') if data is None: return isSpot = (channel.find('spot') >= 0) type = 'spot' if isSpot else 'swap' self.balance[type]['info'] = message if isSpot: if not isinstance(data, list): return for i in range(0, len(data)): timestamp = self.safe_integer(message, 'event_time') self.balance[type]['timestamp'] = timestamp self.balance[type]['datetime'] = self.iso8601(timestamp) balanceDetails = self.safe_value(data[i], 'balance_details', []) for ii in range(0, len(balanceDetails)): rawBalance = balanceDetails[i] account = self.account() currencyId = self.safe_string(rawBalance, 'ccy') code = self.safe_currency_code(currencyId) account['free'] = self.safe_string(rawBalance, 'av_bal') account['used'] = self.safe_string(rawBalance, 'fz_bal') self.balance[type][code] = account else: currencyId = self.safe_string(data, 'currency') code = self.safe_currency_code(currencyId) account = self.account() account['free'] = self.safe_string(data, 'available_balance') account['used'] = self.safe_string(data, 'frozen_balance') self.balance[type][code] = account self.balance[type] = self.safe_balance(self.balance[type]) messageHash = 'balance:' + type client.resolve(self.balance[type], messageHash) async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ https://developer-pro.bitmart.com/en/spot/#public-trade-channel https://developer-pro.bitmart.com/en/futuresv2/#public-trade-channel get the list of most recent trades for a particular symbol :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 :returns dict[]: a list of `trade structures ` """ return await self.watch_trades_for_symbols([symbol], since, limit, params) async def watch_trades_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ https://developer-pro.bitmart.com/en/spot/#public-trade-channel get the list of most recent trades for a list of symbols :param str[] symbols: unified symbol of the market to fetch trades for :param int [since]: timestamp in ms of the earliest trade to fetch :param int [limit]: the maximum amount of trades to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `trade structures ` """ await self.load_markets() marketType = None symbols, marketType, params = self.get_params_for_multiple_sub('watchTradesForSymbols', symbols, limit, params) channelName = 'trade' trades = await self.subscribe_multiple(channelName, marketType, symbols, params) if self.newUpdates: first = self.safe_dict(trades, 0) tradeSymbol = self.safe_string(first, 'symbol') limit = trades.getLimit(tradeSymbol, limit) result = self.filter_by_since_limit(trades, since, limit, 'timestamp', True) if self.handle_option('watchTrades', 'ignoreDuplicates', True): filtered = self.remove_repeated_trades_from_array(result) filtered = self.sort_by(filtered, 'timestamp') return filtered return result def get_params_for_multiple_sub(self, methodName: str, symbols: List[str], limit: Int = None, params={}): symbols = self.market_symbols(symbols, None, False, True) length = len(symbols) if length > 20: raise NotSupported(self.id + ' ' + methodName + '() accepts a maximum of 20 symbols in one request') market = self.market(symbols[0]) marketType = None marketType, params = self.handle_market_type_and_params(methodName, market, params) return [symbols, marketType, params] async def watch_ticker(self, symbol: str, params={}) -> Ticker: """ https://developer-pro.bitmart.com/en/spot/#public-ticker-channel https://developer-pro.bitmart.com/en/futuresv2/#public-ticker-channel watches 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 :returns dict: a `ticker structure ` """ await self.load_markets() symbol = self.symbol(symbol) tickers = await self.watch_tickers([symbol], params) return tickers[symbol] async def watch_tickers(self, symbols: Strings = None, params={}) -> Tickers: """ https://developer-pro.bitmart.com/en/spot/#public-ticker-channel https://developer-pro.bitmart.com/en/futuresv2/#public-ticker-channel watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for all markets of a specific list :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() market = self.get_market_from_symbols(symbols) marketType = None marketType, params = self.handle_market_type_and_params('watchTickers', market, params) ticker = await self.subscribe_multiple('ticker', marketType, symbols, params) if self.newUpdates: tickers: dict = {} tickers[ticker['symbol']] = ticker return tickers return self.filter_by_array(self.tickers, 'symbol', symbols) async def watch_bids_asks(self, symbols: Strings = None, params={}) -> Tickers: """ https://developer-pro.bitmart.com/en/spot/#public-ticker-channel https://developer-pro.bitmart.com/en/futuresv2/#public-ticker-channel watches best bid & ask for symbols :param str[] symbols: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False) firstMarket = self.get_market_from_symbols(symbols) marketType = None marketType, params = self.handle_market_type_and_params('watchBidsAsks', firstMarket, params) url = self.implode_hostname(self.urls['api']['ws'][marketType]['public']) channelType = 'spot' if (marketType == 'spot') else 'futures' actionType = 'op' if (marketType == 'spot') else 'action' rawSubscriptions = [] messageHashes = [] for i in range(0, len(symbols)): market = self.market(symbols[i]) rawSubscriptions.append(channelType + '/ticker:' + market['id']) messageHashes.append('bidask:' + symbols[i]) if marketType != 'spot': rawSubscriptions = [channelType + '/ticker'] request: dict = { 'args': rawSubscriptions, } request[actionType] = 'subscribe' newTickers = await self.watch_multiple(url, messageHashes, request, rawSubscriptions) if self.newUpdates: tickers: dict = {} tickers[newTickers['symbol']] = newTickers return tickers return self.filter_by_array(self.bidsasks, 'symbol', symbols) def handle_bid_ask(self, client: Client, message): table = self.safe_string(message, 'table') isSpot = (table is not None) rawTickers = [] if isSpot: rawTickers = self.safe_list(message, 'data', []) else: rawTickers = [self.safe_value(message, 'data', {})] if not len(rawTickers): return for i in range(0, len(rawTickers)): ticker = self.parse_ws_bid_ask(rawTickers[i]) symbol = ticker['symbol'] self.bidsasks[symbol] = ticker messageHash = 'bidask:' + symbol client.resolve(ticker, messageHash) def parse_ws_bid_ask(self, ticker, market=None): marketId = self.safe_string(ticker, 'symbol') market = self.safe_market(marketId, market) symbol = self.safe_string(market, 'symbol') timestamp = self.safe_integer(ticker, 'ms_t') return self.safe_ticker({ 'symbol': symbol, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'ask': self.safe_string_2(ticker, 'ask_px', 'ask_price'), 'askVolume': self.safe_string_2(ticker, 'ask_sz', 'ask_vol'), 'bid': self.safe_string_2(ticker, 'bid_px', 'bid_price'), 'bidVolume': self.safe_string_2(ticker, 'bid_sz', 'bid_vol'), 'info': ticker, }, market) 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://developer-pro.bitmart.com/en/spot/#private-order-progress https://developer-pro.bitmart.com/en/futuresv2/#private-order-channel :param str symbol: unified market symbol of the market orders were made in :param int [since]: the earliest time in ms to fetch orders for :param int [limit]: the maximum number of order structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `order structures ` """ await self.load_markets() market = None messageHash = 'orders' if symbol is not None: symbol = self.symbol(symbol) market = self.market(symbol) messageHash = 'orders::' + symbol type = 'spot' type, params = self.handle_market_type_and_params('watchOrders', market, params) await self.authenticate(type, params) request = None if type == 'spot': argsRequest = 'spot/user/order:' if symbol is not None: argsRequest += market['id'] else: argsRequest = 'spot/user/orders:ALL_SYMBOLS' request = { 'op': 'subscribe', 'args': [argsRequest], } else: request = { 'action': 'subscribe', 'args': ['futures/order'], } url = self.implode_hostname(self.urls['api']['ws'][type]['private']) newOrders = await self.watch(url, messageHash, self.deep_extend(request, params), messageHash) if self.newUpdates: return newOrders return self.filter_by_symbol_since_limit(self.orders, symbol, since, limit, True) def handle_orders(self, client: Client, message): # # spot # { # "data":[ # { # "symbol": "LTC_USDT", # "notional": '', # "side": "buy", # "last_fill_time": "0", # "ms_t": "1646216634000", # "type": "limit", # "filled_notional": "0.000000000000000000000000000000", # "last_fill_price": "0", # "size": "0.500000000000000000000000000000", # "price": "50.000000000000000000000000000000", # "last_fill_count": "0", # "filled_size": "0.000000000000000000000000000000", # "margin_trading": "0", # "state": "8", # "order_id": "24807076628", # "order_type": "0" # } # ], # "table":"spot/user/order" # } # swap # { # "group":"futures/order", # "data":[ # { # "action":2, # "order":{ # "order_id":"2312045036986775", # "client_order_id":"", # "price":"71.61707928", # "size":"1", # "symbol":"LTCUSDT", # "state":1, # "side":4, # "type":"market", # "leverage":"1", # "open_type":"cross", # "deal_avg_price":"0", # "deal_size":"0", # "create_time":1701625324646, # "update_time":1701625324640, # "plan_order_id":"", # "last_trade":null # } # } # ] # } # orders = self.safe_value(message, 'data') if orders is None: return ordersLength = len(orders) newOrders = [] symbols: dict = {} if ordersLength > 0: limit = self.safe_integer(self.options, 'ordersLimit', 1000) if self.orders is None: self.orders = ArrayCacheBySymbolById(limit) stored = self.orders for i in range(0, len(orders)): order = self.parse_ws_order(orders[i]) stored.append(order) newOrders.append(order) symbol = order['symbol'] symbols[symbol] = True messageHash = 'orders' symbolKeys = list(symbols.keys()) for i in range(0, len(symbolKeys)): symbol = symbolKeys[i] symbolSpecificMessageHash = messageHash + '::' + symbol client.resolve(newOrders, symbolSpecificMessageHash) client.resolve(newOrders, messageHash) def parse_ws_order(self, order: dict, market: Market = None): # # spot # { # "symbol": "LTC_USDT", # "notional": '', # "side": "buy", # "last_fill_time": "0", # "ms_t": "1646216634000", # "type": "limit", # "filled_notional": "0.000000000000000000000000000000", # "last_fill_price": "0", # "size": "0.500000000000000000000000000000", # "price": "50.000000000000000000000000000000", # "last_fill_count": "0", # "filled_size": "0.000000000000000000000000000000", # "margin_trading": "0", # "state": "8", # "order_id": "24807076628", # "order_type": "0" # } # swap # { # "action":2, # "order":{ # "order_id":"2312045036986775", # "client_order_id":"", # "price":"71.61707928", # "size":"1", # "symbol":"LTCUSDT", # "state":1, # "side":4, # "type":"market", # "leverage":"1", # "open_type":"cross", # "deal_avg_price":"0", # "deal_size":"0", # "create_time":1701625324646, # "update_time":1701625324640, # "plan_order_id":"", # "last_trade":null # } # } # action = self.safe_number(order, 'action') isSpot = (action is None) if isSpot: marketId = self.safe_string(order, 'symbol') market = self.safe_market(marketId, market, '_', 'spot') id = self.safe_string(order, 'order_id') clientOrderId = self.safe_string(order, 'clientOid') price = self.safe_string(order, 'price') filled = self.safe_string(order, 'filled_size') amount = self.safe_string(order, 'size') type = self.safe_string(order, 'type') rawState = self.safe_string(order, 'state') status = self.parse_order_status_by_type(market['type'], rawState) timestamp = self.safe_integer(order, 'ms_t') symbol = market['symbol'] side = self.safe_string_lower(order, 'side') return self.safe_order({ 'info': order, 'symbol': symbol, 'id': id, 'clientOrderId': clientOrderId, 'timestamp': None, 'datetime': None, 'lastTradeTimestamp': timestamp, 'type': type, 'timeInForce': None, 'postOnly': None, 'side': side, 'price': price, 'stopPrice': None, 'triggerPrice': None, 'amount': amount, 'cost': None, 'average': None, 'filled': filled, 'remaining': None, 'status': status, 'fee': None, 'trades': None, }, market) else: orderInfo = self.safe_value(order, 'order') marketId = self.safe_string(orderInfo, 'symbol') symbol = self.safe_symbol(marketId, market, '', 'swap') orderId = self.safe_string(orderInfo, 'order_id') timestamp = self.safe_integer(orderInfo, 'create_time') updatedTimestamp = self.safe_integer(orderInfo, 'update_time') lastTrade = self.safe_value(orderInfo, 'last_trade') cachedOrders = self.orders orders = self.safe_value(cachedOrders.hashmap, symbol, {}) cachedOrder = self.safe_value(orders, orderId) trades = None if cachedOrder is not None: trades = self.safe_value(order, 'trades') if lastTrade is not None: if trades is None: trades = [] trades.append(lastTrade) return self.safe_order({ 'info': order, 'symbol': symbol, 'id': orderId, 'clientOrderId': self.safe_string(orderInfo, 'client_order_id'), 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': updatedTimestamp, 'type': self.safe_string(orderInfo, 'type'), 'timeInForce': None, 'postOnly': None, 'side': self.parse_ws_order_side(self.safe_string(orderInfo, 'side')), 'price': self.safe_string(orderInfo, 'price'), 'stopPrice': None, 'triggerPrice': None, 'amount': self.safe_string(orderInfo, 'size'), 'cost': None, 'average': self.safe_string(orderInfo, 'deal_avg_price'), 'filled': self.safe_string(orderInfo, 'deal_size'), 'remaining': None, 'status': self.parse_ws_order_status(self.safe_string(order, 'action')), 'fee': None, 'trades': trades, }, market) def parse_ws_order_status(self, statusId): statuses: dict = { '1': 'closed', # match deal '2': 'open', # submit order '3': 'canceled', # cancel order '4': 'closed', # liquidate cancel order '5': 'canceled', # adl cancel order '6': 'open', # part liquidate '7': 'open', # bankrupty order '8': 'closed', # passive adl match deal '9': 'closed', # active adl match deal } return self.safe_string(statuses, statusId, statusId) def parse_ws_order_side(self, sideId): sides: dict = { '1': 'buy', # buy_open_long '2': 'buy', # buy_close_short '3': 'sell', # sell_close_long '4': 'sell', # sell_open_short } return self.safe_string(sides, sideId, sideId) async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]: """ https://developer-pro.bitmart.com/en/futures/#private-position-channel watch all open positions :param str[]|None symbols: list of unified market symbols :param int [since]: the earliest time in ms to fetch positions :param int [limit]: the maximum number of positions to retrieve :param dict params: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `position structure ` """ await self.load_markets() type = 'swap' await self.authenticate(type, params) symbols = self.market_symbols(symbols, 'swap', True, True, False) messageHash = 'positions' if symbols is not None: messageHash += '::' + ','.join(symbols) subscriptionHash = 'futures/position' request: dict = { 'action': 'subscribe', 'args': ['futures/position'], } url = self.implode_hostname(self.urls['api']['ws'][type]['private']) newPositions = await self.watch(url, messageHash, self.deep_extend(request, params), subscriptionHash) if self.newUpdates: return newPositions return self.filter_by_symbols_since_limit(self.positions, symbols, since, limit) def handle_positions(self, client: Client, message): # # { # "group":"futures/position", # "data":[ # { # "symbol":"LTCUSDT", # "hold_volume":"5", # "position_type":2, # "open_type":2, # "frozen_volume":"0", # "close_volume":"0", # "hold_avg_price":"71.582", # "close_avg_price":"0", # "open_avg_price":"71.582", # "liquidate_price":"0", # "create_time":1701623327513, # "update_time":1701627620439 # }, # { # "symbol":"LTCUSDT", # "hold_volume":"6", # "position_type":1, # "open_type":2, # "frozen_volume":"0", # "close_volume":"0", # "hold_avg_price":"71.681666666666666667", # "close_avg_price":"0", # "open_avg_price":"71.681666666666666667", # "liquidate_price":"0", # "create_time":1701621167225, # "update_time":1701628152614 # } # ] # } # data = self.safe_value(message, 'data', []) if self.positions is None: self.positions = ArrayCacheBySymbolBySide() cache = self.positions newPositions = [] for i in range(0, len(data)): rawPosition = data[i] position = self.parse_ws_position(rawPosition) newPositions.append(position) cache.append(position) messageHashes = self.find_message_hashes(client, 'positions::') for i in range(0, len(messageHashes)): messageHash = messageHashes[i] parts = messageHash.split('::') symbolsString = parts[1] symbols = symbolsString.split(',') positions = self.filter_by_array(newPositions, 'symbol', symbols, False) if not self.is_empty(positions): client.resolve(positions, messageHash) client.resolve(newPositions, 'positions') def parse_ws_position(self, position, market: Market = None): # # { # "symbol":"LTCUSDT", # "hold_volume":"6", # "position_type":1, # "open_type":2, # "frozen_volume":"0", # "close_volume":"0", # "hold_avg_price":"71.681666666666666667", # "close_avg_price":"0", # "open_avg_price":"71.681666666666666667", # "liquidate_price":"0", # "create_time":1701621167225, # "update_time":1701628152614 # } # marketId = self.safe_string(position, 'symbol') market = self.safe_market(marketId, market, None, 'swap') symbol = market['symbol'] openTimestamp = self.safe_integer(position, 'create_time') timestamp = self.safe_integer(position, 'update_time') side = self.safe_integer(position, 'position_type') marginModeId = self.safe_integer(position, 'open_type') return self.safe_position({ 'info': position, 'id': None, 'symbol': symbol, 'timestamp': openTimestamp, 'datetime': self.iso8601(openTimestamp), 'lastUpdateTimestamp': timestamp, 'hedged': None, 'side': 'long' if (side == 1) else 'short', 'contracts': self.safe_number(position, 'hold_volume'), 'contractSize': self.safe_number(market, 'contractSize'), 'entryPrice': self.safe_number(position, 'open_avg_price'), 'markPrice': self.safe_number(position, 'hold_avg_price'), 'lastPrice': None, 'notional': None, 'leverage': None, 'collateral': None, 'initialMargin': None, 'initialMarginPercentage': None, 'maintenanceMargin': None, 'maintenanceMarginPercentage': None, 'unrealizedPnl': None, 'realizedPnl': None, 'liquidationPrice': self.safe_number(position, 'liquidate_price'), 'marginMode': 'isolated' if (marginModeId == 1) else 'cross', 'percentage': None, 'marginRatio': None, 'stopLossPrice': None, 'takeProfitPrice': None, }) def handle_trade(self, client: Client, message): # # spot # { # "table": "spot/trade", # "data": [ # { # "price": "52700.50", # "s_t": 1630982050, # "side": "buy", # "size": "0.00112", # "symbol": "BTC_USDT" # }, # ] # } # # swap # { # "group":"futures/trade:BTCUSDT", # "data":[ # { # "trade_id":6798697637, # "symbol":"BTCUSDT", # "deal_price":"39735.8", # "deal_vol":"2", # "way":1, # "created_at":"2023-12-03T15:48:23.517518538Z", # "m": True, # } # ] # } # data = self.safe_value(message, 'data') if data is None: return symbol = None length = len(data) isSwap = ('group' in message) if isSwap: # in swap, chronologically decreasing: 1709536849322, 1709536848954, for i in range(0, length): index = length - i - 1 symbol = self.handle_trade_loop(data[index]) else: # in spot, chronologically increasing: 1709536771200, 1709536771226, for i in range(0, length): symbol = self.handle_trade_loop(data[i]) client.resolve(self.trades[symbol], 'trade:' + symbol) def handle_trade_loop(self, entry): trade = self.parse_ws_trade(entry) symbol = trade['symbol'] tradesLimit = self.safe_integer(self.options, 'tradesLimit', 1000) if self.safe_value(self.trades, symbol) is None: self.trades[symbol] = ArrayCache(tradesLimit) stored = self.trades[symbol] stored.append(trade) return symbol def parse_ws_trade(self, trade: dict, market: Market = None): # # spot # { # "ms_t": 1740320841473, # "price": "2806.54", # "s_t": 1740320841, # "side": "sell", # "size": "0.77598", # "symbol": "ETH_USDT" # } # # swap # { # "trade_id": "3000000245258661", # "symbol": "ETHUSDT", # "deal_price": "2811.1", # "deal_vol": "1858", # "way": 2, # "m": True, # "created_at": "2025-02-23T13:59:59.646490751Z" # } # marketId = self.safe_string(trade, 'symbol') market = self.safe_market(marketId, market) timestamp = self.safe_integer(trade, 'ms_t') datetime: Str = None if timestamp is None: datetime = self.safe_string(trade, 'created_at') timestamp = self.parse8601(datetime) else: datetime = self.iso8601(timestamp) takerOrMaker = None # True for public trades side = self.safe_string(trade, 'side') buyerMaker = self.safe_bool(trade, 'm') if buyerMaker is not None: if side is None: if buyerMaker: side = 'sell' else: side = 'buy' takerOrMaker = 'taker' return self.safe_trade({ 'info': trade, 'id': self.safe_string(trade, 'trade_id'), 'order': None, 'timestamp': timestamp, 'datetime': datetime, 'symbol': market['symbol'], 'type': None, 'side': side, 'price': self.safe_string_2(trade, 'price', 'deal_price'), 'amount': self.safe_string_2(trade, 'size', 'deal_vol'), 'cost': None, 'takerOrMaker': takerOrMaker, 'fee': None, }, market) def handle_ticker(self, client: Client, message): # # { # "data": [ # { # "base_volume_24h": "78615593.81", # "high_24h": "52756.97", # "last_price": "52638.31", # "low_24h": "50991.35", # "open_24h": "51692.03", # "s_t": 1630981727, # "symbol": "BTC_USDT" # } # ], # "table": "spot/ticker" # } # # { # "data": { # "symbol": "ETHUSDT", # "last_price": "2807.73", # "volume_24": "2227011952", # "range": "0.0273398194664491", # "mark_price": "2807.5", # "index_price": "2808.71047619", # "ask_price": "2808.04", # "ask_vol": "7371", # "bid_price": "2807.28", # "bid_vol": "3561" # }, # "group": "futures/ticker:ETHUSDT@100ms" # } # self.handle_bid_ask(client, message) table = self.safe_string(message, 'table') isSpot = (table is not None) rawTickers = [] if isSpot: rawTickers = self.safe_list(message, 'data', []) else: rawTickers = [self.safe_value(message, 'data', {})] if not len(rawTickers): return for i in range(0, len(rawTickers)): ticker = self.parse_ticker(rawTickers[i]) if isSpot else self.parse_ws_swap_ticker(rawTickers[i]) symbol = ticker['symbol'] self.tickers[symbol] = ticker messageHash = 'ticker:' + symbol client.resolve(ticker, messageHash) def parse_ws_swap_ticker(self, ticker, market: Market = None): # # { # "symbol": "ETHUSDT", # "last_price": "2807.73", # "volume_24": "2227011952", # "range": "0.0273398194664491", # "mark_price": "2807.5", # "index_price": "2808.71047619", # "ask_price": "2808.04", # "ask_vol": "7371", # "bid_price": "2807.28", # "bid_vol": "3561" # } # marketId = self.safe_string(ticker, 'symbol') return self.safe_ticker({ 'symbol': self.safe_symbol(marketId, market, '', 'swap'), 'timestamp': None, 'datetime': None, 'high': None, 'low': None, 'bid': self.safe_string(ticker, 'bid_price'), 'bidVolume': self.safe_string(ticker, 'bid_vol'), 'ask': self.safe_string(ticker, 'ask_price'), 'askVolume': self.safe_string(ticker, 'ask_vol'), 'vwap': None, 'open': None, 'close': None, 'last': self.safe_string(ticker, 'last_price'), 'previousClose': None, 'change': None, 'percentage': None, 'average': None, 'baseVolume': None, 'quoteVolume': self.safe_string(ticker, 'volume_24'), 'info': ticker, 'markPrice': self.safe_string(ticker, 'mark_price'), 'indexPrice': self.safe_string(ticker, 'index_price'), }, market) async def watch_ohlcv(self, symbol: str, timeframe='1m', since: Int = None, limit: Int = None, params={}) -> List[list]: """ https://developer-pro.bitmart.com/en/spot/#public-kline-channel https://developer-pro.bitmart.com/en/futuresv2/#public-klinebin-channel watches historical candlestick data containing the open, high, low, and close price, and the volume of a market :param str symbol: unified symbol of the market to fetch OHLCV data for :param str timeframe: the length of time each candle represents :param int [since]: timestamp in ms of the earliest candle to fetch :param int [limit]: the maximum amount of candles to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() symbol = self.symbol(symbol) market = self.market(symbol) type = 'spot' type, params = self.handle_market_type_and_params('watchOrderBook', market, params) timeframes = self.safe_value(self.options, 'timeframes', {}) interval = self.safe_string(timeframes, timeframe) name = None if type == 'spot': name = 'kline' + interval else: name = 'klineBin' + interval ohlcv = await self.subscribe(name, symbol, type, params) if self.newUpdates: limit = ohlcv.getLimit(symbol, limit) return self.filter_by_since_limit(ohlcv, since, limit, 0, True) def handle_ohlcv(self, client: Client, message): # # { # "data": [ # { # "candle": [ # 1631056350, # "46532.83", # "46555.71", # "46511.41", # "46555.71", # "0.25" # ], # "symbol": "BTC_USDT" # } # ], # "table": "spot/kline1m" # } # swap # { # "group":"futures/klineBin1m:BTCUSDT", # "data":{ # "symbol":"BTCUSDT", # "items":[ # { # "o":"39635.8", # "h":"39636", # "l":"39614.4", # "c":"39629.7", # "v":"31852", # "ts":1701617761 # } # ] # } # } # channel = self.safe_string_2(message, 'table', 'group') isSpot = (channel.find('spot') >= 0) data = self.safe_value(message, 'data') if data is None: return parts = channel.split('/') part1 = self.safe_string(parts, 1, '') interval = part1.replace('kline', '') interval = interval.replace('Bin', '') intervalParts = interval.split(':') interval = self.safe_string(intervalParts, 0) # use a reverse lookup in a static map instead timeframes = self.safe_value(self.options, 'timeframes', {}) timeframe = self.find_timeframe(interval, timeframes) duration = self.parse_timeframe(timeframe) durationInMs = duration * 1000 if isSpot: for i in range(0, len(data)): marketId = self.safe_string(data[i], 'symbol') market = self.safe_market(marketId) symbol = market['symbol'] rawOHLCV = self.safe_value(data[i], 'candle') parsed = self.parse_ohlcv(rawOHLCV, market) parsed[0] = self.parse_to_int(parsed[0] / durationInMs) * durationInMs self.ohlcvs[symbol] = self.safe_value(self.ohlcvs, symbol, {}) stored = self.safe_value(self.ohlcvs[symbol], timeframe) if stored is None: limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) stored = ArrayCacheByTimestamp(limit) self.ohlcvs[symbol][timeframe] = stored stored.append(parsed) messageHash = channel + ':' + marketId client.resolve(stored, messageHash) else: marketId = self.safe_string(data, 'symbol') market = self.safe_market(marketId, None, None, 'swap') symbol = market['symbol'] items = self.safe_value(data, 'items', []) self.ohlcvs[symbol] = self.safe_value(self.ohlcvs, symbol, {}) stored = self.safe_value(self.ohlcvs[symbol], timeframe) if stored is None: limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) stored = ArrayCacheByTimestamp(limit) self.ohlcvs[symbol][timeframe] = stored for i in range(0, len(items)): candle = items[i] parsed = self.parse_ohlcv(candle, market) stored.append(parsed) client.resolve(stored, channel) async def watch_order_book(self, symbol: str, limit: Int = None, params={}) -> OrderBook: """ https://developer-pro.bitmart.com/en/spot/#public-depth-all-channel https://developer-pro.bitmart.com/en/spot/#public-depth-increase-channel https://developer-pro.bitmart.com/en/futuresv2/#public-depth-channel watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data :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 :param str [params.speed]: *futures only* '100ms' or '200ms' :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() options = self.safe_value(self.options, 'watchOrderBook', {}) depth = self.safe_string(options, 'depth', 'depth/increase100') symbol = self.symbol(symbol) market = self.market(symbol) type = 'spot' type, params = self.handle_market_type_and_params('watchOrderBook', market, params) if type == 'swap' and depth == 'depth/increase100': depth = 'depth50' orderbook = await self.subscribe(depth, symbol, type, params) return orderbook.limit() 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): # # { # "asks": [ # ['46828.38', "0.21847"], # ['46830.68', "0.08232"], # ['46832.08', "0.09285"], # ['46837.82', "0.02028"], # ['46839.43', "0.15068"] # ], # "bids": [ # ['46820.78', "0.00444"], # ['46814.33', "0.00234"], # ['46813.50', "0.05021"], # ['46808.14', "0.00217"], # ['46808.04', "0.00013"] # ], # "ms_t": 1631044962431, # "symbol": "BTC_USDT" # } # asks = self.safe_list(message, 'asks', []) bids = self.safe_list(message, 'bids', []) self.handle_deltas(orderbook['asks'], asks) self.handle_deltas(orderbook['bids'], bids) timestamp = self.safe_integer(message, 'ms_t') marketId = self.safe_string(message, 'symbol') symbol = self.safe_symbol(marketId) orderbook['symbol'] = symbol orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) return orderbook def handle_order_book(self, client: Client, message): # # spot depth-all # # { # "data": [ # { # "asks": [ # ['46828.38', "0.21847"], # ['46830.68', "0.08232"], # ... # ], # "bids": [ # ['46820.78', "0.00444"], # ['46814.33', "0.00234"], # ... # ], # "ms_t": 1631044962431, # "symbol": "BTC_USDT" # } # ], # "table": "spot/depth5" # } # # spot increse depth snapshot # # { # "data":[ # { # "asks":[ # ["43652.52", "0.02039"], # ... # ], # "bids":[ # ["43652.51", "0.00500"], # ... # ], # "ms_t":1703376836487, # "symbol":"BTC_USDT", # "type":"snapshot", # or update # "version":2141731 # } # ], # "table":"spot/depth/increase100" # } # # swap # # { # "group":"futures/depth50:BTCUSDT", # "data":{ # "symbol":"BTCUSDT", # "way":1, # "depths":[ # { # "price":"39509.8", # "vol":"2379" # }, # { # "price":"39509.6", # "vol":"6815" # }, # ... # ], # "ms_t":1701566021194 # } # } # isSpot = ('table' in message) datas = [] if isSpot: datas = self.safe_list(message, 'data', datas) else: orderBookEntry = self.safe_dict(message, 'data') if orderBookEntry is not None: datas.append(orderBookEntry) length = len(datas) if length <= 0: return channelName = self.safe_string_2(message, 'table', 'group') # find limit subscribed to limitsToCheck = ['100', '50', '20', '10', '5'] limit = 0 for i in range(0, len(limitsToCheck)): limitString = limitsToCheck[i] if channelName.find(limitString) >= 0: limit = self.parse_to_int(limitString) break if isSpot: channel = channelName.replace('spot/', '') for i in range(0, len(datas)): update = datas[i] marketId = self.safe_string(update, 'symbol') symbol = self.safe_symbol(marketId) if not (symbol in self.orderbooks): ob = self.order_book({}, limit) ob['symbol'] = symbol self.orderbooks[symbol] = ob orderbook = self.orderbooks[symbol] type = self.safe_string(update, 'type') if (type == 'snapshot') or (not(channelName.find('increase') >= 0)): orderbook.reset({}) self.handle_order_book_message(client, update, orderbook) timestamp = self.safe_integer(update, 'ms_t') if orderbook['timestamp'] is None: orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) messageHash = channelName + ':' + marketId client.resolve(orderbook, messageHash) # resolve ForSymbols messageHashForMulti = channel + ':' + symbol client.resolve(orderbook, messageHashForMulti) else: tableParts = channelName.split(':') channel = tableParts[0].replace('futures/', '') data = datas[0] # contract markets always contain only one member depths = data['depths'] marketId = self.safe_string(data, 'symbol') symbol = self.safe_symbol(marketId) if not (symbol in self.orderbooks): ob = self.order_book({}, limit) ob['symbol'] = symbol self.orderbooks[symbol] = ob orderbook = self.orderbooks[symbol] way = self.safe_integer(data, 'way') side = 'bids' if (way == 1) else 'asks' if way == 1: orderbook[side] = Bids([], limit) else: orderbook[side] = Asks([], limit) for i in range(0, len(depths)): depth = depths[i] price = self.safe_number(depth, 'price') amount = self.safe_number(depth, 'vol') orderbookSide = self.safe_value(orderbook, side) orderbookSide.store(price, amount) bidsLength = len(orderbook['bids']) asksLength = len(orderbook['asks']) if (bidsLength == 0) or (asksLength == 0): return timestamp = self.safe_integer(data, 'ms_t') orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) messageHash = channelName client.resolve(orderbook, messageHash) # resolve ForSymbols messageHashForMulti = channel + ':' + symbol client.resolve(orderbook, messageHashForMulti) 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://developer-pro.bitmart.com/en/spot/#public-depth-increase-channel :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 :param str [params.depth]: the type of order book to subscribe to, default is 'depth/increase100', also accepts 'depth5' or 'depth20' or depth50 :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() type = None symbols, type, params = self.get_params_for_multiple_sub('watchOrderBookForSymbols', symbols, limit, params) channel = None channel, params = self.handle_option_and_params(params, 'watchOrderBookForSymbols', 'depth', 'depth/increase100') if type == 'swap' and channel == 'depth/increase100': channel = 'depth50' orderbook = await self.subscribe_multiple(channel, type, symbols, params) return orderbook.limit() async def authenticate(self, type, params={}): self.check_required_credentials() url = self.implode_hostname(self.urls['api']['ws'][type]['private']) messageHash = 'authenticated' client = self.client(url) future = client.future(messageHash) authenticated = self.safe_value(client.subscriptions, messageHash) if authenticated is None: timestamp = str(self.milliseconds()) memo = self.uid path = 'bitmart.WebSocket' auth = timestamp + '#' + memo + '#' + path signature = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256) request = None if type == 'spot': request = { 'op': 'login', 'args': [ self.apiKey, timestamp, signature, ], } else: request = { 'action': 'access', 'args': [ self.apiKey, timestamp, signature, 'web', ], } message = self.extend(request, params) self.watch(url, messageHash, message, messageHash) return await future def handle_subscription_status(self, client: Client, message): # # {"event":"subscribe","channel":"spot/depth:BTC-USDT"} # return message def handle_authenticate(self, client: Client, message): # # spot # {event: "login"} # swap # {action: 'access', success: True} # messageHash = 'authenticated' future = self.safe_value(client.futures, messageHash) future.resolve(True) def handle_error_message(self, client: Client, message): # # {event: "error", message: "Invalid sign", errorCode: 30013} # {"event":"error","message":"Unrecognized request: {\"event\":\"subscribe\",\"channel\":\"spot/depth:BTC-USDT\"}","errorCode":30039} # { # action: '', # group: 'futures/trade:BTCUSDT', # success: False, # request: {action: '', args: ['futures/trade:BTCUSDT']}, # error: 'Invalid action [] for group [futures/trade:BTCUSDT]' # } # errorCode = self.safe_string(message, 'errorCode') error = self.safe_string(message, 'error') try: if errorCode is not None or error is not None: feedback = self.id + ' ' + self.json(message) self.throw_exactly_matched_exception(self.exceptions['exact'], errorCode, feedback) messageString = self.safe_value(message, 'message', error) self.throw_broadly_matched_exception(self.exceptions['broad'], messageString, feedback) action = self.safe_string(message, 'action') if action == 'access': raise AuthenticationError(feedback) raise ExchangeError(feedback) return False except Exception as e: if (isinstance(e, AuthenticationError)): messageHash = 'authenticated' client.reject(e, messageHash) if messageHash in client.subscriptions: del client.subscriptions[messageHash] client.reject(e) return True def handle_message(self, client: Client, message): if self.handle_error_message(client, message): return # # {"event":"error","message":"Unrecognized request: {\"event\":\"subscribe\",\"channel\":\"spot/depth:BTC-USDT\"}","errorCode":30039} # # subscribe events on spot: # # {"event":"subscribe", "topic":"spot/kline1m:BTC_USDT"} # # subscribe on contracts: # # {"action":"subscribe", "group":"futures/klineBin1m:BTCUSDT", "success":true, "request":{"action":"subscribe", "args":["futures/klineBin1m:BTCUSDT"]}} # # regular updates - spot # # { # "table": "spot/depth", # "action": "partial", # "data": [ # { # "instrument_id": "BTC-USDT", # "asks": [ # ["5301.8", "0.03763319", "1"], # ["5302.4", "0.00305", "2"], # ], # "bids": [ # ["5301.7", "0.58911427", "6"], # ["5301.6", "0.01222922", "4"], # ], # "timestamp": "2020-03-16T03:25:00.440Z", # "checksum": -2088736623 # } # ] # } # # regular updates - contracts # # { # group: "futures/klineBin1m:BTCUSDT", # data: { # symbol: "BTCUSDT", # items: [{o: "67944.7", "h": ....}], # }, # } # # {data: '', table: "spot/user/order"} # # the only realiable way(for both spot & swap) is to check 'data' key isDataUpdate = ('data' in message) if not isDataUpdate: event = self.safe_string_2(message, 'event', 'action') if event is not None: methods: dict = { # 'info': self.handleSystemStatus, 'login': self.handle_authenticate, 'access': self.handle_authenticate, 'subscribe': self.handle_subscription_status, } method = self.safe_value(methods, event) if method is not None: method(client, message) else: channel = self.safe_string_2(message, 'table', 'group') methods: dict = { 'depth': self.handle_order_book, 'ticker': self.handle_ticker, 'trade': self.handle_trade, 'kline': self.handle_ohlcv, 'order': self.handle_orders, 'position': self.handle_positions, 'balance': self.handle_balance, 'asset': self.handle_balance, } keys = list(methods.keys()) for i in range(0, len(keys)): key = keys[i] if channel.find(key) >= 0: method = self.safe_value(methods, key) method(client, message)