# -*- 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, ArrayCacheByTimestamp import hashlib from ccxt.base.types import Any, Balances, Int, Num, Order, OrderBook, OrderSide, OrderType, 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 ArgumentsRequired from ccxt.base.errors import BadRequest from ccxt.base.precise import Precise class cex(ccxt.async_support.cex): def describe(self) -> Any: return self.deep_extend(super(cex, self).describe(), { 'has': { 'ws': True, 'watchBalance': True, 'watchTicker': True, 'watchTickers': True, 'watchTrades': True, 'watchTradesForSymbols': False, 'watchMyTrades': True, 'watchOrders': True, 'watchOrderBook': True, 'watchOHLCV': True, 'watchPosition': None, 'createOrderWs': True, 'editOrderWs': True, 'cancelOrderWs': True, 'cancelOrdersWs': True, 'fetchOrderWs': True, 'fetchOpenOrdersWs': True, 'fetchTickerWs': True, 'fetchBalanceWs': True, }, 'urls': { 'api': { 'ws': 'wss://ws.cex.io/ws', }, }, 'options': { 'orderbook': {}, }, 'streaming': { }, 'exceptions': { }, }) def request_id(self): requestId = self.sum(self.safe_integer(self.options, 'requestId', 0), 1) self.options['requestId'] = requestId return str(requestId) async def watch_balance(self, params={}) -> Balances: """ watch balance and get the amount of funds available for trading or funds locked in orders https://cex.io/websocket-api#get-balance :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `balance structure ` """ await self.authenticate(params) messageHash = self.request_id() url = self.urls['api']['ws'] subscribe: dict = { 'e': 'get-balance', 'data': {}, 'oid': self.request_id(), } request = self.deep_extend(subscribe, params) return await self.watch(url, messageHash, request, messageHash, request) def handle_balance(self, client: Client, message): # # { # "e": "get-balance", # "data": { # "balance": { # "BTC": "0.00000000", # "USD": "0.00", # ... # }, # "obalance": { # "BTC": "0.00000000", # "USD": "0.00", # ... # }, # "time": 1663761159605 # }, # "oid": 1, # "ok": "ok" # } # data = self.safe_value(message, 'data', {}) freeBalance = self.safe_value(data, 'balance', {}) usedBalance = self.safe_value(data, 'obalance', {}) result: dict = { 'info': data, } currencyIds = list(freeBalance.keys()) for i in range(0, len(currencyIds)): currencyId = currencyIds[i] account = self.account() account['free'] = self.safe_string(freeBalance, currencyId) account['used'] = self.safe_string(usedBalance, currencyId) code = self.safe_currency_code(currencyId) result[code] = account self.balance = self.safe_balance(result) messageHash = self.safe_string(message, 'oid') client.resolve(self.balance, messageHash) 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. Note: can only watch one symbol at a time. https://cex.io/websocket-api#old-pair-room :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 ` """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] url = self.urls['api']['ws'] messageHash = 'trades' subscriptionHash = 'old:' + symbol self.options['currentWatchTradeSymbol'] = symbol # exchange supports only 1 symbol for self watchTrades channel client = self.safe_value(self.clients, url) if client is not None: subscriptionKeys = list(client.subscriptions.keys()) for i in range(0, len(subscriptionKeys)): subscriptionKey = subscriptionKeys[i] if subscriptionKey == subscriptionHash: continue subscriptionKey = subscriptionKey[0:3] if subscriptionKey == 'old': raise ExchangeError(self.id + ' watchTrades() only supports watching one symbol at a time.') message: dict = { 'e': 'subscribe', 'rooms': ['pair-' + market['base'] + '-' + market['quote']], } request = self.deep_extend(message, params) trades = await self.watch(url, messageHash, request, subscriptionHash) # assing symbol to the trades does not contain symbol information for i in range(0, len(trades)): trades[i]['symbol'] = symbol return self.filter_by_since_limit(trades, since, limit, 'timestamp', True) def handle_trades_snapshot(self, client: Client, message): # # { # "e": "history", # "data": [ # 'buy:1710255706095:444444:71222.2:14892622' # 'sell:1710255658251:42530:71300:14892621' # 'buy:1710252424241:87913:72800:14892620' # ... timestamp descending # ] # } # data = self.safe_list(message, 'data', []) limit = self.safe_integer(self.options, 'tradesLimit', 1000) stored = ArrayCache(limit) symbol = self.safe_string(self.options, 'currentWatchTradeSymbol') if symbol is None: return market = self.market(symbol) dataLength = len(data) for i in range(0, dataLength): index = dataLength - 1 - i rawTrade = data[index] parsed = self.parse_ws_old_trade(rawTrade, market) stored.append(parsed) messageHash = 'trades' self.trades = stored # trades don't have symbol client.resolve(self.trades, messageHash) def parse_ws_old_trade(self, trade, market=None): # # snapshot trade # "sell:1665467367741:3888551:19058.8:14541219" # update trade # ['buy', '1665467516704', '98070', "19057.7", "14541220"] # if not isinstance(trade, list): trade = trade.split(':') side = self.safe_string(trade, 0) timestamp = self.safe_integer(trade, 1) amount = self.safe_string(trade, 2) price = self.safe_string(trade, 3) id = self.safe_string(trade, 4) return self.safe_trade({ 'info': trade, 'id': id, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'symbol': self.safe_string(market, 'symbol'), 'type': None, 'side': side, 'order': None, 'takerOrMaker': None, 'price': price, 'amount': amount, 'cost': None, 'fee': None, }, market) def handle_trade(self, client: Client, message): # # { # "e": "history-update", # "data": [ # ['buy', '1665467516704', '98070', "19057.7", "14541220"] # ] # } # data = self.safe_value(message, 'data', []) stored = self.trades # to do fix self, self.trades is not meant to be used like self dataLength = len(data) for i in range(0, dataLength): index = dataLength - 1 - i rawTrade = data[index] parsed = self.parse_ws_old_trade(rawTrade) stored.append(parsed) messageHash = 'trades' self.trades = stored client.resolve(self.trades, messageHash) async def watch_ticker(self, symbol: str, params={}) -> Ticker: """ https://cex.io/websocket-api#ticker-subscription watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.method]: public or private :returns dict: a `ticker structure ` """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] url = self.urls['api']['ws'] messageHash = 'ticker:' + symbol method = self.safe_string(params, 'method', 'private') # default to private because the specified ticker is received quicker message = { 'e': 'subscribe', 'rooms': [ 'tickers', ], } subscriptionHash = 'tickers' if method == 'private': await self.authenticate() message = { 'e': 'ticker', 'data': [ market['baseId'], market['quoteId'], ], 'oid': self.request_id(), } subscriptionHash = 'ticker:' + symbol request = self.deep_extend(message, params) return await self.watch(url, messageHash, request, subscriptionHash) async def watch_tickers(self, symbols: Strings = None, params={}) -> Tickers: """ https://cex.io/websocket-api#ticker-subscription watches price tickers for multiple markets, statistical information calculated over the past 24 hours for each market :param str[]|None symbols: unified symbols of the markets to fetch the ticker for, all market tickers are returned if not assigned :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a dictionary of `ticker structures ` """ await self.load_markets() symbols = self.market_symbols(symbols) url = self.urls['api']['ws'] messageHash = 'tickers' message: dict = { 'e': 'subscribe', 'rooms': [ 'tickers', ], } request = self.deep_extend(message, params) ticker = await self.watch(url, messageHash, request, messageHash) tickerSymbol = ticker['symbol'] if symbols is not None and not self.in_array(tickerSymbol, symbols): return await self.watch_tickers(symbols, params) if self.newUpdates: result: dict = {} result[tickerSymbol] = ticker return result return self.filter_by_array(self.tickers, 'symbol', symbols) async def fetch_ticker_ws(self, symbol: str, params={}) -> Ticker: """ https://docs.cex.io/#ws-api-ticker-deprecated fetches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the cex api endpoint :returns dict: a `ticker structure ` """ await self.load_markets() market = self.market(symbol) url = self.urls['api']['ws'] messageHash = self.request_id() request = self.extend({ 'e': 'ticker', 'oid': messageHash, 'data': [market['base'], market['quote']], }, params) return await self.watch(url, messageHash, request, messageHash) def handle_ticker(self, client: Client, message): # # { # "e": "tick", # "data": { # "symbol1": "LRC", # "symbol2": "USD", # "price": "0.305", # "open24": "0.301", # "volume": "241421.641700" # } # } # data = self.safe_value(message, 'data', {}) ticker = self.parse_ws_ticker(data) symbol = ticker['symbol'] if symbol is None: return self.tickers[symbol] = ticker messageHash = 'ticker:' + symbol client.resolve(ticker, messageHash) client.resolve(ticker, 'tickers') messageHash = self.safe_string(message, 'oid') if messageHash is not None: client.resolve(ticker, messageHash) def parse_ws_ticker(self, ticker, market=None): # # public # { # "symbol1": "LRC", # "symbol2": "USD", # "price": "0.305", # "open24": "0.301", # "volume": "241421.641700" # } # private # { # "timestamp": "1663764969", # "low": "18756.3", # "high": "19200", # "last": "19200", # "volume": "0.94735907", # "volume30d": "64.61299999", # "bid": 19217.2, # "ask": 19247.5, # "priceChange": "44.3", # "priceChangePercentage": "0.23", # "pair": ["BTC", "USDT"] # } pair = self.safe_value(ticker, 'pair', []) baseId = self.safe_string(ticker, 'symbol1') if baseId is None: baseId = self.safe_string(pair, 0) quoteId = self.safe_string(ticker, 'symbol2') if quoteId is None: quoteId = self.safe_string(pair, 1) base = self.safe_currency_code(baseId) quote = self.safe_currency_code(quoteId) symbol = base + '/' + quote timestamp = self.safe_integer(ticker, 'timestamp') if timestamp is not None: timestamp = timestamp * 1000 return self.safe_ticker({ 'symbol': symbol, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'high': self.safe_string(ticker, 'high'), 'low': self.safe_string(ticker, 'low'), 'bid': self.safe_string(ticker, 'bid'), 'bidVolume': None, 'ask': self.safe_string(ticker, 'ask'), 'askVolume': None, 'vwap': None, 'open': self.safe_string(ticker, 'open24'), 'close': None, 'last': self.safe_string_2(ticker, 'price', 'last'), 'previousClose': None, 'change': self.safe_string(ticker, 'priceChange'), 'percentage': self.safe_string(ticker, 'priceChangePercentage'), 'average': None, 'baseVolume': None, 'quoteVolume': self.safe_string(ticker, 'volume'), 'info': ticker, }, market) async def fetch_balance_ws(self, params={}) -> Balances: """ https://docs.cex.io/#ws-api-get-balance query for balance and get the amount of funds available for trading or funds locked in orders :param dict [params]: extra parameters specific to the cex api endpoint :returns dict: a `balance structure ` """ await self.load_markets() await self.authenticate() url = self.urls['api']['ws'] messageHash = self.request_id() request = self.extend({ 'e': 'get-balance', 'oid': messageHash, }, params) return await self.watch(url, messageHash, request, messageHash) async def watch_orders(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ get the list of orders associated with the user. Note: In CEX.IO system, orders can be present in trade engine or in archive database. There can be time periods(~2 seconds or more), when order is done/canceled, but still not moved to archive database. That means, you cannot see it using calls: archived-orders/open-orders. https://docs.cex.io/#ws-api-open-orders :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 ` """ if symbol is None: raise ArgumentsRequired(self.id + ' watchOrders() requires a symbol argument') await self.load_markets() await self.authenticate(params) url = self.urls['api']['ws'] market = self.market(symbol) symbol = market['symbol'] messageHash = 'orders:' + symbol message: dict = { 'e': 'open-orders', 'data': { 'pair': [ market['baseId'], market['quoteId'], ], }, 'oid': symbol, } request = self.deep_extend(message, params) orders = await self.watch(url, messageHash, request, messageHash, request) if self.newUpdates: limit = orders.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True) async def watch_my_trades(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ get the list of trades associated with the user. Note: In CEX.IO system, orders can be present in trade engine or in archive database. There can be time periods(~2 seconds or more), when order is done/canceled, but still not moved to archive database. That means, you cannot see it using calls: archived-orders/open-orders. https://docs.cex.io/#ws-api-open-orders :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 ` """ if symbol is None: raise ArgumentsRequired(self.id + ' watchMyTrades() requires a symbol argument') await self.load_markets() await self.authenticate(params) url = self.urls['api']['ws'] market = self.market(symbol) messageHash = 'myTrades:' + market['symbol'] subscriptionHash = 'orders:' + market['symbol'] message: dict = { 'e': 'open-orders', 'data': { 'pair': [ market['baseId'], market['quoteId'], ], }, 'oid': market['symbol'], } request = self.deep_extend(message, params) orders = await self.watch(url, messageHash, request, subscriptionHash, request) return self.filter_by_symbol_since_limit(orders, market['symbol'], since, limit) def handle_transaction(self, client: Client, message): data = self.safe_value(message, 'data') symbol2 = self.safe_string(data, 'symbol2') if symbol2 is None: return self.handle_order_update(client, message) self.handle_my_trades(client, message) def handle_my_trades(self, client: Client, message): # # { # "e": "tx", # "data": { # "d": "order:59091012956:a:USD", # "c": "user:up105393824:a:USD", # "a": "0.01", # "ds": 0, # "cs": "15.27", # "user": "up105393824", # "symbol": "USD", # "order": 59091012956, # "amount": "-18.49", # "type": "buy", # "time": "2022-09-24T19:36:18.466Z", # "balance": "15.27", # "id": "59091012966" # } # } # { # "e": "tx", # "data": { # "d": "order:59091012956:a:BTC", # "c": "user:up105393824:a:BTC", # "a": "0.00096420", # "ds": 0, # "cs": "0.00096420", # "user": "up105393824", # "symbol": "BTC", # "symbol2": "USD", # "amount": "0.00096420", # "buy": 59091012956, # "order": 59091012956, # "sell": 59090796005, # "price": 19135, # "type": "buy", # "time": "2022-09-24T19:36:18.466Z", # "balance": "0.00096420", # "fee_amount": "0.05", # "id": "59091012962" # } # } data = self.safe_value(message, 'data', {}) stored = self.myTrades if stored is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) stored = ArrayCacheBySymbolById(limit) self.myTrades = stored trade = self.parse_ws_trade(data) stored.append(trade) messageHash = 'myTrades:' + trade['symbol'] client.resolve(stored, messageHash) def parse_ws_trade(self, trade, market=None): # # { # "d": "order:59091012956:a:BTC", # "c": "user:up105393824:a:BTC", # "a": "0.00096420", # "ds": 0, # "cs": "0.00096420", # "user": "up105393824", # "symbol": "BTC", # "symbol2": "USD", # "amount": "0.00096420", # "buy": 59091012956, # "order": 59091012956, # "sell": 59090796005, # "price": 19135, # "type": "buy", # "time": "2022-09-24T19:36:18.466Z", # "balance": "0.00096420", # "fee_amount": "0.05", # "id": "59091012962" # } # Note symbol and symbol2 are inverse on sell and ammount is in symbol currency. # side = self.safe_string(trade, 'type') price = self.safe_string(trade, 'price') datetime = self.safe_string(trade, 'time') baseId = self.safe_string(trade, 'symbol') quoteId = self.safe_string(trade, 'symbol2') base = self.safe_currency_code(baseId) quote = self.safe_currency_code(quoteId) symbol = base + '/' + quote amount = self.safe_string(trade, 'amount') if side == 'sell': symbol = quote + '/' + base amount = Precise.string_div(amount, price) # due to rounding errors amount in not exact to trade parsedTrade: dict = { 'id': self.safe_string(trade, 'id'), 'order': self.safe_string(trade, 'order'), 'info': trade, 'timestamp': self.parse8601(datetime), 'datetime': datetime, 'symbol': symbol, 'type': None, 'side': side, 'takerOrMaker': None, 'price': price, 'cost': None, 'amount': amount, 'fee': None, } fee = self.safe_string(trade, 'fee_amount') if fee is not None: parsedTrade['fee'] = { 'cost': fee, 'currency': quote, 'rate': None, } return self.safe_trade(parsedTrade, market) def handle_order_update(self, client: Client, message): # # partialExecution # { # "e": "order", # "data": { # "id": "150714937", # "remains": "1000000", # "price": "17513", # "amount": 2000000, As Precision # "time": "1654506118448", # "type": "buy", # "pair": { # "symbol1": "BTC", # "symbol2": "USD" # }, # "fee": "0.15" # } # } # canceled order # { # "e": "order", # "data": { # "id": "6310857", # "remains": "200000000" # "fremains": "2.00000000" # "cancel": True, # "pair": { # "symbol1": "BTC", # "symbol2": "USD" # } # } # } # fulfilledOrder # { # "e": "order", # "data": { # "id": "59098421630", # "remains": "0", # "pair": { # "symbol1": "BTC", # "symbol2": "USD" # } # } # } # { # "e": "tx", # "data": { # "d": "order:59425993014:a:BTC", # "c": "user:up105393824:a:BTC", # "a": "0.00098152", # "ds": 0, # "cs": "0.00098152", # "user": "up105393824", # "symbol": "BTC", # "symbol2": "USD", # "amount": "0.00098152", # "buy": 59425993014, # "order": 59425993014, # "sell": 59425986168, # "price": 19306.6, # "type": "buy", # "time": "2022-10-02T01:11:15.148Z", # "balance": "0.00098152", # "fee_amount": "0.05", # "id": "59425993020" # } # } # data = self.safe_value(message, 'data', {}) isTransaction = self.safe_string(message, 'e') == 'tx' orderId = self.safe_string_2(data, 'id', 'order') remains = self.safe_string(data, 'remains') baseId = self.safe_string(data, 'symbol') quoteId = self.safe_string(data, 'symbol2') pair = self.safe_value(data, 'pair') if pair is not None: baseId = self.safe_string(pair, 'symbol1') quoteId = self.safe_string(pair, 'symbol2') base = self.safe_currency_code(baseId) quote = self.safe_currency_code(quoteId) symbol = base + '/' + quote market = self.safe_market(symbol) remains = self.currency_from_precision(base, remains) if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) self.orders = ArrayCacheBySymbolById(limit) storedOrders = self.orders ordersBySymbol = self.safe_value(storedOrders.hashmap, symbol, {}) order = self.safe_value(ordersBySymbol, orderId) if order is None: order = self.parse_ws_order_update(data, market) order['remaining'] = remains canceled = self.safe_bool(data, 'cancel', False) if canceled: order['status'] = 'canceled' if isTransaction: order['status'] = 'closed' fee = self.safe_number(data, 'fee') if fee is not None: order['fee'] = { 'cost': fee, 'currency': quote, 'rate': None, } timestamp = self.safe_integer(data, 'time') order['timestamp'] = timestamp order['datetime'] = self.iso8601(timestamp) order = self.safe_order(order) storedOrders.append(order) messageHash = 'orders:' + symbol client.resolve(storedOrders, messageHash) def parse_ws_order_update(self, order, market=None): # # { # "id": "150714937", # "remains": "1000000", # "price": "17513", # "amount": 2000000, As Precision # "time": "1654506118448", # "type": "buy", # "pair": { # "symbol1": "BTC", # "symbol2": "USD" # }, # "fee": "0.15" # } # transaction # { # "d": "order:59425993014:a:BTC", # "c": "user:up105393824:a:BTC", # "a": "0.00098152", # "ds": 0, # "cs": "0.00098152", # "user": "up105393824", # "symbol": "BTC", # "symbol2": "USD", # "amount": "0.00098152", # "buy": 59425993014, # "order": 59425993014, # "sell": 59425986168, # "price": 19306.6, # "type": "buy", # "time": "2022-10-02T01:11:15.148Z", # "balance": "0.00098152", # "fee_amount": "0.05", # "id": "59425993020" # } # isTransaction = self.safe_value(order, 'd') is not None remainsPrecision = self.safe_string(order, 'remains') remaining = None if remainsPrecision is not None: remaining = self.currency_from_precision(market['base'], remainsPrecision) amount = self.safe_string(order, 'amount') if not isTransaction: self.currency_from_precision(market['base'], amount) baseId = self.safe_string(order, 'symbol') quoteId = self.safe_string(order, 'symbol2') pair = self.safe_value(order, 'pair') if pair is not None: baseId = self.safe_string(order, 'symbol1') quoteId = self.safe_string(order, 'symbol2') base = self.safe_currency_code(baseId) quote = self.safe_currency_code(quoteId) symbol = None if base is not None and quote is not None: symbol = base + '/' + quote market = self.safe_market(symbol, market) time = self.safe_integer(order, 'time', self.milliseconds()) timestamp = time if isTransaction: timestamp = self.parse8601(time) canceled = self.safe_bool(order, 'cancel', False) status = 'open' if canceled: status = 'canceled' elif isTransaction: status = 'closed' parsedOrder: dict = { 'id': self.safe_string_2(order, 'id', 'order'), 'clientOrderId': None, 'info': order, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': None, 'status': status, 'symbol': symbol, 'type': None, 'timeInForce': None, 'postOnly': None, 'side': self.safe_string(order, 'type'), 'price': self.safe_number(order, 'price'), 'stopPrice': None, 'triggerPrice': None, 'average': None, 'cost': None, 'amount': amount, 'filled': None, 'remaining': remaining, 'fee': { 'cost': self.safe_number_2(order, 'fee', 'fee_amount'), 'currency': quote, 'rate': None, }, 'trades': None, } if isTransaction: parsedOrder['trades'] = self.parse_ws_trade(order, market) return self.safe_order(parsedOrder, market) def from_precision(self, amount, scale): if amount is None: return None precise = Precise(amount) precise.decimals = self.sum(precise.decimals, scale) precise.reduce() return str(precise) def currency_from_precision(self, currency, amount): scale = self.safe_integer(self.currencies[currency], 'precision', 0) return self.from_precision(amount, scale) def handle_orders_snapshot(self, client: Client, message): # # { # "e": "open-orders", # "data": [{ # "id": "59098421630", # "time": "1664062285425", # "type": "buy", # "price": "18920", # "amount": "0.00100000", # "pending": "0.00100000" # }], # "oid": 1, # "ok": "ok" # } # symbol = self.safe_string(message, 'oid') # symbol is set in watchOrders rawOrders = self.safe_value(message, 'data', []) myOrders = self.orders if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) myOrders = ArrayCacheBySymbolById(limit) for i in range(0, len(rawOrders)): rawOrder = rawOrders[i] market = self.safe_market(symbol) order = self.parse_order(rawOrder, market) order['status'] = 'open' myOrders.append(order) self.orders = myOrders messageHash = 'orders:' + symbol ordersLength = len(myOrders) if ordersLength > 0: client.resolve(myOrders, messageHash) 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://cex.io/websocket-api#orderbook-subscribe :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 """ await self.load_markets() await self.authenticate() market = self.market(symbol) symbol = market['symbol'] url = self.urls['api']['ws'] messageHash = 'orderbook:' + symbol depth = 0 if (limit is None) else limit subscribe: dict = { 'e': 'order-book-subscribe', 'data': { 'pair': [ market['baseId'], market['quoteId'], ], 'subscribe': True, 'depth': depth, }, 'oid': self.request_id(), } request = self.deep_extend(subscribe, params) orderbook = await self.watch(url, messageHash, request, messageHash) return orderbook.limit() def handle_order_book_snapshot(self, client: Client, message): # # { # "e": "order-book-subscribe", # "data": { # "timestamp": 1663762032, # "timestamp_ms": 1663762031680, # "bids": [ # [241.947, 155.91626], # [241, 154], # ], # "asks": [ # [242.947, 155.91626], # [243, 154], ], # "pair": "BTC:USDT", # "id": 616267120, # "sell_total": "13.59066946", # "buy_total": "163553.625948" # }, # "oid": "1", # "ok": "ok" # } # data = self.safe_value(message, 'data', {}) pair = self.safe_string(data, 'pair') symbol = self.pair_to_symbol(pair) messageHash = 'orderbook:' + symbol timestamp = self.safe_integer_2(data, 'timestamp_ms', 'timestamp') incrementalId = self.safe_integer(data, 'id') orderbook = self.order_book({}) snapshot = self.parse_order_book(data, symbol, timestamp, 'bids', 'asks') snapshot['nonce'] = incrementalId orderbook.reset(snapshot) self.options['orderbook'][symbol] = { 'incrementalId': incrementalId, } self.orderbooks[symbol] = orderbook client.resolve(orderbook, messageHash) def pair_to_symbol(self, pair): parts = pair.split(':') baseId = self.safe_string(parts, 0) quoteId = self.safe_string(parts, 1) base = self.safe_currency_code(baseId) quote = self.safe_currency_code(quoteId) symbol = base + '/' + quote return symbol def handle_order_book_update(self, client: Client, message): # # { # "e": "md_update", # "data": { # "id": 616267121, # "pair": "BTC:USDT", # "time": 1663762031719, # "bids": [], # "asks": [ # [122, 23] # ] # } # } # data = self.safe_value(message, 'data', {}) incrementalId = self.safe_integer(data, 'id') pair = self.safe_string(data, 'pair', '') symbol = self.pair_to_symbol(pair) storedOrderBook = self.safe_value(self.orderbooks, symbol) messageHash = 'orderbook:' + symbol if incrementalId != storedOrderBook['nonce'] + 1: del client.subscriptions[messageHash] client.reject(self.id + ' watchOrderBook() skipped a message', messageHash) timestamp = self.safe_integer(data, 'time') asks = self.safe_value(data, 'asks', []) bids = self.safe_value(data, 'bids', []) self.handle_deltas(storedOrderBook['asks'], asks) self.handle_deltas(storedOrderBook['bids'], bids) storedOrderBook['timestamp'] = timestamp storedOrderBook['datetime'] = self.iso8601(timestamp) storedOrderBook['nonce'] = incrementalId client.resolve(storedOrderBook, messageHash) def handle_delta(self, bookside, delta): bidAsk = self.parse_bid_ask(delta, 0, 1) 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_ohlcv(self, symbol: str, timeframe='1m', since: Int = None, limit: Int = None, params={}) -> List[list]: """ https://cex.io/websocket-api#minute-data watches historical candlestick data containing the open, high, low, and close price, and the volume of a market. It will return the last 120 minutes with the selected timeframe and then 1m candle updates after that. :param str symbol: unified symbol of the market to fetch OHLCV data for :param str timeframe: the length of time each candle represents. :param int [since]: timestamp in ms of the earliest candle to fetch :param int [limit]: the maximum amount of candles to fetch :param dict [params]: extra parameters specific to the exchange API endpoint :returns int[][]: A list of candles ordered, open, high, low, close, volume """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] messageHash = 'ohlcv:' + symbol url = self.urls['api']['ws'] request: dict = { 'e': 'init-ohlcv', 'i': timeframe, 'rooms': [ 'pair-' + market['baseId'] + '-' + market['quoteId'], ], } ohlcv = await self.watch(url, messageHash, self.extend(request, params), messageHash) if self.newUpdates: limit = ohlcv.getLimit(symbol, limit) return self.filter_by_since_limit(ohlcv, since, limit, 0, True) def handle_init_ohlcv(self, client: Client, message): # # { # "e": "init-ohlcv-data", # "data": [ # [ # 1663660680, # "19396.4", # "19396.4", # "19396.4", # "19396.4", # "1262861" # ], # ... # ], # "pair": "BTC:USDT" # } # pair = self.safe_string(message, 'pair') parts = pair.split(':') baseId = self.safe_string(parts, 0) quoteId = self.safe_string(parts, 1) base = self.safe_currency_code(baseId) quote = self.safe_currency_code(quoteId) symbol = base + '/' + quote market = self.safe_market(symbol) messageHash = 'ohlcv:' + symbol data = self.safe_value(message, 'data', []) limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) stored = ArrayCacheByTimestamp(limit) sorted = self.sort_by(data, 0) for i in range(0, len(sorted)): stored.append(self.parse_ohlcv(sorted[i], market)) if not (symbol in self.ohlcvs): self.ohlcvs[symbol] = {} self.ohlcvs[symbol]['unknown'] = stored client.resolve(stored, messageHash) def handle_ohlcv24(self, client: Client, message): # # { # "e": "ohlcv24", # "data": ['18793.2', '19630', '18793.2', "19104.1", "314157273"], # "pair": "BTC:USDT" # } # return message def handle_ohlcv1m(self, client: Client, message): # # { # "e": "ohlcv1m", # "data": { # "pair": "BTC:USD", # "time": "1665436800", # "o": "19279.6", # "h": "19279.6", # "l": "19266.7", # "c": "19266.7", # "v": 3343884, # "d": 3343884 # } # } # data = self.safe_value(message, 'data', {}) pair = self.safe_string(data, 'pair') symbol = self.pair_to_symbol(pair) messageHash = 'ohlcv:' + symbol ohlcv = [ self.safe_timestamp(data, 'time'), self.safe_number(data, 'o'), self.safe_number(data, 'h'), self.safe_number(data, 'l'), self.safe_number(data, 'c'), self.safe_number(data, 'v'), ] stored = self.safe_value(self.ohlcvs, symbol) stored.append(ohlcv) client.resolve(stored, messageHash) def handle_ohlcv(self, client: Client, message): # # { # "e": "ohlcv", # "data": [ # [1665461100, '19068.2', '19068.2', '19068.2', "19068.2", 268478] # ], # "pair": "BTC:USD" # } # data = self.safe_value(message, 'data', []) pair = self.safe_string(message, 'pair') symbol = self.pair_to_symbol(pair) messageHash = 'ohlcv:' + symbol # stored = self.safe_value(self.ohlcvs, symbol) stored = self.ohlcvs[symbol]['unknown'] for i in range(0, len(data)): ohlcv = [ self.safe_timestamp(data[i], 0), self.safe_number(data[i], 1), self.safe_number(data[i], 2), self.safe_number(data[i], 3), self.safe_number(data[i], 4), self.safe_number(data[i], 5), ] stored.append(ohlcv) dataLength = len(data) if dataLength > 0: client.resolve(stored, messageHash) async def fetch_order_ws(self, id: str, symbol: Str = None, params={}): """ fetches information on an order made by the user https://docs.cex.io/#ws-api-get-order :param str id: the order id :param str symbol: not used by cex fetchOrder :param dict [params]: extra parameters specific to the cex api endpoint :returns dict: An `order structure ` """ await self.load_markets() await self.authenticate() market = None if symbol is not None: market = self.market(symbol) data = self.extend({ 'order_id': str(id), }, params) url = self.urls['api']['ws'] messageHash = self.request_id() request: dict = { 'e': 'get-order', 'oid': messageHash, 'data': data, } response = await self.watch(url, messageHash, request, messageHash) return self.parse_order(response, market) async def fetch_open_orders_ws(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}): """ https://docs.cex.io/#ws-api-open-orders fetch all unfilled currently open orders :param str symbol: unified market symbol :param int [since]: the earliest time in ms to fetch open orders for :param int [limit]: the maximum number of open orders structures to retrieve :param dict [params]: extra parameters specific to the cex api endpoint :returns Order[]: a list of `order structures ` """ if symbol is None: raise ArgumentsRequired(self.id + ' fetchOpenOrdersWs requires a symbol.') await self.load_markets() await self.authenticate() market = self.market(symbol) url = self.urls['api']['ws'] messageHash = self.request_id() data = self.extend({ 'pair': [market['baseId'], market['quoteId']], }, params) request: dict = { 'e': 'open-orders', 'oid': messageHash, 'data': data, } response = await self.watch(url, messageHash, request, messageHash) return self.parse_orders(response, market, since, limit, params) async def create_order_ws(self, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}) -> Order: """ https://docs.cex.io/#ws-api-order-placement create a trade order :param str symbol: unified symbol of the market to create an order in :param str type: 'market' or 'limit' :param str side: 'buy' or 'sell' :param float amount: how much of currency you want to trade in units of base currency :param float price: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders :param dict [params]: extra parameters specific to the kraken api endpoint :param boolean [params.maker_only]: Optional, maker only places an order only if offers best sell(<= max) or buy(>= max) price for self pair, if not order placement will be rejected with an error - "Order is not maker" :returns dict: an `order structure ` """ if price is None: raise BadRequest(self.id + ' createOrderWs requires a price argument') await self.load_markets() await self.authenticate() market = self.market(symbol) url = self.urls['api']['ws'] messageHash = self.request_id() data = self.extend({ 'pair': [market['baseId'], market['quoteId']], 'amount': amount, 'price': price, 'type': side, }, params) request: dict = { 'e': 'place-order', 'oid': messageHash, 'data': data, } rawOrder = await self.watch(url, messageHash, request, messageHash) return self.parse_order(rawOrder, market) async def edit_order_ws(self, id: str, symbol: str, type: OrderType, side: OrderSide, amount: Num = None, price: Num = None, params={}) -> Order: """ edit a trade order https://docs.cex.io/#ws-api-cancel-replace :param str id: order id :param str symbol: unified symbol of the market to create an order in :param str type: 'market' or 'limit' :param str side: 'buy' or 'sell' :param float amount: how much of the currency you want to trade in units of the base currency :param float|None [price]: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders :param dict [params]: extra parameters specific to the cex api endpoint :returns dict: an `order structure ` """ if amount is None: raise ArgumentsRequired(self.id + ' editOrder() requires a amount argument') if price is None: raise ArgumentsRequired(self.id + ' editOrder() requires a price argument') await self.load_markets() await self.authenticate() market = self.market(symbol) data = self.extend({ 'pair': [market['baseId'], market['quoteId']], 'type': side, 'amount': amount, 'price': price, 'order_id': id, }, params) messageHash = self.request_id() url = self.urls['api']['ws'] request: dict = { 'e': 'cancel-replace-order', 'oid': messageHash, 'data': data, } response = await self.watch(url, messageHash, request, messageHash, messageHash) return self.parse_order(response, market) async def cancel_order_ws(self, id: str, symbol: Str = None, params={}): """ https://docs.cex.io/#ws-api-order-cancel cancels an open order :param str id: order id :param str symbol: not used by cex cancelOrder() :param dict [params]: extra parameters specific to the cex api endpoint :returns dict: An `order structure ` """ await self.load_markets() await self.authenticate() market = None if symbol is not None: market = self.market(symbol) data = self.extend({ 'order_id': id, }, params) messageHash = self.request_id() url = self.urls['api']['ws'] request: dict = { 'e': 'cancel-order', 'oid': messageHash, 'data': data, } response = await self.watch(url, messageHash, request, messageHash, messageHash) return self.parse_order(response, market) async def cancel_orders_ws(self, ids: List[str], symbol: Str = None, params={}): """ cancel multiple orders https://docs.cex.io/#ws-api-mass-cancel-place :param str[] ids: order ids :param str symbol: not used by cex cancelOrders() :param dict [params]: extra parameters specific to the cex api endpoint :returns dict: a list of `order structures ` """ if symbol is not None: raise BadRequest(self.id + ' cancelOrderWs does not allow filtering by symbol') await self.load_markets() await self.authenticate() messageHash = self.request_id() data = self.extend({ 'cancel-orders': ids, }, params) url = self.urls['api']['ws'] request: dict = { 'e': 'mass-cancel-place-orders', 'oid': messageHash, 'data': data, } response = await self.watch(url, messageHash, request, messageHash, messageHash) # # { # "cancel-orders": [{ # "order_id": 69202557979, # "fremains": "0.15000000" # }], # "place-orders": [], # "placed-cancelled": [] # } # canceledOrders = self.safe_value(response, 'cancel-orders') return self.parse_orders(canceledOrders, None, None, None, params) def resolve_data(self, client: Client, message): # # "e": "open-orders", # "data": [ # { # "id": "2477098", # "time": "1435927928618", # "type": "buy", # "price": "241.9477", # "amount": "0.02000000", # "pending": "0.02000000" # }, # ... # ], # "oid": "1435927928274_9_open-orders", # "ok": "ok" # } # data = self.safe_value(message, 'data') messageHash = self.safe_string(message, 'oid') client.resolve(data, messageHash) def handle_connected(self, client: Client, message): # # { # "e": "connected" # } # return message def handle_error_message(self, client: Client, message): # # { # "e": "get-balance", # "data": {error: "Please Login"}, # "oid": 1, # "ok": "error" # } # try: data = self.safe_value(message, 'data', {}) error = self.safe_string(data, 'error') event = self.safe_string(message, 'e', '') feedback = self.id + ' ' + event + ' ' + error self.throw_exactly_matched_exception(self.exceptions['exact'], error, feedback) self.throw_broadly_matched_exception(self.exceptions['broad'], error, feedback) raise ExchangeError(feedback) except Exception as error: messageHash = self.safe_string(message, 'oid') future = self.safe_value(client['futures'], messageHash) if future is not None: client.reject(error, messageHash) else: raise error def handle_message(self, client: Client, message): ok = self.safe_string(message, 'ok') if ok == 'error': self.handle_error_message(client, message) return event = self.safe_string(message, 'e') handlers: dict = { 'auth': self.handle_authentication_message, 'connected': self.handle_connected, 'tick': self.handle_ticker, 'ticker': self.handle_ticker, 'init-ohlcv-data': self.handle_init_ohlcv, 'ohlcv24': self.handle_ohlcv24, 'ohlcv1m': self.handle_ohlcv1m, 'ohlcv': self.handle_ohlcv, 'get-balance': self.handle_balance, 'order-book-subscribe': self.handle_order_book_snapshot, 'md_update': self.handle_order_book_update, 'open-orders': self.resolve_data, 'order': self.handle_order_update, 'history-update': self.handle_trade, 'history': self.handle_trades_snapshot, 'tx': self.handle_transaction, 'place-order': self.resolve_data, 'cancel-replace-order': self.resolve_data, 'cancel-order': self.resolve_data, 'mass-cancel-place-orders': self.resolve_data, 'get-order': self.resolve_data, } handler = self.safe_value(handlers, event) if handler is not None: handler(client, message) def handle_authentication_message(self, client: Client, message): # # { # "e": "auth", # "data": { # "ok": "ok" # }, # "ok": "ok", # "timestamp":1448034593 # } # future = self.safe_value(client.futures, 'authenticated') if future is not None: future.resolve(True) async def authenticate(self, params={}): url = self.urls['api']['ws'] client = self.client(url) messageHash = 'authenticated' future = client.future('authenticated') authenticated = self.safe_value(client.subscriptions, messageHash) if authenticated is None: self.check_required_credentials() nonce = str(self.seconds()) auth = nonce + self.apiKey signature = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256) request: dict = { 'e': 'auth', 'auth': { 'key': self.apiKey, 'signature': signature.upper(), 'timestamp': nonce, }, } await self.watch(url, messageHash, self.extend(request, params), messageHash) return await future