# -*- coding: utf-8 -*- # PLEASE DO NOT EDIT THIS FILE, IT IS GENERATED AND WILL BE OVERWRITTEN: # https://github.com/ccxt/ccxt/blob/master/CONTRIBUTING.md#how-to-contribute-code import ccxt.async_support from ccxt.async_support.base.ws.cache import ArrayCache, ArrayCacheBySymbolById, ArrayCacheBySymbolBySide, ArrayCacheByTimestamp import hashlib from ccxt.base.types import Any, Balances, Bool, Int, Market, Order, OrderBook, Position, Str, Strings, Ticker, Tickers, Trade from ccxt.async_support.base.ws.client import Client from typing import List from ccxt.base.errors import ExchangeError from ccxt.base.errors import AuthenticationError from ccxt.base.errors import ArgumentsRequired from ccxt.base.errors import BadRequest from ccxt.base.errors import NotSupported from ccxt.base.errors import RateLimitExceeded from ccxt.base.errors import ChecksumError from ccxt.base.errors import UnsubscribeError from ccxt.base.precise import Precise class coincatch(ccxt.async_support.coincatch): def describe(self) -> Any: return self.deep_extend(super(coincatch, self).describe(), { 'has': { 'ws': True, 'watchTrades': True, 'watchTradesForSymbols': True, 'watchOrderBook': True, 'watchOrderBookForSymbols': True, 'watchOHLCV': True, 'watchOHLCVForSymbols': False, # todo 'watchOrders': True, 'watchMyTrades': False, 'watchTicker': True, 'watchTickers': True, 'watchBalance': True, 'watchPositions': True, }, 'urls': { 'api': { 'ws': { 'public': 'wss://ws.coincatch.com/public/v1/stream', 'private': 'wss://ws.coincatch.com/private/v1/stream', }, }, }, 'options': { 'tradesLimit': 1000, 'OHLCVLimit': 200, 'timeframesForWs': { '1m': '1m', '5m': '5m', '15m': '15m', '30m': '30m', '1h': '1H', '4h': '4H', '12h': '12H', '1d': '1D', '1w': '1W', }, 'watchOrderBook': { 'checksum': True, }, }, 'streaming': { 'ping': self.ping, }, 'exceptions': { 'ws': { 'exact': { '30001': BadRequest, # Channel does not exist '30002': AuthenticationError, # illegal request '30003': BadRequest, # invalid op '30004': AuthenticationError, # User needs to log in '30005': AuthenticationError, # login failed '30006': RateLimitExceeded, # request too many '30007': RateLimitExceeded, # request over limit,connection close '30011': AuthenticationError, # invalid ACCESS_KEY '30012': AuthenticationError, # invalid ACCESS_PASSPHRASE '30013': AuthenticationError, # invalid ACCESS_TIMESTAMP '30014': BadRequest, # Request timestamp expired '30015': AuthenticationError, # {event: 'error', code: 30015, msg: 'Invalid sign'} }, 'broad': {}, }, }, }) def get_market_from_arg(self, entry): instId = self.safe_string(entry, 'instId') instType = self.safe_string(entry, 'instType') baseAndQuote = self.parseSpotMarketId(instId) baseId = baseAndQuote['baseId'] quoteId = baseAndQuote['quoteId'] suffix = '_SPBL' # spot suffix if instType == 'mc': if quoteId == 'USD': suffix = '_DMCBL' else: suffix = '_UMCBL' marketId = self.safe_currency_code(baseId) + self.safe_currency_code(quoteId) + suffix return self.safeMarketCustom(marketId) async def authenticate(self, params={}): self.check_required_credentials() url = self.urls['api']['ws']['private'] client = self.client(url) messageHash = 'authenticated' future = client.future(messageHash) authenticated = self.safe_value(client.subscriptions, messageHash) if authenticated is None: timestamp = str(self.seconds()) auth = timestamp + 'GET' + '/user/verify' signature = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256, 'base64') operation = 'login' request: dict = { 'op': operation, 'args': [ { 'apiKey': self.apiKey, 'passphrase': self.password, 'timestamp': timestamp, 'sign': signature, }, ], } message = self.extend(request, params) self.watch(url, messageHash, message, messageHash) return await future async def watch_public(self, messageHash, subscribeHash, args, params={}): url = self.urls['api']['ws']['public'] request: dict = { 'op': 'subscribe', 'args': [args], } message = self.extend(request, params) return await self.watch(url, messageHash, message, subscribeHash) async def un_watch_public(self, messageHash, args, params={}): url = self.urls['api']['ws']['public'] request: dict = { 'op': 'unsubscribe', 'args': [args], } message = self.extend(request, params) return await self.watch(url, messageHash, message, messageHash) async def watch_private(self, messageHash, subscribeHash, args, params={}): await self.authenticate() url = self.urls['api']['ws']['private'] request: dict = { 'op': 'subscribe', 'args': [args], } message = self.extend(request, params) return await self.watch(url, messageHash, message, subscribeHash) async def watch_private_multiple(self, messageHashes, subscribeHashes, args, params={}): await self.authenticate() url = self.urls['api']['ws']['private'] request: dict = { 'op': 'subscribe', 'args': args, } message = self.extend(request, params) return await self.watch_multiple(url, messageHashes, message, subscribeHashes) def handle_authenticate(self, client: Client, message): # # {event: "login", code: 0} # messageHash = 'authenticated' future = self.safe_value(client.futures, messageHash) future.resolve(True) async def watch_public_multiple(self, messageHashes, subscribeHashes, argsArray, params={}): url = self.urls['api']['ws']['public'] request: dict = { 'op': 'subscribe', 'args': argsArray, } message = self.extend(request, params) return await self.watch_multiple(url, messageHashes, message, subscribeHashes) async def un_watch_channel(self, symbol: str, channel: str, messageHashTopic: str, params={}) -> Any: await self.load_markets() market = self.market(symbol) instType, instId = self.get_public_inst_type_and_id(market) messageHash = 'unsubscribe:' + messageHashTopic + ':' + symbol args: dict = { 'instType': instType, 'channel': channel, 'instId': instId, } return await self.un_watch_public(messageHash, args, params) def get_public_inst_type_and_id(self, market: Market): instId = market['baseId'] + market['quoteId'] instType = None if market['spot']: instType = 'SP' elif market['swap']: instType = 'MC' else: raise NotSupported(self.id + ' supports only spot and swap markets') return [instType, instId] def handle_dmcbl_market_by_message_hashes(self, market: Market, hash: str, client: Client, timeframe: Str = None): marketId = market['id'] messageHashes = self.find_message_hashes(client, hash) # the exchange counts DMCBL markets same market with different quote currencies # for example symbols ETHUSD:ETH and ETH/USD:BTC both have the same marketId ETHUSD_DMCBL # we need to check all markets with the same marketId to find the correct market that is in messageHashes marketsWithCurrentId = self.safe_list(self.markets_by_id, marketId, []) suffix = '' if timeframe is not None: suffix = ':' + timeframe for i in range(0, len(marketsWithCurrentId)): market = marketsWithCurrentId[i] symbol = market['symbol'] messageHash = hash + symbol + suffix if self.in_array(messageHash, messageHashes): return market return market async def watch_ticker(self, symbol: str, params={}) -> Ticker: """ watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market https://coincatch.github.io/github.io/en/spot/#tickers-channel :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.instType]: the type of the instrument to fetch the ticker for, 'SP' for spot markets, 'MC' for futures markets(default is 'SP') :returns dict: a `ticker structure ` """ await self.load_markets() market = self.market(symbol) instType, instId = self.get_public_inst_type_and_id(market) channel = 'ticker' messageHash = channel + ':' + symbol args: dict = { 'instType': instType, 'channel': channel, 'instId': instId, } return await self.watch_public(messageHash, messageHash, args, params) async def un_watch_ticker(self, symbol: str, params={}) -> Any: """ unsubscribe from the ticker channel https://coincatch.github.io/github.io/en/mix/#tickers-channel :param str symbol: unified symbol of the market to unwatch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns any: status of the unwatch request """ await self.load_markets() return await self.un_watch_channel(symbol, 'ticker', 'ticker', params) async def watch_tickers(self, symbols: Strings = None, params={}) -> Tickers: """ watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for all markets of a specific list https://coincatch.github.io/github.io/en/mix/#tickers-channel :param str[] symbols: unified symbol of the market to watch the tickers for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() if symbols is None: symbols = self.symbols topics = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) instType, instId = self.get_public_inst_type_and_id(market) args: dict = { 'instType': instType, 'channel': 'ticker', 'instId': instId, } topics.append(args) messageHashes.append('ticker:' + symbol) tickers = await self.watch_public_multiple(messageHashes, messageHashes, topics, params) if self.newUpdates: result: dict = {} result[tickers['symbol']] = tickers return result return self.filter_by_array(self.tickers, 'symbol', symbols) def handle_ticker(self, client: Client, message): # # action: 'snapshot', # arg: {instType: 'sp', channel: 'ticker', instId: 'ETHUSDT'}, # data: [ # { # instId: 'ETHUSDT', # last: '2421.06', # open24h: '2416.93', # high24h: '2441.47', # low24h: '2352.99', # bestBid: '2421.03', # bestAsk: '2421.06', # baseVolume: '9445.2043', # quoteVolume: '22807159.1148', # ts: 1728131730687, # labeId: 0, # openUtc: '2414.50', # chgUTC: '0.00272', # bidSz: '3.866', # askSz: '0.124' # } # ], # ts: 1728131730688 # arg = self.safe_dict(message, 'arg', {}) market = self.get_market_from_arg(arg) marketId = market['id'] hash = 'ticker:' if marketId.find('_DMCBL') >= 0: market = self.handle_dmcbl_market_by_message_hashes(market, hash, client) data = self.safe_list(message, 'data', []) ticker = self.parse_ws_ticker(self.safe_dict(data, 0, {}), market) symbol = market['symbol'] self.tickers[symbol] = ticker messageHash = hash + symbol client.resolve(self.tickers[symbol], messageHash) def parse_ws_ticker(self, ticker, market=None): # # spot # { # instId: 'ETHUSDT', # last: '2421.06', # open24h: '2416.93', # high24h: '2441.47', # low24h: '2352.99', # bestBid: '2421.03', # bestAsk: '2421.06', # baseVolume: '9445.2043', # quoteVolume: '22807159.1148', # ts: 1728131730687, # labeId: 0, # openUtc: '2414.50', # chgUTC: '0.00272', # bidSz: '3.866', # askSz: '0.124' # } # # swap # { # instId: 'ETHUSDT', # last: '2434.47', # bestAsk: '2434.48', # bestBid: '2434.47', # high24h: '2471.68', # low24h: '2400.01', # priceChangePercent: '0.00674', # capitalRate: '0.000082', # nextSettleTime: 1728489600000, # systemTime: 1728471993602, # markPrice: '2434.46', # indexPrice: '2435.44', # holding: '171450.25', # baseVolume: '1699298.91', # quoteVolume: '4144522832.32', # openUtc: '2439.67', # chgUTC: '-0.00213', # symbolType: 1, # symbolId: 'ETHUSDT_UMCBL', # deliveryPrice: '0', # bidSz: '26.12', # askSz: '49.6' # } # last = self.safe_string(ticker, 'last') timestamp = self.safe_integer_2(ticker, 'ts', 'systemTime') return self.safe_ticker({ 'symbol': market['symbol'], 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'high': self.safe_string(ticker, 'high24h'), 'low': self.safe_string(ticker, 'low24h'), 'bid': self.safe_string(ticker, 'bestBid'), 'bidVolume': self.safe_string(ticker, 'bidSz'), 'ask': self.safe_string(ticker, 'bestAsk'), 'askVolume': self.safe_string(ticker, 'askSz'), 'vwap': None, 'open': self.safe_string_2(ticker, 'open24h', 'openUtc'), 'close': last, 'last': last, 'previousClose': None, 'change': None, 'percentage': Precise.string_mul(self.safe_string(ticker, 'chgUTC'), '100'), 'average': None, 'baseVolume': self.safe_number(ticker, 'baseVolume'), 'quoteVolume': self.safe_number(ticker, 'quoteVolume'), 'indexPrice': self.safe_string(ticker, 'indexPrice'), 'markPrice': self.safe_string(ticker, 'markPrice'), 'info': ticker, }, market) async def watch_ohlcv(self, symbol: str, timeframe='1m', since: Int = None, limit: Int = None, params={}) -> List[list]: """ watches historical candlestick data containing the open, high, low, and close price, and the volume of a market https://coincatch.github.io/github.io/en/spot/#candlesticks-channel :param str symbol: unified symbol of the market to fetch OHLCV data for :param str timeframe: the length of time each candle represents :param int [since]: timestamp in ms of the earliest candle to fetch(not including) :param int [limit]: the maximum amount of candles to fetch(not including) :param dict [params]: extra parameters specific to the exchange API endpoint :param bool [params.instType]: the type of the instrument to fetch the OHLCV data for, 'SP' for spot markets, 'MC' for futures markets(default is 'SP') :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() market = self.market(symbol) timeframes = self.options['timeframesForWs'] channel = 'candle' + self.safe_string(timeframes, timeframe) instType, instId = self.get_public_inst_type_and_id(market) args: dict = { 'instType': instType, 'channel': channel, 'instId': instId, } messageHash = 'ohlcv:' + symbol + ':' + timeframe ohlcv = await self.watch_public(messageHash, messageHash, args, params) if self.newUpdates: limit = ohlcv.getLimit(symbol, limit) return self.filter_by_since_limit(ohlcv, since, limit, 0, True) async def un_watch_ohlcv(self, symbol: str, timeframe='1m', params={}) -> Any: """ unsubscribe from the ohlcv channel https://www.bitget.com/api-doc/spot/websocket/public/Candlesticks-Channel :param str symbol: unified symbol of the market to unwatch the ohlcv for @param timeframe :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() timeframes = self.options['timeframesForWs'] interval = self.safe_string(timeframes, timeframe) channel = 'candle' + interval return await self.un_watch_channel(symbol, channel, 'ohlcv:' + interval, params) def handle_ohlcv(self, client: Client, message): # # { # action: 'update', # arg: {instType: 'sp', channel: 'candle1D', instId: 'ETHUSDT'}, # data: [ # [ # '1728316800000', # '2474.5', # '2478.21', # '2459.8', # '2463.51', # '86.0551' # ] # ], # ts: 1728317607657 # } # arg = self.safe_dict(message, 'arg', {}) market = self.get_market_from_arg(arg) marketId = market['id'] hash = 'ohlcv:' data = self.safe_list(message, 'data', []) channel = self.safe_string(arg, 'channel') klineType = channel[6:] timeframe = self.find_timeframe(klineType) if marketId.find('_DMCBL') >= 0: market = self.handle_dmcbl_market_by_message_hashes(market, hash, client, timeframe) symbol = market['symbol'] if not (symbol in self.ohlcvs): self.ohlcvs[symbol] = {} if not (timeframe in self.ohlcvs[symbol]): limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) self.ohlcvs[symbol][timeframe] = ArrayCacheByTimestamp(limit) stored = self.ohlcvs[symbol][timeframe] for i in range(0, len(data)): candle = self.safe_list(data, i, []) parsed = self.parse_ws_ohlcv(candle, market) stored.append(parsed) messageHash = hash + symbol + ':' + timeframe client.resolve(stored, messageHash) def parse_ws_ohlcv(self, ohlcv, market: Market = None) -> list: # # [ # '1728316800000', # '2474.5', # '2478.21', # '2459.8', # '2463.51', # '86.0551' # ] # return [ self.safe_integer(ohlcv, 0), self.safe_number(ohlcv, 1), self.safe_number(ohlcv, 2), self.safe_number(ohlcv, 3), self.safe_number(ohlcv, 4), self.safe_number(ohlcv, 5), ] async def watch_order_book(self, symbol: str, limit: Int = None, params={}) -> OrderBook: """ watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://coincatch.github.io/github.io/en/spot/#depth-channel :param str symbol: unified symbol of the market to fetch the order book for :param int [limit]: the maximum amount of order book entries to return :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A dictionary of `order book structures ` indexed by market symbols """ return await self.watch_order_book_for_symbols([symbol], limit, params) async def un_watch_order_book(self, symbol: str, params={}) -> Any: """ unsubscribe from the orderbook channel https://coincatch.github.io/github.io/en/spot/#depth-channel :param str symbol: unified symbol of the market to fetch the order book for :param dict [params]: extra parameters specific to the exchange API endpoint :param int [params.limit]: orderbook limit, default is None :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() channel = 'books' limit = self.safe_integer(params, 'limit') if (limit == 5) or (limit == 15): params = self.omit(params, 'limit') channel += str(limit) return await self.un_watch_channel(symbol, channel, channel, params) async def watch_order_book_for_symbols(self, symbols: List[str], limit: Int = None, params={}) -> OrderBook: """ watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data https://coincatch.github.io/github.io/en/spot/#depth-channel @param symbols :param int [limit]: the maximum amount of order book entries to return :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: A dictionary of `order book structures ` indexed by market symbols """ await self.load_markets() symbols = self.market_symbols(symbols) channel = 'books' topics = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) instType, instId = self.get_public_inst_type_and_id(market) args: dict = { 'instType': instType, 'channel': channel, 'instId': instId, } topics.append(args) messageHashes.append(channel + ':' + symbol) orderbook = await self.watch_public_multiple(messageHashes, messageHashes, topics, params) return orderbook.limit() def handle_order_book(self, client: Client, message): # # { # action: 'update', # arg: {instType: 'sp', channel: 'books', instId: 'ETHUSDT'}, # data: [ # { # asks: [[2507.07, 0.4248]], # bids: [[2507.05, 0.1198]], # checksum: -1400923312, # ts: '1728339446908' # } # ], # ts: 1728339446908 # } # arg = self.safe_dict(message, 'arg', {}) market = self.get_market_from_arg(arg) marketId = market['id'] hash = 'books:' if marketId.find('_DMCBL') >= 0: market = self.handle_dmcbl_market_by_message_hashes(market, hash, client) symbol = market['symbol'] channel = self.safe_string(arg, 'channel') messageHash = hash + symbol data = self.safe_list(message, 'data', []) rawOrderBook = self.safe_dict(data, 0) timestamp = self.safe_integer(rawOrderBook, 'ts') incrementalBook = channel if incrementalBook: if not (symbol in self.orderbooks): ob = self.counted_order_book({}) ob['symbol'] = symbol self.orderbooks[symbol] = ob storedOrderBook = self.orderbooks[symbol] asks = self.safe_list(rawOrderBook, 'asks', []) bids = self.safe_list(rawOrderBook, 'bids', []) self.handle_deltas(storedOrderBook['asks'], asks) self.handle_deltas(storedOrderBook['bids'], bids) storedOrderBook['timestamp'] = timestamp storedOrderBook['datetime'] = self.iso8601(timestamp) checksum = self.safe_bool(self.options, 'checksum', True) isSnapshot = self.safe_string(message, 'action') == 'snapshot' if not isSnapshot and checksum: storedAsks = storedOrderBook['asks'] storedBids = storedOrderBook['bids'] asksLength = len(storedAsks) bidsLength = len(storedBids) payloadArray = [] for i in range(0, 25): if i < bidsLength: payloadArray.append(storedBids[i][2][0]) payloadArray.append(storedBids[i][2][1]) if i < asksLength: payloadArray.append(storedAsks[i][2][0]) payloadArray.append(storedAsks[i][2][1]) payload = ':'.join(payloadArray) calculatedChecksum = self.crc32(payload, True) responseChecksum = self.safe_integer(rawOrderBook, 'checksum') if calculatedChecksum != responseChecksum: self.spawn(self.handle_check_sum_error, client, symbol, messageHash) return else: orderbook = self.order_book({}) parsedOrderbook = self.parse_order_book(rawOrderBook, symbol, timestamp) orderbook.reset(parsedOrderbook) self.orderbooks[symbol] = orderbook client.resolve(self.orderbooks[symbol], messageHash) async def handle_check_sum_error(self, client: Client, symbol: str, messageHash: str): await self.un_watch_order_book(symbol) error = ChecksumError(self.id + ' ' + self.orderbook_checksum_message(symbol)) client.reject(error, messageHash) def handle_delta(self, bookside, delta): bidAsk = self.parse_bid_ask(delta, 0, 1) bidAsk.append(delta) bookside.storeArray(bidAsk) def handle_deltas(self, bookside, deltas): for i in range(0, len(deltas)): self.handle_delta(bookside, deltas[i]) async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ get the list of most recent trades for a particular symbol https://coincatch.github.io/github.io/en/spot/#trades-channel :param str symbol: unified symbol of the market to fetch trades for :param int [since]: timestamp in ms of the earliest trade to fetch :param int [limit]: the maximum amount of trades to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `trade structures ` """ return await self.watch_trades_for_symbols([symbol], since, limit, params) async def watch_trades_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ watches information on multiple trades made in a market https://coincatch.github.io/github.io/en/spot/#trades-channel @param symbols :param int [since]: the earliest time in ms to fetch orders for :param int [limit]: the maximum number of trade structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `trade structures ` """ symbolsLength = len(symbols) if symbolsLength == 0: raise ArgumentsRequired(self.id + ' watchTradesForSymbols() requires a non-empty array of symbols') await self.load_markets() symbols = self.market_symbols(symbols) topics = [] messageHashes = [] for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) instType, instId = self.get_public_inst_type_and_id(market) args: dict = { 'instType': instType, 'channel': 'trade', 'instId': instId, } topics.append(args) messageHashes.append('trade:' + symbol) trades = await self.watch_public_multiple(messageHashes, messageHashes, topics, params) if self.newUpdates: first = self.safe_dict(trades, 0) tradeSymbol = self.safe_string(first, 'symbol') limit = trades.getLimit(tradeSymbol, limit) return self.filter_by_since_limit(trades, since, limit, 'timestamp', True) async def un_watch_trades(self, symbol: str, params={}) -> Any: """ unsubscribe from the trades channel https://coincatch.github.io/github.io/en/spot/#trades-channel :param str symbol: unified symbol of the market to unwatch the trades for :param dict [params]: extra parameters specific to the exchange API endpoint :returns any: status of the unwatch request """ await self.load_markets() return await self.un_watch_channel(symbol, 'trade', 'trade', params) def handle_trades(self, client: Client, message): # # { # action: 'update', # arg: {instType: 'sp', channel: 'trade', instId: 'ETHUSDT'}, # data: [['1728341807469', '2421.41', '0.478', 'sell']], # ts: 1728341807482 # } # arg = self.safe_dict(message, 'arg', {}) market = self.get_market_from_arg(arg) marketId = market['id'] hash = 'trade:' if marketId.find('_DMCBL') >= 0: market = self.handle_dmcbl_market_by_message_hashes(market, hash, client) symbol = market['symbol'] if not (symbol in self.trades): limit = self.safe_integer(self.options, 'tradesLimit', 1000) self.trades[symbol] = ArrayCache(limit) stored = self.trades[symbol] data = self.safe_list(message, 'data', []) if data is not None: data = self.sort_by(data, 0) for i in range(0, len(data)): trade = self.safe_list(data, i) parsed = self.parse_ws_trade(trade, market) stored.append(parsed) messageHash = hash + symbol client.resolve(stored, messageHash) def parse_ws_trade(self, trade, market=None) -> Trade: # # [ # '1728341807469', # '2421.41', # '0.478', # 'sell' # ] # timestamp = self.safe_integer(trade, 0) return self.safe_trade({ 'id': None, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'symbol': market['symbol'], 'side': self.safe_string_lower(trade, 3), 'price': self.safe_string(trade, 1), 'amount': self.safe_string(trade, 2), 'cost': None, 'takerOrMaker': None, 'type': None, 'order': None, 'fee': None, 'info': trade, }, market) async def watch_balance(self, params={}) -> Balances: """ watch balance and get the amount of funds available for trading or funds locked in orders https://coincatch.github.io/github.io/en/spot/#account-channel https://coincatch.github.io/github.io/en/mix/#account-channel :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.type]: 'spot' or 'swap'(default is 'spot') :param str [params.instType]: *swap only* 'umcbl' or 'dmcbl'(default is 'umcbl') :returns dict: a `balance structure ` """ type = None type, params = self.handle_market_type_and_params('watchBalance', None, params) instType = 'spbl' # must be lower case for spot if type == 'swap': instType = 'umcbl' channel = 'account' instType, params = self.handle_option_and_params(params, 'watchBalance', 'instType', instType) args: dict = { 'instType': instType, 'channel': channel, 'instId': 'default', } messageHash = 'balance:' + instType.lower() return await self.watch_private(messageHash, messageHash, args, params) def handle_balance(self, client: Client, message): # # spot # { # action: 'snapshot', # arg: {instType: 'spbl', channel: 'account', instId: 'default'}, # data: [ # { # coinId: '3', # coinName: 'ETH', # available: '0.0000832', # frozen: '0', # lock: '0' # } # ], # ts: 1728464548725 # } # # # swap # { # action: 'snapshot', # arg: {instType: 'dmcbl', channel: 'account', instId: 'default'}, # data: [ # { # marginCoin: 'ETH', # locked: '0.00000000', # available: '0.00001203', # maxOpenPosAvailable: '0.00001203', # maxTransferOut: '0.00001203', # equity: '0.00001203', # usdtEquity: '0.029092328738', # coinDisplayName: 'ETH' # } # ], # ts: 1728650777643 # } # data = self.safe_list(message, 'data', []) for i in range(0, len(data)): rawBalance = data[i] currencyId = self.safe_string_2(rawBalance, 'coinName', 'marginCoin') code = self.safe_currency_code(currencyId) account = self.balance[code] if (code in self.balance) else self.account() freeQuery = 'maxTransferOut' if ('maxTransferOut' in rawBalance) else 'available' account['free'] = self.safe_string(rawBalance, freeQuery) account['total'] = self.safe_string(rawBalance, 'equity') account['used'] = self.safe_string(rawBalance, 'frozen') self.balance[code] = account self.balance = self.safe_balance(self.balance) arg = self.safe_dict(message, 'arg') instType = self.safe_string_lower(arg, 'instType') messageHash = 'balance:' + instType client.resolve(self.balance, messageHash) async def watch_orders(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ watches information on multiple orders made by the user https://coincatch.github.io/github.io/en/spot/#order-channel https://coincatch.github.io/github.io/en/mix/#order-channel https://coincatch.github.io/github.io/en/mix/#plan-order-channel :param str symbol: unified market symbol of the market orders were made in :param int [since]: the earliest time in ms to fetch orders for :param int [limit]: the maximum number of order structures to retrieve :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.type]: 'spot' or 'swap' :param str [params.instType]: *swap only* 'umcbl' or 'dmcbl'(default is 'umcbl') :param bool [params.trigger]: *swap only* whether to watch trigger orders(default is False) :returns dict[]: a list of `order structures ` """ methodName = 'watchOrders' await self.load_markets() market = None marketId = None if symbol is not None: market = self.market(symbol) symbol = market['symbol'] marketId = market['id'] marketType = None marketType, params = self.handle_market_type_and_params(methodName, market, params) instType = 'spbl' instId = marketId if marketType == 'spot': if symbol is None: raise ArgumentsRequired(self.id + ' ' + methodName + '() requires a symbol argument for ' + marketType + ' markets.') else: instId = 'default' instType = 'umcbl' if symbol is None: instType, params = self.handle_option_and_params(params, methodName, 'instType', instType) else: if marketId.find('_DMCBL') >= 0: instType = 'dmcbl' channel = 'orders' isTrigger = self.safe_bool(params, 'trigger') if isTrigger: channel = 'ordersAlgo' # channel does not return any data params = self.omit(params, 'trigger') args: dict = { 'instType': instType, 'channel': channel, 'instId': instId, } messageHash = 'orders' if symbol is not None: messageHash += ':' + symbol orders = await self.watch_private(messageHash, messageHash, args, params) if self.newUpdates: limit = orders.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True) def handle_order(self, client: Client, message): # # spot # # { # action: 'snapshot', # arg: {instType: 'spbl', channel: 'orders', instId: 'ETHUSDT_SPBL'}, # data: [ # { # instId: 'ETHUSDT_SPBL', # ordId: '1228627925964996608', # clOrdId: 'f0cccf74-c535-4523-a53d-dbe3b9958559', # px: '2000', # sz: '0.001', # notional: '2', # ordType: 'limit', # force: 'normal', # side: 'buy', # accFillSz: '0', # avgPx: '0', # status: 'new', # cTime: 1728653645030, # uTime: 1728653645030, # orderFee: [], # eps: 'API' # } # ], # ts: 1728653645046 # } # # swap # # { # action: 'snapshot', # arg: {instType: 'umcbl', channel: 'orders', instId: 'default'}, # data: [ # { # accFillSz: '0', # cTime: 1728653796976, # clOrdId: '1228628563272753152', # eps: 'API', # force: 'normal', # hM: 'single_hold', # instId: 'ETHUSDT_UMCBL', # lever: '5', # low: False, # notionalUsd: '20', # ordId: '1228628563188867072', # ordType: 'limit', # orderFee: [], # posSide: 'net', # px: '2000', # side: 'buy', # status: 'new', # sz: '0.01', # tS: 'buy_single', # tdMode: 'cross', # tgtCcy: 'USDT', # uTime: 1728653796976 # } # ], # ts: 1728653797002 # } # # arg = self.safe_dict(message, 'arg', {}) instType = self.safe_string(arg, 'instType') argInstId = self.safe_string(arg, 'instId') marketType = None if instType == 'spbl': marketType = 'spot' else: marketType = 'swap' data = self.safe_list(message, 'data', []) if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) self.orders = ArrayCacheBySymbolById(limit) hash = 'orders' stored = self.orders symbol: Str = None for i in range(0, len(data)): order = data[i] marketId = self.safe_string(order, 'instId', argInstId) market = self.safe_market(marketId, None, None, marketType) parsed = self.parse_ws_order(order, market) stored.append(parsed) symbol = parsed['symbol'] messageHash = 'orders:' + symbol client.resolve(stored, messageHash) client.resolve(stored, hash) def parse_ws_order(self, order: dict, market: Market = None) -> Order: # # spot # { # instId: 'ETHUSDT_SPBL', # ordId: '1228627925964996608', # clOrdId: 'f0cccf74-c535-4523-a53d-dbe3b9958559', # px: '2000', # sz: '0.001', # notional: '2', # ordType: 'limit', # force: 'normal', # side: 'buy', # accFillSz: '0', # avgPx: '0', # status: 'new', # cTime: 1728653645030, # uTime: 1728653645030, # orderFee: orderFee: [{fee: '0', feeCcy: 'USDT'}], # eps: 'API' # } # # swap # { # accFillSz: '0', # cTime: 1728653796976, # clOrdId: '1228628563272753152', # eps: 'API', # force: 'normal', # hM: 'single_hold', # instId: 'ETHUSDT_UMCBL', # lever: '5', # low: False, # notionalUsd: '20', # ordId: '1228628563188867072', # ordType: 'limit', # orderFee: [{fee: '0', feeCcy: 'USDT'}], # posSide: 'net', # px: '2000', # side: 'buy', # status: 'new', # sz: '0.01', # tS: 'buy_single', # tdMode: 'cross', # tgtCcy: 'USDT', # uTime: 1728653796976 # } # marketId = self.safe_string(order, 'instId') settleId = self.safe_string(order, 'tgtCcy') market = self.safeMarketCustom(marketId, market, settleId) timestamp = self.safe_integer(order, 'cTime') symbol = market['symbol'] rawStatus = self.safe_string(order, 'status') orderFee = self.safe_list(order, 'orderFee', []) fee = self.safe_dict(orderFee, 0) feeCost = Precise.string_mul(self.safe_string(fee, 'fee'), '-1') feeCurrency = self.safe_string(fee, 'feeCcy') price = self.omit_zero(self.safe_string(order, 'px')) priceAvg = self.omit_zero(self.safe_string(order, 'avgPx')) if price is None: price = priceAvg type = self.safe_string_lower(order, 'ordType') return self.safe_order({ 'id': self.safe_string(order, 'ordId'), 'clientOrderId': self.safe_string(order, 'clOrdId'), 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': None, 'lastUpdateTimestamp': self.safe_integer(order, 'uTime'), 'status': self.parse_order_status(rawStatus), 'symbol': symbol, 'type': type, 'timeInForce': self.parseOrderTimeInForce(self.safe_string_lower(order, 'force')), 'side': self.safe_string_lower(order, 'side'), 'price': price, 'average': self.safe_string(order, 'avgPx'), 'amount': self.safe_string(order, 'sz'), 'filled': self.safe_string(order, 'accFillSz'), 'remaining': None, 'triggerPrice': None, 'takeProfitPrice': None, 'stopLossPrice': None, 'cost': self.safe_string(order, 'notional'), 'trades': None, 'fee': { 'currency': feeCurrency, 'cost': feeCost, }, 'reduceOnly': self.safe_bool(order, 'low'), 'postOnly': None, 'info': order, }, market) async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]: """ watch all open positions https://coincatch.github.io/github.io/en/mix/#positions-channel :param str[]|None symbols: list of unified market symbols @param since @param limit :param dict params: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `position structure ` """ await self.load_markets() symbols = self.market_symbols(symbols, 'swap') messageHashes = [] hash = 'positions' instTypes = [] if symbols is not None: for i in range(0, len(symbols)): symbol = symbols[i] market = self.market(symbol) instType = self.get_private_inst_type(market) if not self.in_array(instType, instTypes): instTypes.append(instType) messageHashes.append(hash + '::' + symbol) else: instTypes = ['umcbl', 'dmcbl'] messageHashes.append(hash) args = [] subscribeHashes = [] for i in range(0, len(instTypes)): instType = instTypes[i] arg: dict = { 'instType': instType, 'channel': hash, 'instId': 'default', } subscribeHashes.append(hash + '::' + instType) args.append(arg) newPositions = await self.watch_private_multiple(messageHashes, subscribeHashes, args, params) if self.newUpdates: return newPositions return self.filter_by_symbols_since_limit(newPositions, symbols, since, limit, True) def get_private_inst_type(self, market: Market): marketId = market['id'] if marketId.find('_DMCBL') >= 0: return 'dmcbl' return 'umcbl' def handle_positions(self, client: Client, message): # # { # action: 'snapshot', # arg: {instType: 'umcbl', channel: 'positions', instId: 'default'}, # data: [ # { # posId: '1221355728745619456', # instId: 'ETHUSDT_UMCBL', # instName: 'ETHUSDT', # marginCoin: 'USDT', # margin: '5.27182', # marginMode: 'crossed', # holdSide: 'long', # holdMode: 'single_hold', # total: '0.01', # available: '0.01', # locked: '0', # averageOpenPrice: '2635.91', # leverage: 5, # achievedProfits: '0', # upl: '-0.0267', # uplRate: '-0.005064664576', # liqPx: '-3110.66866033', # keepMarginRate: '0.0033', # marginRate: '0.002460827254', # cTime: '1726919818102', # uTime: '1728919604312', # markPrice: '2633.24', # autoMargin: 'off' # } # ], # ts: 1728919604329 # } # if self.positions is None: self.positions = ArrayCacheBySymbolBySide() cache = self.positions rawPositions = self.safe_value(message, 'data', []) dataLength = len(rawPositions) if dataLength == 0: return newPositions = [] symbols = [] for i in range(0, len(rawPositions)): rawPosition = rawPositions[i] position = self.parse_ws_position(rawPosition) symbols.append(position['symbol']) newPositions.append(position) cache.append(position) hash = 'positions' messageHashes = self.find_message_hashes(client, hash) for i in range(0, len(messageHashes)): messageHash = messageHashes[i] parts = messageHash.split('::') symbol = parts[1] if self.in_array(symbol, symbols): positionsForSymbol = [] for j in range(0, len(newPositions)): position = newPositions[j] if position['symbol'] == symbol: positionsForSymbol.append(position) client.resolve(positionsForSymbol, messageHash) client.resolve(newPositions, hash) def parse_ws_position(self, position, market=None): # # { # posId: '1221355728745619456', # instId: 'ETHUSDT_UMCBL', # instName: 'ETHUSDT', # marginCoin: 'USDT', # margin: '5.27182', # marginMode: 'crossed', # holdSide: 'long', # holdMode: 'single_hold', # total: '0.01', # available: '0.01', # locked: '0', # averageOpenPrice: '2635.91', # leverage: 5, # achievedProfits: '0', # upl: '-0.0267', # uplRate: '-0.005064664576', # liqPx: '-3110.66866033', # keepMarginRate: '0.0033', # marginRate: '0.002460827254', # cTime: '1726919818102', # uTime: '1728919604312', # markPrice: '2633.24', # autoMargin: 'off' # } # marketId = self.safe_string(position, 'symbol') settleId = self.safe_string(position, 'marginCoin') market = self.safeMarketCustom(marketId, market, settleId) timestamp = self.safe_integer(position, 'cTime') marginModeId = self.safe_string(position, 'marginMode') marginMode = self.get_supported_mapping(marginModeId, { 'crossed': 'cross', 'isolated': 'isolated', }) isHedged: Bool = None holdMode = self.safe_string(position, 'holdMode') if holdMode == 'double_hold': isHedged = True elif holdMode == 'single_hold': isHedged = False percentageDecimal = self.safe_string(position, 'uplRate') percentage = Precise.string_mul(percentageDecimal, '100') margin = self.safe_number(position, 'margin') return self.safe_position({ 'symbol': market['symbol'], 'id': None, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'contracts': self.safe_number(position, 'total'), 'contractSize': None, 'side': self.safe_string_lower(position, 'holdSide'), 'notional': margin, # todo check 'leverage': self.safe_integer(position, 'leverage'), 'unrealizedPnl': self.safe_number(position, 'upl'), 'realizedPnl': self.safe_number(position, 'achievedProfits'), 'collateral': None, # todo check 'entryPrice': self.safe_number(position, 'averageOpenPrice'), 'markPrice': self.safe_number(position, 'markPrice'), 'liquidationPrice': self.safe_number(position, 'liqPx'), 'marginMode': marginMode, 'hedged': isHedged, 'maintenanceMargin': None, # todo check 'maintenanceMarginPercentage': self.safe_number(position, 'keepMarginRate'), 'initialMargin': margin, # todo check 'initialMarginPercentage': None, 'marginRatio': self.safe_number(position, 'marginRate'), 'lastUpdateTimestamp': self.safe_integer(position, 'uTime'), 'lastPrice': None, 'stopLossPrice': None, 'takeProfitPrice': None, 'percentage': percentage, 'info': position, }) def handle_error_message(self, client: Client, message): # # {event: "error", code: 30001, msg: "Channel does not exist"} # event = self.safe_string(message, 'event') try: if event == 'error': code = self.safe_string(message, 'code') feedback = self.id + ' ' + self.json(message) self.throw_exactly_matched_exception(self.exceptions['ws']['exact'], code, feedback) msg = self.safe_string(message, 'msg', '') self.throw_broadly_matched_exception(self.exceptions['ws']['broad'], msg, feedback) raise ExchangeError(feedback) return False except Exception as e: if isinstance(e, AuthenticationError): messageHash = 'authenticated' client.reject(e, messageHash) if messageHash in client.subscriptions: del client.subscriptions[messageHash] else: client.reject(e) return True def handle_message(self, client: Client, message): # todo handle with subscribe and unsubscribe if self.handle_error_message(client, message): return content = self.safe_string(message, 'message') if content == 'pong': self.handle_pong(client, message) return if message == 'pong': self.handle_pong(client, message) return event = self.safe_string(message, 'event') if event == 'login': self.handle_authenticate(client, message) return if event == 'subscribe': self.handle_subscription_status(client, message) return if event == 'unsubscribe': self.handle_un_subscription_status(client, message) return data = self.safe_dict(message, 'arg', {}) channel = self.safe_string(data, 'channel') if channel == 'ticker': self.handle_ticker(client, message) if channel.find('candle') >= 0: self.handle_ohlcv(client, message) if channel.find('books') >= 0: self.handle_order_book(client, message) if channel == 'trade': self.handle_trades(client, message) if channel == 'account': self.handle_balance(client, message) if (channel == 'orders') or (channel == 'ordersAlgo'): self.handle_order(client, message) if channel == 'positions': self.handle_positions(client, message) def ping(self, client: Client): return 'ping' def handle_pong(self, client: Client, message): client.lastPong = self.milliseconds() return message def handle_subscription_status(self, client: Client, message): return message def handle_un_subscription_status(self, client: Client, message): argsList = self.safe_list(message, 'args') if argsList is None: argsList = [self.safe_dict(message, 'arg', {})] for i in range(0, len(argsList)): arg = argsList[i] channel = self.safe_string(arg, 'channel') if channel == 'books': self.handle_order_book_un_subscription(client, message) elif channel == 'trade': self.handle_trades_un_subscription(client, message) elif channel == 'ticker': self.handle_ticker_un_subscription(client, message) elif channel.startswith('candle'): self.handle_ohlcv_un_subscription(client, message) return message def handle_order_book_un_subscription(self, client: Client, message): arg = self.safe_dict(message, 'arg', {}) instType = self.safe_string_lower(arg, 'instType') type = 'spot' if (instType == 'sp') else 'swap' instId = self.safe_string(arg, 'instId') market = self.safe_market(instId, None, None, type) symbol = market['symbol'] messageHash = 'unsubscribe:orderbook:' + market['symbol'] subMessageHash = 'orderbook:' + symbol if symbol in self.orderbooks: del self.orderbooks[symbol] if subMessageHash in client.subscriptions: del client.subscriptions[subMessageHash] if messageHash in client.subscriptions: del client.subscriptions[messageHash] error = UnsubscribeError(self.id + ' orderbook ' + symbol) client.reject(error, subMessageHash) client.resolve(True, messageHash) def handle_trades_un_subscription(self, client: Client, message): arg = self.safe_dict(message, 'arg', {}) instType = self.safe_string_lower(arg, 'instType') type = 'spot' if (instType == 'sp') else 'swap' instId = self.safe_string(arg, 'instId') market = self.safe_market(instId, None, None, type) symbol = market['symbol'] messageHash = 'unsubscribe:trade:' + market['symbol'] subMessageHash = 'trade:' + symbol if symbol in self.trades: del self.trades[symbol] if subMessageHash in client.subscriptions: del client.subscriptions[subMessageHash] if messageHash in client.subscriptions: del client.subscriptions[messageHash] error = UnsubscribeError(self.id + ' trades ' + symbol) client.reject(error, subMessageHash) client.resolve(True, messageHash) def handle_ticker_un_subscription(self, client: Client, message): arg = self.safe_dict(message, 'arg', {}) instType = self.safe_string_lower(arg, 'instType') type = 'spot' if (instType == 'sp') else 'swap' instId = self.safe_string(arg, 'instId') market = self.safe_market(instId, None, None, type) symbol = market['symbol'] messageHash = 'unsubscribe:ticker:' + market['symbol'] subMessageHash = 'ticker:' + symbol if symbol in self.tickers: del self.tickers[symbol] if subMessageHash in client.subscriptions: del client.subscriptions[subMessageHash] if messageHash in client.subscriptions: del client.subscriptions[messageHash] error = UnsubscribeError(self.id + ' ticker ' + symbol) client.reject(error, subMessageHash) client.resolve(True, messageHash) def handle_ohlcv_un_subscription(self, client: Client, message): arg = self.safe_dict(message, 'arg', {}) instType = self.safe_string_lower(arg, 'instType') type = 'spot' if (instType == 'sp') else 'swap' instId = self.safe_string(arg, 'instId') channel = self.safe_string(arg, 'channel') interval = channel.replace('candle', '') timeframes = self.safe_dict(self.options, 'timeframesForWs') timeframe = self.find_timeframe(interval, timeframes) market = self.safe_market(instId, None, None, type) symbol = market['symbol'] messageHash = 'unsubscribe:ohlcv:' + timeframe + ':' + market['symbol'] subMessageHash = 'ohlcv:' + symbol + ':' + timeframe if symbol in self.ohlcvs: if timeframe in self.ohlcvs[symbol]: del self.ohlcvs[symbol][timeframe] self.clean_unsubscription(client, subMessageHash, messageHash)