<?php

namespace ccxt\pro;

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

use Exception; // a common import
use \React\Async;
use \React\Promise\PromiseInterface;

class xt extends \ccxt\async\xt {

    public function describe(): mixed {
        return $this->deep_extend(parent::describe(), array(
            'has' => array(
                'ws' => true,
                'watchOHLCV' => true,
                'watchOrderBook' => true,
                'watchTicker' => true,
                'watchTickers' => true,
                'watchTrades' => true,
                'watchTradesForSymbols' => false,
                'watchBalance' => true,
                'watchOrders' => true,
                'watchMyTrades' => true,
                'watchPositions' => true,
            ),
            'urls' => array(
                'api' => array(
                    'ws' => array(
                        'spot' => 'wss://stream.xt.com',
                        'contract' => 'wss://fstream.xt.com/ws',
                    ),
                ),
            ),
            'options' => array(
                'tradesLimit' => 1000,
                'ordersLimit' => 1000,
                'OHLCVLimit' => 1000,
                'watchTicker' => array(
                    'method' => 'ticker',  // agg_ticker (contract only)
                ),
                'watchTickers' => array(
                    'method' => 'tickers',  // agg_tickers (contract only)
                ),
                'watchPositions' => array(
                    'type' => 'swap',
                    'fetchPositionsSnapshot' => true,
                    'awaitPositionsSnapshot' => true,
                ),
            ),
            'streaming' => array(
                'keepAlive' => 20000,
                'ping' => array($this, 'ping'),
            ),
            'token' => null,
        ));
    }

    public function get_listen_key(bool $isContract) {
        return Async\async(function () use ($isContract) {
            /**
             * @ignore
             * required for private endpoints
             * @param {string} $isContract true for contract trades
             *
             * @see https://doc.xt.com/#websocket_privategetToken
             * @see https://doc.xt.com/#futures_user_websocket_v2base
             *
             * @return {string} listen key / access $token
             */
            $this->check_required_credentials();
            $tradeType = $isContract ? 'contract' : 'spot';
            $url = $this->urls['api']['ws'][$tradeType];
            if (!$isContract) {
                $url = $url . '/private';
            }
            $client = $this->client($url);
            $token = $this->safe_string($client->subscriptions, 'token');
            if ($token === null) {
                if ($isContract) {
                    $response = Async\await($this->privateLinearGetFutureUserV1UserListenKey ());
                    //
                    //    {
                    //        returnCode => '0',
                    //        msgInfo => 'success',
                    //        error => null,
                    //        $result => '3BC1D71D6CF96DA3458FC35B05B633351684511731128'
                    //    }
                    //
                    $client->subscriptions['token'] = $this->safe_string($response, 'result');
                } else {
                    $response = Async\await($this->privateSpotPostWsToken ());
                    //
                    //    {
                    //        "rc" => 0,
                    //        "mc" => "SUCCESS",
                    //        "ma" => array(),
                    //        "result" => {
                    //            "token" => "eyJhbqGciOiJSUzI1NiJ9.eyJhY2NvdW50SWQiOiIyMTQ2Mjg1MzIyNTU5Iiwic3ViIjoibGh4dDRfMDAwMUBzbmFwbWFpbC5jYyIsInNjb3BlIjoiYXV0aCIsImlzcyI6Inh0LmNvbSIsImxhc3RBdXRoVGltZSI6MTY2MzgxMzY5MDk1NSwic2lnblR5cGUiOiJBSyIsInVzZXJOYW1lIjoibGh4dDRfMDAwMUBzbmFwbWFpbC5jYyIsImV4cCI6MTY2NjQwNTY5MCwiZGV2aWNlIjoidW5rbm93biIsInVzZXJJZCI6MjE0NjI4NTMyMjU1OX0.h3zJlJBQrK2x1HvUxsKivnn6PlSrSDXXXJ7WqHAYSrN2CG5XPTKc4zKnTVoYFbg6fTS0u1fT8wH7wXqcLWXX71vm0YuP8PCvdPAkUIq4-HyzltbPr5uDYd0UByx0FPQtq1exvsQGe7evXQuDXx3SEJXxEqUbq_DNlXPTq_JyScI",
                    //            "refreshToken" => "eyJhbGciOiqJSUzI1NiJ9.eyJhY2NvdW50SWQiOiIyMTQ2Mjg1MzIyNTU5Iiwic3ViIjoibGh4dDRfMDAwMUBzbmFwbWFpbC5jYyIsInNjb3BlIjoicmVmcmVzaCIsImlzcyI6Inh0LmNvbSIsImxhc3RBdXRoVGltZSI6MTY2MzgxMzY5MDk1NSwic2lnblR5cGUiOiJBSyIsInVzZXJOYW1lIjoibGh4dDRfMDAwMUBzbmFwbWFpbC5jYyIsImV4cCI6MTY2NjQwNTY5MCwiZGV2aWNlIjoidW5rbm93biIsInVzZXJJZCI6MjE0NjI4NTMyMjU1OX0.Fs3YVm5YrEOzzYOSQYETSmt9iwxUHBovh2u73liv1hLUec683WGfktA_s28gMk4NCpZKFeQWFii623FvdfNoteXR0v1yZ2519uNvNndtuZICDdv3BQ4wzW1wIHZa1skxFfqvsDnGdXpjqu9UFSbtHwxprxeYfnxChNk4ssei430"
                    //        }
                    //    }
                    //
                    $result = $this->safe_dict($response, 'result');
                    $client->subscriptions['token'] = $this->safe_string($result, 'accessToken');
                }
            }
            return $client->subscriptions['token'];
        }) ();
    }

    public function get_cache_index($orderbook, $cache) {
        // return the first index of the $cache that can be applied to the $orderbook or -1 if not possible
        $nonce = $this->safe_integer($orderbook, 'nonce');
        $firstDelta = $this->safe_value($cache, 0);
        $firstDeltaNonce = $this->safe_integer_2($firstDelta, 'i', 'u');
        if ($nonce < $firstDeltaNonce - 1) {
            return -1;
        }
        for ($i = 0; $i < count($cache); $i++) {
            $delta = $cache[$i];
            $deltaNonce = $this->safe_integer_2($delta, 'i', 'u');
            if ($deltaNonce >= $nonce) {
                return $i;
            }
        }
        return count($cache);
    }

    public function handle_delta($orderbook, $delta) {
        $orderbook['nonce'] = $this->safe_integer_2($delta, 'i', 'u');
        $obAsks = $this->safe_list($delta, 'a', array());
        $obBids = $this->safe_list($delta, 'b', array());
        $bids = $orderbook['bids'];
        $asks = $orderbook['asks'];
        for ($i = 0; $i < count($obBids); $i++) {
            $bid = $obBids[$i];
            $price = $this->safe_number($bid, 0);
            $quantity = $this->safe_number($bid, 1);
            $bids->store ($price, $quantity);
        }
        for ($i = 0; $i < count($obAsks); $i++) {
            $ask = $obAsks[$i];
            $price = $this->safe_number($ask, 0);
            $quantity = $this->safe_number($ask, 1);
            $asks->store ($price, $quantity);
        }
        // $this->handleBidAsks (storedBids, $bids);
        // $this->handleBidAsks (storedAsks, $asks);
    }

    public function subscribe(string $name, string $access, string $methodName, ?array $market = null, ?array $symbols = null, $params = array ()) {
        return Async\async(function () use ($name, $access, $methodName, $market, $symbols, $params) {
            /**
             * @ignore
             * Connects to a websocket channel
             *
             * @see https://doc.xt.com/#websocket_privaterequestFormat
             * @see https://doc.xt.com/#futures_market_websocket_v2base
             *
             * @$param {string} $name name of the channel
             * @$param {string} $access public or private
             * @$param {string} $methodName the $name of the CCXT class method
             * @$param {array} [$market] CCXT $market
             * @$param {string[]} [$symbols] unified $market $symbols
             * @$param {array} $params extra parameters specific to the xt api
             * @return {array} data from the websocket stream
             */
            $privateAccess = $access === 'private';
            $type = null;
            list($type, $params) = $this->handle_market_type_and_params($methodName, $market, $params);
            $isContract = ($type !== 'spot');
            $subscribe = array(
                'method' => $isContract ? 'SUBSCRIBE' : 'subscribe',
                'id' => $this->number_to_string($this->milliseconds()) . $name,  // call back ID
            );
            if ($privateAccess) {
                if (!$isContract) {
                    $subscribe['params'] = array( $name );
                    $subscribe['listenKey'] = Async\await($this->get_listen_key($isContract));
                } else {
                    $listenKey = Async\await($this->get_listen_key($isContract));
                    $param = $name . '@' . $listenKey;
                    $subscribe['params'] = array( $param );
                }
            } else {
                $subscribe['params'] = array( $name );
            }
            $tradeType = $isContract ? 'contract' : 'spot';
            $messageHash = $name . '::' . $tradeType;
            if ($symbols !== null) {
                $messageHash = $messageHash . '::' . implode(',', $symbols);
            }
            $request = $this->extend($subscribe, $params);
            $tail = $access;
            if ($isContract) {
                $tail = $privateAccess ? 'user' : 'market';
            }
            $url = $this->urls['api']['ws'][$tradeType] . '/' . $tail;
            return Async\await($this->watch($url, $messageHash, $request, $messageHash));
        }) ();
    }

    public function watch_ticker(string $symbol, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $params) {
            /**
             * watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific $market
             *
             * @see https://doc.xt.com/#websocket_publictickerRealTime
             * @see https://doc.xt.com/#futures_market_websocket_v2tickerRealTime
             * @see https://doc.xt.com/#futures_market_websocket_v2aggTickerRealTime
             *
             * @param {string} $symbol unified $symbol of the $market to fetch the ticker for
             * @param {array} $params extra parameters specific to the xt api endpoint
             * @param {string} [$params->method] 'agg_ticker' (contract only) or 'ticker', default = 'ticker' - the endpoint that will be streamed
             * @return {array} a {@link https://docs.ccxt.com/en/latest/manual.html#ticker-structure ticker structure}
             */
            Async\await($this->load_markets());
            $market = $this->market($symbol);
            $options = $this->safe_dict($this->options, 'watchTicker');
            $defaultMethod = $this->safe_string($options, 'method', 'ticker');
            $method = $this->safe_string($params, 'method', $defaultMethod);
            $name = $method . '@' . $market['id'];
            return Async\await($this->subscribe($name, 'public', 'watchTicker', $market, null, $params));
        }) ();
    }

    public function watch_tickers(?array $symbols = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $params) {
            /**
             * watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific $market
             *
             * @see https://doc.xt.com/#websocket_publicallTicker
             * @see https://doc.xt.com/#futures_market_websocket_v2allTicker
             * @see https://doc.xt.com/#futures_market_websocket_v2allAggTicker
             *
             * @param {string} [$symbols] unified $market $symbols
             * @param {array} $params extra parameters specific to the xt api endpoint
             * @param {string} [$params->method] 'agg_tickers' (contract only) or 'tickers', default = 'tickers' - the endpoint that will be streamed
             * @return {array} a {@link https://docs.ccxt.com/en/latest/manual.html#ticker-structure ticker structure}
             */
            Async\await($this->load_markets());
            $options = $this->safe_dict($this->options, 'watchTickers');
            $defaultMethod = $this->safe_string($options, 'method', 'tickers');
            $name = $this->safe_string($params, 'method', $defaultMethod);
            $market = null;
            if ($symbols !== null) {
                $market = $this->market($symbols[0]);
            }
            $tickers = Async\await($this->subscribe($name, 'public', 'watchTickers', $market, $symbols, $params));
            if ($this->newUpdates) {
                return $tickers;
            }
            return $this->filter_by_array($this->tickers, 'symbol', $symbols);
        }) ();
    }

    public function watch_ohlcv(string $symbol, $timeframe = '1m', ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $timeframe, $since, $limit, $params) {
            /**
             * watches historical candlestick data containing the open, high, low, and close price, and the volume of a $market
             *
             * @see https://doc.xt.com/#websocket_publicsymbolKline
             * @see https://doc.xt.com/#futures_market_websocket_v2symbolKline
             *
             * @param {string} $symbol unified $symbol of the $market to fetch OHLCV data for
             * @param {string} $timeframe 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, or 1M
             * @param {int} [$since] not used by xt watchOHLCV
             * @param {int} [$limit] not used by xt watchOHLCV
             * @param {array} $params extra parameters specific to the xt api endpoint
             * @return {int[][]} A list of candles ordered, open, high, low, close, volume
             */
            Async\await($this->load_markets());
            $market = $this->market($symbol);
            $name = 'kline@' . $market['id'] . ',' . $timeframe;
            $ohlcv = Async\await($this->subscribe($name, 'public', 'watchOHLCV', $market, null, $params));
            if ($this->newUpdates) {
                $limit = $ohlcv->getLimit ($symbol, $limit);
            }
            return $this->filter_by_since_limit($ohlcv, $since, $limit, 0, true);
        }) ();
    }

    public function watch_trades(string $symbol, ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $since, $limit, $params) {
            /**
             * get the list of most recent $trades for a particular $symbol
             *
             * @see https://doc.xt.com/#websocket_publicdealRecord
             * @see https://doc.xt.com/#futures_market_websocket_v2dealRecord
             *
             * @param {string} $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 {array} $params extra parameters specific to the xt api endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/en/latest/manual.html?#public-$trades trade structures~
             */
            Async\await($this->load_markets());
            $market = $this->market($symbol);
            $name = 'trade@' . $market['id'];
            $trades = Async\await($this->subscribe($name, 'public', 'watchTrades', $market, null, $params));
            if ($this->newUpdates) {
                $limit = $trades->getLimit ($symbol, $limit);
            }
            return $this->filter_by_since_limit($trades, $since, $limit, 'timestamp');
        }) ();
    }

    public function watch_order_book(string $symbol, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $limit, $params) {
            /**
             * watches information on open orders with bid (buy) and ask (sell) prices, volumes and other data
             *
             * @see https://doc.xt.com/#websocket_publiclimitDepth
             * @see https://doc.xt.com/#websocket_publicincreDepth
             * @see https://doc.xt.com/#futures_market_websocket_v2limitDepth
             * @see https://doc.xt.com/#futures_market_websocket_v2increDepth
             *
             * @param {string} $symbol unified $symbol of the $market to fetch the order book for
             * @param {int} [$limit] not used by xt watchOrderBook
             * @param {array} $params extra parameters specific to the xt api endpoint
             * @param {int} [$params->levels] 5, 10, 20, or 50
             * @return {array} A dictionary of {@link https://docs.ccxt.com/en/latest/manual.html#order-book-structure order book structures} indexed by $market symbols
             */
            Async\await($this->load_markets());
            $market = $this->market($symbol);
            $levels = $this->safe_string($params, 'levels');
            $params = $this->omit($params, 'levels');
            $name = 'depth_update@' . $market['id'];
            if ($levels !== null) {
                $name = 'depth@' . $market['id'] . ',' . $levels;
            }
            $orderbook = Async\await($this->subscribe($name, 'public', 'watchOrderBook', $market, null, $params));
            return $orderbook->limit ();
        }) ();
    }

    public function watch_orders(?string $symbol = null, ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $since, $limit, $params) {
            /**
             * watches information on multiple $orders made by the user
             *
             * @see https://doc.xt.com/#websocket_privateorderChange
             * @see https://doc.xt.com/#futures_user_websocket_v2order
             *
             * @param {string} [$symbol] unified $market $symbol
             * @param {int} [$since] not used by xt watchOrders
             * @param {int} [$limit] the maximum number of $orders to return
             * @param {array} $params extra parameters specific to the xt api endpoint
             * @return {array[]} a list of {@link https://docs.ccxt.com/en/latest/manual.html#order-structure order structures}
             */
            Async\await($this->load_markets());
            $name = 'order';
            $market = null;
            if ($symbol !== null) {
                $market = $this->market($symbol);
            }
            $orders = Async\await($this->subscribe($name, 'private', 'watchOrders', $market, null, $params));
            if ($this->newUpdates) {
                $limit = $orders->getLimit ($symbol, $limit);
            }
            return $this->filter_by_since_limit($orders, $since, $limit, 'timestamp');
        }) ();
    }

    public function watch_my_trades(?string $symbol = null, ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $since, $limit, $params) {
            /**
             * watches information on multiple $trades made by the user
             *
             * @see https://doc.xt.com/#websocket_privateorderDeal
             * @see https://doc.xt.com/#futures_user_websocket_v2trade
             *
             * @param {string} $symbol unified $market $symbol of the $market orders were made in
             * @param {int} [$since] the earliest time in ms to fetch orders for
             * @param {int} [$limit] the maximum number of  orde structures to retrieve
             * @param {array} $params extra parameters specific to the kucoin api endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=trade-structure trade structures~
             */
            Async\await($this->load_markets());
            $name = 'trade';
            $market = null;
            if ($symbol !== null) {
                $market = $this->market($symbol);
            }
            $trades = Async\await($this->subscribe($name, 'private', 'watchMyTrades', $market, null, $params));
            if ($this->newUpdates) {
                $limit = $trades->getLimit ($symbol, $limit);
            }
            return $this->filter_by_since_limit($trades, $since, $limit, 'timestamp');
        }) ();
    }

    public function watch_balance($params = array ()): PromiseInterface {
        return Async\async(function () use ($params) {
            /**
             * watches information on multiple orders made by the user
             *
             * @see https://doc.xt.com/#websocket_privatebalanceChange
             * @see https://doc.xt.com/#futures_user_websocket_v2balance
             *
             * @param {array} $params extra parameters specific to the xt api endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=balance-structure balance structures~
             */
            Async\await($this->load_markets());
            $name = 'balance';
            return Async\await($this->subscribe($name, 'private', 'watchBalance', null, null, $params));
        }) ();
    }

    public function watch_positions(?array $symbols = null, ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $since, $limit, $params) {
            /**
             *
             * @see https://doc.xt.com/#futures_user_websocket_v2position
             *
             * watch all open positions
             * @param {string[]|null} $symbols list of unified market $symbols
             * @param {number} [$since] $since timestamp
             * @param {number} [$limit] $limit
             * @param {array} $params extra parameters specific to the exchange API endpoint
             * @return {array[]} a list of {@link https://docs.ccxt.com/en/latest/manual.html#position-structure position structure}
             */
            Async\await($this->load_markets());
            $url = $this->urls['api']['ws']['contract'] . '/' . 'user';
            $client = $this->client($url);
            $this->set_positions_cache($client);
            $fetchPositionsSnapshot = $this->handle_option('watchPositions', 'fetchPositionsSnapshot', true);
            $awaitPositionsSnapshot = $this->handle_option('watchPositions', 'awaitPositionsSnapshot', true);
            $cache = $this->positions;
            if ($fetchPositionsSnapshot && $awaitPositionsSnapshot && $this->is_empty($cache)) {
                $snapshot = Async\await($client->future ('fetchPositionsSnapshot'));
                return $this->filter_by_symbols_since_limit($snapshot, $symbols, $since, $limit, true);
            }
            $name = 'position';
            $newPositions = Async\await($this->subscribe($name, 'private', 'watchPositions', null, null, $params));
            if ($this->newUpdates) {
                return $newPositions;
            }
            return $this->filter_by_symbols_since_limit($cache, $symbols, $since, $limit, true);
        }) ();
    }

    public function set_positions_cache(Client $client) {
        if ($this->positions === null) {
            $this->positions = new ArrayCacheBySymbolBySide ();
        }
        $fetchPositionsSnapshot = $this->handle_option('watchPositions', 'fetchPositionsSnapshot');
        if ($fetchPositionsSnapshot) {
            $messageHash = 'fetchPositionsSnapshot';
            if (!(is_array($client->futures) && array_key_exists($messageHash, $client->futures))) {
                $client->future ($messageHash);
                $this->spawn(array($this, 'load_positions_snapshot'), $client, $messageHash);
            }
        }
    }

    public function load_positions_snapshot($client, $messageHash) {
        return Async\async(function () use ($client, $messageHash) {
            $positions = Async\await($this->fetch_positions(null));
            $this->positions = new ArrayCacheBySymbolBySide ();
            $cache = $this->positions;
            for ($i = 0; $i < count($positions); $i++) {
                $position = $positions[$i];
                $contracts = $this->safe_number($position, 'contracts', 0);
                if ($contracts > 0) {
                    $cache->append ($position);
                }
            }
            // don't remove the $future from the .futures $cache
            $future = $client->futures[$messageHash];
            $future->resolve ($cache);
            $client->resolve ($cache, 'position::contract');
        }) ();
    }

    public function handle_position($client, $message) {
        //
        //    {
        //      topic => 'position',
        //      event => 'position',
        //      $data => {
        //        accountId => 245296,
        //        accountType => 0,
        //        symbol => 'eth_usdt',
        //        contractType => 'PERPETUAL',
        //        positionType => 'CROSSED',
        //        positionSide => 'LONG',
        //        positionSize => '1',
        //        closeOrderSize => '0',
        //        availableCloseSize => '1',
        //        realizedProfit => '-0.0121',
        //        entryPrice => '2637.87',
        //        openOrderSize => '1',
        //        isolatedMargin => '2.63787',
        //        openOrderMarginFrozen => '2.78832014',
        //        underlyingType => 'U_BASED',
        //        leverage => 10,
        //        welfareAccount => false,
        //        profitFixedLatest => array(),
        //        closeProfit => '0.0000',
        //        totalFee => '-0.0158',
        //        totalFundFee => '0.0037',
        //        markPrice => '2690.96'
        //      }
        //    }
        //
        if ($this->positions === null) {
            $this->positions = new ArrayCacheBySymbolBySide ();
        }
        $cache = $this->positions;
        $data = $this->safe_dict($message, 'data', array());
        $position = $this->parse_position($data);
        $cache->append ($position);
        $messageHashes = $this->find_message_hashes($client, 'position::contract');
        for ($i = 0; $i < count($messageHashes); $i++) {
            $messageHash = $messageHashes[$i];
            $parts = explode('::', $messageHash);
            $symbolsString = $parts[1];
            $symbols = explode(',', $symbolsString);
            $positions = $this->filter_by_array(array( $position ), 'symbol', $symbols, false);
            if (!$this->is_empty($positions)) {
                $client->resolve ($positions, $messageHash);
            }
        }
        $client->resolve (array( $position ), 'position::contract');
    }

    public function handle_ticker(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        topic => 'ticker',
        //        $event => 'ticker@btc_usdt',
        //        $data => {
        //           s => 'btc_usdt',            // $symbol
        //           t => 1683501935877,         // time(Last transaction time)
        //           $cv => '-82.67',             // priceChangeValue(24 hour price change)
        //           cr => '-0.0028',            // priceChangeRate 24-hour price change (percentage)
        //           o => '28823.87',            // open price
        //           c => '28741.20',            // close price
        //           h => '29137.64',            // highest price
        //           l => '28660.93',            // lowest price
        //           q => '6372.601573',         // quantity
        //           v => '184086075.2772391'    // volume
        //        }
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "ticker",
        //        "event" => "ticker@btc_usdt",
        //        "data" => {
        //            "s" => "btc_index",  // trading pair
        //            "o" => "49000",      // opening price
        //            "c" => "50000",      // closing price
        //            "h" => "0.1",        // highest price
        //            "l" => "0.1",        // lowest price
        //            "a" => "0.1",        // volume
        //            "v" => "0.1",        // turnover
        //            "ch" => "0.21",      // quote change
        //            "t" => 123124124     // timestamp
        //       }
        //    }
        //
        // agg_ticker (contract)
        //
        //    {
        //        "topic" => "agg_ticker",
        //        "event" => "agg_ticker@btc_usdt",
        //        "data" => {
        //            "s" => "btc_index",          // trading pair
        //            "o" => "49000",              // opening price
        //            "c" => "50000",              // closing price
        //            "h" => "0.1",                // highest price
        //            "l" => "0.1",                // lowest price
        //            "a" => "0.1",                // volume
        //            "v" => "0.1",                // turnover
        //            "ch" => "0.21",              // quote change
        //            "i" => "0.21" ,              // index price
        //            "m" => "0.21",               // mark price
        //            "bp" => "0.21",              // bid price
        //            "ap" => "0.21" ,             // ask price
        //            "t" => 123124124             // timestamp
        //       }
        //    }
        //
        $data = $this->safe_dict($message, 'data');
        $marketId = $this->safe_string($data, 's');
        if ($marketId !== null) {
            $cv = $this->safe_string($data, 'cv');
            $isSpot = $cv !== null;
            $ticker = $this->parse_ticker($data);
            $symbol = $ticker['symbol'];
            $this->tickers[$symbol] = $ticker;
            $event = $this->safe_string($message, 'event');
            $messageHashTail = $isSpot ? 'spot' : 'contract';
            $messageHash = $event . '::' . $messageHashTail;
            $client->resolve ($ticker, $messageHash);
        }
        return $message;
    }

    public function handle_tickers(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        topic => 'tickers',
        //        event => 'tickers',
        //        $data => array(
        //            array(
        //                s => 'elon_usdt',
        //                t => 1683502958381,
        //                cv => '-0.0000000125',
        //                cr => '-0.0495',
        //                o => '0.0000002522',
        //                c => '0.0000002397',
        //                h => '0.0000002690',
        //                l => '0.0000002371',
        //                q => '3803783034.0000000000',
        //                v => '955.3260820022'
        //            ),
        //            ...
        //        )
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "tickers",
        //        "event" => "tickers",
        //        "data" => array(
        //            {
        //                "s" => "btc_index",  // trading pair
        //                "o" => "49000",      // opening price
        //                "c" => "50000",      // closing price
        //                "h" => "0.1",        // highest price
        //                "l" => "0.1",        // lowest price
        //                "a" => "0.1",        // volume
        //                "v" => "0.1",        // turnover
        //                "ch" => "0.21",      // quote change
        //                "t" => 123124124     // timestamp
        //            }
        //        )
        //    }
        //
        // agg_ticker (contract)
        //
        //    {
        //        "topic" => "agg_tickers",
        //        "event" => "agg_tickers",
        //        "data" => array(
        //            {
        //                "s" => "btc_index",          // trading pair
        //                "o" => "49000",              // opening price
        //                "c" => "50000",              // closing price
        //                "h" => "0.1",                // highest price
        //                "l" => "0.1",                // lowest price
        //                "a" => "0.1",                // volume
        //                "v" => "0.1",                // turnover
        //                "ch" => "0.21",              // quote change
        //                "i" => "0.21" ,              // index price
        //                "m" => "0.21",               // mark price
        //                "bp" => "0.21",              // bid price
        //                "ap" => "0.21" ,             // ask price
        //                "t" => 123124124             // timestamp
        //            }
        //        )
        //    }
        //
        $data = $this->safe_list($message, 'data', array());
        $firstTicker = $this->safe_dict($data, 0);
        $spotTest = $this->safe_string_2($firstTicker, 'cv', 'aq');
        $tradeType = ($spotTest !== null) ? 'spot' : 'contract';
        $newTickers = array();
        for ($i = 0; $i < count($data); $i++) {
            $tickerData = $data[$i];
            $ticker = $this->parse_ticker($tickerData);
            $symbol = $ticker['symbol'];
            $this->tickers[$symbol] = $ticker;
            $newTickers[] = $ticker;
        }
        $messageHashStart = $this->safe_string($message, 'topic') . '::' . $tradeType;
        $messageHashes = $this->find_message_hashes($client, $messageHashStart . '::');
        for ($i = 0; $i < count($messageHashes); $i++) {
            $messageHash = $messageHashes[$i];
            $parts = explode('::', $messageHash);
            $symbolsString = $parts[2];
            $symbols = explode(',', $symbolsString);
            $tickers = $this->filter_by_array($newTickers, 'symbol', $symbols);
            $tickersSymbols = is_array($tickers) ? array_keys($tickers) : array();
            $numTickers = count($tickersSymbols);
            if ($numTickers > 0) {
                $client->resolve ($tickers, $messageHash);
            }
        }
        $client->resolve ($this->tickers, $messageHashStart);
        return $message;
    }

    public function handle_ohlcv(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        "topic" => "kline",
        //        "event" => "kline@btc_usdt,5m",
        //        "data" => {
        //            "s" => "btc_usdt",        // $symbol
        //            "t" => 1656043200000,     // time
        //            "i" => "5m",              // interval
        //            "o" => "44000",           // open price
        //            "c" => "50000",           // close price
        //            "h" => "52000",           // highest price
        //            "l" => "36000",           // lowest price
        //            "q" => "34.2",            // qty(quantity)
        //            "v" => "230000"           // volume
        //        }
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "kline",
        //        "event" => "kline@btc_usdt,5m",
        //        "data" => {
        //            "s" => "btc_index",      // trading pair
        //            "o" => "49000",          // opening price
        //            "c" => "50000",          // closing price
        //            "h" => "0.1",            // highest price
        //            "l" => "0.1",            // lowest price
        //            "a" => "0.1",            // volume
        //            "v" => "0.1",            // turnover
        //            "ch" => "0.21",          // quote change
        //            "t" => 123124124         // timestamp
        //        }
        //    }
        //
        $data = $this->safe_dict($message, 'data', array());
        $marketId = $this->safe_string($data, 's');
        if ($marketId !== null) {
            $timeframe = $this->safe_string($data, 'i');
            $tradeType = (is_array($data) && array_key_exists('q', $data)) ? 'spot' : 'contract';
            $market = $this->safe_market($marketId, null, null, $tradeType);
            $symbol = $market['symbol'];
            $parsed = $this->parse_ohlcv($data, $market);
            $this->ohlcvs[$symbol] = $this->safe_dict($this->ohlcvs, $symbol, array());
            $stored = $this->safe_value($this->ohlcvs[$symbol], $timeframe);
            if ($stored === null) {
                $limit = $this->safe_integer($this->options, 'OHLCVLimit', 1000);
                $stored = new ArrayCacheByTimestamp ($limit);
                $this->ohlcvs[$symbol][$timeframe] = $stored;
            }
            $stored->append ($parsed);
            $event = $this->safe_string($message, 'event');
            $messageHash = $event . '::' . $tradeType;
            $client->resolve ($stored, $messageHash);
        }
        return $message;
    }

    public function handle_trade(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        topic => 'trade',
        //        $event => 'trade@btc_usdt',
        //        $data => {
        //            s => 'btc_usdt',
        //            $i => '228825383103928709',
        //            t => 1684258222702,
        //            p => '27003.65',
        //            q => '0.000796',
        //            b => true
        //        }
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "trade",
        //        "event" => "trade@btc_usdt",
        //        "data" => {
        //            "s" => "btc_index",  // trading pair
        //            "p" => "50000",      // price
        //            "a" => "0.1"         // Quantity
        //            "m" => "BID"         // Deal side  BID:Buy ASK:Sell
        //            "t" => 123124124     // timestamp
        //        }
        //    }
        //
        $data = $this->safe_dict($message, 'data');
        $marketId = $this->safe_string_lower($data, 's');
        if ($marketId !== null) {
            $trade = $this->parse_trade($data);
            $i = $this->safe_string($data, 'i');
            $tradeType = ($i !== null) ? 'spot' : 'contract';
            $market = $this->safe_market($marketId, null, null, $tradeType);
            $symbol = $market['symbol'];
            $event = $this->safe_string($message, 'event');
            $tradesArray = $this->safe_value($this->trades, $symbol);
            if ($tradesArray === null) {
                $tradesLimit = $this->safe_integer($this->options, 'tradesLimit', 1000);
                $tradesArray = new ArrayCache ($tradesLimit);
                $this->trades[$symbol] = $tradesArray;
            }
            $tradesArray->append ($trade);
            $messageHash = $event . '::' . $tradeType;
            $client->resolve ($tradesArray, $messageHash);
        }
        return $message;
    }

    public function handle_order_book(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        "topic" => "depth",
        //        "event" => "depth@btc_usdt,20",
        //        "data" => {
        //            "s" => "btc_usdt",        // $symbol
        //            "fi" => 1681433733351,    // firstUpdateId = previous lastUpdateId + 1
        //            "i" => 1681433733371,     // updateId
        //            "a" => [                  // $asks(sell order)
        //                [                   // [0]$price, [1]$quantity
        //                    "34000",        // $price
        //                    "1.2"           // $quantity
        //                ],
        //                array(
        //                    "34001",
        //                    "2.3"
        //                )
        //            ],
        //            "b" => array(                   // $bids(buy order)
        //                array(
        //                    "32000",
        //                    "0.2"
        //                ),
        //                array(
        //                    "31000",
        //                    "0.5"
        //                )
        //            )
        //        }
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "depth",
        //        "event" => "depth@btc_usdt,20",
        //        "data" => {
        //            s => "btc_usdt",
        //            pu => "548111455664",
        //            fu => "548111455665",
        //            u => "548111455667",
        //            a => array(
        //                array(
        //                    "26841.5",
        //                    "50210",
        //                ),
        //            ),
        //            b => array(
        //                array(
        //                    "26841",
        //                    "67075",
        //                ),
        //            ),
        //            t => 1684530667083,
        //        }
        //    }
        //
        $data = $this->safe_dict($message, 'data');
        $marketId = $this->safe_string($data, 's');
        if ($marketId !== null) {
            $event = $this->safe_string($message, 'event');
            $splitEvent = explode(',', $event);
            $event = $this->safe_string($splitEvent, 0);
            $tradeType = (is_array($data) && array_key_exists('fu', $data)) ? 'contract' : 'spot';
            $market = $this->safe_market($marketId, null, null, $tradeType);
            $symbol = $market['symbol'];
            $obAsks = $this->safe_list($data, 'a');
            $obBids = $this->safe_list($data, 'b');
            $messageHash = $event . '::' . $tradeType;
            if (!(is_array($this->orderbooks) && array_key_exists($symbol, $this->orderbooks))) {
                $subscription = $this->safe_dict($client->subscriptions, $messageHash, array());
                $limit = $this->safe_integer($subscription, 'limit');
                $this->orderbooks[$symbol] = $this->order_book(array(), $limit);
            }
            $orderbook = $this->orderbooks[$symbol];
            $nonce = $this->safe_integer($orderbook, 'nonce');
            if ($nonce === null) {
                $cacheLength = count($orderbook->cache);
                $snapshotDelay = $this->handle_option('watchOrderBook', 'snapshotDelay', 25);
                if ($cacheLength === $snapshotDelay) {
                    $this->spawn(array($this, 'load_order_book'), $client, $messageHash, $symbol);
                }
                $orderbook->cache[] = $data;
                return;
            }
            if ($obAsks !== null) {
                $asks = $orderbook['asks'];
                for ($i = 0; $i < count($obAsks); $i++) {
                    $ask = $obAsks[$i];
                    $price = $this->safe_number($ask, 0);
                    $quantity = $this->safe_number($ask, 1);
                    $asks->store ($price, $quantity);
                }
            }
            if ($obBids !== null) {
                $bids = $orderbook['bids'];
                for ($i = 0; $i < count($obBids); $i++) {
                    $bid = $obBids[$i];
                    $price = $this->safe_number($bid, 0);
                    $quantity = $this->safe_number($bid, 1);
                    $bids->store ($price, $quantity);
                }
            }
            $timestamp = $this->safe_integer($data, 't');
            $orderbook['nonce'] = $this->safe_integer_2($data, 'i', 'u');
            $orderbook['timestamp'] = $timestamp;
            $orderbook['datetime'] = $this->iso8601($timestamp);
            $orderbook['symbol'] = $symbol;
            $client->resolve ($orderbook, $messageHash);
        }
    }

    public function parse_ws_order_trade(array $trade, ?array $market = null) {
        //
        //    {
        //        "s" => "btc_usdt",                         // symbol
        //        "t" => 1656043204763,                      // time happened time
        //        "i" => "6216559590087220004",              // orderId,
        //        "ci" => "test123",                         // clientOrderId
        //        "st" => "PARTIALLY_FILLED",                // state
        //        "sd" => "BUY",                             // side BUY/SELL
        //        "eq" => "2",                               // executedQty executed quantity
        //        "ap" => "30000",                           // avg price
        //        "f" => "0.002"                             // fee
        //    }
        //
        // contract
        //
        //    {
        //        "symbol" => "btc_usdt",                    // Trading pair
        //        "orderId" => "1234",                       // Order Id
        //        "origQty" => "34244",                      // Original Quantity
        //        "avgPrice" => "123",                       // Quantity
        //        "price" => "1111",                         // Average price
        //        "executedQty" => "34244",                  // Volume (Cont)
        //        "orderSide" => "BUY",                      // BUY, SELL
        //        "positionSide" => "LONG",                  // LONG, SHORT
        //        "marginFrozen" => "123",                   // Occupied margin
        //        "sourceType" => "default",                 // DEFAULT:normal order,ENTRUST:plan commission,PROFIR:Take Profit and Stop Loss
        //        "sourceId" : "1231231",                  // Triggering conditions ID
        //        "state" => "",                             // state:NEW：New order (unfilled);PARTIALLY_FILLED:Partial deal;PARTIALLY_CANCELED:Partial revocation;FILLED:Filled;CANCELED:Cancled;REJECTED:Order failed;EXPIRED：Expired
        //        "createTime" => 1731231231,                // CreateTime
        //        "clientOrderId" => "204788317630342726"
        //    }
        //
        $marketId = $this->safe_string($trade, 's');
        $tradeType = (is_array($trade) && array_key_exists('symbol', $trade)) ? 'contract' : 'spot';
        $market = $this->safe_market($marketId, $market, null, $tradeType);
        $timestamp = $this->safe_string($trade, 't');
        return $this->safe_trade(array(
            'info' => $trade,
            'id' => null,
            'timestamp' => $timestamp,
            'datetime' => $this->iso8601($timestamp),
            'symbol' => $market['symbol'],
            'order' => $this->safe_string($trade, 'i', 'orderId'),
            'type' => $this->parse_order_status($this->safe_string($trade, 'st', 'state')),
            'side' => $this->safe_string_lower($trade, 'sd', 'orderSide'),
            'takerOrMaker' => null,
            'price' => $this->safe_number($trade, 'price'),
            'amount' => $this->safe_string($trade, 'origQty'),
            'cost' => null,
            'fee' => array(
                'currency' => null,
                'cost' => $this->safe_number($trade, 'f'),
                'rate' => null,
            ),
        ), $market);
    }

    public function parse_ws_order(array $order, ?array $market = null) {
        //
        // spot
        //
        //    {
        //        "s" => "btc_usdt",                // symbol
        //        "bc" => "btc",                    // base currency
        //        "qc" => "usdt",                   // quotation currency
        //        "t" => 1656043204763,             // happened time
        //        "ct" => 1656043204663,            // create time
        //        "i" => "6216559590087220004",     // $order id,
        //        "ci" => "test123",                // client $order id
        //        "st" => "PARTIALLY_FILLED",       // state NEW/PARTIALLY_FILLED/FILLED/CANCELED/REJECTED/EXPIRED
        //        "sd" => "BUY",                    // side BUY/SELL
        //        "tp" => "LIMIT",                  // type LIMIT/MARKET
        //        "oq" =>  "4"                      // original quantity
        //        "oqq" =>  48000,                  // original quotation quantity
        //        "eq" => "2",                      // executed quantity
        //        "lq" => "2",                      // remaining quantity
        //        "p" => "4000",                    // price
        //        "ap" => "30000",                  // avg price
        //        "f":"0.002"                     // fee
        //    }
        //
        // contract
        //
        //    {
        //        "symbol" => "btc_usdt",                    // Trading pair
        //        "orderId" => "1234",                       // Order Id
        //        "origQty" => "34244",                      // Original Quantity
        //        "avgPrice" => "123",                       // Quantity
        //        "price" => "1111",                         // Average price
        //        "executedQty" => "34244",                  // Volume (Cont)
        //        "orderSide" => "BUY",                      // BUY, SELL
        //        "positionSide" => "LONG",                  // LONG, SHORT
        //        "marginFrozen" => "123",                   // Occupied margin
        //        "sourceType" => "default",                 // DEFAULT:normal $order,ENTRUST:plan commission,PROFIR:Take Profit and Stop Loss
        //        "sourceId" : "1231231",                  // Triggering conditions ID
        //        "state" => "",                             // state:NEW：New $order (unfilled);PARTIALLY_FILLED:Partial deal;PARTIALLY_CANCELED:Partial revocation;FILLED:Filled;CANCELED:Cancled;REJECTED:Order failed;EXPIRED：Expired
        //        "createTime" => 1731231231,                // CreateTime
        //        "clientOrderId" => "204788317630342726"
        //    }
        //
        $marketId = $this->safe_string_2($order, 's', 'symbol');
        $tradeType = (is_array($order) && array_key_exists('symbol', $order)) ? 'contract' : 'spot';
        $market = $this->safe_market($marketId, $market, null, $tradeType);
        $timestamp = $this->safe_integer_2($order, 'ct', 'createTime');
        return $this->safe_order(array(
            'info' => $order,
            'id' => $this->safe_string_2($order, 'i', 'orderId'),
            'clientOrderId' => $this->safe_string_2($order, 'ci', 'clientOrderId'),
            'timestamp' => $timestamp,
            'datetime' => $this->iso8601($timestamp),
            'lastTradeTimestamp' => null,
            'symbol' => $market['symbol'],
            'type' => $market['type'],
            'timeInForce' => null,
            'postOnly' => null,
            'side' => $this->safe_string_lower_2($order, 'sd', 'orderSide'),
            'price' => $this->safe_number_2($order, 'p', 'price'),
            'stopPrice' => null,
            'stopLoss' => null,
            'takeProfit' => null,
            'amount' => $this->safe_string_2($order, 'oq', 'origQty'),
            'filled' => $this->safe_string_2($order, 'eq', 'executedQty'),
            'remaining' => $this->safe_string($order, 'lq'),
            'cost' => null,
            'average' => $this->safe_string_2($order, 'ap', 'avgPrice'),
            'status' => $this->parse_order_status($this->safe_string($order, 'st', 'state')),
            'fee' => array(
                'currency' => null,
                'cost' => $this->safe_number($order, 'f'),
            ),
            'trades' => null,
        ), $market);
    }

    public function handle_order(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        "topic" => "order",
        //        "event" => "order",
        //        "data" => {
        //            "s" => "btc_usdt",                // symbol
        //            "t" => 1656043204763,             // time happened time
        //            "i" => "6216559590087220004",     // orderId,
        //            "ci" => "test123",                // clientOrderId
        //            "st" => "PARTIALLY_FILLED",       // state
        //            "sd" => "BUY",                    // side BUY/SELL
        //            "eq" => "2",                      // executedQty executed quantity
        //            "ap" => "30000",                  // avg price
        //            "f" => "0.002"                    // fee
        //        }
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "order",
        //        "event" => "order@123456",
        //        "data" => {
        //             "symbol" => "btc_usdt",                    // Trading pair
        //             "orderId" => "1234",                       // Order Id
        //             "origQty" => "34244",                      // Original Quantity
        //             "avgPrice" => "123",                       // Quantity
        //             "price" => "1111",                         // Average price
        //             "executedQty" => "34244",                  // Volume (Cont)
        //             "orderSide" => "BUY",                      // BUY, SELL
        //             "positionSide" => "LONG",                  // LONG, SHORT
        //             "marginFrozen" => "123",                   // Occupied margin
        //             "sourceType" => "default",                 // DEFAULT:normal $order,ENTRUST:plan commission,PROFIR:Take Profit and Stop Loss
        //             "sourceId" : "1231231",                  // Triggering conditions ID
        //             "state" => "",                             // state:NEW：New $order (unfilled);PARTIALLY_FILLED:Partial deal;PARTIALLY_CANCELED:Partial revocation;FILLED:Filled;CANCELED:Cancled;REJECTED:Order failed;EXPIRED：Expired
        //             "createTime" => 1731231231,                // CreateTime
        //             "clientOrderId" => "204788317630342726"
        //           }
        //    }
        //
        $orders = $this->orders;
        if ($orders === null) {
            $limit = $this->safe_integer($this->options, 'ordersLimit');
            $orders = new ArrayCacheBySymbolById ($limit);
            $this->orders = $orders;
        }
        $order = $this->safe_dict($message, 'data', array());
        $marketId = $this->safe_string_2($order, 's', 'symbol');
        if ($marketId !== null) {
            $tradeType = (is_array($order) && array_key_exists('symbol', $order)) ? 'contract' : 'spot';
            $market = $this->safe_market($marketId, null, null, $tradeType);
            $parsed = $this->parse_ws_order($order, $market);
            $orders->append ($parsed);
            $client->resolve ($orders, 'order::' . $tradeType);
        }
        return $message;
    }

    public function handle_balance(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        topic => 'balance',
        //        event => 'balance',
        //        $data => {
        //            a => 3513677381884,
        //            t => 1684250056775,
        //            c => 'usdt',
        //            b => '7.71000000',
        //            f => '0.00000000',
        //            z => 'SPOT'
        //        }
        //    }
        //
        // contract
        //
        //    {
        //        "topic" => "balance",
        //        "event" => "balance@123456",
        //        "data" => {
        //            "coin" => "usdt",
        //            "underlyingType" => 1,                          // 1:Coin-M,2:USDT-M
        //            "walletBalance" => "123",                       // Balance
        //            "openOrderMarginFrozen" => "123",               // Frozen order
        //            "isolatedMargin" => "213",                      // Isolated Margin
        //            "crossedMargin" => "0"                          // Crossed Margin
        //            "availableBalance" => '2.256114450000000000',
        //            "coupon" => '0',
        //            "bonus" => '0'
        //        }
        //    }
        //
        $data = $this->safe_dict($message, 'data', array());
        $currencyId = $this->safe_string_2($data, 'c', 'coin');
        $code = $this->safe_currency_code($currencyId);
        $account = $this->account();
        $account['free'] = $this->safe_string($data, 'availableBalance');
        $account['used'] = $this->safe_string($data, 'f');
        $account['total'] = $this->safe_string_2($data, 'b', 'walletBalance');
        $this->balance[$code] = $account;
        $this->balance = $this->safe_balance($this->balance);
        $tradeType = (is_array($data) && array_key_exists('coin', $data)) ? 'contract' : 'spot';
        $client->resolve ($this->balance, 'balance::' . $tradeType);
    }

    public function handle_my_trades(Client $client, array $message) {
        //
        // spot
        //
        //    {
        //        "topic" => "trade",
        //        "event" => "trade",
        //        "data" => {
        //            "s" => "btc_usdt",                // symbol
        //            "t" => 1656043204763,             // time
        //            "i" => "6316559590087251233",     // tradeId
        //            "oi" => "6216559590087220004",    // orderId
        //            "p" => "30000",                   // trade price
        //            "q" => "3",                       // qty quantity
        //            "v" => "90000"                    // volume trade amount
        //        }
        //    }
        //
        // contract
        //
        //    {
        //       "topic" => "trade",
        //       "event" => "trade@123456",
        //       "data" => {
        //            "symbol" => 'btc_usdt',
        //            "orderSide" => 'SELL',
        //            "positionSide" => 'LONG',
        //            "orderId" => '231485367663419328',
        //            "price" => '27152.7',
        //            "quantity" => '33',
        //            "marginUnfrozen" => '2.85318000',
        //            "timestamp" => 1684892412565
        //        }
        //    }
        //
        $data = $this->safe_dict($message, 'data', array());
        $stored = $this->myTrades;
        if ($stored === null) {
            $limit = $this->safe_integer($this->options, 'tradesLimit', 1000);
            $stored = new ArrayCacheBySymbolById ($limit);
            $this->myTrades = $stored;
        }
        $parsedTrade = $this->parse_trade($data);
        $market = $this->market($parsedTrade['symbol']);
        $stored->append ($parsedTrade);
        $tradeType = $market['contract'] ? 'contract' : 'spot';
        $client->resolve ($stored, 'trade::' . $tradeType);
    }

    public function handle_message(Client $client, $message) {
        $event = $this->safe_string($message, 'event');
        if ($event === 'pong') {
            $client->onPong ();
        } elseif ($event !== null) {
            $topic = $this->safe_string($message, 'topic');
            $methods = array(
                'kline' => array($this, 'handle_ohlcv'),
                'depth' => array($this, 'handle_order_book'),
                'depth_update' => array($this, 'handle_order_book'),
                'ticker' => array($this, 'handle_ticker'),
                'agg_ticker' => array($this, 'handle_ticker'),
                'tickers' => array($this, 'handle_tickers'),
                'agg_tickers' => array($this, 'handle_tickers'),
                'balance' => array($this, 'handle_balance'),
                'order' => array($this, 'handle_order'),
                'position' => array($this, 'handle_position'),
            );
            $method = $this->safe_value($methods, $topic);
            if ($topic === 'trade') {
                $data = $this->safe_dict($message, 'data');
                if ((is_array($data) && array_key_exists('oi', $data)) || (is_array($data) && array_key_exists('orderId', $data))) {
                    $method = array($this, 'handle_my_trades');
                } else {
                    $method = array($this, 'handle_trade');
                }
            }
            if ($method !== null) {
                $method($client, $message);
            }
        }
    }

    public function ping(Client $client) {
        $client->lastPong = $this->milliseconds();
        return 'ping';
    }

    public function handle_error_message(Client $client, array $message) {
        //
        //    {
        //        "id" => "123",
        //        "code" => 401,
        //        "msg" => "token expire"
        //    }
        //
        $msg = $this->safe_string($message, 'msg');
        if (($msg === 'invalid_listen_key') || ($msg === 'token expire')) {
            $client->subscriptions['token'] = null;
            $this->get_listen_key(true);
            return;
        }
        $client->reject ($message);
    }
}
