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

class apex extends \ccxt\async\apex {

    public function describe(): mixed {
        return $this->deep_extend(parent::describe(), array(
            'has' => array(
                'ws' => true,
                'watchTicker' => true,
                'watchTickers' => true,
                'watchOrderBook' => true,
                'watchOrders' => true,
                'watchTrades' => true,
                'watchTradesForSymbols' => false,
                'watchPositions' => true,
                'watchMyTrades' => true,
                'watchBalance' => false,
                'watchOHLCV' => true,
            ),
            'urls' => array(
                'logo' => 'https://omni.apex.exchange/assets/logo_content-CY9uyFbz.svg',
                'api' => array(
                    'ws' => array(
                        'public' => 'wss://quote.omni.apex.exchange/realtime_public?v=2',
                        'private' => 'wss://quote.omni.apex.exchange/realtime_private?v=2',
                    ),
                ),
                'test' => array(
                    'ws' => array(
                        'public' => 'wss://qa-quote.omni.apex.exchange/realtime_public?v=2',
                        'private' => 'wss://qa-quote.omni.apex.exchange/realtime_private?v=2',
                    ),
                ),
                'www' => 'https://apex.exchange/',
                'doc' => 'https://api-docs.pro.apex.exchange',
                'fees' => 'https://apex-pro.gitbook.io/apex-pro/apex-omni-live-now/trading-perpetual-contracts/trading-fees',
                'referral' => 'https://omni.apex.exchange/trade',
            ),
            'options' => array(),
            'streaming' => array(
                'ping' => array($this, 'ping'),
                'keepAlive' => 18000,
            ),
        ));
    }

    public function watch_trades(string $symbol, ?int $since = null, ?int $limit = null, $params = array ()): PromiseInterface {
        return Async\async(function () use ($symbol, $since, $limit, $params) {
            /**
             * watches information on multiple trades made in a market
             *
             * @see https://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @param {string} $symbol unified market $symbol of the market trades were made in
             * @param {int} [$since] the earliest time in ms to fetch trades for
             * @param {int} [$limit] the maximum number of trade structures to retrieve
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=trade-structure 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 list of $symbols
             *
             * @see https://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @param {string[]} $symbols unified $symbol of the $market to fetch $trades for
             * @param {int} [$since] timestamp in ms of the earliest trade to fetch
             * @param {int} [$limit] the maximum amount of $trades to fetch
             * @param {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);
            $symbolsLength = count($symbols);
            if ($symbolsLength === 0) {
                throw new ArgumentsRequired($this->id . ' watchTradesForSymbols() requires a non-empty array of symbols');
            }
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['public'] . '&timestamp=' . $timeStamp;
            $topics = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $market = $this->market($symbol);
                $topic = 'recentlyTrade.H.' . $market['id2'];
                $topics[] = $topic;
                $messageHash = 'trade:' . $symbol;
                $messageHashes[] = $messageHash;
            }
            $trades = Async\await($this->watch_topics($url, $messageHashes, $topics, $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 handle_trades(Client $client, $message) {
        //
        //     {
        //         "topic" => "recentlyTrade.H.BTCUSDT",
        //         "type" => "snapshot",
        //         "ts" => 1672304486868,
        //         "data" => array(
        //             array(
        //                 "T" => 1672304486865,
        //                 "s" => "BTCUSDT",
        //                 "S" => "Buy",
        //                 "v" => "0.001",
        //                 "p" => "16578.50",
        //                 "L" => "PlusTick",
        //                 "i" => "20f43950-d8dd-5b31-9112-a178eb6023ef",
        //                 "BT" => false
        //             ),
        //             // sorted by newest first
        //         )
        //     }
        //
        $data = $this->safe_value($message, 'data', array());
        $topic = $this->safe_string($message, 'topic');
        $trades = $data;
        $parts = explode('.', $topic);
        $marketId = $this->safe_string($parts, 2);
        $market = $this->safe_market($marketId, null, null);
        $symbol = $market['symbol'];
        $stored = $this->safe_value($this->trades, $symbol);
        if ($stored === null) {
            $limit = $this->safe_integer($this->options, 'tradesLimit', 1000);
            $stored = new ArrayCache ($limit);
            $this->trades[$symbol] = $stored;
        }
        $length = count($trades);
        for ($j = 0; $j < $length; $j++) {
            $index = $length - $j - 1;
            $parsed = $this->parse_ws_trade($trades[$index], $market);
            $stored->append ($parsed);
        }
        $messageHash = 'trade' . ':' . $symbol;
        $client->resolve ($stored, $messageHash);
    }

    public function parse_ws_trade($trade, $market = null) {
        //
        // public
        //    {
        //         "T" => 1672304486865,
        //         "s" => "BTCUSDT",
        //         "S" => "Buy",
        //         "v" => "0.001",
        //         "p" => "16578.50",
        //         "L" => "PlusTick",
        //         "i" => "20f43950-d8dd-5b31-9112-a178eb6023af",
        //         "BT" => false
        //     }
        //
        $id = $this->safe_string_n($trade, array( 'i', 'id', 'v' ));
        $marketId = $this->safe_string_n($trade, array( 's', 'symbol' ));
        $market = $this->safe_market($marketId, $market, null);
        $symbol = $market['symbol'];
        $timestamp = $this->safe_integer_n($trade, array( 't', 'T', 'createdAt' ));
        $side = $this->safe_string_lower_n($trade, array( 'S', 'side' ));
        $price = $this->safe_string_n($trade, array( 'p', 'price' ));
        $amount = $this->safe_string_n($trade, array( 'q', 'v', 'size' ));
        return $this->safe_trade(array(
            'id' => $id,
            'info' => $trade,
            'timestamp' => $timestamp,
            'datetime' => $this->iso8601($timestamp),
            'symbol' => $symbol,
            'order' => null,
            'type' => null,
            'side' => $side,
            'takerOrMaker' => null,
            'price' => $price,
            'amount' => $amount,
            'cost' => null,
            'fee' => null,
        ), $market);
    }

    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://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @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://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @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
             */
            Async\await($this->load_markets());
            $symbolsLength = count($symbols);
            if ($symbolsLength === 0) {
                throw new ArgumentsRequired($this->id . ' watchOrderBookForSymbols() requires a non-empty array of symbols');
            }
            $symbols = $this->market_symbols($symbols);
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['public'] . '&timestamp=' . $timeStamp;
            $topics = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $market = $this->market($symbol);
                if ($limit === null) {
                    $limit = 25;
                }
                $topic = 'orderBook' . (string) $limit . '.H.' . $market['id2'];
                $topics[] = $topic;
                $messageHash = 'orderbook:' . $symbol;
                $messageHashes[] = $messageHash;
            }
            $orderbook = Async\await($this->watch_topics($url, $messageHashes, $topics, $params));
            return $orderbook->limit ();
        }) ();
    }

    public function watch_topics($url, $messageHashes, $topics, $params = array ()) {
        return Async\async(function () use ($url, $messageHashes, $topics, $params) {
            $request = array(
                'op' => 'subscribe',
                'args' => $topics,
            );
            $message = $this->extend($request, $params);
            return Async\await($this->watch_multiple($url, $messageHashes, $message, $messageHashes));
        }) ();
    }

    public function handle_order_book(Client $client, $message) {
        //
        //     {
        //         "topic" => "orderbook25.H.BTCUSDT",
        //         "type" => "snapshot",
        //         "ts" => 1672304484978,
        //         "data" => {
        //             "s" => "BTCUSDT",
        //             "b" => array(
        //                 ...,
        //                 array(
        //                     "16493.50",
        //                     "0.006"
        //                 ),
        //                 array(
        //                     "16493.00",
        //                     "0.100"
        //                 )
        //             ),
        //             "a" => array(
        //                 array(
        //                     "16611.00",
        //                     "0.029"
        //                 ),
        //                 array(
        //                     "16612.00",
        //                     "0.213"
        //                 ),
        //             ),
        //             "u" => 18521288,
        //             "seq" => 7961638724
        //         }
        //     }
        //
        $type = $this->safe_string($message, 'type');
        $isSnapshot = ($type === 'snapshot');
        $data = $this->safe_dict($message, 'data', array());
        $marketId = $this->safe_string($data, 's');
        $market = $this->safe_market($marketId, null, null);
        $symbol = $market['symbol'];
        $timestamp = $this->safe_integer_product($message, 'ts', 0.001);
        if (!(is_array($this->orderbooks) && array_key_exists($symbol, $this->orderbooks))) {
            $this->orderbooks[$symbol] = $this->order_book();
        }
        $orderbook = $this->orderbooks[$symbol];
        if ($isSnapshot) {
            $snapshot = $this->parse_order_book($data, $symbol, $timestamp, 'b', 'a');
            $orderbook->reset ($snapshot);
        } else {
            $asks = $this->safe_list($data, 'a', array());
            $bids = $this->safe_list($data, 'b', array());
            $this->handle_deltas($orderbook['asks'], $asks);
            $this->handle_deltas($orderbook['bids'], $bids);
            $orderbook['timestamp'] = $timestamp;
            $orderbook['datetime'] = $this->iso8601($timestamp);
        }
        $messageHash = 'orderbook' . ':' . $symbol;
        $this->orderbooks[$symbol] = $orderbook;
        $client->resolve ($orderbook, $messageHash);
    }

    public function handle_delta($bookside, $delta) {
        $bidAsk = $this->parse_bid_ask($delta, 0, 1);
        $bookside->storeArray ($bidAsk);
    }

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

    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://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @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'];
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['public'] . '&timestamp=' . $timeStamp;
            $messageHash = 'ticker:' . $symbol;
            $topic = 'instrumentInfo' . '.H.' . $market['id2'];
            $topics = array( $topic );
            return Async\await($this->watch_topics($url, array( $messageHash ), $topics, $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 all markets of a specific list
             *
             * @see https://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @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());
            $symbols = $this->market_symbols($symbols, null, false);
            $messageHashes = array();
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['public'] . '&timestamp=' . $timeStamp;
            $topics = [ ];
            for ($i = 0; $i < count($symbols); $i++) {
                $symbol = $symbols[$i];
                $market = $this->market($symbol);
                $topic = 'instrumentInfo' . '.H.' . $market['id2'];
                $topics[] = $topic;
                $messageHash = 'ticker:' . $symbol;
                $messageHashes[] = $messageHash;
            }
            $ticker = Async\await($this->watch_topics($url, $messageHashes, $topics, $params));
            if ($this->newUpdates) {
                $result = array();
                $result[$ticker['symbol']] = $ticker;
                return $result;
            }
            return $this->filter_by_array($this->tickers, 'symbol', $symbols);
        }) ();
    }

    public function handle_ticker(Client $client, $message) {
        // "topic":"instrumentInfo.H.BTCUSDT",
        //     "type":"snapshot",
        //     "data":array(
        //     "symbol":"BTCUSDT",
        //         "lastPrice":"21572.5",
        //         "price24hPcnt":"-0.0194318181818182",
        //         "highPrice24h":"25306.5",
        //         "lowPrice24h":"17001.5",
        //         "turnover24h":"1334891.4545",
        //         "volume24h":"64.896",
        //         "nextFundingTime":"2022-08-26T08:00:00Z",
        //         "oraclePrice":"21412.060000000002752512",
        //         "indexPrice":"21409.82",
        //         "openInterest":"49.598",
        //         "tradeCount":"0",
        //         "fundingRate":"0.0000125",
        //         "predictedFundingRate":"0.0000125"
        // ),
        //     "cs":44939063,
        //     "ts":1661500091955487
        // }
        $topic = $this->safe_string($message, 'topic', '');
        $updateType = $this->safe_string($message, 'type', '');
        $data = $this->safe_dict($message, 'data', array());
        $symbol = null;
        $parsed = null;
        if (($updateType === 'snapshot')) {
            $parsed = $this->parse_ticker($data);
            $symbol = $parsed['symbol'];
        } elseif ($updateType === 'delta') {
            $topicParts = explode('.', $topic);
            $topicLength = count($topicParts);
            $marketId = $this->safe_string($topicParts, $topicLength - 1);
            $market = $this->safe_market($marketId, null, null);
            $symbol = $market['symbol'];
            $ticker = $this->safe_dict($this->tickers, $symbol, array());
            $rawTicker = $this->safe_dict($ticker, 'info', array());
            $merged = $this->extend($rawTicker, $data);
            $parsed = $this->parse_ticker($merged);
        }
        $timestamp = $this->safe_integer_product($message, 'ts', 0.001);
        $parsed['timestamp'] = $timestamp;
        $parsed['datetime'] = $this->iso8601($timestamp);
        $this->tickers[$symbol] = $parsed;
        $messageHash = 'ticker:' . $symbol;
        $client->resolve ($this->tickers[$symbol], $messageHash);
    }

    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://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @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
             */
            $params['callerMethodName'] = 'watchOHLCV';
            $result = Async\await($this->watch_ohlcv_for_symbols(array( array( $symbol, $timeframe ) ), $since, $limit, $params));
            return $result[$symbol][$timeframe];
        }) ();
    }

    public function watch_ohlcv_for_symbols(array $symbolsAndTimeframes, ?int $since = null, ?int $limit = null, $params = array ()) {
        return Async\async(function () use ($symbolsAndTimeframes, $since, $limit, $params) {
            /**
             * watches historical candlestick $data containing the open, high, low, and close price, and the volume of a $market
             *
             * @see https://api-docs.pro.apex.exchange/#websocket-v3-for-omni-websocket-endpoint
             *
             * @param {string[][]} $symbolsAndTimeframes array of arrays containing unified symbols and timeframes to fetch OHLCV $data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]
             * @param {int} [$since] timestamp in ms of the earliest candle to fetch
             * @param {int} [$limit] the maximum amount of candles to fetch
             * @param {array} [$params] extra parameters specific to the exchange API endpoint
             * @return {array} A list of candles ordered, open, high, low, close, volume
             */
            Async\await($this->load_markets());
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['public'] . '&timestamp=' . $timeStamp;
            $rawHashes = array();
            $messageHashes = array();
            for ($i = 0; $i < count($symbolsAndTimeframes); $i++) {
                $data = $symbolsAndTimeframes[$i];
                $symbolString = $this->safe_string($data, 0);
                $market = $this->market($symbolString);
                $symbolString = $market['id2'];
                $unfiedTimeframe = $this->safe_string($data, 1, '1');
                $timeframeId = $this->safe_string($this->timeframes, $unfiedTimeframe, $unfiedTimeframe);
                $rawHashes[] = 'candle.' . $timeframeId . '.' . $symbolString;
                $messageHashes[] = 'ohlcv::' . $symbolString . '::' . $unfiedTimeframe;
            }
            list($symbol, $timeframe, $stored) = Async\await($this->watch_topics($url, $messageHashes, $rawHashes, $params));
            if ($this->newUpdates) {
                $limit = $stored->getLimit ($symbol, $limit);
            }
            $filtered = $this->filter_by_since_limit($stored, $since, $limit, 0, true);
            return $this->create_ohlcv_object($symbol, $timeframe, $filtered);
        }) ();
    }

    public function handle_ohlcv(Client $client, $message) {
        //
        //     {
        //         "topic" => "candle.5.BTCUSDT",
        //         "data" => array(
        //             {
        //                 "start" => 1672324800000,
        //                 "end" => 1672325099999,
        //                 "interval" => "5",
        //                 "open" => "16649.5",
        //                 "close" => "16677",
        //                 "high" => "16677",
        //                 "low" => "16608",
        //                 "volume" => "2.081",
        //                 "turnover" => "34666.4005",
        //                 "confirm" => false,
        //                 "timestamp" => 1672324988882
        //             }
        //         ),
        //         "ts" => 1672324988882,
        //         "type" => "snapshot"
        //     }
        //
        $data = $this->safe_value($message, 'data', array());
        $topic = $this->safe_string($message, 'topic');
        $topicParts = explode('.', $topic);
        $topicLength = count($topicParts);
        $timeframeId = $this->safe_string($topicParts, 1);
        $timeframe = $this->find_timeframe($timeframeId);
        $marketId = $this->safe_string($topicParts, $topicLength - 1);
        $isSpot = mb_strpos($client->url, 'spot') > -1;
        $marketType = $isSpot ? 'spot' : 'contract';
        $market = $this->safe_market($marketId, null, null, $marketType);
        $symbol = $market['symbol'];
        $ohlcvsByTimeframe = $this->safe_value($this->ohlcvs, $symbol);
        if ($ohlcvsByTimeframe === null) {
            $this->ohlcvs[$symbol] = array();
        }
        if ($this->safe_value($ohlcvsByTimeframe, $timeframe) === null) {
            $limit = $this->safe_integer($this->options, 'OHLCVLimit', 1000);
            $this->ohlcvs[$symbol][$timeframe] = new ArrayCacheByTimestamp ($limit);
        }
        $stored = $this->ohlcvs[$symbol][$timeframe];
        for ($i = 0; $i < count($data); $i++) {
            $parsed = $this->parse_ws_ohlcv($data[$i]);
            $stored->append ($parsed);
        }
        $messageHash = 'ohlcv::' . $symbol . '::' . $timeframe;
        $resolveData = array( $symbol, $timeframe, $stored );
        $client->resolve ($resolveData, $messageHash);
    }

    public function parse_ws_ohlcv($ohlcv, $market = null): array {
        //
        //     {
        //         "start" => 1670363160000,
        //         "end" => 1670363219999,
        //         "interval" => "1",
        //         "open" => "16987.5",
        //         "close" => "16987.5",
        //         "high" => "16988",
        //         "low" => "16987.5",
        //         "volume" => "23.511",
        //         "turnover" => "399396.344",
        //         "confirm" => false,
        //         "timestamp" => 1670363219614
        //     }
        //
        return array(
            $this->safe_integer($ohlcv, 'start'),
            $this->safe_number($ohlcv, 'open'),
            $this->safe_number($ohlcv, 'high'),
            $this->safe_number($ohlcv, 'low'),
            $this->safe_number($ohlcv, 'close'),
            $this->safe_number_2($ohlcv, 'volume', 'turnover'),
        );
    }

    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://api-docs.pro.apex.exchange/#private-websocket
             *
             * @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
             * @param {boolean} [$params->unifiedMargin] use unified margin account
             * @return {array[]} a list of ~@link https://docs.ccxt.com/#/?id=order-structure order structures~
             */
            $messageHash = 'myTrades';
            Async\await($this->load_markets());
            if ($symbol !== null) {
                $symbol = $this->symbol($symbol);
                $messageHash .= ':' . $symbol;
            }
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['private'] . '&timestamp=' . $timeStamp;
            Async\await($this->authenticate($url));
            $trades = Async\await($this->watch_topics($url, array( $messageHash ), array( 'myTrades' ), $params));
            if ($this->newUpdates) {
                $limit = $trades->getLimit ($symbol, $limit);
            }
            return $this->filter_by_symbol_since_limit($trades, $symbol, $since, $limit, true);
        }) ();
    }

    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://api-docs.pro.apex.exchange/#private-websocket
             *
             * watch all open positions
             * @param {string[]} [$symbols] list of unified market $symbols
             * @param {int} [$since] the earliest time in ms to fetch positions for
             * @param {int} [$limit] the maximum number of positions to retrieve
             * @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());
            $messageHash = '';
            if (!$this->is_empty($symbols)) {
                $symbols = $this->market_symbols($symbols);
                $messageHash = '::' . implode(',', $symbols);
            }
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['private'] . '&timestamp=' . $timeStamp;
            $messageHash = 'positions' . $messageHash;
            $client = $this->client($url);
            Async\await($this->authenticate($url));
            $this->set_positions_cache($client, $symbols);
            $cache = $this->positions;
            if ($cache === null) {
                $snapshot = Async\await($client->future ('fetchPositionsSnapshot'));
                return $this->filter_by_symbols_since_limit($snapshot, $symbols, $since, $limit, true);
            }
            $topics = array( 'positions' );
            $newPositions = Async\await($this->watch_topics($url, array( $messageHash ), $topics, $params));
            if ($this->newUpdates) {
                return $newPositions;
            }
            return $this->filter_by_symbols_since_limit($cache, $symbols, $since, $limit, true);
        }) ();
    }

    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://api-docs.pro.apex.exchange/#private-websocket
             *
             * @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());
            $messageHash = 'orders';
            if ($symbol !== null) {
                $symbol = $this->symbol($symbol);
                $messageHash .= ':' . $symbol;
            }
            $timeStamp = (string) $this->milliseconds();
            $url = $this->urls['api']['ws']['private'] . '&timestamp=' . $timeStamp;
            Async\await($this->authenticate($url));
            $topics = array( 'orders' );
            $orders = Async\await($this->watch_topics($url, array( $messageHash ), $topics, $params));
            if ($this->newUpdates) {
                $limit = $orders->getLimit ($symbol, $limit);
            }
            return $this->filter_by_symbol_since_limit($orders, $symbol, $since, $limit, true);
        }) ();
    }

    public function handle_my_trades(Client $client, $lists) {
        // array(
        //     {
        //         "symbol":"ETH-USDT",
        //         "side":"BUY",
        //         "orderId":"2048046080",
        //         "fee":"0.625000",
        //         "liquidity":"TAKER",
        //         "accountId":"1024000",
        //         "createdAt":1652185521361,
        //         "isOpen":true,
        //         "size":"0.500",
        //         "price":"2500.0",
        //         "quoteAmount":"1250.0000",
        //         "id":"2048000182272",
        //         "updatedAt":1652185678345
        //     }
        // )
        if ($this->myTrades === null) {
            $limit = $this->safe_integer($this->options, 'tradesLimit', 1000);
            $this->myTrades = new ArrayCacheBySymbolById ($limit);
        }
        $trades = $this->myTrades;
        $symbols = array();
        for ($i = 0; $i < count($lists); $i++) {
            $rawTrade = $lists[$i];
            $parsed = null;
            $parsed = $this->parse_ws_trade($rawTrade);
            $symbol = $parsed['symbol'];
            $symbols[$symbol] = true;
            $trades->append ($parsed);
        }
        $keys = is_array($symbols) ? array_keys($symbols) : array();
        for ($i = 0; $i < count($keys); $i++) {
            $currentMessageHash = 'myTrades:' . $keys[$i];
            $client->resolve ($trades, $currentMessageHash);
        }
        // non-$symbol specific
        $messageHash = 'myTrades';
        $client->resolve ($trades, $messageHash);
    }

    public function handle_order(Client $client, $lists) {
        // array(
        //     {
        //         "symbol":"ETH-USDT",
        //         "cumSuccessFillFee":"0.625000",
        //         "trailingPercent":"0",
        //         "type":"LIMIT",
        //         "unfillableAt":1654779600000,
        //         "isDeleverage":false,
        //         "createdAt":1652185521339,
        //         "price":"2500.0",
        //         "cumSuccessFillValue":"0",
        //         "id":"2048046080",
        //         "cancelReason":"",
        //         "timeInForce":1,
        //         "updatedAt":1652185521392,
        //         "limitFee":"0.625000",
        //         "side":"BUY",
        //         "clientOrderId":"522843990",
        //         "triggerPrice":"",
        //         "expiresAt":1654779600000,
        //         "cumSuccessFillSize":"0",
        //         "accountId":"1024000",
        //         "size":"0.500",
        //         "reduceOnly":false,
        //         "isLiquidate":false,
        //         "remainingSize":"0.000",
        //         "status":"PENDING"
        //     }
        // )
        if ($this->orders === null) {
            $limit = $this->safe_integer($this->options, 'ordersLimit', 1000);
            $this->orders = new ArrayCacheBySymbolById ($limit);
        }
        $orders = $this->orders;
        $symbols = array();
        for ($i = 0; $i < count($lists); $i++) {
            $parsed = null;
            $parsed = $this->parse_order($lists[$i]);
            $symbol = $parsed['symbol'];
            $symbols[$symbol] = true;
            $orders->append ($parsed);
        }
        $symbolsArray = is_array($symbols) ? array_keys($symbols) : array();
        for ($i = 0; $i < count($symbolsArray); $i++) {
            $currentMessageHash = 'orders:' . $symbolsArray[$i];
            $client->resolve ($orders, $currentMessageHash);
        }
        $messageHash = 'orders';
        $client->resolve ($orders, $messageHash);
    }

    public function set_positions_cache(Client $client, ?array $symbols = null) {
        if ($this->positions !== null) {
            return;
        }
        $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) {
            // one ws channel gives $positions for all types, for snapshot must load all $positions
            $fetchFunctions = array(
                $this->fetch_positions(null),
            );
            $promises = Async\await(Promise\all($fetchFunctions));
            $this->positions = new ArrayCacheBySymbolBySide ();
            $cache = $this->positions;
            for ($i = 0; $i < count($promises); $i++) {
                $positions = $promises[$i];
                for ($ii = 0; $ii < count($positions); $ii++) {
                    $position = $positions[$ii];
                    $cache->append ($position);
                }
            }
            // don't remove the $future from the .futures $cache
            $future = $client->futures[$messageHash];
            $future->resolve ($cache);
            $client->resolve ($cache, 'positions');
        }) ();
    }

    public function handle_positions($client, $lists) {
        //
        // array(
        //     {
        //         "symbol":"ETH-USDT",
        //         "exitPrice":"0",
        //         "side":"LONG",
        //         "maxSize":"2820.000",
        //         "sumOpen":"1.820",
        //         "sumClose":"0.000",
        //         "netFunding":"0.000000",
        //         "entryPrice":"2500.000000000000000000",
        //         "accountId":"1024000",
        //         "createdAt":1652179377769,
        //         "size":"1.820",
        //         "realizedPnl":"0",
        //         "closedAt":1652185521392,
        //         "updatedAt":1652185521392
        //     }
        // )
        //
        // each account is connected to a different endpoint
        // and has exactly one subscriptionhash which is the account type
        if ($this->positions === null) {
            $this->positions = new ArrayCacheBySymbolBySide ();
        }
        $cache = $this->positions;
        $newPositions = array();
        for ($i = 0; $i < count($lists); $i++) {
            $rawPosition = $lists[$i];
            $position = $this->parse_position($rawPosition);
            $side = $this->safe_string($position, 'side');
            // hacky solution to handle closing $positions
            // without crashing, we should handle this properly later
            $newPositions[] = $position;
            if ($side === null || $side === '') {
                // closing update, adding both sides to "reset" both sides
                // since we don't know which $side is being closed
                $position['side'] = 'long';
                $cache->append ($position);
                $position['side'] = 'short';
                $cache->append ($position);
                $position['side'] = null;
            } else {
                // regular update
                $cache->append ($position);
            }
        }
        $messageHashes = $this->find_message_hashes($client, 'positions::');
        for ($i = 0; $i < count($messageHashes); $i++) {
            $messageHash = $messageHashes[$i];
            $parts = explode('::', $messageHash);
            $symbolsString = $parts[1];
            $symbols = explode(',', $symbolsString);
            $positions = $this->filter_by_array($newPositions, 'symbol', $symbols, false);
            if (!$this->is_empty($positions)) {
                $client->resolve ($positions, $messageHash);
            }
        }
        $client->resolve ($newPositions, 'positions');
    }

    public function authenticate($url, $params = array ()) {
        return Async\async(function () use ($url, $params) {
            $this->check_required_credentials();
            $timestamp = (string) $this->milliseconds();
            $request_path = '/ws/accounts';
            $http_method = 'GET';
            $messageString = ($timestamp . $http_method . $request_path);
            $signature = $this->hmac($this->encode($messageString), $this->encode(base64_encode($this->secret)), 'sha256', 'base64');
            $messageHash = 'authenticated';
            $client = $this->client($url);
            $future = $client->reusableFuture ($messageHash);
            $authenticated = $this->safe_value($client->subscriptions, $messageHash);
            if ($authenticated === null) {
                // auth sign
                $request = array(
                    'type' => 'login',
                    'topics' => array( 'ws_zk_accounts_v3' ),
                    'httpMethod' => $http_method,
                    'requestPath' => $request_path,
                    'apiKey' => $this->apiKey,
                    'passphrase' => $this->password,
                    'timestamp' => $timestamp,
                    'signature' => $signature,
                );
                $message = array(
                    'op' => 'login',
                    'args' => array( json_encode ($request) ),
                );
                $this->watch($url, $messageHash, $message, $messageHash);
            }
            return Async\await($future);
        }) ();
    }

    public function handle_error_message(Client $client, $message): Bool {
        //
        //   {
        //       "success" => false,
        //       "ret_msg" => "error:invalid $op",
        //       "conn_id" => "5e079fdd-9c7f-404d-9dbf-969d650838b5",
        //       "request" => array( $op => '', args => null )
        //   }
        //
        // auth $error
        //
        //   {
        //       "success" => false,
        //       "ret_msg" => "error:USVC1111",
        //       "conn_id" => "e73770fb-a0dc-45bd-8028-140e20958090",
        //       "request" => {
        //         "op" => "auth",
        //         "args" => array(
        //           "9rFT6uR4uz9Imkw4Wx",
        //           "1653405853543",
        //           "542e71bd85597b4db0290f0ce2d13ed1fd4bb5df3188716c1e9cc69a879f7889"
        //         )
        //   }
        //
        //   array( $code => '-10009', desc => "Invalid period!" )
        //
        //   {
        //       "reqId":"1",
        //       "retCode":170131,
        //       "retMsg":"Insufficient balance.",
        //       "op":"order.create",
        //       "data":array(
        //
        //       ),
        //       "header":array(
        //           "X-Bapi-Limit":"20",
        //           "X-Bapi-Limit-Status":"19",
        //           "X-Bapi-Limit-Reset-Timestamp":"1714236608944",
        //           "Traceid":"3d7168a137bf32a947b7e5e6a575ac7f",
        //           "Timenow":"1714236608946"
        //       ),
        //       "connId":"cojifin88smerbj9t560-406"
        //   }
        //
        $code = $this->safe_string_n($message, array( 'code', 'ret_code', 'retCode' ));
        try {
            if ($code !== null && $code !== '0') {
                $feedback = $this->id . ' ' . $this->json($message);
                $this->throw_exactly_matched_exception($this->exceptions['exact'], $code, $feedback);
                $msg = $this->safe_string_2($message, 'retMsg', 'ret_msg');
                $this->throw_broadly_matched_exception($this->exceptions['broad'], $msg, $feedback);
                throw new ExchangeError($feedback);
            }
            $success = $this->safe_value($message, 'success');
            if ($success !== null && !$success) {
                $ret_msg = $this->safe_string($message, 'ret_msg');
                $request = $this->safe_value($message, 'request', array());
                $op = $this->safe_string($request, 'op');
                if ($op === 'auth') {
                    throw new AuthenticationError('Authentication failed => ' . $ret_msg);
                } else {
                    throw new ExchangeError($this->id . ' ' . $ret_msg);
                }
            }
            return false;
        } catch (Exception $error) {
            if ($error instanceof AuthenticationError) {
                $messageHash = 'authenticated';
                $client->reject ($error, $messageHash);
                if (is_array($client->subscriptions) && array_key_exists($messageHash, $client->subscriptions)) {
                    unset($client->subscriptions[$messageHash]);
                }
            } else {
                $messageHash = $this->safe_string($message, 'reqId');
                $client->reject ($error, $messageHash);
            }
            return true;
        }
    }

    public function handle_message(Client $client, $message) {
        if ($this->handle_error_message($client, $message)) {
            return;
        }
        $topic = $this->safe_string_2($message, 'topic', 'op', '');
        $methods = array(
            'ws_zk_accounts_v3' => array($this, 'handle_account'),
            'orderBook' => array($this, 'handle_order_book'),
            'depth' => array($this, 'handle_order_book'),
            'candle' => array($this, 'handle_ohlcv'),
            'kline' => array($this, 'handle_ohlcv'),
            'ticker' => array($this, 'handle_ticker'),
            'instrumentInfo' => array($this, 'handle_ticker'),
            'trade' => array($this, 'handle_trades'),
            'recentlyTrade' => array($this, 'handle_trades'),
            'pong' => array($this, 'handle_pong'),
            'auth' => array($this, 'handle_authenticate'),
        );
        $exacMethod = $this->safe_value($methods, $topic);
        if ($exacMethod !== null) {
            $exacMethod($client, $message);
            return;
        }
        $keys = is_array($methods) ? array_keys($methods) : array();
        for ($i = 0; $i < count($keys); $i++) {
            $key = $keys[$i];
            if (mb_strpos($topic, $keys[$i]) !== false) {
                $method = $methods[$key];
                $method($client, $message);
                return;
            }
        }
        // unified auth acknowledgement
        $type = $this->safe_string($message, 'type');
        if ($type === 'AUTH_RESP') {
            $this->handle_authenticate($client, $message);
        }
    }

    public function ping(Client $client) {
        $timeStamp = $this->milliseconds();
        $client->lastPong = $timeStamp;
        return array(
            'args' => array( (string) $timeStamp ),
            'op' => 'ping',
        );
    }

    public function handle_pong(Client $client, $message) {
        //
        //   {
        //       "success" => true,
        //       "ret_msg" => "pong",
        //       "conn_id" => "db3158a0-8960-44b9-a9de-ac350ee13158",
        //       "request" => array( op => "ping", args => null )
        //   }
        //
        //   array( pong => 1653296711335 )
        //
        $client->lastPong = $this->safe_integer($message, 'pong');
        return $message;
    }

    public function handle_account(Client $client, $message) {
        $contents = $this->safe_dict($message, 'contents', array());
        $fills = $this->safe_list($contents, 'fills', array());
        if ($fills !== null) {
            $this->handle_my_trades($client, $fills);
        }
        $positions = $this->safe_list($contents, 'positions', array());
        if ($positions !== null) {
            $this->handle_positions($client, $positions);
        }
        $orders = $this->safe_list($contents, 'orders', array());
        if ($orders !== null) {
            $this->handle_order($client, $orders);
        }
    }

    public function handle_authenticate(Client $client, $message) {
        //
        //    {
        //        "success" => true,
        //        "ret_msg" => '',
        //        "op" => "auth",
        //        "conn_id" => "ce3dpomvha7dha97tvp0-2xh"
        //    }
        //
        $success = $this->safe_value($message, 'success');
        $code = $this->safe_integer($message, 'retCode');
        $messageHash = 'authenticated';
        if ($success || $code === 0) {
            $future = $this->safe_value($client->futures, $messageHash);
            $future->resolve (true);
        } else {
            $error = new AuthenticationError ($this->id . ' ' . $this->json($message));
            $client->reject ($error, $messageHash);
            if (is_array($client->subscriptions) && array_key_exists($messageHash, $client->subscriptions)) {
                unset($client->subscriptions[$messageHash]);
            }
        }
        return $message;
    }

    public function handle_subscription_status(Client $client, $message) {
        //
        //    {
        //        "topic" => "kline",
        //        "event" => "sub",
        //        "params" => array(
        //          "symbol" => "LTCUSDT",
        //          "binary" => "false",
        //          "klineType" => "1m",
        //          "symbolName" => "LTCUSDT"
        //        ),
        //        "code" => "0",
        //        "msg" => "Success"
        //    }
        //
        return $message;
    }
}
