<?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 ccxt\ExchangeError;
use ccxt\ArgumentsRequired;
use \React\Async;
use \React\Promise\PromiseInterface;

class kucoinfutures extends \ccxt\async\kucoinfutures {

    public function describe(): mixed {
        return $this->deep_extend(parent::describe(), array(
            'has' => array(
                'ws' => true,
                'watchLiquidations' => false,
                'watchLiquidatinsForSymbols' => false,
                'watchMyLiquidations' => null,
                'watchMyLiquidationsForSymbols' => null,
                'watchTicker' => true,
                'watchTickers' => true,
                'watchBidsAsks' => true,
                'watchTrades' => true,
                'watchOHLCV' => true,
                'watchOrderBook' => true,
                'watchOrders' => true,
                'watchBalance' => true,
                'watchPosition' => true,
                'watchPositions' => false,
                'watchPositionForSymbols' => false,
                'watchTradesForSymbols' => true,
                'watchOrderBookForSymbols' => true,
            ),
            'options' => array(
                'timeframes' => array(
                    '1m' => '1min',
                    '3m' => '1min',
                    '5m' => '5min',
                    '15m' => '15min',
                    '30m' => '30min',
                    '1h' => '1hour',
                    '2h' => '2hour',
                    '4h' => '4hour',
                    '8h' => '8hour',
                    '12h' => '12hour',
                    '1d' => '1day',
                    '1w' => '1week',
                    '1M' => '1month',
                ),
                'accountsByType' => array(
                    'swap' => 'future',
                    'cross' => 'margin',
                    // 'spot' => ,
                    // 'margin' => ,
                    // 'main' => ,
                    // 'funding' => ,
                    // 'future' => ,
                    // 'mining' => ,
                    // 'trade' => ,
                    // 'contract' => ,
                    // 'pool' => ,
                ),
                'tradesLimit' => 1000,
                'watchOrderBook' => array(
                    'snapshotDelay' => 20,
                    'snapshotMaxRetries' => 3,
                ),
                'watchPosition' => array(
                    'fetchPositionSnapshot' => true, // or false
                    'awaitPositionSnapshot' => true, // whether to wait for the position snapshot before providing updates
                ),
            ),
            'streaming' => array(
                // kucoin does not support built-in ws protocol-level ping-pong
                // instead it requires a custom json-based text ping-pong
                // https://docs.kucoin.com/#ping
                'ping' => array($this, 'ping'),
            ),
        ));
    }

    public function negotiate($privateChannel, $params = array ()) {
        return Async\async(function () use ($privateChannel, $params) {
            $connectId = $privateChannel ? 'private' : 'public';
            $urls = $this->safe_value($this->options, 'urls', array());
            $future = $this->safe_value($urls, $connectId);
            if ($future !== null) {
                return Async\await($future);
            }
            // we store an awaitable to the url
            // so that multiple calls don't asynchronously
            // fetch different $urls and overwrite each other
            $urls[$connectId] = $this->spawn(array($this, 'negotiate_helper'), $privateChannel, $params); // we have to wait here otherwsie in c# will not work
            $this->options['urls'] = $urls;
            $future = $urls[$connectId];
            return Async\await($future);
        }) ();
    }

    public function negotiate_helper($privateChannel, $params = array ()) {
        return Async\async(function () use ($privateChannel, $params) {
            $response = null;
            $connectId = $privateChannel ? 'private' : 'public';
            try {
                if ($privateChannel) {
                    $response = Async\await($this->futuresPrivatePostBulletPrivate ($params));
                    //
                    //     {
                    //         "code" => "200000",
                    //         "data" => {
                    //             "instanceServers" => array(
                    //                 {
                    //                     "pingInterval" =>  50000,
                    //                     "endpoint" => "wss://push-private.kucoin.com/endpoint",
                    //                     "protocol" => "websocket",
                    //                     "encrypt" => true,
                    //                     "pingTimeout" => 10000
                    //                 }
                    //             ),
                    //             "token" => "2neAiuYvAU61ZDXANAGAsiL4-iAExhsBXZxftpOeh_55i3Ysy2q2LEsEWU64mdzUOPusi34M_wGoSf7iNyEWJ1UQy47YbpY4zVdzilNP-Bj3iXzrjjGlWtiYB9J6i9GjsxUuhPw3BlrzazF6ghq4Lzf7scStOz3KkxjwpsOBCH4=.WNQmhZQeUKIkh97KYgU0Lg=="
                    //         }
                    //     }
                    //
                } else {
                    $response = Async\await($this->futuresPublicPostBulletPublic ($params));
                }
                $data = $this->safe_value($response, 'data', array());
                $instanceServers = $this->safe_value($data, 'instanceServers', array());
                $firstInstanceServer = $this->safe_value($instanceServers, 0);
                $pingInterval = $this->safe_integer($firstInstanceServer, 'pingInterval');
                $endpoint = $this->safe_string($firstInstanceServer, 'endpoint');
                $token = $this->safe_string($data, 'token');
                $result = $endpoint . '?' . $this->urlencode(array(
                    'token' => $token,
                    'privateChannel' => $privateChannel,
                    'connectId' => $connectId,
                ));
                $client = $this->client($result);
                $client->keepAlive = $pingInterval;
                return $result;
            } catch (Exception $e) {
                $future = $this->safe_value($this->options['urls'], $connectId);
                $future->reject ($e);
                unset($this->options['urls'][$connectId]);
            }
            return null;
        }) ();
    }

    public function request_id() {
        $requestId = $this->sum($this->safe_integer($this->options, 'requestId', 0), 1);
        $this->options['requestId'] = $requestId;
        return $requestId;
    }

    public function subscribe($url, $messageHash, $subscriptionHash, $subscription, $params = array ()) {
        return Async\async(function () use ($url, $messageHash, $subscriptionHash, $subscription, $params) {
            $requestId = (string) $this->request_id();
            $request = array(
                'id' => $requestId,
                'type' => 'subscribe',
                'topic' => $subscriptionHash,
                'response' => true,
            );
            $message = $this->extend($request, $params);
            $subscriptionRequest = array(
                'id' => $requestId,
            );
            if ($subscription === null) {
                $subscription = $subscriptionRequest;
            } else {
                $subscription = $this->extend($subscriptionRequest, $subscription);
            }
            return Async\await($this->watch($url, $messageHash, $message, $subscriptionHash, $subscription));
        }) ();
    }

    public function subscribe_multiple($url, $messageHashes, $topic, $subscriptionHashes, $subscriptionArgs, $params = array ()) {
        return Async\async(function () use ($url, $messageHashes, $topic, $subscriptionHashes, $subscriptionArgs, $params) {
            $requestId = (string) $this->request_id();
            $request = array(
                'id' => $requestId,
                'type' => 'subscribe',
                'topic' => $topic,
                'response' => true,
            );
            return Async\await($this->watch_multiple($url, $messageHashes, $this->extend($request, $params), $subscriptionHashes, $subscriptionArgs));
        }) ();
    }

    public function un_subscribe_multiple($url, $messageHashes, $topic, $subscriptionHashes, $params = array (), ?array $subscription = null) {
        return Async\async(function () use ($url, $messageHashes, $topic, $subscriptionHashes, $params, $subscription) {
            $requestId = (string) $this->request_id();
            $request = array(
                'id' => $requestId,
                'type' => 'unsubscribe',
                'topic' => $topic,
                'response' => true,
            );
            $message = $this->extend($request, $params);
            if ($subscription !== null) {
                $subscription[$requestId] = $requestId;
            }
            $client = $this->client($url);
            for ($i = 0; $i < count($subscriptionHashes); $i++) {
                $subscriptionHash = $subscriptionHashes[$i];
                if (!(is_array($client->subscriptions) && array_key_exists($subscriptionHash, $client->subscriptions))) {
                    $client->subscriptions[$requestId] = $subscriptionHash;
                }
            }
            return Async\await($this->watch_multiple($url, $messageHashes, $message, $subscriptionHashes, $subscription));
        }) ();
    }

    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://www.kucoin.com/docs/websocket/futures-trading/public-channels/get-ticker
             *
             * @param {string} $symbol unified $symbol of the $market to fetch the ticker for
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} a ~@link https://docs.ccxt.com/#/?id=ticker-structure ticker structure~
             */
            Async\await($this->load_markets());
            $market = $this->market($symbol);
            $symbol = $market['symbol'];
            $params['callerMethodName'] = 'watchTicker';
            $tickers = Async\await($this->watch_tickers(array( $symbol ), $params));
            return $tickers[$symbol];
        }) ();
    }

    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 all markets of a specific list
             * @param {string[]} $symbols unified symbol of the market to fetch the $ticker for
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} a ~@link https://docs.ccxt.com/#/?id=$ticker-structure $ticker structure~
             */
            Async\await($this->load_markets());
            $ticker = Async\await($this->watch_multi_request('watchTickers', '/contractMarket/ticker:', $symbols, $params));
            if ($this->newUpdates) {
                $tickers = array();
                $tickers[$ticker['symbol']] = $ticker;
                return $tickers;
            }
            return $this->filter_by_array($this->tickers, 'symbol', $symbols);
        }) ();
    }

    public function handle_ticker(Client $client, $message) {
        //
        // $ticker (v1)
        //
        //    {
        //     "subject" => "ticker",
        //     "topic" => "/contractMarket/ticker:XBTUSDM",
        //     "data" => {
        //         "symbol" => "XBTUSDM", //Market of the symbol
        //         "sequence" => 45, //Sequence number which is used to judge the continuity of the pushed messages
        //         "side" => "sell", //Transaction side of the last traded taker order
        //         "price" => "3600.0", //Filled price
        //         "size" => 16, //Filled quantity
        //         "tradeId" => "5c9dcf4170744d6f5a3d32fb", //Order ID
        //         "bestBidSize" => 795, //Best bid size
        //         "bestBidPrice" => "3200.0", //Best bid
        //         "bestAskPrice" => "3600.0", //Best ask size
        //         "bestAskSize" => 284, //Best ask
        //         "ts" => 1553846081210004941 //Filled time - nanosecond
        //     }
        //    }
        //
        $data = $this->safe_value($message, 'data', array());
        $marketId = $this->safe_value($data, 'symbol');
        $market = $this->safe_market($marketId, null, '-');
        $ticker = $this->parse_ticker($data, $market);
        $this->tickers[$market['symbol']] = $ticker;
        $client->resolve ($ticker, $this->get_message_hash('ticker', $market['symbol']));
    }

    public function watch_bids_asks(?array $symbols = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $params) {
            /**
             *
             * @see https://www.kucoin.com/docs/websocket/futures-trading/public-channels/get-$ticker-v2
             *
             * watches best bid & ask for $symbols
             * @param {string[]} $symbols unified symbol of the market to fetch the $ticker for
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} a ~@link https://docs.ccxt.com/#/?id=$ticker-structure $ticker structure~
             */
            $ticker = Async\await($this->watch_multi_request('watchBidsAsks', '/contractMarket/tickerV2:', $symbols, $params));
            if ($this->newUpdates) {
                $tickers = array();
                $tickers[$ticker['symbol']] = $ticker;
                return $tickers;
            }
            return $this->filter_by_array($this->bidsasks, 'symbol', $symbols);
        }) ();
    }

    public function watch_multi_request($methodName, string $channelName, ?array $symbols = null, $params = array ()) {
        return Async\async(function () use ($methodName, $channelName, $symbols, $params) {
            Async\await($this->load_markets());
            list($methodName, $params) = $this->handle_param_string($params, 'callerMethodName', $methodName);
            $isBidsAsks = ($methodName === 'watchBidsAsks');
            $symbols = $this->market_symbols($symbols, null, false, true, false);
            $length = count($symbols);
            if ($length > 100) {
                throw new ArgumentsRequired($this->id . ' ' . $methodName . '() accepts a maximum of 100 symbols');
            }
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $market = $this->market($symbol);
                $prefix = $isBidsAsks ? 'bidask' : 'ticker';
                $messageHashes[] = $this->get_message_hash($prefix, $market['symbol']);
            }
            $url = Async\await($this->negotiate(false));
            $marketIds = $this->market_ids($symbols);
            $joined = implode(',', $marketIds);
            $requestId = (string) $this->request_id();
            $request = array(
                'id' => $requestId,
                'type' => 'subscribe',
                'topic' => $channelName . $joined,
                'response' => true,
            );
            $subscription = array(
                'id' => $requestId,
            );
            return Async\await($this->watch_multiple($url, $messageHashes, $this->extend($request, $params), $messageHashes, $subscription));
        }) ();
    }

    public function handle_bid_ask(Client $client, $message) {
        //
        // arrives one $symbol dict
        //
        // {
        //   "subject" => "tickerV2",
        //   "topic" => "/contractMarket/tickerV2:XBTUSDM",
        //   "data" => {
        //     "symbol" => "XBTUSDM", //Market of the $symbol
        //     "bestBidSize" => 795, // Best bid size
        //     "bestBidPrice" => 3200.0, // Best bid
        //     "bestAskPrice" => 3600.0, // Best ask
        //     "bestAskSize" => 284, // Best ask size
        //     "ts" => 1553846081210004941 // Filled time - nanosecond
        //   }
        // }
        //
        $parsedTicker = $this->parse_ws_bid_ask($message);
        $symbol = $parsedTicker['symbol'];
        $this->bidsasks[$symbol] = $parsedTicker;
        $client->resolve ($parsedTicker, $this->get_message_hash('bidask', $symbol));
    }

    public function parse_ws_bid_ask($ticker, $market = null) {
        $data = $this->safe_dict($ticker, 'data', array());
        $marketId = $this->safe_string($data, 'symbol');
        $market = $this->safe_market($marketId, $market);
        $symbol = $this->safe_string($market, 'symbol');
        $timestamp = $this->safe_integer_product($data, 'ts', 0.000001);
        return $this->safe_ticker(array(
            'symbol' => $symbol,
            'timestamp' => $timestamp,
            'datetime' => $this->iso8601($timestamp),
            'ask' => $this->safe_number($data, 'bestAskPrice'),
            'askVolume' => $this->safe_number($data, 'bestAskSize'),
            'bid' => $this->safe_number($data, 'bestBidPrice'),
            'bidVolume' => $this->safe_number($data, 'bestBidSize'),
            'info' => $ticker,
        ), $market);
    }

    public function watch_position(?string $symbol = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $params) {
            /**
             * watch open positions for a specific $symbol
             *
             * @see https://docs.kucoin.com/futures/#position-change-events
             *
             * @param {string|null} $symbol unified $market $symbol
             * @param {array} $params extra parameters specific to the exchange API endpoint
             * @return {array} a {@link https://docs.ccxt.com/en/latest/manual.html#position-structure position structure}
             */
            if ($symbol === null) {
                throw new ArgumentsRequired($this->id . ' watchPosition() requires a $symbol argument');
            }
            Async\await($this->load_markets());
            $url = Async\await($this->negotiate(true));
            $market = $this->market($symbol);
            $topic = '/contract/position:' . $market['id'];
            $request = array(
                'privateChannel' => true,
            );
            $messageHash = 'position:' . $market['symbol'];
            $client = $this->client($url);
            $this->set_position_cache($client, $symbol);
            $fetchPositionSnapshot = $this->handle_option('watchPosition', 'fetchPositionSnapshot', true);
            $awaitPositionSnapshot = $this->handle_option('watchPosition', 'awaitPositionSnapshot', true);
            $currentPosition = $this->get_current_position($symbol);
            if ($fetchPositionSnapshot && $awaitPositionSnapshot && $currentPosition === null) {
                $snapshot = Async\await($client->future ('fetchPositionSnapshot:' . $symbol));
                return $snapshot;
            }
            return Async\await($this->subscribe($url, $messageHash, $topic, null, $this->extend($request, $params)));
        }) ();
    }

    public function get_current_position($symbol) {
        if ($this->positions === null) {
            return null;
        }
        $cache = $this->positions.hashmap;
        $symbolCache = $this->safe_value($cache, $symbol, array());
        $values = is_array($symbolCache) ? array_values($symbolCache) : array();
        return $this->safe_value($values, 0);
    }

    public function set_position_cache(Client $client, string $symbol) {
        $fetchPositionSnapshot = $this->handle_option('watchPosition', 'fetchPositionSnapshot', false);
        if ($fetchPositionSnapshot) {
            $messageHash = 'fetchPositionSnapshot:' . $symbol;
            if (!(is_array($client->futures) && array_key_exists($messageHash, $client->futures))) {
                $client->future ($messageHash);
                $this->spawn(array($this, 'load_position_snapshot'), $client, $messageHash, $symbol);
            }
        }
    }

    public function load_position_snapshot($client, $messageHash, $symbol) {
        return Async\async(function () use ($client, $messageHash, $symbol) {
            $position = Async\await($this->fetch_position($symbol));
            $this->positions = new ArrayCacheBySymbolById ();
            $cache = $this->positions;
            $cache->append ($position);
            // don't remove the $future from the .futures $cache
            $future = $client->futures[$messageHash];
            $future->resolve ($cache);
            $client->resolve ($position, 'position:' . $symbol);
        }) ();
    }

    public function handle_position(Client $client, $message) {
        //
        // Position Changes Caused Operations
        //    {
        //        "type" => "message",
        //        "userId" => "5c32d69203aa676ce4b543c7", // Deprecated, will detele later
        //        "channelType" => "private",
        //        "topic" => "/contract/position:XBTUSDM",
        //        "subject" => "position.change",
        //        "data" => {
        //            "realisedGrossPnl" => 0E-8, //Accumulated realised profit and loss
        //            "symbol" => "XBTUSDM", //Symbol
        //            "crossMode" => false, //Cross mode or not
        //            "liquidationPrice" => 1000000.0, //Liquidation price
        //            "posLoss" => 0E-8, //Manually added margin amount
        //            "avgEntryPrice" => 7508.22, //Average entry price
        //            "unrealisedPnl" => -0.00014735, //Unrealised profit and loss
        //            "markPrice" => 7947.83, //Mark price
        //            "posMargin" => 0.00266779, //Position margin
        //            "autoDeposit" => false, //Auto deposit margin or not
        //            "riskLimit" => 100000, //Risk limit
        //            "unrealisedCost" => 0.00266375, //Unrealised value
        //            "posComm" => 0.00000392, //Bankruptcy cost
        //            "posMaint" => 0.00001724, //Maintenance margin
        //            "posCost" => 0.00266375, //Position value
        //            "maintMarginReq" => 0.005, //Maintenance margin rate
        //            "bankruptPrice" => 1000000.0, //Bankruptcy price
        //            "realisedCost" => 0.00000271, //Currently accumulated realised $position value
        //            "markValue" => 0.00251640, //Mark value
        //            "posInit" => 0.00266375, //Position margin
        //            "realisedPnl" => -0.00000253, //Realised profit and losts
        //            "maintMargin" => 0.00252044, //Position margin
        //            "realLeverage" => 1.06, //Leverage of the order
        //            "changeReason" => "positionChange", //changeReason:marginChange、positionChange、liquidation、autoAppendMarginStatusChange、adl
        //            "currentCost" => 0.00266375, //Current $position value
        //            "openingTimestamp" => 1558433191000, //Open time
        //            "currentQty" => -20, //Current $position
        //            "delevPercentage" => 0.52, //ADL ranking percentile
        //            "currentComm" => 0.00000271, //Current commission
        //            "realisedGrossCost" => 0E-8, //Accumulated reliased gross profit value
        //            "isOpen" => true, //Opened $position or not
        //            "posCross" => 1.2E-7, //Manually added margin
        //            "currentTimestamp" => 1558506060394, //Current timestamp
        //            "unrealisedRoePcnt" => -0.0553, //Rate of return on investment
        //            "unrealisedPnlPcnt" => -0.0553, //Position profit and loss ratio
        //            "settleCurrency" => "XBT" //Currency used to clear and settle the trades
        //        }
        //    }
        // Position Changes Caused by Mark Price
        //    {
        //        "userId" => "5cd3f1a7b7ebc19ae9558591", // Deprecated, will detele later
        //        "topic" => "/contract/position:XBTUSDM",
        //        "subject" => "position.change",
        //          "data" => {
        //              "markPrice" => 7947.83,                   //Mark price
        //              "markValue" => 0.00251640,                 //Mark value
        //              "maintMargin" => 0.00252044,              //Position margin
        //              "realLeverage" => 10.06,                   //Leverage of the order
        //              "unrealisedPnl" => -0.00014735,           //Unrealised profit and lost
        //              "unrealisedRoePcnt" => -0.0553,           //Rate of return on investment
        //              "unrealisedPnlPcnt" => -0.0553,            //Position profit and loss ratio
        //              "delevPercentage" => 0.52,             //ADL ranking percentile
        //              "currentTimestamp" => 1558087175068,      //Current timestamp
        //              "settleCurrency" => "XBT"                 //Currency used to clear and settle the trades
        //          }
        //    }
        //  Funding Settlement
        //    {
        //        "userId" => "xbc453tg732eba53a88ggyt8c", // Deprecated, will detele later
        //        "topic" => "/contract/position:XBTUSDM",
        //        "subject" => "position.settlement",
        //        "data" => {
        //            "fundingTime" => 1551770400000,          //Funding time
        //            "qty" => 100,                            //Position siz
        //            "markPrice" => 3610.85,                 //Settlement price
        //            "fundingRate" => -0.002966,             //Funding rate
        //            "fundingFee" => -296,                   //Funding fees
        //            "ts" => 1547697294838004923,             //Current time (nanosecond)
        //            "settleCurrency" => "XBT"                //Currency used to clear and settle the trades
        //        }
        //    }
        // Adjustmet result of risk limit level
        //     {
        //         "userId" => "xbc453tg732eba53a88ggyt8c",
        //         "topic" => "/contract/position:ADAUSDTM",
        //         "subject" => "position.adjustRiskLimit",
        //         "data" => {
        //           "success" => true, // Successful or not
        //           "riskLimitLevel" => 1, // Current risk limit level
        //           "msg" => "" // Failure reason
        //         }
        //     }
        //
        $topic = $this->safe_string($message, 'topic', '');
        $parts = explode(':', $topic);
        $marketId = $this->safe_string($parts, 1);
        $symbol = $this->safe_symbol($marketId, null, '');
        $cache = $this->positions;
        $currentPosition = $this->get_current_position($symbol);
        $messageHash = 'position:' . $symbol;
        $data = $this->safe_value($message, 'data', array());
        $newPosition = $this->parse_position($data);
        $keys = is_array($newPosition) ? array_keys($newPosition) : array();
        for ($i = 0; $i < count($keys); $i++) {
            $key = $keys[$i];
            if ($newPosition[$key] === null) {
                unset($newPosition[$key]);
            }
        }
        $position = $this->extend($currentPosition, $newPosition);
        $cache->append ($position);
        $client->resolve ($position, $messageHash);
    }

    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://docs.kucoin.com/futures/#execution-data
             *
             * @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 exchange API endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=public-trades trade structures~
             */
            return Async\await($this->watch_trades_for_symbols(array( $symbol ), $since, $limit, $params));
        }) ();
    }

    public function watch_trades_for_symbols(array $symbols, ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $since, $limit, $params) {
            /**
             * get the list of most recent $trades for a particular $symbol
             * @param {string[]} $symbols
             * @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 exchange API endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=public-$trades trade structures~
             */
            $symbolsLength = count($symbols);
            if ($symbolsLength === 0) {
                throw new ArgumentsRequired($this->id . ' watchTradesForSymbols() requires a non-empty array of symbols');
            }
            Async\await($this->load_markets());
            $symbols = $this->market_symbols($symbols);
            $url = Async\await($this->negotiate(false));
            $symbols = $this->market_symbols($symbols);
            $marketIds = $this->market_ids($symbols);
            $topic = '/contractMarket/execution:' . implode(',', $marketIds);
            $subscriptionHashes = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $marketId = $marketIds[$i];
                $messageHashes[] = 'trades:' . $symbol;
                $subscriptionHashes[] = '/contractMarket/execution:' . $marketId;
            }
            $trades = Async\await($this->subscribe_multiple($url, $messageHashes, $topic, $subscriptionHashes, null, $params));
            if ($this->newUpdates) {
                $first = $this->safe_value($trades, 0);
                $tradeSymbol = $this->safe_string($first, 'symbol');
                $limit = $trades->getLimit ($tradeSymbol, $limit);
            }
            return $this->filter_by_since_limit($trades, $since, $limit, 'timestamp', true);
        }) ();
    }

    public function un_watch_trades(string $symbol, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $params) {
            /**
             * unWatches trades stream
             *
             * @see https://docs.kucoin.com/futures/#execution-data
             *
             * @param {string} $symbol unified $symbol of the market to fetch trades for
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=public-trades trade structures~
             */
            return Async\await($this->un_watch_trades_for_symbols(array( $symbol ), $params));
        }) ();
    }

    public function un_watch_trades_for_symbols(array $symbols, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $params) {
            /**
             * get the list of most recent trades for a particular $symbol
             * @param {string[]} $symbols
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=public-trades trade structures~
             */
            Async\await($this->load_markets());
            $symbols = $this->market_symbols($symbols, null, false);
            $url = Async\await($this->negotiate(false));
            $symbols = $this->market_symbols($symbols);
            $marketIds = $this->market_ids($symbols);
            $topic = '/contractMarket/execution:' . implode(',', $marketIds);
            $subscriptionHashes = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $messageHashes[] = 'unsubscribe:trades:' . $symbol;
                $subscriptionHashes[] = 'trades:' . $symbol;
            }
            $subscription = array(
                'messageHashes' => $messageHashes,
                'subMessageHashes' => $subscriptionHashes,
                'topic' => 'trades',
                'unsubscribe' => true,
                'symbols' => $symbols,
            );
            return Async\await($this->un_subscribe_multiple($url, $messageHashes, $topic, $messageHashes, $params, $subscription));
        }) ();
    }

    public function handle_trade(Client $client, $message) {
        //
        //    {
        //        "type" => "message",
        //        "topic" => "/contractMarket/execution:ADAUSDTM",
        //        "subject" => "match",
        //        "data" => {
        //            "makerUserId" => "62286a4d720edf0001e81961",
        //            "symbol" => "ADAUSDTM",
        //            "sequence" => 41320766,
        //            "side" => "sell",
        //            "size" => 2,
        //            "price" => 0.35904,
        //            "takerOrderId" => "636dd9da9857ba00010cfa44",
        //            "makerOrderId" => "636dd9c8df149d0001e62bc8",
        //            "takerUserId" => "6180be22b6ab210001fa3371",
        //            "tradeId" => "636dd9da0000d400d477eca7",
        //            "ts" => 1668143578987357700
        //        }
        //    }
        //
        $data = $this->safe_value($message, 'data', array());
        $trade = $this->parse_trade($data);
        $symbol = $trade['symbol'];
        $trades = $this->safe_value($this->trades, $symbol);
        if ($trades === null) {
            $limit = $this->safe_integer($this->options, 'tradesLimit', 1000);
            $trades = new ArrayCache ($limit);
            $this->trades[$symbol] = $trades;
        }
        $trades->append ($trade);
        $messageHash = 'trades:' . $symbol;
        $client->resolve ($trades, $messageHash);
        return $message;
    }

    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) {
            /**
             *
             * @see https://www.kucoin.com/docs/websocket/futures-trading/public-channels/klines
             *
             * watches historical candlestick data containing the open, high, low, and close price, and the volume of a market
             * @param {string} $symbol unified $symbol of the market to fetch OHLCV data for
             * @param {string} $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 {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {int[][]} A list of candles ordered, open, high, low, close, volume
             */
            Async\await($this->load_markets());
            $symbol = $this->symbol($symbol);
            $url = Async\await($this->negotiate(false));
            $marketId = $this->market_id($symbol);
            $timeframes = $this->safe_dict($this->options, 'timeframes');
            $timeframeId = $this->safe_string($timeframes, $timeframe, $timeframe);
            $topic = '/contractMarket/limitCandle:' . $marketId . '_' . $timeframeId;
            $messageHash = 'ohlcv::' . $symbol . '_' . $timeframe;
            $ohlcv = Async\await($this->subscribe($url, $messageHash, $topic, null, $params));
            if ($this->newUpdates) {
                $limit = $ohlcv->getLimit ($symbol, $limit);
            }
            return $this->filter_by_since_limit($ohlcv, $since, $limit, 0, true);
        }) ();
    }

    public function handle_ohlcv(Client $client, $message) {
        //
        //    {
        //        "topic":"/contractMarket/limitCandle:LTCUSDTM_1min",
        //        "type":"message",
        //        "data":array(
        //            "symbol":"LTCUSDTM",
        //            "candles":array(
        //                "1715470980",
        //                "81.38",
        //                "81.38",
        //                "81.38",
        //                "81.38",
        //                "61.0",
        //                "61"
        //            ),
        //            "time":1715470994801
        //        ),
        //        "subject":"candle.stick"
        //    }
        //
        $topic = $this->safe_string($message, 'topic');
        $parts = explode('_', $topic);
        $timeframeId = $this->safe_string($parts, 1);
        $data = $this->safe_dict($message, 'data');
        $timeframes = $this->safe_dict($this->options, 'timeframes');
        $timeframe = $this->find_timeframe($timeframeId, $timeframes);
        $marketId = $this->safe_string($data, 'symbol');
        $symbol = $this->safe_symbol($marketId);
        $messageHash = 'ohlcv::' . $symbol . '_' . $timeframe;
        $ohlcv = $this->safe_list($data, 'candles');
        $parsed = array(
            $this->safe_integer($ohlcv, 0),
            $this->safe_number($ohlcv, 1),
            $this->safe_number($ohlcv, 2),
            $this->safe_number($ohlcv, 3),
            $this->safe_number($ohlcv, 4),
            $this->safe_number($ohlcv, 6), // Note value 5 is incorrect and will be fixed in subsequent versions of kucoin
        );
        $this->ohlcvs[$symbol] = $this->safe_dict($this->ohlcvs, $symbol, array());
        if (!(is_array($this->ohlcvs[$symbol]) && array_key_exists($timeframe, $this->ohlcvs[$symbol]))) {
            $limit = $this->safe_integer($this->options, 'OHLCVLimit', 1000);
            $this->ohlcvs[$symbol][$timeframe] = new ArrayCacheByTimestamp ($limit);
        }
        $stored = $this->ohlcvs[$symbol][$timeframe];
        $stored->append ($parsed);
        $client->resolve ($stored, $messageHash);
    }

    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
             *   1. After receiving the websocket Level 2 data flow, cache the data.
             *   2. Initiate a REST request to get the snapshot data of Level 2 order book.
             *   3. Playback the cached Level 2 data flow.
             *   4. Apply the new Level 2 data flow to the local snapshot to ensure that the sequence of the new Level 2 update lines up with the sequence of the previous Level 2 data. Discard all the message prior to that sequence, and then playback the change to snapshot.
             *   5. Update the level2 full data based on sequence according to the size. If the price is 0, ignore the messages and update the sequence. If the size=0, update the sequence and remove the price of which the size is 0 out of level 2. For other cases, please update the price.
             *   6. If the sequence of the newly pushed message does not line up to the sequence of the last message, you could pull through REST Level 2 message request to get the updated messages. Please note that the difference between the start and end parameters cannot exceed 500.
             *
             * @see https://docs.kucoin.com/futures/#level-2-market-data
             *
             * @param {string} $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 {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} A dictionary of ~@link https://docs.ccxt.com/#/?id=order-book-structure order book structures~ indexed by market symbols
             */
            return Async\await($this->watch_order_book_for_symbols(array( $symbol ), $limit, $params));
        }) ();
    }

    public function watch_order_book_for_symbols(array $symbols, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $limit, $params) {
            /**
             * watches information on open orders with bid (buy) and ask (sell) prices, volumes and other data
             *
             * @see https://docs.kucoin.com/futures/#level-2-market-data
             *
             * @param {string[]} $symbols unified array of $symbols
             * @param {int} [$limit] the maximum amount of order book entries to return
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} A dictionary of ~@link https://docs.ccxt.com/#/?id=order-book-structure order book structures~ indexed by market $symbols
             */
            $symbolsLength = count($symbols);
            if ($symbolsLength === 0) {
                throw new ArgumentsRequired($this->id . ' watchOrderBookForSymbols() requires a non-empty array of symbols');
            }
            if ($limit !== null) {
                if (($limit !== 20) && ($limit !== 100)) {
                    throw new ExchangeError($this->id . " watchOrderBook 'limit' argument must be null, 20 or 100");
                }
            }
            Async\await($this->load_markets());
            $symbols = $this->market_symbols($symbols);
            $marketIds = $this->market_ids($symbols);
            $url = Async\await($this->negotiate(false));
            $topic = '/contractMarket/level2:' . implode(',', $marketIds);
            $subscriptionArgs = array(
                'limit' => $limit,
            );
            $subscriptionHashes = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $marketId = $marketIds[$i];
                $messageHashes[] = 'orderbook:' . $symbol;
                $subscriptionHashes[] = '/contractMarket/level2:' . $marketId;
            }
            $orderbook = Async\await($this->subscribe_multiple($url, $messageHashes, $topic, $subscriptionHashes, $subscriptionArgs, $params));
            return $orderbook->limit ();
        }) ();
    }

    public function un_watch_order_book(string $symbol, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $params) {
            /**
             * unWatches information on open orders with bid (buy) and ask (sell) prices, volumes and other data
             *
             * @see https://docs.kucoin.com/futures/#level-2-market-data
             *
             * @param {string} $symbol unified $symbol of the market to fetch the order book for
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} A dictionary of ~@link https://docs.ccxt.com/#/?id=order-book-structure order book structures~ indexed by market symbols
             */
            return Async\await($this->un_watch_order_book_for_symbols(array( $symbol ), $params));
        }) ();
    }

    public function un_watch_order_book_for_symbols(array $symbols, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbols, $params) {
            /**
             * unWatches information on open orders with bid (buy) and ask (sell) prices, volumes and other data
             * @param {string[]} $symbols unified array of $symbols
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} A dictionary of ~@link https://docs.ccxt.com/#/?id=order-book-structure order book structures~ indexed by market $symbols
             */
            Async\await($this->load_markets());
            $symbols = $this->market_symbols($symbols);
            $marketIds = $this->market_ids($symbols);
            $url = Async\await($this->negotiate(false));
            $topic = '/contractMarket/level2:' . implode(',', $marketIds);
            $subscriptionHashes = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $messageHashes[] = 'unsubscribe:orderbook:' . $symbol;
                $subscriptionHashes[] = 'orderbook:' . $symbol;
            }
            $subscription = array(
                'messageHashes' => $messageHashes,
                'symbols' => $symbols,
                'unsubscribe' => true,
                'topic' => 'orderbook',
                'subMessageHashes' => $subscriptionHashes,
            );
            return Async\await($this->un_subscribe_multiple($url, $messageHashes, $topic, $messageHashes, $params, $subscription));
        }) ();
    }

    public function handle_delta($orderbook, $delta) {
        $orderbook['nonce'] = $this->safe_integer($delta, 'sequence');
        $timestamp = $this->safe_integer($delta, 'timestamp');
        $orderbook['timestamp'] = $timestamp;
        $orderbook['datetime'] = $this->iso8601($timestamp);
        $change = $this->safe_value($delta, 'change', array());
        $splitChange = explode(',', $change);
        $price = $this->safe_number($splitChange, 0);
        $side = $this->safe_string($splitChange, 1);
        $quantity = $this->safe_number($splitChange, 2);
        $type = ($side === 'buy') ? 'bids' : 'asks';
        $value = array( $price, $quantity );
        if ($type === 'bids') {
            $storedBids = $orderbook['bids'];
            $storedBids->storeArray ($value);
        } else {
            $storedAsks = $orderbook['asks'];
            $storedAsks->storeArray ($value);
        }
    }

    public function handle_deltas($bookside, $deltas) {
        for ($i = 0; $i < count($deltas); $i++) {
            $this->handle_delta($bookside, $deltas[$i]);
        }
    }

    public function handle_order_book(Client $client, $message) {
        //
        // initial snapshot is fetched with ccxt's fetchOrderBook
        // the feed does not include a snapshot, just the deltas
        //
        //    {
        //        "type" => "message",
        //        "topic" => "/contractMarket/level2:ADAUSDTM",
        //        "subject" => "level2",
        //        "data" => {
        //            "sequence" => 1668059586457,
        //            "change" => "0.34172,sell,456", // type, side, quantity
        //            "timestamp" => 1668573023223
        //        }
        //    }
        //
        $data = $this->safe_value($message, 'data');
        $topic = $this->safe_string($message, 'topic');
        $topicParts = explode(':', $topic);
        $marketId = $this->safe_string($topicParts, 1);
        $symbol = $this->safe_symbol($marketId, null, '-');
        $messageHash = 'orderbook:' . $symbol;
        if (!(is_array($this->orderbooks) && array_key_exists($symbol, $this->orderbooks))) {
            $subscriptionArgs = $this->safe_dict($client->subscriptions, $topic, array());
            $limit = $this->safe_integer($subscriptionArgs, 'limit');
            $this->orderbooks[$symbol] = $this->order_book(array(), $limit);
        }
        $storedOrderBook = $this->orderbooks[$symbol];
        $nonce = $this->safe_integer($storedOrderBook, 'nonce');
        $deltaEnd = $this->safe_integer($data, 'sequence');
        if ($nonce === null) {
            $cacheLength = count($storedOrderBook->cache);
            $topicPartsNew = explode(':', $topic);
            $topicSymbol = $this->safe_string($topicPartsNew, 1);
            $topicChannel = $this->safe_string($topicPartsNew, 0);
            $subscriptions = is_array($client->subscriptions) ? array_keys($client->subscriptions) : array();
            $subscription = null;
            for ($i = 0; $i < count($subscriptions); $i++) {
                $key = $subscriptions[$i];
                if ((mb_strpos($key, $topicSymbol) !== false) && (mb_strpos($key, $topicChannel) !== false)) {
                    $subscription = $client->subscriptions[$key];
                    break;
                }
            }
            $limit = $this->safe_integer($subscription, 'limit');
            $snapshotDelay = $this->handle_option('watchOrderBook', 'snapshotDelay', 5);
            if ($cacheLength === $snapshotDelay) {
                $this->spawn(array($this, 'load_order_book'), $client, $messageHash, $symbol, $limit, array());
            }
            $storedOrderBook->cache[] = $data;
            return;
        } elseif ($nonce >= $deltaEnd) {
            return;
        }
        $this->handle_delta($storedOrderBook, $data);
        $client->resolve ($storedOrderBook, $messageHash);
    }

    public function get_cache_index($orderbook, $cache) {
        $firstDelta = $this->safe_value($cache, 0);
        $nonce = $this->safe_integer($orderbook, 'nonce');
        $firstDeltaStart = $this->safe_integer($firstDelta, 'sequence');
        if ($nonce < $firstDeltaStart - 1) {
            return -1;
        }
        for ($i = 0; $i < count($cache); $i++) {
            $delta = $cache[$i];
            $deltaStart = $this->safe_integer($delta, 'sequence');
            if ($nonce < $deltaStart - 1) {
                return $i;
            }
        }
        return count($cache);
    }

    public function handle_system_status(Client $client, $message) {
        //
        // todo => answer the question whether handleSystemStatus should be renamed
        // and unified for any usage pattern that
        // involves system status and maintenance updates
        //
        //     {
        //         "id" => "1578090234088", // connectId
        //         "type" => "welcome",
        //     }
        //
        return $message;
    }

    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://docs.kucoin.com/futures/#trade-$orders-according-to-the-$market
             *
             * @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 order structures to retrieve
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=order-structure order structures~
             */
            Async\await($this->load_markets());
            $url = Async\await($this->negotiate(true));
            $topic = '/contractMarket/tradeOrders';
            $request = array(
                'privateChannel' => true,
            );
            $messageHash = 'orders';
            if ($symbol !== null) {
                $market = $this->market($symbol);
                $symbol = $market['symbol'];
                $messageHash = $messageHash . ':' . $symbol;
            }
            $orders = Async\await($this->subscribe($url, $messageHash, $topic, null, $this->extend($request, $params)));
            if ($this->newUpdates) {
                $limit = $orders->getLimit ($symbol, $limit);
            }
            return $this->filter_by_symbol_since_limit($orders, $symbol, $since, $limit, true);
        }) ();
    }

    public function parse_ws_order_status($status) {
        $statuses = array(
            'open' => 'open',
            'filled' => 'closed',
            'match' => 'open',
            'update' => 'open',
            'canceled' => 'canceled',
        );
        return $this->safe_string($statuses, $status, $status);
    }

    public function parse_ws_order($order, $market = null) {
        //
        //         "symbol" => "XCAD-USDT",
        //     {
        //         "orderType" => "limit",
        //         "side" => "buy",
        //         "orderId" => "6249167327218b000135e749",
        //         "type" => "canceled",
        //         "orderTime" => 1648957043065280224,
        //         "size" => "100.452",
        //         "filledSize" => "0",
        //         "price" => "2.9635",
        //         "clientOid" => "buy-XCAD-USDT-1648957043010159",
        //         "remainSize" => "0",
        //         "status" => "done",
        //         "ts" => 1648957054031001037
        //     }
        //
        $id = $this->safe_string($order, 'orderId');
        $clientOrderId = $this->safe_string($order, 'clientOid');
        $orderType = $this->safe_string_lower($order, 'orderType');
        $price = $this->safe_string($order, 'price');
        $filled = $this->safe_string($order, 'filledSize');
        $amount = $this->safe_string($order, 'size');
        $rawType = $this->safe_string($order, 'type');
        $status = $this->parse_ws_order_status($rawType);
        $timestamp = $this->safe_integer_product($order, 'orderTime', 0.000001);
        $marketId = $this->safe_string($order, 'symbol');
        $market = $this->safe_market($marketId, $market);
        $symbol = $market['symbol'];
        $side = $this->safe_string_lower($order, 'side');
        return $this->safe_order(array(
            'info' => $order,
            'symbol' => $symbol,
            'id' => $id,
            'clientOrderId' => $clientOrderId,
            'timestamp' => $timestamp,
            'datetime' => $this->iso8601($timestamp),
            'lastTradeTimestamp' => null,
            'type' => $orderType,
            'timeInForce' => null,
            'postOnly' => null,
            'side' => $side,
            'price' => $price,
            'stopPrice' => null,
            'amount' => $amount,
            'cost' => null,
            'average' => null,
            'filled' => $filled,
            'remaining' => null,
            'status' => $status,
            'fee' => null,
            'trades' => null,
        ), $market);
    }

    public function handle_order(Client $client, $message) {
        $messageHash = 'orders';
        $data = $this->safe_value($message, 'data');
        $parsed = $this->parse_ws_order($data);
        $symbol = $this->safe_string($parsed, 'symbol');
        $orderId = $this->safe_string($parsed, 'id');
        if ($symbol !== null) {
            if ($this->orders === null) {
                $limit = $this->safe_integer($this->options, 'ordersLimit', 1000);
                $this->orders = new ArrayCacheBySymbolById ($limit);
            }
            $cachedOrders = $this->orders;
            $orders = $this->safe_value($cachedOrders->hashmap, $symbol, array());
            $order = $this->safe_value($orders, $orderId);
            if ($order !== null) {
                // todo add others to calculate average etc
                $stopPrice = $this->safe_value($order, 'stopPrice');
                if ($stopPrice !== null) {
                    $parsed['stopPrice'] = $stopPrice;
                }
                if ($order['status'] === 'closed') {
                    $parsed['status'] = 'closed';
                }
            }
            $cachedOrders->append ($parsed);
            $client->resolve ($this->orders, $messageHash);
            $symbolSpecificMessageHash = $messageHash . ':' . $symbol;
            $client->resolve ($this->orders, $symbolSpecificMessageHash);
        }
    }

    public function watch_balance($params = array ()): PromiseInterface {
        return Async\async(function () use ($params) {
            /**
             * watch balance and get the amount of funds available for trading or funds locked in orders
             *
             * @see https://docs.kucoin.com/futures/#account-balance-events
             *
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} a ~@link https://docs.ccxt.com/#/?id=balance-structure balance structure~
             */
            Async\await($this->load_markets());
            $url = Async\await($this->negotiate(true));
            $topic = '/contractAccount/wallet';
            $request = array(
                'privateChannel' => true,
            );
            $subscription = array(
                'method' => array($this, 'handle_balance_subscription'),
            );
            $messageHash = 'balance';
            return Async\await($this->subscribe($url, $messageHash, $topic, $subscription, $this->extend($request, $params)));
        }) ();
    }

    public function handle_balance(Client $client, $message) {
        //
        //    {
        //        "id" => "6375553193027a0001f6566f",
        //        "type" => "message",
        //        "topic" => "/contractAccount/wallet",
        //        "userId" => "613a896885d8660006151f01",
        //        "channelType" => "private",
        //        "subject" => "availableBalance.change",
        //        "data" => {
        //            "currency" => "USDT",
        //            "holdBalance" => "0.0000000000",
        //            "availableBalance" => "14.0350281903",
        //            "timestamp" => "1668633905657"
        //        }
        //    }
        //
        $data = $this->safe_value($message, 'data', array());
        $this->balance['info'] = $data;
        $currencyId = $this->safe_string($data, 'currency');
        $code = $this->safe_currency_code($currencyId);
        $account = $this->account();
        $account['free'] = $this->safe_string($data, 'availableBalance');
        $account['used'] = $this->safe_string($data, 'holdBalance');
        $this->balance[$code] = $account;
        $this->balance = $this->safe_balance($this->balance);
        $client->resolve ($this->balance, 'balance');
    }

    public function handle_balance_subscription(Client $client, $message, $subscription) {
        $this->spawn(array($this, 'fetch_balance_snapshot'), $client, $message);
    }

    public function fetch_balance_snapshot($client, $message) {
        return Async\async(function () use ($client, $message) {
            Async\await($this->load_markets());
            $this->check_required_credentials();
            $messageHash = 'balance';
            $selectedType = $this->safe_string_2($this->options, 'watchBalance', 'defaultType', 'swap'); // spot, margin, main, funding, future, mining, trade, contract, pool
            $params = array(
                'type' => $selectedType,
            );
            $snapshot = Async\await($this->fetch_balance($params));
            //
            //    {
            //        "info" => {
            //            "code" => "200000",
            //            "data" => array(
            //                "accountEquity" => 0.0350281903,
            //                "unrealisedPNL" => 0,
            //                "marginBalance" => 0.0350281903,
            //                "positionMargin" => 0,
            //                "orderMargin" => 0,
            //                "frozenFunds" => 0,
            //                "availableBalance" => 0.0350281903,
            //                "currency" => "USDT"
            //            }
            //        ),
            //        "timestamp" => null,
            //        "datetime" => null,
            //        "USDT" => array(
            //            "free" => 0.0350281903,
            //            "used" => 0,
            //            "total" => 0.0350281903
            //        ),
            //        "free" => array(
            //            "USDT" => 0.0350281903
            //        ),
            //        "used" => array(
            //            "USDT" => 0
            //        ),
            //        "total" => {
            //            "USDT" => 0.0350281903
            //        }
            //    }
            //
            $keys = is_array($snapshot) ? array_keys($snapshot) : array();
            for ($i = 0; $i < count($keys); $i++) {
                $code = $keys[$i];
                if ($code !== 'free' && $code !== 'used' && $code !== 'total' && $code !== 'timestamp' && $code !== 'datetime' && $code !== 'info') {
                    $this->balance[$code] = $snapshot[$code];
                }
            }
            $this->balance['info'] = $this->safe_value($snapshot, 'info', array());
            $client->resolve ($this->balance, $messageHash);
        }) ();
    }

    public function handle_subject(Client $client, $message) {
        //
        //    {
        //        "type" => "message",
        //        "topic" => "/contractMarket/level2:ADAUSDTM",
        //        "subject" => "level2",
        //        "data" => {
        //            "sequence" => 1668059586457,
        //            "change" => "0.34172,sell,456", // type, side, quantity
        //            "timestamp" => 1668573023223
        //        }
        //    }
        //
        $subject = $this->safe_string($message, 'subject');
        $methods = array(
            'level2' => array($this, 'handle_order_book'),
            'ticker' => array($this, 'handle_ticker'),
            'candle.stick' => array($this, 'handle_ohlcv'),
            'tickerV2' => array($this, 'handle_bid_ask'),
            'availableBalance.change' => array($this, 'handle_balance'),
            'match' => array($this, 'handle_trade'),
            'orderChange' => array($this, 'handle_order'),
            'orderUpdated' => array($this, 'handle_order'),
            'position.change' => array($this, 'handle_position'),
            'position.settlement' => array($this, 'handle_position'),
            'position.adjustRiskLimit' => array($this, 'handle_position'),
        );
        $method = $this->safe_value($methods, $subject);
        if ($method !== null) {
            $method($client, $message);
        }
    }

    public function get_message_hash(string $elementName, ?string $symbol = null) {
        // $elementName can be 'ticker', 'bidask', ...
        if ($symbol !== null) {
            return $elementName . '@' . $symbol;
        } else {
            return $elementName . 's@all';
        }
    }

    public function ping(Client $client) {
        // kucoin does not support built-in ws protocol-level ping-pong
        // instead it requires a custom json-based text ping-pong
        // https://docs.kucoin.com/#ping
        $id = (string) $this->request_id();
        return array(
            'id' => $id,
            'type' => 'ping',
        );
    }

    public function handle_pong(Client $client, $message) {
        // https://docs.kucoin.com/#ping
        $client->lastPong = $this->milliseconds();
        return $message;
    }

    public function handle_error_message(Client $client, $message): Bool {
        //
        //    {
        //        "id" => "64d8732c856851144bded10d",
        //        "type" => "error",
        //        "code" => 401,
        //        "data" => "token is expired"
        //    }
        //
        $data = $this->safe_string($message, 'data', '');
        if ($data === 'token is expired') {
            $type = 'public';
            if (mb_strpos($client->url, 'connectId=private') !== false) {
                $type = 'private';
            }
            $this->options['urls'][$type] = null;
        }
        $this->handle_errors(1, '', $client->url, '', array(), $data, $message, array(), array());
        return true;
    }

    public function handle_subscription_status(Client $client, $message) {
        //
        //     {
        //         "id" => "1578090438322",
        //         "type" => "ack"
        //     }
        //
        $id = $this->safe_string($message, 'id');
        if (!(is_array($client->subscriptions) && array_key_exists($id, $client->subscriptions))) {
            return;
        }
        $subscriptionHash = $this->safe_string($client->subscriptions, $id);
        $subscription = $this->safe_value($client->subscriptions, $subscriptionHash);
        unset($client->subscriptions[$id]);
        $method = $this->safe_value($subscription, 'method');
        if ($method !== null) {
            $method($client, $message, $subscription);
        }
        $isUnSub = $this->safe_bool($subscription, 'unsubscribe', false);
        if ($isUnSub) {
            $messageHashes = $this->safe_list($subscription, 'messageHashes', array());
            $subMessageHashes = $this->safe_list($subscription, 'subMessageHashes', array());
            for ($i = 0; $i < count($messageHashes); $i++) {
                $messageHash = $messageHashes[$i];
                $subHash = $subMessageHashes[$i];
                $this->clean_unsubscription($client, $subHash, $messageHash);
            }
            $this->clean_cache($subscription);
        }
    }

    public function handle_message(Client $client, $message) {
        $type = $this->safe_string($message, 'type');
        $methods = array(
            // 'heartbeat' => $this->handleHeartbeat,
            'welcome' => array($this, 'handle_system_status'),
            'message' => array($this, 'handle_subject'),
            'pong' => array($this, 'handle_pong'),
            'error' => array($this, 'handle_error_message'),
            'ack' => array($this, 'handle_subscription_status'),
        );
        $method = $this->safe_value($methods, $type);
        if ($method !== null) {
            $method($client, $message);
        }
    }
}
