# -*- 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, Bool, Int, Market, Order, OrderBook, Position, Str, Strings, Ticker, Tickers, FundingRate, FundingRates, Trade
from ccxt.async_support.base.ws.client import Client
from typing import List
from ccxt.base.errors import ExchangeError
from ccxt.base.errors import AuthenticationError
from ccxt.base.errors import ArgumentsRequired
from ccxt.base.errors import 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,
                'watchFundingRate': True,
                'watchFundingRates': True,
                'watchTicker': True,
                'watchTickers': True,
                'watchBidsAsks': True,
                'watchOrderBook': True,
                'watchOrderBookForSymbols': True,
                'watchOrders': True,
                'watchTrades': True,
                'watchTradesForSymbols': True,
                'watchOHLCV': True,
                'watchPosition': 'emulated',
                'watchPositions': True,
                'unWatchBidsAsks': False,  # the same channel
                'unWatchOHLCV': True,
                'unWatchOrderBook': True,
                'unWatchOrderBookForSymbols': True,
                'unWatchOrders': True,
                'unWatchPositions': True,
                'unWatchTicker': True,
                'unWatchTickers': True,
                'unWatchTrades': True,
                'unWatchTradesForSymbols': 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, unifiedName, channel, symbol, type, params={}):
        market = self.market(symbol)
        url = self.implode_hostname(self.urls['api']['ws'][type]['public'])
        request = {}
        messageHash = None
        rawHash = None
        unsubscribe = self.safe_bool(params, 'unsubscribe', False)
        prefix = ''
        requestOp = 'subscribe'
        if unsubscribe:
            params = self.omit(params, 'unsubscribe')
            prefix = 'unsubscribe::'
            requestOp = 'unsubscribe'
        messageHash = unifiedName + '::' + symbol
        if type == 'spot':
            rawHash = 'spot/' + channel + ':' + market['id']
            request = {
                'op': requestOp,
                'args': [rawHash],
            }
        else:
            rawHash = 'futures/' + channel + ':' + market['id']
            speed = self.safe_string(params, 'speed')
            if speed is not None:
                params = self.omit(params, 'speed')
                messageHash += ':' + speed
            request = {
                'action': requestOp,
                'args': [rawHash],
            }
        messageHash = prefix + messageHash
        return await self.watch(url, messageHash, self.deep_extend(request, params), messageHash)

    async def subscribe_multiple(self, unifiedName: str, 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 = []
        unsubscribe = self.safe_bool(params, 'unsubscribe', False)
        prefix = ''
        requestOp = 'subscribe'
        if unsubscribe:
            params = self.omit(params, 'unsubscribe')
            prefix = 'unsubscribe::'
            requestOp = 'unsubscribe'
        for i in range(0, len(symbols)):
            market = self.market(symbols[i])
            rawHash = channelType + '/' + channel + ':' + market['id']
            messageHash = prefix + unifiedName + '::' + market['symbol']
            messageHashes.append(messageHash)
            if not self.subscription_exists_for_hash(url, messageHash):
                rawSubscriptions.append(rawHash)
        # 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] = requestOp
        return await self.watch_multiple(url, messageHashes, self.deep_extend(request, params), messageHashes)

    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 <https://docs.ccxt.com/?id=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(params, 'watchBalance', 'fetchBalanceSnapshot', True)
        awaitBalanceSnapshot, params = self.handle_option_and_params(params, '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_dict(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
        if messageHash in client.futures:
            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":"1701632345416",
        #              "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_list(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 <https://docs.ccxt.com/?id=public-trades>`
        """
        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
        https://developer-pro.bitmart.com/en/futuresv2/#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 <https://docs.ccxt.com/?id=public-trades>`
        """
        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('trade', 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

    async def un_watch_trades(self, symbol: str, params={}) -> Any:
        """
        unWatches from the stream channel

        https://developer-pro.bitmart.com/en/spot/#public-trade-channel
        https://developer-pro.bitmart.com/en/futuresv2/#public-trade-channel

        :param str symbol: unified symbol of the market to fetch trades for
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict[]: a list of `trade structures <https://docs.ccxt.com/?id=public-trades>`
        """
        return await self.un_watch_trades_for_symbols([symbol], params)

    async def un_watch_trades_for_symbols(self, symbols: List[str], params={}) -> Any:
        """
        unsubscribes from the trades channel

        https://developer-pro.bitmart.com/en/spot/#public-trade-channel
        https://developer-pro.bitmart.com/en/futuresv2/#public-trade-channel

        :param str[] symbols: unified symbol of the market to fetch trades for
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict[]: a list of `trade structures <https://docs.ccxt.com/?id=public-trades>`
        """
        await self.load_markets()
        marketType = None
        symbols, marketType, params = self.get_params_for_multiple_sub('unWatchTradesForSymbols', symbols, None, params)
        channelName = 'trade'
        params = self.extend(params, {'unsubscribe': True})
        return await self.subscribe_multiple('trade', channelName, marketType, symbols, params)

    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 <https://docs.ccxt.com/?id=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 <https://docs.ccxt.com/?id=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', '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 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 a specific market

        https://developer-pro.bitmart.com/en/spot/#public-ticker-channel
        https://developer-pro.bitmart.com/en/futuresv2/#public-ticker-channel

        :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 <https://docs.ccxt.com/?id=ticker-structure>`
        """
        return await self.un_watch_tickers([symbol], params)

    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://developer-pro.bitmart.com/en/spot/#public-ticker-channel
        https://developer-pro.bitmart.com/en/futuresv2/#public-ticker-channel

        :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 <https://docs.ccxt.com/?id=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)
        params = self.extend(params, {'unsubscribe': True})
        return await self.subscribe_multiple('ticker', 'ticker', marketType, symbols, params)

    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 <https://docs.ccxt.com/?id=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/bookTicker' if (marketType == 'spot') else 'futures/ticker'
        actionType = 'op' if (marketType == 'spot') else 'action'
        rawSubscriptions = []
        messageHashes = []
        for i in range(0, len(symbols)):
            market = self.market(symbols[i])
            rawHash = channelType + ':' + market['id']
            messageHash = 'bidask::' + market['symbol']
            messageHashes.append(messageHash)
            if not self.subscription_exists_for_hash(url, messageHash):
                rawSubscriptions.append(rawHash)
        if marketType != 'spot':
            rawSubscriptions = [channelType]
        request: dict = {
            'args': rawSubscriptions,
        }
        request[actionType] = 'subscribe'
        newTickers = await self.watch_multiple(url, messageHashes, request, messageHashes)
        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_dict(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 <https://docs.ccxt.com/?id=order-structure>`
        """
        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)

    async def un_watch_orders(self, symbol: Str = None, params={}) -> Any:
        """
        unWatches 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 dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict[]: a list of `order structures <https://docs.ccxt.com/?id=order-structure>`
        """
        await self.load_markets()
        market = None
        messageHash = 'unsubscribe::orders'
        if symbol is not None:
            symbol = self.symbol(symbol)
            market = self.market(symbol)
            if market['swap']:
                raise NotSupported(self.id + ' unWatchOrders() does not support a symbol for swap markets, unWatch from all markets only')
            messageHash += '::' + 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': 'unsubscribe',
                'args': [argsRequest],
            }
        else:
            request = {
                'action': 'unsubscribe',
                'args': ['futures/order'],
            }
        url = self.implode_hostname(self.urls['api']['ws'][type]['private'])
        return await self.watch(url, messageHash, self.deep_extend(request, params), messageHash)

    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_list(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_dict(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_dict(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 <https://docs.ccxt.com/en/latest/manual.html#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)

    async def un_watch_positions(self, symbols: Strings = None, params={}) -> Any:
        """
        unWatches all open positions

        https://developer-pro.bitmart.com/en/futures/#private-position-channel

        :param str[] [symbols]: list of unified market symbols
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict: status of the unwatch request
        """
        if symbols is not None:
            length = len(symbols)
            if length > 0:
                raise NotSupported(self.id + ' unWatchPositions() does not support a list of symbols, unWatch from all markets only')
        await self.load_markets()
        request: dict = {
            'action': 'unsubscribe',
            'args': ['futures/position'],
        }
        messageHash = 'unsubscribe::positions'
        url = self.implode_hostname(self.urls['api']['ws']['swap']['private'])
        return await self.watch(url, messageHash, self.deep_extend(request, params), messageHash)

    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_list(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_list(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_dict(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: str = '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('watchOHLCV', market, params)
        timeframes = self.safe_dict(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('ohlcv', name, symbol, type, params)
        if self.newUpdates:
            limit = ohlcv.getLimit(symbol, limit)
        return self.filter_by_since_limit(ohlcv, since, limit, 0, True)

    async def un_watch_ohlcv(self, symbol: str, timeframe: str = '1m', params={}) -> Any:
        """
        unWatches historical candlestick data containing the open, high, low, and close price, and the volume of a market

        https://developer-pro.bitmart.com/en/spot/#public-kline-channel
        https://developer-pro.bitmart.com/en/futuresv2/#public-klinebin-channel

        :param str symbol: unified symbol of the market to fetch OHLCV data for
        :param str timeframe: the length of time each candle represents
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns int[][]: A list of candles ordered, open, high, low, close, volume
        """
        await self.load_markets()
        symbol = self.symbol(symbol)
        market = self.market(symbol)
        type = 'spot'
        type, params = self.handle_market_type_and_params('unWatchOHLCV', market, params)
        timeframes = self.safe_dict(self.options, 'timeframes', {})
        interval = self.safe_string(timeframes, timeframe)
        name = None
        if type == 'spot':
            name = 'kline' + interval
        else:
            name = 'klineBin' + interval
        params = self.extend(params, {'unsubscribe': True})
        return await self.subscribe('ohlcv', name, symbol, type, params)

    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_dict(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_list(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 = 'ohlcv::' + symbol
                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_list(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, 'ohlcv::' + symbol)

    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 <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        await self.load_markets()
        options = self.safe_dict(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('orderbook', depth, symbol, type, params)
        return orderbook.limit()

    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://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

        :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 <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        await self.load_markets()
        options = self.safe_dict(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('unWatchOrderBook', market, params)
        if type == 'swap' and depth == 'depth/increase100':
            depth = 'depth50'
        params = self.extend(params, {'unsubscribe': True})
        return await self.subscribe('orderbook', depth, symbol, type, params)

    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:
            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 = 'orderbook::' + symbol
                client.resolve(orderbook, messageHash)
        else:
            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 = 'orderbook::' + symbol
            client.resolve(orderbook, messageHash)

    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 <https://docs.ccxt.com/?id=order-book-structure>` 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('orderbook', channel, type, symbols, params)
        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://developer-pro.bitmart.com/en/spot/#public-depth-increase-channel

        :param str[] symbols: unified array of symbols
        :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 <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        await self.load_markets()
        type = None
        symbols, type, params = self.get_params_for_multiple_sub('unWatchOrderBookForSymbols', symbols, None, params)
        channel = None
        channel, params = self.handle_option_and_params(params, 'unWatchOrderBookForSymbols', 'depth', 'depth/increase100')
        if type == 'swap' and channel == 'depth/increase100':
            channel = 'depth50'
        params = self.extend(params, {'unsubscribe': True})
        return await self.subscribe_multiple('orderbook', channel, type, symbols, params)

    async def watch_funding_rate(self, symbol: str, params={}) -> FundingRate:
        """
        watch the current funding rate

        https://developer-pro.bitmart.com/en/futuresv2/#public-funding-rate-channel

        :param str symbol: unified market symbol
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict: a `funding rate structure <https://docs.ccxt.com/?id=funding-rate-structure>`
        """
        await self.load_markets()
        symbol = self.symbol(symbol)
        fundingRate = await self.watch_funding_rates([symbol], params)
        return fundingRate[symbol]

    async def watch_funding_rates(self, symbols: Strings = None, params={}) -> FundingRates:
        """
        watch the funding rate for multiple markets

        https://developer-pro.bitmart.com/en/futuresv2/#public-funding-rate-channel

        :param str[] symbols: a list of unified market symbols
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict: a dictionary of `funding rate structures <https://docs.ccxt.com/?id=funding-rate-structure>`, indexed by market symbols
        """
        if symbols is None:
            raise ArgumentsRequired(self.id + ' watchFundingRates() requires an array of symbols')
        await self.load_markets()
        market = self.get_market_from_symbols(symbols)
        marketType = None
        marketType, params = self.handle_market_type_and_params('watchFundingRates', market, params)
        fundingRate = await self.subscribe_multiple('fundingRate', 'fundingRate', marketType, symbols, params)
        if self.newUpdates:
            fundingRates: dict = {}
            fundingRates[fundingRate['symbol']] = fundingRate
            return fundingRates
        return self.filter_by_array(self.fundingRates, 'symbol', symbols)

    def handle_funding_rate(self, client: Client, message):
        #
        #     {
        #         "data": {
        #             "symbol": "BTCUSDT",
        #             "fundingRate": "0.0000561",
        #             "fundingTime": 1770978448000,
        #             "nextFundingRate": "-0.0000195",
        #             "nextFundingTime": 1770998400000,
        #             "funding_upper_limit": "0.0375",
        #             "funding_lower_limit": "-0.0375",
        #             "ts": 1770978448970
        #         },
        #         "group": "futures/fundingRate:BTCUSDT"
        #     }
        #
        data = self.safe_dict(message, 'data', {})
        fundingRate = self.parse_funding_rate(data)
        symbol = fundingRate['symbol']
        self.fundingRates[symbol] = fundingRate
        messageHash = 'fundingRate::' + symbol
        client.resolve(fundingRate, messageHash)

    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.reusableFuture(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) -> Bool:
        #
        #    {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')
        # Duplicate-subscription notice errorCode 90008: bitmart's WS rejects
        # a re-subscribe attempt on a topic that's already active on self
        # connection, but the original subscription keeps delivering data —
        # so treat it. Without self short-circuit, the generic
        # client.reject below kills every unrelated in-flight future —
        # e.g. a watchOHLCV waiting on its kline subscription gets rejected
        # by an orderbook 90008 raised on the same socket.
        if errorCode == '90008':
            return False
        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_un_subscription(self, client: Client, message):
        #
        # spot
        #     {
        #         "topic": "spot/ticker:ETH_USDT",
        #         "event": "unsubscribe"
        #     }
        #
        # swap
        #     {
        #         "action": "unsubscribe",
        #         "group": "futures/ticker:ETHUSDT",
        #         "success": True,
        #         "request": {
        #             "action": "unsubscribe",
        #             "args": [
        #                 "futures/ticker:ETHUSDT"
        #             ]
        #         }
        #     }
        #
        messageTopic = self.safe_string_2(message, 'topic', 'group')
        unSubMessageTopic = 'unsubscribe::' + messageTopic
        # one message includes info about one unsubscription only even if we requested multiple
        # so we can not just create subscription object in unWatch method and use it here
        # we need to reconstruct subscription params from the messageTopic
        subscription = self.get_un_sub_params(messageTopic)
        subHash = self.safe_string(subscription, 'subHash')
        unsubHash = 'unsubscribe::' + subHash
        subHashIsPrefix = self.safe_bool(subscription, 'subHashIsPrefix', False)
        # clean up both ways of storing subscription and unsubscription
        self.clean_unsubscription(client, subHash, unsubHash, subHashIsPrefix)
        self.clean_unsubscription(client, messageTopic, unSubMessageTopic, subHashIsPrefix)
        self.clean_cache(subscription)

    def get_un_sub_params(self, messageTopic):
        parts = messageTopic.split(':')
        channel = self.safe_string(parts, 0)
        marketTypeAndTopic = channel.split('/')
        rawMarketType = self.safe_string_lower(marketTypeAndTopic, 0)
        marketType = self.parse_market_type(rawMarketType)
        topic = self.safe_string(marketTypeAndTopic, 1)
        thirdPart = self.safe_string(marketTypeAndTopic, 2)
        if thirdPart is not None:
            topic += '/' + thirdPart
        marketId = self.safe_string(parts, 1)
        symbols = []
        symbol = None
        subHash = topic
        hashDelimiter = ':'
        subHashIsPrefix = False
        parsedTopic = self.parse_topic(topic)
        if (parsedTopic == 'orders') or (parsedTopic == 'positions'):
            subHash = parsedTopic
            hashDelimiter = '::'
        if (marketId is not None) and (marketId != 'ALL_SYMBOLS'):
            # if marketId is defined, we have a single symbol subscription
            delimiter = '_' if (marketType == 'spot') else ''
            market = self.safe_market(marketId, None, delimiter, marketType)
            symbol = market['symbol']
            subHash += hashDelimiter + symbol
            symbols.append(symbol)
        else:
            subHashIsPrefix = True  # need to clean all subHashes with self prefix
        symbolsAndTimeframes = []
        if topic.startswith('kline'):
            interval = topic.replace('kline', '')
            if interval.startswith('Bin'):
                # swap market
                interval = interval.replace('Bin', '')
            timeframes = self.safe_dict(self.options, 'timeframes', {})
            timeframe = self.find_timeframe(interval, timeframes)
            symbolAndTimeframe = [symbol, timeframe]
            symbolsAndTimeframes.append(symbolAndTimeframe)
        result = {
            'topic': parsedTopic,
            'symbols': symbols,
            'subHash': subHash,
            'symbolsAndTimeframes': symbolsAndTimeframes,
            'subHashIsPrefix': subHashIsPrefix,
        }
        return result

    def parse_topic(self, topic):
        if topic.startswith('depth'):
            return 'orderbook'
        if topic.startswith('kline'):
            return 'ohlcv'
        topics = {
            'ticker': 'ticker',
            'trade': 'trades',
            'user/order': 'orders',
            'user/orders': 'orders',
            'order': 'orders',
            'position': 'positions',
        }
        return self.safe_string(topics, topic, topic)

    def parse_market_type(self, marketType: str):
        types = {
            'spot': 'spot',
            'futures': 'swap',
        }
        return self.safe_string(types, marketType, marketType)

    def subscription_exists_for_hash(self, url: str, hash: str):
        client = self.client(url)
        return(hash in client.subscriptions)

    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,
                    'unsubscribe': self.handle_un_subscription,
                }
                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,
                'bookTicker': self.handle_bid_ask,
                '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,
            }
            if channel.find('fundingRate') >= 0:
                self.handle_funding_rate(client, message)
            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)
