# -*- coding: utf-8 -*-

# PLEASE DO NOT EDIT THIS FILE, IT IS GENERATED AND WILL BE OVERWRITTEN:
# https://github.com/ccxt/ccxt/blob/master/CONTRIBUTING.md#how-to-contribute-code

import ccxt.async_support
from ccxt.async_support.base.ws.cache import ArrayCache, ArrayCacheBySymbolById, ArrayCacheBySymbolBySide, ArrayCacheByTimestamp
import hashlib
from ccxt.base.types import Any, Balances, Bool, Int, Liquidation, Num, Order, OrderBook, OrderSide, OrderType, 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 BadRequest
from ccxt.base.errors import InvalidNonce


class okx(ccxt.async_support.okx):

    def describe(self) -> Any:
        return self.deep_extend(super(okx, self).describe(), {
            'has': {
                'ws': True,
                'watchTicker': True,
                'watchMarkPrice': True,
                'watchMarkPrices': True,
                'watchTickers': True,
                'watchBidsAsks': True,
                'watchOrderBook': True,
                'watchTrades': True,
                'watchTradesForSymbols': True,
                'watchOrderBookForSymbols': True,
                'watchBalance': True,
                'watchLiquidations': 'emulated',
                'watchLiquidationsForSymbols': True,
                'watchMyLiquidations': 'emulated',
                'watchMyLiquidationsForSymbols': True,
                'watchOHLCV': True,
                'watchOHLCVForSymbols': True,
                'watchOrders': True,
                'watchMyTrades': True,
                'watchPositions': True,
                'watchFundingRate': True,
                'watchFundingRates': True,
                'createOrderWs': True,
                'editOrderWs': True,
                'cancelOrderWs': True,
                'cancelOrdersWs': True,
                'cancelAllOrdersWs': True,
            },
            'urls': {
                'api': {
                    'ws': 'wss://ws.okx.com:8443/ws/v5',
                },
                'test': {
                    'ws': 'wss://wspap.okx.com:8443/ws/v5',
                },
            },
            'options': {
                'watchOrderBook': {
                    'checksum': True,
                    #
                    # bbo-tbt
                    # 1. Newly added channel that sends tick-by-tick Level 1 data
                    # 2. All API users can subscribe
                    # 3. Public depth channel, verification not required
                    #
                    # books-l2-tbt
                    # 1. Only users who're VIP5 and above can subscribe
                    # 2. Identity verification required before subscription
                    #
                    # books50-l2-tbt
                    # 1. Only users who're VIP4 and above can subscribe
                    # 2. Identity verification required before subscription
                    #
                    # books
                    # 1. All API users can subscribe
                    # 2. Public depth channel, verification not required
                    #
                    # books5
                    # 1. All API users can subscribe
                    # 2. Public depth channel, verification not required
                    # 3. Data feeds will be delivered every 100ms(vs. every 200ms now)
                    #
                    'depth': 'books',
                },
                'watchBalance': 'spot',  # margin, futures, swap
                'watchTicker': {
                    'channel': 'tickers',  # tickers, sprd-tickers, index-tickers, block-tickers
                },
                'watchTickers': {
                    'channel': 'tickers',  # tickers, sprd-tickers, index-tickers, block-tickers
                },
                'watchOrders': {
                    'type': 'ANY',  # SPOT, MARGIN, SWAP, FUTURES, OPTION, ANY
                },
                'watchMyTrades': {
                    'type': 'ANY',  # SPOT, MARGIN, SWAP, FUTURES, OPTION, ANY
                },
                'createOrderWs': {
                    'op': 'batch-orders',  # order, batch-orders
                },
                'editOrderWs': {
                    'op': 'amend-order',  # amend-order, batch-amend-orders
                },
                'ws': {
                    # 'inflate': True,
                },
            },
            'streaming': {
                # okex does not support built-in ws protocol-level ping-pong
                # instead it requires a custom text-based ping-pong
                'ping': self.ping,
                'keepAlive': 18000,
            },
        })

    def get_url(self, channel: str, access='public'):
        # for context: https://www.okx.com/help-center/changes-to-v5-api-websocket-subscription-parameter-and-url
        isSandbox = self.options['sandboxMode']
        sandboxSuffix = '?brokerId=9999' if isSandbox else ''
        isBusiness = (access == 'business')
        isPublic = (access == 'public')
        url = self.urls['api']['ws']
        if isBusiness or (channel.find('candle') > -1) or (channel == 'orders-algo'):
            return url + '/business' + sandboxSuffix
        elif isPublic:
            return url + '/public' + sandboxSuffix
        return url + '/private' + sandboxSuffix

    async def subscribe_multiple(self, access, channel, symbols: Strings = None, params={}):
        await self.load_markets()
        if symbols is None:
            symbols = self.symbols
        symbols = self.market_symbols(symbols)
        url = self.get_url(channel, access)
        messageHashes = []
        args = []
        for i in range(0, len(symbols)):
            marketId = self.market_id(symbols[i])
            arg: dict = {
                'channel': channel,
                'instId': marketId,
            }
            args.append(self.extend(arg, params))
            messageHashes.append(channel + '::' + symbols[i])
        request: dict = {
            'op': 'subscribe',
            'args': args,
        }
        return await self.watch_multiple(url, messageHashes, request, messageHashes)

    async def subscribe(self, access, messageHash, channel, symbol, params={}):
        await self.load_markets()
        url = self.get_url(channel, access)
        firstArgument: dict = {
            'channel': channel,
        }
        if symbol is not None:
            market = self.market(symbol)
            messageHash += ':' + market['id']
            firstArgument['instId'] = market['id']
        request: dict = {
            'op': 'subscribe',
            'args': [
                self.deep_extend(firstArgument, params),
            ],
        }
        return await self.watch(url, messageHash, request, messageHash)

    async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-trades-channel
        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-all-trades-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://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-trades-channel
        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-all-trades-channel

        get the list of most recent trades for a particular symbol
        :param str symbols:
        :param int [since]: timestamp in ms of the earliest trade to fetch
        :param int [limit]: the maximum amount of trades to fetch
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.channel]: the channel to subscribe to, trades by default. Can be 'trades' and 'trades-all'
        :returns dict[]: a list of `trade structures <https://docs.ccxt.com/?id=public-trades>`
        """
        symbolsLength = len(symbols)
        if symbolsLength == 0:
            raise ArgumentsRequired(self.id + ' watchTradesForSymbols() requires a non-empty array of symbols')
        await self.load_markets()
        symbols = self.market_symbols(symbols)
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchTrades', 'channel', 'trades')
        topics = []
        messageHashes = []
        for i in range(0, len(symbols)):
            symbol = symbols[i]
            messageHashes.append(channel + ':' + symbol)
            marketId = self.market_id(symbol)
            topic: dict = {
                'channel': channel,
                'instId': marketId,
            }
            topics.append(topic)
        request: dict = {
            'op': 'subscribe',
            'args': topics,
        }
        access = 'public'
        if channel == 'trades-all':
            access = 'business'
            await self.authenticate({'access': access})
        url = self.get_url(channel, access)
        trades = await self.watch_multiple(url, messageHashes, request, messageHashes)
        if self.newUpdates:
            first = self.safe_value(trades, 0)
            tradeSymbol = self.safe_string(first, 'symbol')
            limit = trades.getLimit(tradeSymbol, limit)
        return self.filter_by_since_limit(trades, since, limit, 'timestamp', True)

    async def un_watch_trades_for_symbols(self, symbols: List[str], params={}) -> Any:
        """
        unWatches from the stream channel
        :param str[] symbols:
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.channel]: the channel to subscribe to, trades by default. Can be trades, trades-all
        :returns dict[]: a list of `trade structures <https://docs.ccxt.com/?id=public-trades>`
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols, None, False)
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchTrades', 'channel', 'trades')
        topics = []
        messageHashes = []
        for i in range(0, len(symbols)):
            symbol = symbols[i]
            messageHashes.append('unsubscribe:' + channel + ':' + symbol)
            marketId = self.market_id(symbol)
            topic: dict = {
                'channel': channel,
                'instId': marketId,
            }
            topics.append(topic)
        request: dict = {
            'op': 'unsubscribe',
            'args': topics,
        }
        access = 'public'
        if channel == 'trades-all':
            access = 'business'
            await self.authenticate({'access': access})
        url = self.get_url(channel, access)
        return await self.watch_multiple(url, messageHashes, request, messageHashes)

    async def un_watch_trades(self, symbol: str, params={}) -> Any:
        """
        unWatches from the stream 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)

    def handle_trades(self, client: Client, message):
        #
        #     {
        #         "arg": {channel: "trades", instId: "BTC-USDT"},
        #         "data": [
        #             {
        #                 "instId": "BTC-USDT",
        #                 "tradeId": "216970876",
        #                 "px": "31684.5",
        #                 "sz": "0.00001186",
        #                 "side": "buy",
        #                 "ts": "1626531038288"
        #             }
        #         ]
        #     }
        #     {
        #         "arg": {
        #             "channel": "trades-all",
        #             "instId": "BTC-USDT"
        #         },
        #         "data": [
        #             {
        #                 "instId": "BTC-USDT",
        #                 "tradeId": "130639474",
        #                 "px": "42219.9",
        #                 "sz": "0.12060306",
        #                 "side": "buy",
        #                 "source": "0",
        #                 "ts": "1630048897897"
        #             }
        #         ]
        #     }
        #
        arg = self.safe_value(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        marketId = self.safe_string(arg, 'instId')
        symbol = self.safe_symbol(marketId)
        data = self.safe_value(message, 'data', [])
        tradesLimit = self.safe_integer(self.options, 'tradesLimit', 1000)
        for i in range(0, len(data)):
            trade = self.parse_trade(data[i])
            messageHash = channel + ':' + symbol
            stored = self.safe_value(self.trades, symbol)
            if stored is None:
                stored = ArrayCache(tradesLimit)
                self.trades[symbol] = stored
            stored.append(trade)
            client.resolve(stored, messageHash)

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

        https://www.okx.com/docs-v5/en/#public-data-websocket-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>`
        """
        symbol = self.symbol(symbol)
        fr = await self.watch_funding_rates([symbol], params)
        return fr[symbol]

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

        https://www.okx.com/docs-v5/en/#public-data-websocket-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 rates 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()
        symbols = self.market_symbols(symbols)
        channel = 'funding-rate'
        topics = []
        messageHashes = []
        for i in range(0, len(symbols)):
            symbol = symbols[i]
            messageHashes.append(channel + ':' + symbol)
            marketId = self.market_id(symbol)
            topic: dict = {
                'channel': channel,
                'instId': marketId,
            }
            topics.append(topic)
        request: dict = {
            'op': 'subscribe',
            'args': topics,
        }
        url = self.get_url(channel, 'public')
        fundingRate = await self.watch_multiple(url, messageHashes, request, messageHashes)
        if self.newUpdates:
            symbol = self.safe_string(fundingRate, 'symbol')
            result: dict = {}
            result[symbol] = fundingRate
            return result
        return self.filter_by_array(self.fundingRates, 'symbol', symbols)

    def handle_funding_rate(self, client: Client, message):
        #
        # "data":[
        #     {
        #        "fundingRate":"0.0001875391284828",
        #        "fundingTime":"1700726400000",
        #        "instId":"BTC-USD-SWAP",
        #        "instType":"SWAP",
        #        "method": "next_period",
        #        "maxFundingRate":"0.00375",
        #        "minFundingRate":"-0.00375",
        #        "nextFundingRate":"0.0002608059239328",
        #        "nextFundingTime":"1700755200000",
        #        "premium": "0.0001233824646391",
        #        "settFundingRate":"0.0001699799259033",
        #        "settState":"settled",
        #        "ts":"1700724675402"
        #     }
        # ]
        #
        data = self.safe_list(message, 'data', [])
        for i in range(0, len(data)):
            rawfr = data[i]
            fundingRate = self.parse_funding_rate(rawfr)
            symbol = fundingRate['symbol']
            self.fundingRates[symbol] = fundingRate
            client.resolve(fundingRate, 'funding-rate' + ':' + fundingRate['symbol'])

    async def watch_ticker(self, symbol: str, params={}) -> Ticker:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-tickers-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
        :param str [params.channel]: the channel to subscribe to, tickers by default. Can be tickers, sprd-tickers, index-tickers, block-tickers
        :returns dict: a `ticker structure <https://docs.ccxt.com/?id=ticker-structure>`
        """
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchTicker', 'channel', 'tickers')
        params['channel'] = channel
        market = self.market(symbol)
        symbol = market['symbol']
        ticker = await self.watch_tickers([symbol], params)
        return self.safe_value(ticker, symbol)

    async def un_watch_ticker(self, symbol: str, params={}) -> Any:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-tickers-channel

        unWatches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market
        :param str symbol: unified symbol of the market to fetch the ticker for
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.channel]: the channel to subscribe to, tickers by default. Can be tickers, sprd-tickers, index-tickers, block-tickers
        :returns dict: a `ticker structure <https://docs.ccxt.com/?id=ticker-structure>`
        """
        return await self.un_watch_tickers([symbol], params)

    async def watch_tickers(self, symbols: Strings = None, params={}) -> Tickers:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-tickers-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
        :param str [params.channel]: the channel to subscribe to, tickers by default. Can be tickers, sprd-tickers, index-tickers, block-tickers
        :returns dict: a `ticker structure <https://docs.ccxt.com/?id=ticker-structure>`
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols, None, False)
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchTickers', 'channel', 'tickers')
        newTickers = await self.subscribe_multiple('public', channel, symbols, params)
        if self.newUpdates:
            return newTickers
        return self.filter_by_array(self.tickers, 'symbol', symbols)

    async def watch_mark_price(self, symbol: str, params={}) -> Ticker:
        """

        https://www.okx.com/docs-v5/en/#public-data-websocket-mark-price-channel

        watches a mark price
        :param str symbol: unified symbol of the market to fetch the ticker for
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.channel]: the channel to subscribe to, tickers by default. Can be tickers, sprd-tickers, index-tickers, block-tickers
        :returns dict: a `ticker structure <https://docs.ccxt.com/?id=ticker-structure>`
        """
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchMarkPrice', 'channel', 'mark-price')
        params['channel'] = channel
        market = self.market(symbol)
        symbol = market['symbol']
        ticker = await self.watch_mark_prices([symbol], params)
        return ticker[symbol]

    async def watch_mark_prices(self, symbols: Strings = None, params={}) -> Tickers:
        """

        https://www.okx.com/docs-v5/en/#public-data-websocket-mark-price-channel

        watches mark prices
        :param str[] [symbols]: unified symbol of the market to fetch the ticker for
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.channel]: the channel to subscribe to, tickers by default. Can be tickers, sprd-tickers, index-tickers, block-tickers
        :returns dict: a `ticker structure <https://docs.ccxt.com/?id=ticker-structure>`
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols, None, False)
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchMarkPrices', 'channel', 'mark-price')
        newTickers = await self.subscribe_multiple('public', channel, symbols, params)
        if self.newUpdates:
            return newTickers
        return self.filter_by_array(self.tickers, 'symbol', symbols)

    async def un_watch_tickers(self, symbols: Strings = None, params={}) -> Any:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-tickers-channel

        unWatches 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
        :param str [params.channel]: the channel to subscribe to, tickers by default. Can be tickers, sprd-tickers, index-tickers, block-tickers
        :returns dict: a `ticker structure <https://docs.ccxt.com/?id=ticker-structure>`
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols, None, False)
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchTickers', 'channel', 'tickers')
        topics = []
        messageHashes = []
        for i in range(0, len(symbols)):
            symbol = symbols[i]
            messageHashes.append('unsubscribe:ticker:' + symbol)
            marketId = self.market_id(symbol)
            topic: dict = {
                'channel': channel,
                'instId': marketId,
            }
            topics.append(topic)
        request: dict = {
            'op': 'unsubscribe',
            'args': topics,
        }
        url = self.get_url(channel, 'public')
        return await self.watch_multiple(url, messageHashes, request, messageHashes)

    def handle_ticker(self, client: Client, message):
        #
        #     {
        #         "arg": {channel: "tickers", instId: "BTC-USDT"},
        #         "data": [
        #             {
        #                 "instType": "SPOT",
        #                 "instId": "BTC-USDT",
        #                 "last": "31500.1",
        #                 "lastSz": "0.00001754",
        #                 "askPx": "31500.1",
        #                 "askSz": "0.00998144",
        #                 "bidPx": "31500",
        #                 "bidSz": "3.05652439",
        #                 "open24h": "31697",
        #                 "high24h": "32248",
        #                 "low24h": "31165.6",
        #                 "sodUtc0": "31385.5",
        #                 "sodUtc8": "32134.9",
        #                 "volCcy24h": "503403597.38138519",
        #                 "vol24h": "15937.10781721",
        #                 "ts": "1626526618762"
        #             }
        #         ]
        #     }
        #
        self.handle_bid_ask(client, message)
        arg = self.safe_value(message, 'arg', {})
        marketId = self.safe_string(arg, 'instId')
        market = self.safe_market(marketId, None, '-')
        symbol = market['symbol']
        channel = self.safe_string(arg, 'channel')
        data = self.safe_value(message, 'data', [])
        newTickers: dict = {}
        for i in range(0, len(data)):
            ticker = self.parse_ticker(data[i])
            self.tickers[symbol] = ticker
            newTickers[symbol] = ticker
        messageHash = channel + '::' + symbol
        client.resolve(newTickers, messageHash)

    async def watch_bids_asks(self, symbols: Strings = None, params={}) -> Tickers:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-tickers-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)
        channel = None
        channel, params = self.handle_option_and_params(params, 'watchBidsAsks', 'channel', 'tickers')
        url = self.get_url(channel, 'public')
        messageHashes = []
        args = []
        for i in range(0, len(symbols)):
            marketId = self.market_id(symbols[i])
            arg: dict = {
                'channel': channel,
                'instId': marketId,
            }
            args.append(self.extend(arg, params))
            messageHashes.append('bidask::' + symbols[i])
        request: dict = {
            'op': 'subscribe',
            'args': args,
        }
        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):
        #
        #     {
        #         "arg": {channel: "tickers", instId: "BTC-USDT"},
        #         "data": [
        #             {
        #                 "instType": "SPOT",
        #                 "instId": "BTC-USDT",
        #                 "last": "31500.1",
        #                 "lastSz": "0.00001754",
        #                 "askPx": "31500.1",
        #                 "askSz": "0.00998144",
        #                 "bidPx": "31500",
        #                 "bidSz": "3.05652439",
        #                 "open24h": "31697",
        #                 "high24h": "32248",
        #                 "low24h": "31165.6",
        #                 "sodUtc0": "31385.5",
        #                 "sodUtc8": "32134.9",
        #                 "volCcy24h": "503403597.38138519",
        #                 "vol24h": "15937.10781721",
        #                 "ts": "1626526618762"
        #             }
        #         ]
        #     }
        #
        data = self.safe_list(message, 'data', [])
        ticker = self.safe_dict(data, 0, {})
        parsedTicker = self.parse_ws_bid_ask(ticker)
        symbol = parsedTicker['symbol']
        self.bidsasks[symbol] = parsedTicker
        messageHash = 'bidask::' + symbol
        client.resolve(parsedTicker, messageHash)

    def parse_ws_bid_ask(self, ticker, market=None):
        marketId = self.safe_string(ticker, 'instId')
        market = self.safe_market(marketId, market)
        symbol = self.safe_string(market, 'symbol')
        timestamp = self.safe_integer(ticker, 'ts')
        return self.safe_ticker({
            'symbol': symbol,
            'timestamp': timestamp,
            'datetime': self.iso8601(timestamp),
            'ask': self.safe_string(ticker, 'askPx'),
            'askVolume': self.safe_string(ticker, 'askSz'),
            'bid': self.safe_string(ticker, 'bidPx'),
            'bidVolume': self.safe_string(ticker, 'bidSz'),
            'info': ticker,
        }, market)

    async def watch_liquidations_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Liquidation]:
        """
        watch the public liquidations of a trading pair

        https://www.okx.com/docs-v5/en/#public-data-websocket-liquidation-orders-channel

        :param str symbols:
        :param int [since]: the earliest time in ms to fetch liquidations for
        :param int [limit]: the maximum number of liquidation structures to retrieve
        :param dict [params]: exchange specific parameters for the okx api endpoint
        :returns dict: an array of `liquidation structures <https://github.com/ccxt/ccxt/wiki/Manual#liquidation-structure>`
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols, None, True, True)
        messageHash = 'liquidations'
        messageHashes = []
        if symbols is not None:
            for i in range(0, len(symbols)):
                symbol = symbols[i]
                messageHashes.append(messageHash + '::' + symbol)
        else:
            messageHashes.append(messageHash)
        market = self.get_market_from_symbols(symbols)
        type = None
        type, params = self.handle_market_type_and_params('watchliquidationsForSymbols', market, params)
        channel = 'liquidation-orders'
        if type == 'spot':
            type = 'SWAP'
        elif type == 'future':
            type = 'futures'
        uppercaseType = type.upper()
        request = {
            'op': 'subscribe',
            'args': [
                {
                    'channel': channel,
                    'instType': uppercaseType,
                },
            ],
        }
        url = self.get_url(channel, 'public')
        newLiquidations = await self.watch_multiple(url, messageHashes, request, messageHashes)
        if self.newUpdates:
            return newLiquidations
        return self.filter_by_symbols_since_limit(self.liquidations, symbols, since, limit, True)

    def handle_liquidation(self, client: Client, message):
        #
        #    {
        #        "arg": {
        #            "channel": "liquidation-orders",
        #            "instType": "SWAP"
        #        },
        #        "data": [
        #            {
        #                "details": [
        #                    {
        #                        "bkLoss": "0",
        #                        "bkPx": "0.007831",
        #                        "ccy": "",
        #                        "posSide": "short",
        #                        "side": "buy",
        #                        "sz": "13",
        #                        "ts": "1692266434010"
        #                    }
        #                ],
        #                "instFamily": "IOST-USDT",
        #                "instId": "IOST-USDT-SWAP",
        #                "instType": "SWAP",
        #                "uly": "IOST-USDT"
        #            }
        #        ]
        #    }
        #
        rawLiquidations = self.safe_list(message, 'data', [])
        for i in range(0, len(rawLiquidations)):
            rawLiquidation = rawLiquidations[i]
            liquidation = self.parse_ws_liquidation(rawLiquidation)
            symbol = self.safe_string(liquidation, 'symbol')
            if self.liquidations is None:
                limit = self.safe_integer(self.options, 'liquidationsLimit', 1000)
                self.liquidations = ArrayCache(limit)
            cache = self.liquidations
            cache.append(liquidation)
            client.resolve([liquidation], 'liquidations')
            client.resolve([liquidation], 'liquidations::' + symbol)

    async def watch_my_liquidations_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Liquidation]:
        """
        watch the private liquidations of a trading pair

        https://www.okx.com/docs-v5/en/#trading-account-websocket-balance-and-position-channel

        :param str[] symbols:
        :param int [since]: the earliest time in ms to fetch liquidations for
        :param int [limit]: the maximum number of liquidation structures to retrieve
        :param dict [params]: exchange specific parameters for the okx api endpoint
        :returns dict: an array of `liquidation structures <https://github.com/ccxt/ccxt/wiki/Manual#liquidation-structure>`
        """
        await self.load_markets()
        isTrigger = self.safe_value_2(params, 'stop', 'trigger', False)
        params = self.omit(params, ['stop', 'trigger'])
        await self.authenticate({'access': 'business' if isTrigger else 'private'})
        symbols = self.market_symbols(symbols, None, True, True)
        messageHash = 'myLiquidations'
        messageHashes = []
        if symbols is not None:
            for i in range(0, len(symbols)):
                symbol = symbols[i]
                messageHashes.append(messageHash + '::' + symbol)
        else:
            messageHashes.append(messageHash)
        channel = 'balance_and_position'
        request: dict = {
            'op': 'subscribe',
            'args': [
                {
                    'channel': channel,
                },
            ],
        }
        url = self.get_url(channel, 'private')
        newLiquidations = await self.watch_multiple(url, messageHashes, self.deep_extend(request, params), messageHashes)
        if self.newUpdates:
            return newLiquidations
        return self.filter_by_symbols_since_limit(self.liquidations, symbols, since, limit, True)

    def handle_my_liquidation(self, client: Client, message):
        #
        #    {
        #        "arg": {
        #            "channel": "balance_and_position",
        #            "uid": "77982378738415879"
        #        },
        #        "data": [{
        #            "pTime": "1597026383085",
        #            "eventType": "snapshot",
        #            "balData": [{
        #                "ccy": "BTC",
        #                "cashBal": "1",
        #                "uTime": "1597026383085"
        #            }],
        #            "posData": [{
        #                "posId": "1111111111",
        #                "tradeId": "2",
        #                "instId": "BTC-USD-191018",
        #                "instType": "FUTURES",
        #                "mgnMode": "cross",
        #                "posSide": "long",
        #                "pos": "10",
        #                "ccy": "BTC",
        #                "posCcy": "",
        #                "avgPx": "3320",
        #                "uTIme": "1597026383085"
        #            }],
        #            "trades": [{
        #                "instId": "BTC-USD-191018",
        #                "tradeId": "2",
        #            }]
        #        }]
        #    }
        #
        rawLiquidations = self.safe_list(message, 'data', [])
        for i in range(0, len(rawLiquidations)):
            rawLiquidation = rawLiquidations[i]
            eventType = self.safe_string(rawLiquidation, 'eventType')
            if eventType != 'liquidation':
                return
            liquidation = self.parse_ws_my_liquidation(rawLiquidation)
            symbol = self.safe_string(liquidation, 'symbol')
            if self.liquidations is None:
                limit = self.safe_integer(self.options, 'liquidationsLimit', 1000)
                self.liquidations = ArrayCache(limit)
            cache = self.liquidations
            cache.append(liquidation)
            client.resolve([liquidation], 'myLiquidations')
            client.resolve([liquidation], 'myLiquidations::' + symbol)

    def parse_ws_my_liquidation(self, liquidation, market=None):
        #
        #    {
        #        "pTime": "1597026383085",
        #        "eventType": "snapshot",
        #        "balData": [{
        #            "ccy": "BTC",
        #            "cashBal": "1",
        #            "uTime": "1597026383085"
        #        }],
        #        "posData": [{
        #            "posId": "1111111111",
        #            "tradeId": "2",
        #            "instId": "BTC-USD-191018",
        #            "instType": "FUTURES",
        #            "mgnMode": "cross",
        #            "posSide": "long",
        #            "pos": "10",
        #            "ccy": "BTC",
        #            "posCcy": "",
        #            "avgPx": "3320",
        #            "uTIme": "1597026383085"
        #        }],
        #        "trades": [{
        #            "instId": "BTC-USD-191018",
        #            "tradeId": "2",
        #        }]
        #    }
        #
        posData = self.safe_list(liquidation, 'posData', [])
        firstPosData = self.safe_dict(posData, 0, {})
        marketId = self.safe_string(firstPosData, 'instId')
        market = self.safe_market(marketId, market)
        timestamp = self.safe_integer(firstPosData, 'uTIme')
        return self.safe_liquidation({
            'info': liquidation,
            'symbol': self.safe_symbol(marketId, market),
            'contracts': self.safe_number(firstPosData, 'pos'),
            'contractSize': self.safe_number(market, 'contractSize'),
            'price': self.safe_number(liquidation, 'avgPx'),
            'baseValue': None,
            'quoteValue': None,
            'timestamp': timestamp,
            'datetime': self.iso8601(timestamp),
        })

    def parse_ws_liquidation(self, liquidation, market=None):
        #
        # public liquidation
        #    {
        #        "details": [
        #            {
        #                "bkLoss": "0",
        #                "bkPx": "0.007831",
        #                "ccy": "",
        #                "posSide": "short",
        #                "side": "buy",
        #                "sz": "13",
        #                "ts": "1692266434010"
        #            }
        #        ],
        #        "instFamily": "IOST-USDT",
        #        "instId": "IOST-USDT-SWAP",
        #        "instType": "SWAP",
        #        "uly": "IOST-USDT"
        #    }
        #
        details = self.safe_list(liquidation, 'details', [])
        liquidationDetails = self.safe_dict(details, 0, {})
        marketId = self.safe_string(liquidation, 'instId')
        market = self.safe_market(marketId, market)
        timestamp = self.safe_integer(liquidationDetails, 'ts')
        return self.safe_liquidation({
            'info': liquidation,
            'symbol': self.safe_symbol(marketId, market),
            'contracts': self.safe_number(liquidationDetails, 'sz'),
            'contractSize': self.safe_number(market, 'contractSize'),
            'price': self.safe_number(liquidationDetails, 'bkPx'),
            'side': self.safe_string(liquidationDetails, 'side'),
            'baseValue': None,
            'quoteValue': None,
            'timestamp': timestamp,
            'datetime': self.iso8601(timestamp),
        })

    async def watch_ohlcv(self, symbol: str, timeframe: str = '1m', since: Int = None, limit: Int = None, params={}) -> List[list]:
        """
        watches historical candlestick data containing the open, high, low, and close price, and the volume of a market
        :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)
        interval = self.safe_string(self.timeframes, timeframe, timeframe)
        name = 'candle' + interval
        ohlcv = await self.subscribe('public', name, name, symbol, 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:
        """
        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 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()
        return await self.un_watch_ohlcv_for_symbols([[symbol, timeframe]], params)

    async def watch_ohlcv_for_symbols(self, symbolsAndTimeframes: List[List[str]], since: Int = None, limit: Int = None, params={}):
        """
        watches historical candlestick data containing the open, high, low, and close price, and the volume of a market
        :param str[][] symbolsAndTimeframes: array of arrays containing unified symbols and timeframes to fetch OHLCV data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]
        :param int [since]: timestamp in ms of the earliest candle to fetch
        :param int [limit]: the maximum amount of candles to fetch
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns int[][]: A list of candles ordered, open, high, low, close, volume
        """
        symbolsLength = len(symbolsAndTimeframes)
        if symbolsLength == 0 or not isinstance(symbolsAndTimeframes[0], list):
            raise ArgumentsRequired(self.id + " watchOHLCVForSymbols() requires a an array of symbols and timeframes, like  [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]")
        await self.load_markets()
        topics = []
        messageHashes = []
        for i in range(0, len(symbolsAndTimeframes)):
            symbolAndTimeframe = symbolsAndTimeframes[i]
            sym = symbolAndTimeframe[0]
            tf = symbolAndTimeframe[1]
            marketId = self.market_id(sym)
            interval = self.safe_string(self.timeframes, tf, tf)
            channel = 'candle' + interval
            topic: dict = {
                'channel': channel,
                'instId': marketId,
            }
            topics.append(topic)
            messageHashes.append('multi:' + channel + ':' + sym)
        request: dict = {
            'op': 'subscribe',
            'args': topics,
        }
        url = self.get_url('candle', 'public')
        symbol, timeframe, candles = await self.watch_multiple(url, messageHashes, request, messageHashes)
        if self.newUpdates:
            limit = candles.getLimit(symbol, limit)
        filtered = self.filter_by_since_limit(candles, since, limit, 0, True)
        return self.create_ohlcv_object(symbol, timeframe, filtered)

    async def un_watch_ohlcv_for_symbols(self, symbolsAndTimeframes: List[List[str]], params={}) -> Any:
        """
        unWatches historical candlestick data containing the open, high, low, and close price, and the volume of a market
        :param str[][] symbolsAndTimeframes: array of arrays containing unified symbols and timeframes to fetch OHLCV data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns int[][]: A list of candles ordered, open, high, low, close, volume
        """
        symbolsLength = len(symbolsAndTimeframes)
        if symbolsLength == 0 or not isinstance(symbolsAndTimeframes[0], list):
            raise ArgumentsRequired(self.id + " watchOHLCVForSymbols() requires a an array of symbols and timeframes, like  [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]")
        await self.load_markets()
        topics = []
        messageHashes = []
        for i in range(0, len(symbolsAndTimeframes)):
            symbolAndTimeframe = symbolsAndTimeframes[i]
            sym = symbolAndTimeframe[0]
            tf = symbolAndTimeframe[1]
            marketId = self.market_id(sym)
            interval = self.safe_string(self.timeframes, tf, tf)
            channel = 'candle' + interval
            topic: dict = {
                'channel': channel,
                'instId': marketId,
            }
            topics.append(topic)
            messageHashes.append('unsubscribe:multi:' + channel + ':' + sym)
        request: dict = {
            'op': 'unsubscribe',
            'args': topics,
        }
        url = self.get_url('candle', 'public')
        return await self.watch_multiple(url, messageHashes, request, messageHashes)

    def handle_ohlcv(self, client: Client, message):
        #
        #     {
        #         "arg": {channel: "candle1m", instId: "BTC-USDT"},
        #         "data": [
        #             [
        #                 "1626690720000",
        #                 "31334",
        #                 "31334",
        #                 "31334",
        #                 "31334",
        #                 "0.0077",
        #                 "241.2718"
        #             ]
        #         ]
        #     }
        #
        arg = self.safe_value(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        data = self.safe_value(message, 'data', [])
        marketId = self.safe_string(arg, 'instId')
        market = self.safe_market(marketId)
        symbol = market['symbol']
        interval = channel.replace('candle', '')
        # use a reverse lookup in a static map instead
        timeframe = self.find_timeframe(interval)
        for i in range(0, len(data)):
            parsed = self.parse_ohlcv(data[i], market)
            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 + ':' + market['id']
            client.resolve(stored, messageHash)
            # for multiOHLCV we need special object, to other "multi"
            # methods, because OHLCV response item does not contain symbol
            # or timeframe, thus otherwise it would be unrecognizable
            messageHashForMulti = 'multi:' + channel + ':' + symbol
            client.resolve([symbol, timeframe, stored], messageHashForMulti)

    async def watch_order_book(self, symbol: str, limit: Int = None, params={}) -> OrderBook:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-order-book-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.depth]: okx order book depth, can be books, books5, books-l2-tbt, books50-l2-tbt, bbo-tbt
        :returns dict: A dictionary of `order book structures <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        #
        # bbo-tbt
        # 1. Newly added channel that sends tick-by-tick Level 1 data
        # 2. All API users can subscribe
        # 3. Public depth channel, verification not required
        #
        # books-l2-tbt
        # 1. Only users who're VIP5 and above can subscribe
        # 2. Identity verification required before subscription
        #
        # books50-l2-tbt
        # 1. Only users who're VIP4 and above can subscribe
        # 2. Identity verification required before subscription
        #
        # books
        # 1. All API users can subscribe
        # 2. Public depth channel, verification not required
        #
        # books5
        # 1. All API users can subscribe
        # 2. Public depth channel, verification not required
        # 3. Data feeds will be delivered every 100ms(vs. every 200ms now)
        #
        return await self.watch_order_book_for_symbols([symbol], limit, params)

    async def watch_order_book_for_symbols(self, symbols: List[str], limit: Int = None, params={}) -> OrderBook:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-order-book-channel

        watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data
        :param str[] symbols: unified array of symbols
        :param int [limit]: 1,5, 400, 50(l2-tbt, vip4+) or 40000(vip5+) the maximum amount of order book entries to return
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.depth]: okx order book depth, can be books, books5, books-l2-tbt, books50-l2-tbt, bbo-tbt
        :returns dict: A dictionary of `order book structures <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols)
        depth = None
        depth, params = self.handle_option_and_params(params, 'watchOrderBook', 'depth', 'books')
        if limit is not None:
            if limit == 1:
                depth = 'bbo-tbt'
            elif limit > 1 and limit <= 5:
                depth = 'books5'
            elif limit == 50:
                depth = 'books50-l2-tbt'  # Make sure you have VIP4 and above
            elif limit == 400:
                depth = 'books'
        if (depth == 'books-l2-tbt') or (depth == 'books50-l2-tbt'):
            if not self.check_required_credentials(False):
                raise AuthenticationError(self.id + ' watchOrderBook/watchOrderBookForSymbols requires authentication for self depth. Add credentials or change the depth option to books or books5')
            await self.authenticate({'access': 'public'})
        topics = []
        messageHashes = []
        for i in range(0, len(symbols)):
            symbol = symbols[i]
            messageHashes.append(depth + ':' + symbol)
            marketId = self.market_id(symbol)
            topic: dict = {
                'channel': depth,
                'instId': marketId,
            }
            topics.append(topic)
        request: dict = {
            'op': 'subscribe',
            'args': topics,
        }
        url = self.get_url(depth, 'public')
        orderbook = await self.watch_multiple(url, messageHashes, request, messageHashes)
        return orderbook.limit()

    async def un_watch_order_book_for_symbols(self, symbols: List[str], params={}) -> Any:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-order-book-channel

        unWatches information on open orders with bid(buy) and ask(sell) prices, volumes and other data
        :param str[] symbols: unified array of symbols
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param int [params.limit]: the maximum amount of order book entries to return
        :param str [params.depth]: okx order book depth, can be books, books5, books-l2-tbt, books50-l2-tbt, bbo-tbt
        :returns dict: A dictionary of `order book structures <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        await self.load_markets()
        symbols = self.market_symbols(symbols, None, False)
        depth = None
        depth, params = self.handle_option_and_params(params, 'watchOrderBook', 'depth', 'books')
        limit = self.safe_integer(params, 'limit')
        if limit is not None:
            if limit == 1:
                depth = 'bbo-tbt'
            elif limit > 1 and limit <= 5:
                depth = 'books5'
            elif limit == 50:
                depth = 'books50-l2-tbt'  # Make sure you have VIP4 and above
            elif limit == 400:
                depth = 'books'
        topics = []
        subMessageHashes = []
        messageHashes = []
        for i in range(0, len(symbols)):
            symbol = symbols[i]
            subMessageHashes.append(depth + ':' + symbol)
            messageHashes.append('unsubscribe:orderbook:' + symbol)
            marketId = self.market_id(symbol)
            topic: dict = {
                'channel': depth,
                'instId': marketId,
            }
            topics.append(topic)
        request: dict = {
            'op': 'unsubscribe',
            'args': topics,
        }
        url = self.get_url(depth, 'public')
        return await self.watch_multiple(url, messageHashes, request, messageHashes)

    async def un_watch_order_book(self, symbol: str, params={}) -> Any:
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-order-book-channel

        unWatches information on open orders with bid(buy) and ask(sell) prices, volumes and other data
        :param str symbol: unified array of symbols
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param int [params.limit]: the maximum amount of order book entries to return
        :param str [params.depth]: okx order book depth, can be books, books5, books-l2-tbt, books50-l2-tbt, bbo-tbt
        :returns dict: A dictionary of `order book structures <https://docs.ccxt.com/?id=order-book-structure>` indexed by market symbols
        """
        return await self.un_watch_order_book_for_symbols([symbol], params)

    def handle_delta(self, bookside, delta):
        #
        #     [
        #         "31685",  # price
        #         "0.78069158",  # amount
        #         "0",  # liquidated orders
        #         "17"  # orders
        #     ]
        #
        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, messageHash, market=None):
        #
        #     {
        #         "asks": [
        #             ['31738.3', '0.05973179', "0", "3"],
        #             ['31738.5', '0.11035404', "0", "2"],
        #             ['31739.6', '0.01', "0", "1"],
        #         ],
        #         "bids": [
        #             ['31738.2', '0.67557666', "0", "9"],
        #             ['31738', '0.02466947', "0", "2"],
        #             ['31736.3', '0.01705046', "0", "2"],
        #         ],
        #         "instId": "BTC-USDT",
        #         "ts": "1626537446491"
        #         "checksum": -855196043,
        #         "prevSeqId": 123456,
        #         "seqId": 123457
        #     }
        #
        asks = self.safe_value(message, 'asks', [])
        bids = self.safe_value(message, 'bids', [])
        storedAsks = orderbook['asks']
        storedBids = orderbook['bids']
        self.handle_deltas(storedAsks, asks)
        self.handle_deltas(storedBids, bids)
        marketId = self.safe_string(message, 'instId')
        symbol = self.safe_symbol(marketId, market)
        seqId = self.safe_integer(message, 'seqId')
        prevSeqId = self.safe_integer(message, 'prevSeqId')
        nonce = orderbook['nonce']
        error = None
        if prevSeqId is not None and prevSeqId != -1 and nonce != prevSeqId:
            error = InvalidNonce(self.id + ' watchOrderBook received invalid nonce')
        if error is not None:
            del client.subscriptions[messageHash]
            if symbol is not None:
                del self.orderbooks[symbol]
            client.reject(error, messageHash)
        timestamp = self.safe_integer(message, 'ts')
        orderbook['nonce'] = seqId
        orderbook['timestamp'] = timestamp
        orderbook['datetime'] = self.iso8601(timestamp)
        return orderbook

    def handle_order_book(self, client: Client, message):
        #
        # snapshot
        #
        #     {
        #         "arg": {channel: 'books-l2-tbt', instId: "BTC-USDT"},
        #         "action": "snapshot",
        #         "data": [
        #             {
        #                 "asks": [
        #                     ['31685', '0.78069158', "0", "17"],
        #                     ['31685.1', '0.0001', "0", "1"],
        #                     ['31685.6', '0.04543165', "0", "1"],
        #                 ],
        #                 "bids": [
        #                     ['31684.9', '0.01', "0", "1"],
        #                     ['31682.9', '0.0001', "0", "1"],
        #                     ['31680.7', '0.01', "0", "1"],
        #                 ],
        #                 "ts": "1626532416403",
        #                 "checksum": -1023440116
        #             }
        #         ]
        #     }
        #
        # update
        #
        #     {
        #         "arg": {channel: 'books-l2-tbt', instId: "BTC-USDT"},
        #         "action": "update",
        #         "data": [
        #             {
        #                 "asks": [
        #                     ['31657.7', '0', "0", "0"],
        #                     ['31659.7', '0.01', "0", "1"],
        #                     ['31987.3', '0.01', "0", "1"]
        #                 ],
        #                 "bids": [
        #                     ['31642.9', '0.50296385', "0", "4"],
        #                     ['31639.9', '0', "0", "0"],
        #                     ['31638.7', '0.01', "0", "1"],
        #                 ],
        #                 "ts": "1626535709008",
        #                 "checksum": 830931827
        #             }
        #         ]
        #     }
        #
        # books5
        #
        #     {
        #         "arg": {channel: "books5", instId: "BTC-USDT"},
        #         "data": [
        #             {
        #                 "asks": [
        #                     ['31738.3', '0.05973179', "0", "3"],
        #                     ['31738.5', '0.11035404', "0", "2"],
        #                     ['31739.6', '0.01', "0", "1"],
        #                 ],
        #                 "bids": [
        #                     ['31738.2', '0.67557666', "0", "9"],
        #                     ['31738', '0.02466947', "0", "2"],
        #                     ['31736.3', '0.01705046', "0", "2"],
        #                 ],
        #                 "instId": "BTC-USDT",
        #                 "ts": "1626537446491"
        #             }
        #         ]
        #     }
        #
        # bbo-tbt
        #
        #     {
        #         "arg":{
        #             "channel":"bbo-tbt",
        #             "instId":"BTC-USDT"
        #         },
        #         "data":[
        #             {
        #                 "asks":[["36232.2","1.8826134","0","17"]],
        #                 "bids":[["36232.1","0.00572212","0","2"]],
        #                 "ts":"1651826598363"
        #             }
        #         ]
        #     }
        #
        arg = self.safe_dict(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        action = self.safe_string(message, 'action')
        data = self.safe_list(message, 'data', [])
        marketId = self.safe_string(arg, 'instId')
        market = self.safe_market(marketId)
        symbol = market['symbol']
        depths: dict = {
            'bbo-tbt': 1,
            'books': 400,
            'books5': 5,
            'books-l2-tbt': 400,
            'books50-l2-tbt': 50,
        }
        limit = self.safe_integer(depths, channel)
        messageHash = channel + ':' + symbol
        if action == 'snapshot':
            for i in range(0, len(data)):
                update = data[i]
                orderbook = self.order_book({}, limit)
                self.orderbooks[symbol] = orderbook
                orderbook['symbol'] = symbol
                self.handle_order_book_message(client, update, orderbook, messageHash)
                client.resolve(orderbook, messageHash)
        elif action == 'update':
            if symbol in self.orderbooks:
                orderbook = self.orderbooks[symbol]
                for i in range(0, len(data)):
                    update = data[i]
                    self.handle_order_book_message(client, update, orderbook, messageHash, market)
                    client.resolve(orderbook, messageHash)
        elif (channel == 'books5') or (channel == 'bbo-tbt'):
            if not (symbol in self.orderbooks):
                self.orderbooks[symbol] = self.order_book({}, limit)
            orderbook = self.orderbooks[symbol]
            for i in range(0, len(data)):
                update = data[i]
                timestamp = self.safe_integer(update, 'ts')
                snapshot = self.parse_order_book(update, symbol, timestamp, 'bids', 'asks', 0, 1)
                orderbook.reset(snapshot)
                client.resolve(orderbook, messageHash)
        return message

    async def authenticate(self, params={}):
        self.check_required_credentials()
        access = self.safe_string(params, 'access', 'private')
        params = self.omit(params, ['access'])
        url = self.get_url('users', access)
        messageHash = 'authenticated'
        client = self.client(url)
        future = client.reusableFuture(messageHash)
        authenticated = self.safe_value(client.subscriptions, messageHash)
        if authenticated is None:
            timestamp = str(self.seconds())
            method = 'GET'
            path = '/users/self/verify'
            auth = timestamp + method + path
            signature = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256, 'base64')
            operation = 'login'
            request: dict = {
                'op': operation,
                'args': [
                    {
                        'apiKey': self.apiKey,
                        'passphrase': self.password,
                        'timestamp': timestamp,
                        'sign': signature,
                    },
                ],
            }
            # Only add params['access'] to prevent sending custom parameters, such.
            if 'access' in params:
                request['access'] = params['access']
            self.watch(url, messageHash, request, messageHash)
        return await future

    async def watch_balance(self, params={}) -> Balances:
        """

        https://www.okx.com/docs-v5/en/#trading-account-websocket-account-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()
        await self.authenticate()
        return await self.subscribe('private', 'account', 'account', None, params)

    def handle_balance_and_position(self, client: Client, message):
        self.handle_my_liquidation(client, message)

    def handle_balance(self, client: Client, message):
        #
        #     {
        #         arg: {
        #             channel: 'account',
        #             uid: '158635852459810816'
        #         },
        #         eventType: 'snapshot',
        #         curPage: 1,
        #         lastPage: True,
        #         data: [
        #             {
        #                 adjEq: '100942.13298468884',
        #                 availEq: '98461.99074635761',
        #                 borrowFroz: '409.3702076072169',
        #                 delta: '',
        #                 deltaLever: '',
        #                 deltaNeutralStatus: '',
        #                 details: [
        #                     {
        #                         accAvgPx: '',
        #                         autoLendAmt: '0',
        #                         autoLendMtAmt: '0',
        #                         autoLendStatus: 'unsupported',
        #                         autoStakingStatus: 'unsupported',
        #                         availBal: '2900',
        #                         availEq: '2900',
        #                         borrowFroz: '0',
        #                         cashBal: '2900',
        #                         ccy: 'TUSD',
        #                         clSpotInUseAmt: '',
        #                         coinUsdPrice: '200.026',
        #                         colBorrAutoConversion: '0',
        #                         colRes: '0',
        #                         collateralEnabled: False,
        #                         collateralRestrict: False,
        #                         crossLiab: '0',
        #                         disEq: '0',
        #                         eq: '2900',
        #                         eqUsd: '580075.4',
        #                         fixedBal: '0',
        #                         frozenBal: '0',
        #                         frpType: '0',
        #                         imr: '',
        #                         interest: '0',
        #                         isoEq: '0',
        #                         isoLiab: '0',
        #                         isoUpl: '0',
        #                         liab: '0',
        #                         maxLoan: '0',
        #                         maxSpotInUseAmt: '',
        #                         mgnRatio: '',
        #                         mmr: '',
        #                         notionalLever: '',
        #                         openAvgPx: '',
        #                         ordFrozen: '0',
        #                         rewardBal: '',
        #                         smtSyncEq: '0',
        #                         spotBal: '',
        #                         spotCopyTradingEq: '0',
        #                         spotInUseAmt: '',
        #                         spotIsoBal: '0',
        #                         spotUpl: '',
        #                         spotUplRatio: '',
        #                         stgyEq: '0',
        #                         totalPnl: '',
        #                         totalPnlRatio: '',
        #                         twap: '0',
        #                         uTime: '1752243427083',
        #                         upl: '0',
        #                         uplLiab: '0'
        #                     }
        #                 ],
        #                 imr: '2480.1422383312265',
        #                 isoEq: '0',
        #                 mgnRatio: '1814.5924297007082',
        #                 mmr: '52.196024182958055',
        #                 notionalUsd: '4634.0971302081125',
        #                 notionalUsdForBorrow: '2046.8510380360844',
        #                 notionalUsdForFutures: '101.11322817202809',
        #                 notionalUsdForOption: '0',
        #                 notionalUsdForSwap: '2486.1328640000006',
        #                 ordFroz: '1208.3566666666668',
        #                 totalEq: '711018.8951315446',
        #                 uTime: '1773837554158',
        #                 upl: '9.244336372701904'
        #             }
        #         ]
        #     }
        #
        arg = self.safe_value(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        balance = self.parseTradingBalance(message)
        newBalance = self.deep_extend(self.balance, balance)
        self.balance = self.safe_balance(newBalance)
        client.resolve(self.balance, channel)

    def order_to_trade(self, order, market=None):
        info = self.safe_value(order, 'info', {})
        timestamp = self.safe_integer(info, 'fillTime')
        feeMarketId = self.safe_string(info, 'fillFeeCcy')
        isTaker = self.safe_string(info, 'execType', '') == 'T'
        return self.safe_trade({
            'info': info,
            'timestamp': timestamp,
            'datetime': self.iso8601(timestamp),
            'symbol': self.safe_string(order, 'symbol'),
            'id': self.safe_string(info, 'tradeId'),
            'order': self.safe_string(order, 'id'),
            'type': self.safe_string(order, 'type'),
            'takerOrMaker': 'taker' if (isTaker) else 'maker',
            'side': self.safe_string(order, 'side'),
            'price': self.safe_number(info, 'fillPx'),
            'amount': self.safe_number(info, 'fillSz'),
            'cost': self.safe_number(order, 'cost'),
            'fee': {
                'cost': self.safe_number(info, 'fillFee'),
                'currency': self.safe_currency_code(feeMarketId),
            },
        }, market)

    async def watch_my_trades(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]:
        """
        watches information on multiple trades made by the user

        https://www.okx.com/docs-v5/en/#order-book-trading-trade-ws-order-channel

        :param str [symbol]: unified market symbol of the market trades were made in
        :param int [since]: the earliest time in ms to fetch trades for
        :param int [limit]: the maximum number of trade structures to retrieve
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param bool [params.trigger]: True if fetching trigger or conditional trades
        :param str [params.type]: 'spot', 'swap', 'future', 'option', 'ANY', 'SPOT', 'MARGIN', 'SWAP', 'FUTURES' or 'OPTION'
        :param str [params.marginMode]: 'cross' or 'isolated', for automatically setting the type to spot margin
        :returns dict[]: a list of `trade structures <https://docs.ccxt.com/?id=trade-structure>`
        """
        # By default, receive order updates from any instrument type
        type = None
        type, params = self.handle_option_and_params(params, 'watchMyTrades', 'type', 'ANY')
        isTrigger = self.safe_bool_2(params, 'trigger', 'stop', False)
        params = self.omit(params, ['trigger', 'stop'])
        await self.load_markets()
        await self.authenticate({'access': 'business' if isTrigger else 'private'})
        channel = 'orders-algo' if isTrigger else 'orders'
        messageHash = channel + '::myTrades'
        market = None
        if symbol is not None:
            market = self.market(symbol)
            symbol = market['symbol']
            type = market['type']
            messageHash = messageHash + '::' + symbol
        if type == 'future':
            type = 'futures'
        uppercaseType = type.upper()
        marginMode = None
        marginMode, params = self.handle_margin_mode_and_params('watchMyTrades', params)
        if uppercaseType == 'SPOT':
            if marginMode is not None:
                uppercaseType = 'MARGIN'
        request: dict = {
            'instType': uppercaseType,
        }
        orders = await self.subscribe('private', messageHash, channel, None, self.extend(request, params))
        if self.newUpdates:
            limit = orders.getLimit(symbol, limit)
        return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True)

    async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]:
        """

        https://www.okx.com/docs-v5/en/#trading-account-websocket-positions-channel

        watch all open positions
        :param str[] [symbols]: list of unified market symbols
        :param int [since]: timestamp in ms of the earliest position to fetch
        :param int [limit]: the maximum number of positions to fetch
        :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()
        await self.authenticate(params)
        symbols = self.market_symbols(symbols)
        request: dict = {
            'instType': 'ANY',
        }
        channel = 'positions'
        newPositions = None
        if symbols is None:
            arg: dict = {
                'channel': 'positions',
                'instType': 'ANY',
            }
            args = [self.extend(arg, params)]
            nonSymbolRequest: dict = {
                'op': 'subscribe',
                'args': args,
            }
            url = self.get_url(channel, 'private')
            newPositions = await self.watch(url, channel, nonSymbolRequest, channel)
        else:
            newPositions = await self.subscribe_multiple('private', channel, symbols, self.extend(request, params))
        if self.newUpdates:
            return newPositions
        return self.filter_by_symbols_since_limit(self.positions, symbols, since, limit, True)

    def handle_positions(self, client, message):
        #
        #    {
        #        arg: {
        #            channel: 'positions',
        #            instType: 'ANY',
        #            instId: 'XRP-USDT-SWAP',
        #            uid: '464737184507959869'
        #        },
        #        data: [{
        #            adl: '1',
        #            availPos: '',
        #            avgPx: '0.52668',
        #            baseBal: '',
        #            baseBorrowed: '',
        #            baseInterest: '',
        #            bizRefId: '',
        #            bizRefType: '',
        #            cTime: '1693151444408',
        #            ccy: 'USDT',
        #            closeOrderAlgo: [],
        #            deltaBS: '',
        #            deltaPA: '',
        #            gammaBS: '',
        #            gammaPA: '',
        #            idxPx: '0.52683',
        #            imr: '17.564000000000004',
        #            instId: 'XRP-USDT-SWAP',
        #            instType: 'SWAP',
        #            interest: '',
        #            last: '0.52691',
        #            lever: '3',
        #            liab: '',
        #            liabCcy: '',
        #            liqPx: '0.3287514731020614',
        #            margin: '',
        #            markPx: '0.52692',
        #            mgnMode: 'cross',
        #            mgnRatio: '69.00363001456147',
        #            mmr: '0.26346',
        #            notionalUsd: '52.68620388000001',
        #            optVal: '',
        #            pTime: '1693151906023',
        #            pendingCloseOrdLiabVal: '',
        #            pos: '1',
        #            posCcy: '',
        #            posId: '616057041198907393',
        #            posSide: 'net',
        #            quoteBal: '',
        #            quoteBorrowed: '',
        #            quoteInterest: '',
        #            spotInUseAmt: '',
        #            spotInUseCcy: '',
        #            thetaBS: '',
        #            thetaPA: '',
        #            tradeId: '138745402',
        #            uTime: '1693151444408',
        #            upl: '0.0240000000000018',
        #            uplLastPx: '0.0229999999999952',
        #            uplRatio: '0.0013670539986328',
        #            uplRatioLastPx: '0.001310093415356',
        #            usdPx: '',
        #            vegaBS: '',
        #            vegaPA: ''
        #        }]
        #    }
        #
        arg = self.safe_value(message, 'arg', {})
        marketId = self.safe_string(arg, 'instId')
        market = self.safe_market(marketId, None, '-')
        symbol = market['symbol']
        channel = self.safe_string(arg, 'channel', '')
        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_position(rawPosition)
            if position['contracts'] == 0 and rawPosition['posSide'] == 'net':
                position['side'] = 'long'
                shortPosition = self.clone(position)
                shortPosition['side'] = 'short'
                cache.append(shortPosition)
                newPositions.append(shortPosition)
            newPositions.append(position)
            cache.append(position)
        messageHash = channel
        if symbol is not None:
            messageHash = channel + '::' + symbol
        client.resolve(newPositions, messageHash)

    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://www.okx.com/docs-v5/en/#order-book-trading-trade-ws-order-channel

        :param str [symbol]: unified market symbol of the market the orders were made in
        :param int [since]: the earliest time in ms to fetch orders for
        :param int [limit]: the maximum number of order structures to retrieve
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param bool [params.trigger]: True if fetching trigger or conditional orders
        :param str [params.type]: 'spot', 'swap', 'future', 'option', 'ANY', 'SPOT', 'MARGIN', 'SWAP', 'FUTURES' or 'OPTION'
        :param str [params.marginMode]: 'cross' or 'isolated', for automatically setting the type to spot margin
        :returns dict[]: a list of `order structures <https://docs.ccxt.com/?id=order-structure>`
        """
        type = None
        # By default, receive order updates from any instrument type
        type, params = self.handle_option_and_params(params, 'watchOrders', 'type', 'ANY')
        isTrigger = self.safe_value_2(params, 'stop', 'trigger', False)
        params = self.omit(params, ['stop', 'trigger'])
        await self.load_markets()
        await self.authenticate({'access': 'business' if isTrigger else 'private'})
        market = None
        if symbol is not None:
            market = self.market(symbol)
            symbol = market['symbol']
            type = market['type']
        if type == 'future':
            type = 'futures'
        uppercaseType = type.upper()
        marginMode = None
        marginMode, params = self.handle_margin_mode_and_params('watchOrders', params)
        if uppercaseType == 'SPOT':
            if marginMode is not None:
                uppercaseType = 'MARGIN'
        request: dict = {
            'instType': uppercaseType,
        }
        channel = 'orders-algo' if isTrigger else 'orders'
        orders = await self.subscribe('private', channel, channel, symbol, self.extend(request, params))
        if self.newUpdates:
            limit = orders.getLimit(symbol, limit)
        return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True)

    def handle_orders(self, client: Client, message):
        #
        #     {
        #         "arg":{
        #             "channel":"orders",
        #             "instType":"SPOT"
        #         },
        #         "data":[
        #             {
        #                 "accFillSz":"0",
        #                 "amendResult":"",
        #                 "avgPx":"",
        #                 "cTime":"1634548275191",
        #                 "category":"normal",
        #                 "ccy":"",
        #                 "clOrdId":"e847386590ce4dBC330547db94a08ba0",
        #                 "code":"0",
        #                 "execType":"",
        #                 "fee":"0",
        #                 "feeCcy":"USDT",
        #                 "fillFee":"0",
        #                 "fillFeeCcy":"",
        #                 "fillNotionalUsd":"",
        #                 "fillPx":"",
        #                 "fillSz":"0",
        #                 "fillTime":"",
        #                 "instId":"ETH-USDT",
        #                 "instType":"SPOT",
        #                 "lever":"",
        #                 "msg":"",
        #                 "notionalUsd":"451.4516256",
        #                 "ordId":"370257534141235201",
        #                 "ordType":"limit",
        #                 "pnl":"0",
        #                 "posSide":"",
        #                 "px":"60000",
        #                 "rebate":"0",
        #                 "rebateCcy":"ETH",
        #                 "reqId":"",
        #                 "side":"sell",
        #                 "slOrdPx":"",
        #                 "slTriggerPx":"",
        #                 "state":"live",
        #                 "sz":"0.007526",
        #                 "tag":"",
        #                 "tdMode":"cash",
        #                 "tgtCcy":"",
        #                 "tpOrdPx":"",
        #                 "tpTriggerPx":"",
        #                 "tradeId":"",
        #                 "uTime":"1634548275191"
        #             }
        #         ]
        #     }
        #
        self.handle_my_trades(client, message)
        arg = self.safe_value(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        orders = self.safe_value(message, 'data', [])
        ordersLength = len(orders)
        if ordersLength > 0:
            limit = self.safe_integer(self.options, 'ordersLimit', 1000)
            if self.orders is None:
                self.orders = ArrayCacheBySymbolById(limit)
                self.triggerOrders = ArrayCacheBySymbolById(limit)
            stored = self.triggerOrders if (channel == 'orders-algo') else self.orders
            marketIds = []
            parsed = self.parse_orders(orders)
            for i in range(0, len(parsed)):
                order = parsed[i]
                stored.append(order)
                symbol = order['symbol']
                market = self.market(symbol)
                marketIds.append(market['id'])
            client.resolve(stored, channel)
            for i in range(0, len(marketIds)):
                messageHash = channel + ':' + marketIds[i]
                client.resolve(stored, messageHash)

    def handle_my_trades(self, client: Client, message):
        #
        #     {
        #         "arg":{
        #             "channel":"orders",
        #             "instType":"SPOT"
        #         },
        #         "data":[
        #             {
        #                 "accFillSz":"0",
        #                 "amendResult":"",
        #                 "avgPx":"",
        #                 "cTime":"1634548275191",
        #                 "category":"normal",
        #                 "ccy":"",
        #                 "clOrdId":"e847386590ce4dBC330547db94a08ba0",
        #                 "code":"0",
        #                 "execType":"",
        #                 "fee":"0",
        #                 "feeCcy":"USDT",
        #                 "fillFee":"0",
        #                 "fillFeeCcy":"",
        #                 "fillNotionalUsd":"",
        #                 "fillPx":"",
        #                 "fillSz":"0",
        #                 "fillTime":"",
        #                 "instId":"ETH-USDT",
        #                 "instType":"SPOT",
        #                 "lever":"",
        #                 "msg":"",
        #                 "notionalUsd":"451.4516256",
        #                 "ordId":"370257534141235201",
        #                 "ordType":"limit",
        #                 "pnl":"0",
        #                 "posSide":"",
        #                 "px":"60000",
        #                 "rebate":"0",
        #                 "rebateCcy":"ETH",
        #                 "reqId":"",
        #                 "side":"sell",
        #                 "slOrdPx":"",
        #                 "slTriggerPx":"",
        #                 "state":"live",
        #                 "sz":"0.007526",
        #                 "tag":"",
        #                 "tdMode":"cash",
        #                 "tgtCcy":"",
        #                 "tpOrdPx":"",
        #                 "tpTriggerPx":"",
        #                 "tradeId":"",
        #                 "uTime":"1634548275191"
        #             }
        #         ]
        #     }
        #
        arg = self.safe_value(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        rawOrders = self.safe_value(message, 'data', [])
        filteredOrders = []
        # filter orders with no last trade id
        for i in range(0, len(rawOrders)):
            rawOrder = rawOrders[i]
            tradeId = self.safe_string(rawOrder, 'tradeId', '')
            if len(tradeId) > 0:
                order = self.parse_order(rawOrder)
                filteredOrders.append(order)
        tradesLength = len(filteredOrders)
        if tradesLength == 0:
            return
        if self.myTrades is None:
            limit = self.safe_integer(self.options, 'tradesLimit', 1000)
            self.myTrades = ArrayCacheBySymbolById(limit)
        myTrades = self.myTrades
        symbols: dict = {}
        for i in range(0, len(filteredOrders)):
            rawTrade = filteredOrders[i]
            trade = self.order_to_trade(rawTrade)
            myTrades.append(trade)
            symbol = trade['symbol']
            symbols[symbol] = True
        messageHash = channel + '::myTrades'
        client.resolve(self.myTrades, messageHash)
        tradeSymbols = list(symbols.keys())
        for i in range(0, len(tradeSymbols)):
            symbolMessageHash = messageHash + '::' + tradeSymbols[i]
            client.resolve(self.myTrades, symbolMessageHash)

    def request_id(self):
        ts = str(self.milliseconds())
        randomNumber = self.rand_number(4)
        randomPart = str(randomNumber)
        return ts + randomPart

    async def create_order_ws(self, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}) -> Order:
        """

        https://www.okx.com/docs-v5/en/#websocket-api-trade-place-order

        create a trade order
        :param str symbol: unified symbol of the market to create an order in
        :param str type: 'market' or 'limit'
        :param str side: 'buy' or 'sell'
        :param float amount: how much of currency you want to trade in units of base currency
        :param float|None [price]: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param boolean params['test']: test order, default False
        :returns dict: an `order structure <https://docs.ccxt.com/?id=order-structure>`
        """
        await self.load_markets()
        await self.authenticate()
        url = self.get_url('private', 'private')
        messageHash = self.request_id()
        op = None
        op, params = self.handle_option_and_params(params, 'createOrderWs', 'op', 'batch-orders')
        args = self.create_order_request(symbol, type, side, amount, price, params)
        market = self.market(symbol)
        instIdCode = self.safe_integer(market, 'instIdCode')
        if instIdCode is not None:
            del args['instId']
            args['instIdCode'] = instIdCode
        ordType = self.safe_string(args, 'ordType')
        if (ordType == 'trigger') or (ordType == 'conditional') or (type == 'oco') or (type == 'move_order_stop') or (type == 'iceberg') or (type == 'twap'):
            raise BadRequest(self.id + ' createOrderWs() does not support algo trading. self.options["createOrderWs"]["op"] must be either order or batch-order')
        if (op != 'order') and (op != 'batch-orders'):
            raise BadRequest(self.id + ' createOrderWs() does not support algo trading. self.options["createOrderWs"]["op"] must be either order or privatePostTradeOrder or privatePostTradeOrderAlgo')
        request: dict = {
            'id': messageHash,
            'op': op,
            'args': [args],
        }
        return await self.watch(url, messageHash, request, messageHash)

    def handle_place_orders(self, client: Client, message):
        #
        #  batch-orders/order/cancel-order
        #    {
        #        "id": "1689281055",
        #        "op": "batch-orders",
        #        "code": "0",
        #        "msg": '',
        #        "data": [{
        #            "tag": "e847386590ce4dBC",
        #            "ordId": "599823446566084608",
        #            "clOrdId": "e847386590ce4dBCb939511604f394b0",
        #            "sCode": "0",
        #            "sMsg": "Order successfully placed."
        #        },
        #        ...
        #        ]
        #    }
        #
        messageHash = self.safe_string(message, 'id')
        args = self.safe_value(message, 'data', [])
        # filter out partial errors
        args = self.filter_by(args, 'sCode', '0')
        # if empty means request failed and handle error
        if self.is_empty(args):
            method = self.safe_string(message, 'op')
            stringMsg = self.json(message)
            self.handle_errors(1, '', client.url, method, {}, stringMsg, message, {}, {})
        orders = self.parse_orders(args, None, None, None)
        first = self.safe_dict(orders, 0, {})
        client.resolve(first, messageHash)

    async def edit_order_ws(self, id: str, symbol: str, type: OrderType, side: OrderSide, amount: Num = None, price: Num = None, params={}) -> Order:
        """
        edit a trade order

        https://www.okx.com/docs-v5/en/#order-book-trading-trade-ws-amend-order
        https://www.okx.com/docs-v5/en/#order-book-trading-trade-ws-amend-multiple-orders

        :param str id: order id
        :param str symbol: unified symbol of the market to create an order in
        :param str type: 'market' or 'limit'
        :param str side: 'buy' or 'sell'
        :param float amount: how much of the currency you want to trade in units of the base currency
        :param float|None [price]: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict: an `order structure <https://docs.ccxt.com/?id=order-structure>`
        """
        await self.load_markets()
        await self.authenticate()
        url = self.get_url('private', 'private')
        messageHash = self.request_id()
        op = None
        op, params = self.handle_option_and_params(params, 'editOrderWs', 'op', 'amend-order')
        args = self.edit_order_request(id, symbol, type, side, amount, price, params)
        market = self.market(symbol)
        instIdCode = self.safe_integer(market, 'instIdCode')
        if instIdCode is not None:
            del args['instId']
            args['instIdCode'] = instIdCode
        request: dict = {
            'id': messageHash,
            'op': op,
            'args': [args],
        }
        return await self.watch(url, messageHash, self.extend(request, params), messageHash)

    async def cancel_order_ws(self, id: str, symbol: Str = None, params={}) -> Order:
        """

        https://okx-docs.github.io/apidocs/websocket_api/en/#cancel-order-trade

        cancel multiple orders
        :param str id: order id
        :param str symbol: unified market symbol, default is None
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :param str [params.clOrdId]: client order id
        :returns dict: an list of `order structures <https://docs.ccxt.com/?id=order-structure>`
        """
        if symbol is None:
            raise BadRequest(self.id + ' cancelOrderWs() requires a symbol argument')
        await self.load_markets()
        await self.authenticate()
        url = self.get_url('private', 'private')
        messageHash = self.request_id()
        clientOrderId = self.safe_string_2(params, 'clOrdId', 'clientOrderId')
        params = self.omit(params, ['clientOrderId', 'clOrdId'])
        market = self.market(symbol)
        instIdCode = self.safe_integer(market, 'instIdCode')
        arg: dict = {
            'instIdCode': instIdCode,
        }
        if clientOrderId is not None:
            arg['clOrdId'] = clientOrderId
        else:
            arg['ordId'] = id
        request: dict = {
            'id': messageHash,
            'op': 'cancel-order',
            'args': [self.extend(arg, params)],
        }
        return await self.watch(url, messageHash, request, messageHash)

    async def cancel_orders_ws(self, ids: List[str], symbol: Str = None, params={}):
        """

        https://www.okx.com/docs-v5/en/#order-book-trading-trade-ws-mass-cancel-order

        cancel multiple orders
        :param str[] ids: order ids
        :param str symbol: unified market symbol, default is None
        :param dict [params]: extra parameters specific to the exchange API endpoint
        :returns dict: an list of `order structures <https://docs.ccxt.com/?id=order-structure>`
        """
        idsLength: number = len(ids)
        if idsLength > 20:
            raise BadRequest(self.id + ' cancelOrdersWs() accepts up to 20 ids at a time')
        if symbol is None:
            raise BadRequest(self.id + ' cancelOrdersWs() requires a symbol argument')
        await self.load_markets()
        await self.authenticate()
        url = self.get_url('private', 'private')
        messageHash = self.request_id()
        args = []
        market = self.market(symbol)
        instIdCode = self.safe_integer(market, 'instIdCode')
        instParams = {
            'instIdCode': instIdCode,
        }
        for i in range(0, idsLength):
            arg: dict = self.extend(instParams, {
                'ordId': ids[i],
            })
            args.append(arg)
        request: dict = {
            'id': messageHash,
            'op': 'batch-cancel-orders',
            'args': args,
        }
        return await self.watch(url, messageHash, self.deep_extend(request, params), messageHash)

    async def cancel_all_orders_ws(self, symbol: Str = None, params={}) -> List[Order]:
        """

        https://docs.okx.com/websockets/#message-cancelAll

        cancel all open orders of a type. Only applicable to Option in Portfolio Margin mode, and MMP privilege is required.
        :param str symbol: unified market symbol, only orders in the market of self symbol are cancelled when symbol is not None
        :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>`
        """
        if symbol is None:
            raise BadRequest(self.id + ' cancelAllOrdersWs() requires a symbol argument')
        await self.load_markets()
        await self.authenticate()
        market = self.market(symbol)
        if market['type'] != 'option':
            raise BadRequest(self.id + ' cancelAllOrdersWs is only applicable to Option in Portfolio Margin mode, and MMP privilege is required.')
        url = self.get_url('private', 'private')
        messageHash = self.request_id()
        request: dict = {
            'id': messageHash,
            'op': 'mass-cancel',
            'args': [self.extend({
                'instType': 'OPTION',
                'instFamily': market['id'],
            }, params)],
        }
        return await self.watch(url, messageHash, request, messageHash)

    def handle_cancel_all_orders(self, client: Client, message):
        #
        #    {
        #        "id": "1512",
        #        "op": "mass-cancel",
        #        "data": [
        #            {
        #                "result": True
        #            }
        #        ],
        #        "code": "0",
        #        "msg": ""
        #    }
        #
        messageHash = self.safe_string(message, 'id')
        data = self.safe_value(message, 'data', [])
        client.resolve(data, messageHash)

    def handle_subscription_status(self, client: Client, message):
        #
        #     {event: 'subscribe', arg: {channel: "tickers", instId: "BTC-USDT"}}
        #
        # channel = self.safe_string(message, "channel")
        # client.subscriptions[channel] = message
        return message

    def handle_authenticate(self, client: Client, message):
        #
        #     {event: "login", success: True}
        #
        future = self.safe_value(client.futures, 'authenticated')
        future.resolve(True)

    def ping(self, client: Client):
        # OKX does not support the built-in WebSocket protocol-level ping-pong.
        # Instead, it requires a custom text-based ping-pong mechanism.
        return 'ping'

    def handle_pong(self, client: Client, message):
        client.lastPong = self.milliseconds()
        return message

    def handle_error_message(self, client: Client, message) -> Bool:
        #
        #     {event: 'error', msg: "Illegal request: {"op":"subscribe","args":["spot/ticker:BTC-USDT"]}", code: "60012"}
        #     {event: 'error", msg: "channel:ticker,instId:BTC-USDT doesn"t exist", code: "60018"}
        #     {"event":"error","msg":"Illegal request: {\\"id\\":\\"17321173472466905\\",\\"op\\":\\"amend-order\\",\\"args\\":[{\\"instId\\":\\"ETH-USDC\\",\\"ordId\\":\\"2000345622407479296\\",\\"newSz\\":\\"0.050857\\",\\"newPx\\":\\"2949.4\\",\\"postOnly\\":true}],\\"postOnly\\":true}","code":"60012","connId":"0808af6c"}
        #
        errorCode = self.safe_string(message, 'code')
        try:
            if errorCode and errorCode != '0':
                feedback = self.id + ' ' + self.json(message)
                if errorCode != '1':
                    self.throw_exactly_matched_exception(self.exceptions['exact'], errorCode, feedback)
                messageString = self.safe_value(message, 'msg')
                if messageString is not None:
                    self.throw_broadly_matched_exception(self.exceptions['broad'], messageString, feedback)
                else:
                    data = self.safe_list(message, 'data', [])
                    for i in range(0, len(data)):
                        d = data[i]
                        errorCode = self.safe_string(d, 'sCode')
                        if errorCode is not None:
                            self.throw_exactly_matched_exception(self.exceptions['exact'], errorCode, feedback)
                        messageString = self.safe_value(d, 'sMsg')
                        if messageString is not None:
                            self.throw_broadly_matched_exception(self.exceptions['broad'], messageString, feedback)
                raise ExchangeError(feedback)
        except Exception as e:
            # if the message contains an id, it means it is a response to a request
            # so we only reject that promise, instead of deleting all futures, destroying the authentication future
            id = self.safe_string(message, 'id')
            if id is None:
                # try to parse it from the stringified json inside msg
                msg = self.safe_string(message, 'msg')
                if msg is not None and msg.startswith('Illegal request: {'):
                    stringifiedJson = msg.replace('Illegal request: ', '')
                    parsedJson = self.parse_json(stringifiedJson)
                    id = self.safe_string(parsedJson, 'id')
            if id is not None:
                client.reject(e, id)
                return False
            client.reject(e)
            return False
        return True

    def handle_message(self, client: Client, message):
        if not self.handle_error_message(client, message):
            return
        #
        #     {event: 'subscribe', arg: {channel: "tickers", instId: "BTC-USDT"}}
        #     {event: 'login", msg: '", code: "0"}
        #
        #     {
        #         "arg": {channel: "tickers", instId: "BTC-USDT"},
        #         "data": [
        #             {
        #                 "instType": "SPOT",
        #                 "instId": "BTC-USDT",
        #                 "last": "31500.1",
        #                 "lastSz": "0.00001754",
        #                 "askPx": "31500.1",
        #                 "askSz": "0.00998144",
        #                 "bidPx": "31500",
        #                 "bidSz": "3.05652439",
        #                 "open24h": "31697",
        #                 "high24h": "32248",
        #                 "low24h": "31165.6",
        #                 "sodUtc0": "31385.5",
        #                 "sodUtc8": "32134.9",
        #                 "volCcy24h": "503403597.38138519",
        #                 "vol24h": "15937.10781721",
        #                 "ts": "1626526618762"
        #             }
        #         ]
        #     }
        #
        #     {event: 'error', msg: "Illegal request: {"op":"subscribe","args":["spot/ticker:BTC-USDT"]}", code: "60012"}
        #     {event: 'error", msg: "channel:ticker,instId:BTC-USDT doesn"t exist", code: "60018"}
        #     {event: 'error', msg: "Invalid OK_ACCESS_KEY", code: "60005"}
        #     {
        #         "event": "error",
        #         "msg": "Illegal request: {"op":"login","args":["de89b035-b233-44b2-9a13-0ccdd00bda0e","7KUcc8YzQhnxBE3K","1626691289","H57N99mBt5NvW8U19FITrPdOxycAERFMaapQWRqLaSE="]}",
        #         "code": "60012"
        #     }
        #
        #
        #
        if message == 'pong':
            self.handle_pong(client, message)
            return
        # table = self.safe_string(message, 'table')
        # if table is None:
        event = self.safe_string_2(message, 'event', 'op')
        if event is not None:
            methods: dict = {
                # 'info': self.handleSystemStatus,
                # 'book': 'handleOrderBook',
                'login': self.handle_authenticate,
                'subscribe': self.handle_subscription_status,
                'unsubscribe': self.handle_unsubscription,
                'order': self.handle_place_orders,
                'batch-orders': self.handle_place_orders,
                'amend-order': self.handle_place_orders,
                'batch-amend-orders': self.handle_place_orders,
                'cancel-order': self.handle_place_orders,
                'mass-cancel': self.handle_cancel_all_orders,
            }
            method = self.safe_value(methods, event)
            if method is not None:
                method(client, message)
        else:
            arg = self.safe_value(message, 'arg', {})
            channel = self.safe_string(arg, 'channel')
            methods: dict = {
                'bbo-tbt': self.handle_order_book,  # newly added channel that sends tick-by-tick Level 1 data, all API users can subscribe, public depth channel, verification not required
                'books': self.handle_order_book,  # all API users can subscribe, public depth channel, verification not required
                'books5': self.handle_order_book,  # all API users can subscribe, public depth channel, verification not required, data feeds will be delivered every 100ms(vs. every 200ms now)
                'books50-l2-tbt': self.handle_order_book,  # only users who're VIP4 and above can subscribe, identity verification required before subscription
                'books-l2-tbt': self.handle_order_book,  # only users who're VIP5 and above can subscribe, identity verification required before subscription
                'tickers': self.handle_ticker,
                'mark-price': self.handle_ticker,
                'positions': self.handle_positions,
                'index-tickers': self.handle_ticker,
                'sprd-tickers': self.handle_ticker,
                'block-tickers': self.handle_ticker,
                'trades': self.handle_trades,
                'trades-all': self.handle_trades,
                'account': self.handle_balance,
                'funding-rate': self.handle_funding_rate,
                # 'margin_account': self.handle_balance,
                'orders': self.handle_orders,
                'orders-algo': self.handle_orders,
                'liquidation-orders': self.handle_liquidation,
                'balance_and_position': self.handle_balance_and_position,
            }
            method = self.safe_value(methods, channel)
            if method is None:
                if channel.find('candle') == 0:
                    self.handle_ohlcv(client, message)
            else:
                method(client, message)

    def handle_un_subscription_trades(self, client: Client, symbol: str, channel: str):
        subMessageHash = channel + ':' + symbol
        messageHash = 'unsubscribe:' + subMessageHash
        self.clean_unsubscription(client, subMessageHash, messageHash)
        if symbol in self.trades:
            del self.trades[symbol]

    def handle_unsubscription_order_book(self, client: Client, symbol: str, channel: str):
        subMessageHash = channel + ':' + symbol
        messageHash = 'unsubscribe:orderbook:' + symbol
        self.clean_unsubscription(client, subMessageHash, messageHash)
        if symbol in self.orderbooks:
            del self.orderbooks[symbol]

    def handle_unsubscription_ohlcv(self, client: Client, symbol: str, channel: str):
        tf = channel.replace('candle', '')
        timeframe = self.find_timeframe(tf)
        subMessageHash = 'multi:' + channel + ':' + symbol
        messageHash = 'unsubscribe:' + subMessageHash
        self.clean_unsubscription(client, subMessageHash, messageHash)
        if timeframe in self.ohlcvs[symbol]:
            del self.ohlcvs[symbol][timeframe]

    def handle_unsubscription_ticker(self, client: Client, symbol: str, channel):
        subMessageHash = channel + '::' + symbol
        messageHash = 'unsubscribe:ticker:' + symbol
        self.clean_unsubscription(client, subMessageHash, messageHash)
        if symbol in self.tickers:
            del self.tickers[symbol]

    def handle_unsubscription(self, client: Client, message):
        #
        # {
        #     "event": "unsubscribe",
        #     "arg": {
        #       "channel": "tickers",
        #       "instId": "LTC-USD-200327"
        #     },
        #     "connId": "a4d3ae55"
        # }
        # arg might be an array or list
        arg = self.safe_dict(message, 'arg', {})
        channel = self.safe_string(arg, 'channel', '')
        marketId = self.safe_string(arg, 'instId')
        symbol = self.safe_symbol(marketId)
        if channel == 'trades' or channel == 'trades-all':
            self.handle_un_subscription_trades(client, symbol, channel)
        elif channel.startswith('bbo') or channel.startswith('book'):
            self.handle_unsubscription_order_book(client, symbol, channel)
        elif channel.find('tickers') > -1:
            self.handle_unsubscription_ticker(client, symbol, channel)
        elif channel.startswith('candle'):
            self.handle_unsubscription_ohlcv(client, symbol, channel)
