# -*- 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 from ccxt.base.types import 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 AuthenticationError from ccxt.base.errors import PermissionDenied from ccxt.base.errors import AccountSuspended from ccxt.base.errors import BadRequest from ccxt.base.errors import BadSymbol from ccxt.base.errors import InsufficientFunds from ccxt.base.errors import InvalidOrder from ccxt.base.errors import OrderNotFound from ccxt.base.errors import NotSupported from ccxt.base.errors import RateLimitExceeded from ccxt.base.errors import ExchangeNotAvailable from ccxt.base.errors import InvalidNonce from ccxt.base.precise import Precise class kraken(ccxt.async_support.kraken): def describe(self): return self.deep_extend(super(kraken, self).describe(), { 'has': { 'ws': True, 'watchBalance': False, # no such type of subscription 2021-01-05 'watchMyTrades': True, 'watchOHLCV': True, 'watchOrderBook': True, 'watchOrderBookForSymbols': True, 'watchOrders': True, 'watchTicker': True, 'watchTickers': True, 'watchTrades': True, 'watchTradesForSymbols': True, 'createOrderWs': True, 'editOrderWs': True, 'cancelOrderWs': True, 'cancelOrdersWs': True, 'cancelAllOrdersWs': True, # 'watchHeartbeat': True, # 'watchStatus': True, }, 'urls': { 'api': { 'ws': { 'public': 'wss://ws.kraken.com', 'private': 'wss://ws-auth.kraken.com', 'beta': 'wss://beta-ws.kraken.com', 'beta-private': 'wss://beta-ws-auth.kraken.com', }, }, }, # 'versions': { # 'ws': '0.2.0', # }, 'options': { 'tradesLimit': 1000, 'OHLCVLimit': 1000, 'ordersLimit': 1000, 'symbolsByOrderId': {}, 'checksum': True, }, 'exceptions': { 'ws': { 'exact': { 'Event(s) not found': BadRequest, }, 'broad': { 'Already subscribed': BadRequest, 'Currency pair not in ISO 4217-A3 format': BadSymbol, 'Malformed request': BadRequest, 'Pair field must be an array': BadRequest, 'Pair field unsupported for self subscription type': BadRequest, 'Pair(s) not found': BadSymbol, 'Subscription book depth must be an integer': BadRequest, 'Subscription depth not supported': BadRequest, 'Subscription field must be an object': BadRequest, 'Subscription name invalid': BadRequest, 'Subscription object unsupported field': BadRequest, 'Subscription ohlc interval must be an integer': BadRequest, 'Subscription ohlc interval not supported': BadRequest, 'Subscription ohlc requires interval': BadRequest, 'EAccount:Invalid permissions': PermissionDenied, 'EAuth:Account temporary disabled': AccountSuspended, 'EAuth:Account unconfirmed': AuthenticationError, 'EAuth:Rate limit exceeded': RateLimitExceeded, 'EAuth:Too many requests': RateLimitExceeded, 'EDatabase: Internal error(to be deprecated)': ExchangeError, 'EGeneral:Internal error[:]': ExchangeError, 'EGeneral:Invalid arguments': BadRequest, 'EOrder:Cannot open opposing position': InvalidOrder, 'EOrder:Cannot open position': InvalidOrder, 'EOrder:Insufficient funds(insufficient user funds)': InsufficientFunds, 'EOrder:Insufficient margin(exchange does not have sufficient funds to allow margin trading)': InsufficientFunds, 'EOrder:Invalid price': InvalidOrder, 'EOrder:Margin allowance exceeded': InvalidOrder, 'EOrder:Margin level too low': InvalidOrder, 'EOrder:Margin position size exceeded(client would exceed the maximum position size for self pair)': InvalidOrder, 'EOrder:Order minimum not met(volume too low)': InvalidOrder, 'EOrder:Orders limit exceeded': InvalidOrder, 'EOrder:Positions limit exceeded': InvalidOrder, 'EOrder:Rate limit exceeded': RateLimitExceeded, 'EOrder:Scheduled orders limit exceeded': InvalidOrder, 'EOrder:Unknown position': OrderNotFound, 'EOrder:Unknown order': OrderNotFound, 'EOrder:Invalid order': InvalidOrder, 'EService:Deadline elapsed': ExchangeNotAvailable, 'EService:Market in cancel_only mode': NotSupported, 'EService:Market in limit_only mode': NotSupported, 'EService:Market in post_only mode': NotSupported, 'EService:Unavailable': ExchangeNotAvailable, 'ETrade:Invalid request': BadRequest, }, }, }, }) async def create_order_ws(self, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}) -> Order: """ :see: https://docs.kraken.com/websockets/#message-addOrder 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 fullfilled, in units of the quote currency, ignored in market orders :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: an `order structure ` """ await self.load_markets() token = await self.authenticate() market = self.market(symbol) url = self.urls['api']['ws']['private'] requestId = self.request_id() messageHash = requestId request = { 'event': 'addOrder', 'token': token, 'reqid': requestId, 'ordertype': type, 'type': side, 'pair': market['wsId'], 'volume': self.amount_to_precision(symbol, amount), } request, params = self.orderRequest('createOrderWs', symbol, type, request, price, params) return await self.watch(url, messageHash, self.extend(request, params), messageHash) def handle_create_edit_order(self, client, message): # # createOrder # { # "descr": "sell 0.00010000 XBTUSDT @ market", # "event": "addOrderStatus", # "reqid": 1, # "status": "ok", # "txid": "OAVXZH-XIE54-JCYYDG" # } # editOrder # { # "descr": "order edited price = 9000.00000000", # "event": "editOrderStatus", # "originaltxid": "O65KZW-J4AW3-VFS74A", # "reqid": 3, # "status": "ok", # "txid": "OTI672-HJFAO-XOIPPK" # } # order = self.parse_order(message) messageHash = self.safe_value(message, 'reqid') client.resolve(order, messageHash) async def edit_order_ws(self, id: str, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}) -> Order: """ edit a trade order :see: https://docs.kraken.com/websockets/#message-editOrder :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 [price]: the price at which the order is to be fullfilled, in units of the quote currency, ignored in market orders :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: an `order structure ` """ await self.load_markets() token = await self.authenticate() market = self.market(symbol) url = self.urls['api']['ws']['private'] requestId = self.request_id() messageHash = requestId request = { 'event': 'editOrder', 'token': token, 'reqid': requestId, 'orderid': id, 'pair': market['wsId'], 'volume': self.amount_to_precision(symbol, amount), } request, params = self.orderRequest('editOrderWs', symbol, type, request, price, params) return await self.watch(url, messageHash, self.extend(request, params), messageHash) async def cancel_orders_ws(self, ids: List[str], symbol: Str = None, params={}): """ :see: https://docs.kraken.com/websockets/#message-cancelOrder cancel multiple orders :param str[] ids: order ids :param str symbol: unified market symbol, default is None :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: an list of `order structures ` """ await self.load_markets() token = await self.authenticate() url = self.urls['api']['ws']['private'] requestId = self.request_id() messageHash = requestId request = { 'event': 'cancelOrder', 'token': token, 'reqid': requestId, 'txid': ids, } return await self.watch(url, messageHash, self.extend(request, params), messageHash) async def cancel_order_ws(self, id: str, symbol: Str = None, params={}) -> Order: """ :see: https://docs.kraken.com/websockets/#message-cancelOrder cancels an open order :param str id: order id :param str symbol: unified symbol of the market the order was made in :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: An `order structure ` """ await self.load_markets() token = await self.authenticate() url = self.urls['api']['ws']['private'] requestId = self.request_id() messageHash = requestId clientOrderId = self.safe_value_2(params, 'userref', 'clientOrderId', id) params = self.omit(params, ['userref', 'clientOrderId']) request = { 'event': 'cancelOrder', 'token': token, 'reqid': requestId, 'txid': [clientOrderId], } return await self.watch(url, messageHash, self.extend(request, params), messageHash) def handle_cancel_order(self, client, message): # # success # { # "event": "cancelOrderStatus", # "status": "ok" # "reqid": 1, # } # reqId = self.safe_value(message, 'reqid') client.resolve(message, reqId) async def cancel_all_orders_ws(self, symbol: Str = None, params={}): """ :see: https://docs.kraken.com/websockets/#message-cancelAll cancel all open orders :param str symbol: unified market symbol, only orders in the market of self symbol are cancelled when symbol is not None :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `order structures ` """ if symbol is not None: raise NotSupported(self.id + ' cancelAllOrdersWs() does not support cancelling orders in a specific market.') await self.load_markets() token = await self.authenticate() url = self.urls['api']['ws']['private'] requestId = self.request_id() messageHash = requestId request = { 'event': 'cancelAll', 'token': token, 'reqid': requestId, } return await self.watch(url, messageHash, self.extend(request, params), messageHash) def handle_cancel_all_orders(self, client, message): # # { # "count": 2, # "event": "cancelAllStatus", # "status": "ok", # "reqId": 1 # } # reqId = self.safe_value(message, 'reqid') client.resolve(message, reqId) def handle_ticker(self, client, message, subscription): # # [ # 0, # channelID # { # "a": ["5525.40000", 1, "1.000"], # ask, wholeAskVolume, askVolume # "b": ["5525.10000", 1, "1.000"], # bid, wholeBidVolume, bidVolume # "c": ["5525.10000", "0.00398963"], # closing price, volume # "h": ["5783.00000", "5783.00000"], # high price today, high price 24h ago # "l": ["5505.00000", "5505.00000"], # low price today, low price 24h ago # "o": ["5760.70000", "5763.40000"], # open price today, open price 24h ago # "p": ["5631.44067", "5653.78939"], # vwap today, vwap 24h ago # "t": [11493, 16267], # number of trades today, 24 hours ago # "v": ["2634.11501494", "3591.17907851"], # volume today, volume 24 hours ago # }, # "ticker", # "XBT/USD" # ] # wsName = message[3] market = self.safe_value(self.options['marketsByWsName'], wsName) symbol = market['symbol'] messageHash = self.get_message_hash('ticker', None, symbol) ticker = message[1] vwap = self.safe_string(ticker['p'], 0) quoteVolume = None baseVolume = self.safe_string(ticker['v'], 0) if baseVolume is not None and vwap is not None: quoteVolume = Precise.string_mul(baseVolume, vwap) last = self.safe_string(ticker['c'], 0) result = self.safe_ticker({ 'symbol': symbol, 'timestamp': None, 'datetime': None, 'high': self.safe_string(ticker['h'], 0), 'low': self.safe_string(ticker['l'], 0), 'bid': self.safe_string(ticker['b'], 0), 'bidVolume': self.safe_string(ticker['b'], 2), 'ask': self.safe_string(ticker['a'], 0), 'askVolume': self.safe_string(ticker['a'], 2), 'vwap': vwap, 'open': self.safe_string(ticker['o'], 0), 'close': last, 'last': last, 'previousClose': None, 'change': None, 'percentage': None, 'average': None, 'baseVolume': baseVolume, 'quoteVolume': quoteVolume, 'info': ticker, }) self.tickers[symbol] = result client.resolve(result, messageHash) def handle_trades(self, client: Client, message, subscription): # # [ # 0, # channelID # [ # price volume time side type misc # ["5541.20000", "0.15850568", "1534614057.321596", "s", "l", ""], # ["6060.00000", "0.02455000", "1534614057.324998", "b", "l", ""], # ], # "trade", # "XBT/USD" # ] # wsName = self.safe_string(message, 3) name = self.safe_string(message, 2) market = self.safe_value(self.options['marketsByWsName'], wsName) symbol = market['symbol'] messageHash = self.get_message_hash(name, None, symbol) stored = self.safe_value(self.trades, symbol) if stored is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) stored = ArrayCache(limit) self.trades[symbol] = stored trades = self.safe_value(message, 1, []) parsed = self.parse_trades(trades, market) for i in range(0, len(parsed)): stored.append(parsed[i]) client.resolve(stored, messageHash) def handle_ohlcv(self, client: Client, message, subscription): # # [ # 216, # channelID # [ # "1574454214.962096", # Time, seconds since epoch # "1574454240.000000", # End timestamp of the interval # "0.020970", # Open price at midnight UTC # "0.020970", # Intraday high price # "0.020970", # Intraday low price # "0.020970", # Closing price at midnight UTC # "0.020970", # Volume weighted average price # "0.08636138", # Accumulated volume today # 1, # Number of trades today # ], # "ohlc-1", # Channel Name of subscription # "ETH/XBT", # Asset pair # ] # info = self.safe_value(subscription, 'subscription', {}) interval = self.safe_integer(info, 'interval') name = self.safe_string(info, 'name') wsName = self.safe_string(message, 3) market = self.safe_value(self.options['marketsByWsName'], wsName) symbol = market['symbol'] timeframe = self.find_timeframe(interval) duration = self.parse_timeframe(timeframe) if timeframe is not None: candle = self.safe_value(message, 1) messageHash = name + ':' + timeframe + ':' + wsName timestamp = self.safe_float(candle, 1) timestamp -= duration ts = self.parse_to_int(timestamp * 1000) result = [ ts, self.safe_float(candle, 2), self.safe_float(candle, 3), self.safe_float(candle, 4), self.safe_float(candle, 5), self.safe_float(candle, 7), ] self.ohlcvs[symbol] = self.safe_value(self.ohlcvs, symbol, {}) stored = self.safe_value(self.ohlcvs[symbol], timeframe) if stored is None: limit = self.safe_integer(self.options, 'OHLCVLimit', 1000) stored = ArrayCacheByTimestamp(limit) self.ohlcvs[symbol][timeframe] = stored stored.append(result) client.resolve(stored, messageHash) def request_id(self): # their support said that reqid must be an int32, not documented reqid = self.sum(self.safe_integer(self.options, 'reqid', 0), 1) self.options['reqid'] = reqid return reqid async def watch_public(self, name, symbol, params={}): await self.load_markets() market = self.market(symbol) wsName = self.safe_value(market['info'], 'wsname') messageHash = name + ':' + wsName url = self.urls['api']['ws']['public'] requestId = self.request_id() subscribe = { 'event': 'subscribe', 'reqid': requestId, 'pair': [ wsName, ], 'subscription': { 'name': name, }, } request = self.deep_extend(subscribe, params) return await self.watch(url, messageHash, request, messageHash) 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 :param str symbol: unified symbol of the market to fetch the ticker for :param dict [params]: extra parameters specific to the exchange API endpoint :returns dict: a `ticker structure ` """ await self.load_markets() symbol = self.symbol(symbol) tickers = await self.watch_tickers([symbol], params) return tickers[symbol] 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 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 :returns dict: a `ticker structure ` """ await self.load_markets() symbols = self.market_symbols(symbols, None, False) ticker = await self.watch_multi_helper('ticker', 'ticker', symbols, None, params) if self.newUpdates: result = {} result[ticker['symbol']] = ticker return result return self.filter_by_array(self.tickers, 'symbol', symbols) 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 :see: https://docs.kraken.com/websockets/#message-trade :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]: """ :see: https://docs.kraken.com/websockets/#message-trade get the list of most recent trades for a list of symbols :param str[] 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 dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of `trade structures ` """ trades = await self.watch_multi_helper('trade', 'trade', symbols, None, params) if self.newUpdates: first = self.safe_list(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 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 :see: https://docs.kraken.com/websockets/#message-book :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 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 :see: https://docs.kraken.com/websockets/#message-book :param str[] symbols: unified array of 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 """ request = {} if limit is not None: if self.in_array(limit, [10, 25, 100, 500, 1000]): request['subscription'] = { 'depth': limit, # default 10, valid options 10, 25, 100, 500, 1000 } else: raise NotSupported(self.id + ' watchOrderBook accepts limit values of 10, 25, 100, 500 and 1000 only') orderbook = await self.watch_multi_helper('orderbook', 'book', symbols, {'limit': limit}, self.extend(request, params)) return orderbook.limit() 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 :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() name = 'ohlc' market = self.market(symbol) symbol = market['symbol'] wsName = self.safe_value(market['info'], 'wsname') messageHash = name + ':' + timeframe + ':' + wsName url = self.urls['api']['ws']['public'] requestId = self.request_id() subscribe = { 'event': 'subscribe', 'reqid': requestId, 'pair': [ wsName, ], 'subscription': { 'name': name, 'interval': self.safe_value(self.timeframes, timeframe, timeframe), }, } request = self.deep_extend(subscribe, params) ohlcv = await self.watch(url, messageHash, request, messageHash) if self.newUpdates: limit = ohlcv.getLimit(symbol, limit) return self.filter_by_since_limit(ohlcv, since, limit, 0, True) async def load_markets(self, reload=False, params={}): markets = await super(kraken, self).load_markets(reload, params) marketsByWsName = self.safe_value(self.options, 'marketsByWsName') if (marketsByWsName is None) or reload: marketsByWsName = {} for i in range(0, len(self.symbols)): symbol = self.symbols[i] market = self.markets[symbol] if market['darkpool']: info = self.safe_value(market, 'info', {}) altname = self.safe_string(info, 'altname') wsName = altname[0:3] + '/' + altname[3:] marketsByWsName[wsName] = market else: info = self.safe_value(market, 'info', {}) wsName = self.safe_string(info, 'wsname') marketsByWsName[wsName] = market self.options['marketsByWsName'] = marketsByWsName return markets async def watch_heartbeat(self, params={}): await self.load_markets() event = 'heartbeat' url = self.urls['api']['ws']['public'] return await self.watch(url, event) def handle_heartbeat(self, client: Client, message): # # every second(approx) if no other updates are sent # # {"event": "heartbeat"} # event = self.safe_string(message, 'event') client.resolve(message, event) def handle_order_book(self, client: Client, message, subscription): # # first message(snapshot) # # [ # 1234, # channelID # { # "as": [ # ["5541.30000", "2.50700000", "1534614248.123678"], # ["5541.80000", "0.33000000", "1534614098.345543"], # ["5542.70000", "0.64700000", "1534614244.654432"] # ], # "bs": [ # ["5541.20000", "1.52900000", "1534614248.765567"], # ["5539.90000", "0.30000000", "1534614241.769870"], # ["5539.50000", "5.00000000", "1534613831.243486"] # ] # }, # "book-10", # "XBT/USD" # ] # # subsequent updates # # [ # 1234, # { # optional # "a": [ # ["5541.30000", "2.50700000", "1534614248.456738"], # ["5542.50000", "0.40100000", "1534614248.456738"] # ] # }, # { # optional # "b": [ # ["5541.30000", "0.00000000", "1534614335.345903"] # ] # }, # "book-10", # "XBT/USD" # ] # messageLength = len(message) wsName = message[messageLength - 1] bookDepthString = message[messageLength - 2] parts = bookDepthString.split('-') depth = self.safe_integer(parts, 1, 10) market = self.safe_value(self.options['marketsByWsName'], wsName) symbol = market['symbol'] timestamp = None messageHash = self.get_message_hash('orderbook', None, symbol) # if self is a snapshot if 'as' in message[1]: # todo get depth from marketsByWsName self.orderbooks[symbol] = self.order_book({}, depth) orderbook = self.orderbooks[symbol] sides = { 'as': 'asks', 'bs': 'bids', } keys = list(sides.keys()) for i in range(0, len(keys)): key = keys[i] side = sides[key] bookside = orderbook[side] deltas = self.safe_value(message[1], key, []) timestamp = self.custom_handle_deltas(bookside, deltas, timestamp) orderbook['symbol'] = symbol orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) client.resolve(orderbook, messageHash) else: orderbook = self.orderbooks[symbol] # else, if self is an orderbook update a = None b = None c = None if messageLength == 5: a = self.safe_value(message[1], 'a', []) b = self.safe_value(message[2], 'b', []) c = self.safe_integer(message[1], 'c') c = self.safe_integer(message[2], 'c', c) else: c = self.safe_integer(message[1], 'c') if 'a' in message[1]: a = self.safe_value(message[1], 'a', []) else: b = self.safe_value(message[1], 'b', []) storedAsks = orderbook['asks'] storedBids = orderbook['bids'] example = None if a is not None: timestamp = self.custom_handle_deltas(storedAsks, a, timestamp) example = self.safe_value(a, 0) if b is not None: timestamp = self.custom_handle_deltas(storedBids, b, timestamp) example = self.safe_value(b, 0) # don't remove self line or I will poop on your face orderbook.limit() checksum = self.safe_bool(self.options, 'checksum', True) if checksum: priceString = self.safe_string(example, 0) amountString = self.safe_string(example, 1) priceParts = priceString.split('.') amountParts = amountString.split('.') priceLength = len(priceParts[1]) - 0 amountLength = len(amountParts[1]) - 0 payloadArray = [] if c is not None: for i in range(0, 10): formatted = self.format_number(storedAsks[i][0], priceLength) + self.format_number(storedAsks[i][1], amountLength) payloadArray.append(formatted) for i in range(0, 10): formatted = self.format_number(storedBids[i][0], priceLength) + self.format_number(storedBids[i][1], amountLength) payloadArray.append(formatted) payload = ''.join(payloadArray) localChecksum = self.crc32(payload, False) if localChecksum != c: error = InvalidNonce(self.id + ' invalid checksum') client.reject(error, messageHash) return orderbook['symbol'] = symbol orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) client.resolve(orderbook, messageHash) def format_number(self, n, length): stringNumber = self.number_to_string(n) parts = stringNumber.split('.') integer = self.safe_string(parts, 0) decimals = self.safe_string(parts, 1, '') paddedDecimals = decimals.ljust(length, '0') joined = integer + paddedDecimals i = 0 while(joined[i] == '0'): i += 1 if i > 0: return joined[i:] else: return joined def custom_handle_deltas(self, bookside, deltas, timestamp=None): for j in range(0, len(deltas)): delta = deltas[j] price = self.parse_number(delta[0]) amount = self.parse_number(delta[1]) oldTimestamp = timestamp if timestamp else 0 timestamp = max(oldTimestamp, self.parse_to_int(float(delta[2]) * 1000)) bookside.store(price, amount) return timestamp def handle_system_status(self, client: Client, message): # # todo: answer the question whether handleSystemStatus should be renamed # and unified for any usage pattern that # involves system status and maintenance updates # # { # "connectionID": 15527282728335292000, # "event": "systemStatus", # "status": "online", # online|maintenance|(custom status tbd) # "version": "0.2.0" # } # return message async def authenticate(self, params={}): url = self.urls['api']['ws']['private'] client = self.client(url) authenticated = 'authenticated' subscription = self.safe_value(client.subscriptions, authenticated) if subscription is None: response = await self.privatePostGetWebSocketsToken(params) # # { # "error":[], # "result":{ # "token":"xeAQ\/RCChBYNVh53sTv1yZ5H4wIbwDF20PiHtTF+4UI", # "expires":900 # } # } # subscription = self.safe_value(response, 'result') client.subscriptions[authenticated] = subscription return self.safe_string(subscription, 'token') async def watch_private(self, name, symbol: Str = None, since: Int = None, limit: Int = None, params={}): await self.load_markets() token = await self.authenticate() subscriptionHash = name messageHash = name if symbol is not None: symbol = self.symbol(symbol) messageHash += ':' + symbol url = self.urls['api']['ws']['private'] requestId = self.request_id() subscribe = { 'event': 'subscribe', 'reqid': requestId, 'subscription': { 'name': name, 'token': token, }, } request = self.deep_extend(subscribe, params) result = await self.watch(url, messageHash, request, subscriptionHash) if self.newUpdates: limit = result.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(result, symbol, since, limit) async def watch_my_trades(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ watches information on multiple trades made by the user :param str 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 dict [params]: extra parameters specific to the exchange API endpoint :returns dict[]: a list of [trade structures]{@link https://docs.ccxt.com/#/?id=trade-structure """ return await self.watch_private('ownTrades', symbol, since, limit, params) def handle_my_trades(self, client: Client, message, subscription=None): # # [ # [ # { # "TT5UC3-GOIRW-6AZZ6R": { # "cost": "1493.90107", # "fee": "3.88415", # "margin": "0.00000", # "ordertxid": "OTLAS3-RRHUF-NDWH5A", # "ordertype": "market", # "pair": "XBT/USDT", # "postxid": "TKH2SE-M7IF5-CFI7LT", # "price": "6851.50005", # "time": "1586822919.335498", # "type": "sell", # "vol": "0.21804000" # } # }, # { # "TIY6G4-LKLAI-Y3GD4A": { # "cost": "22.17134", # "fee": "0.05765", # "margin": "0.00000", # "ordertxid": "ODQXS7-MOLK6-ICXKAA", # "ordertype": "market", # "pair": "ETH/USD", # "postxid": "TKH2SE-M7IF5-CFI7LT", # "price": "169.97999", # "time": "1586340530.895739", # "type": "buy", # "vol": "0.13043500" # } # }, # ], # "ownTrades", # {sequence: 1} # ] # allTrades = self.safe_value(message, 0, []) allTradesLength = len(allTrades) if allTradesLength > 0: if self.myTrades is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) self.myTrades = ArrayCache(limit) stored = self.myTrades symbols = {} for i in range(0, len(allTrades)): trades = self.safe_value(allTrades, i, {}) ids = list(trades.keys()) for j in range(0, len(ids)): id = ids[j] trade = trades[id] parsed = self.parse_ws_trade(self.extend({'id': id}, trade)) stored.append(parsed) symbol = parsed['symbol'] symbols[symbol] = True name = 'ownTrades' client.resolve(self.myTrades, name) keys = list(symbols.keys()) for i in range(0, len(keys)): messageHash = name + ':' + keys[i] client.resolve(self.myTrades, messageHash) def parse_ws_trade(self, trade, market=None): # # { # "id": "TIMIRG-WUNNE-RRJ6GT", # injected from outside # "ordertxid": "OQRPN2-LRHFY-HIFA7D", # "postxid": "TKH2SE-M7IF5-CFI7LT", # "pair": "USDCUSDT", # "time": 1586340086.457, # "type": "sell", # "ordertype": "market", # "price": "0.99860000", # "cost": "22.16892001", # "fee": "0.04433784", # "vol": "22.20000000", # "margin": "0.00000000", # "misc": '' # } # # { # "id": "TIY6G4-LKLAI-Y3GD4A", # "cost": "22.17134", # "fee": "0.05765", # "margin": "0.00000", # "ordertxid": "ODQXS7-MOLK6-ICXKAA", # "ordertype": "market", # "pair": "ETH/USD", # "postxid": "TKH2SE-M7IF5-CFI7LT", # "price": "169.97999", # "time": "1586340530.895739", # "type": "buy", # "vol": "0.13043500" # } # wsName = self.safe_string(trade, 'pair') market = self.safe_value(self.options['marketsByWsName'], wsName, market) symbol = None orderId = self.safe_string(trade, 'ordertxid') id = self.safe_string_2(trade, 'id', 'postxid') timestamp = self.safe_timestamp(trade, 'time') side = self.safe_string(trade, 'type') type = self.safe_string(trade, 'ordertype') price = self.safe_float(trade, 'price') amount = self.safe_float(trade, 'vol') cost = None fee = None if 'fee' in trade: currency = None if market is not None: currency = market['quote'] fee = { 'cost': self.safe_float(trade, 'fee'), 'currency': currency, } if market is not None: symbol = market['symbol'] if price is not None: if amount is not None: cost = price * amount return { 'id': id, 'order': orderId, 'info': trade, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'symbol': symbol, 'type': type, 'side': side, 'takerOrMaker': None, 'price': price, 'amount': amount, 'cost': cost, 'fee': fee, } async def watch_orders(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]: """ :see: https://docs.kraken.com/websockets/#message-openOrders watches information on multiple orders made by the user :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 orde structures to retrieve :param dict [params]: maximum number of orderic to the exchange API endpoint :returns dict[]: a list of `order structures ` """ return await self.watch_private('openOrders', symbol, since, limit, params) def handle_orders(self, client: Client, message, subscription=None): # # [ # [ # { # "OGTT3Y-C6I3P-XRI6HX": { # "cost": "0.00000", # "descr": { # "close": "", # "leverage": "0:1", # "order": "sell 10.00345345 XBT/EUR @ limit 34.50000 with 0:1 leverage", # "ordertype": "limit", # "pair": "XBT/EUR", # "price": "34.50000", # "price2": "0.00000", # "type": "sell" # }, # "expiretm": "0.000000", # "fee": "0.00000", # "limitprice": "34.50000", # "misc": "", # "oflags": "fcib", # "opentm": "0.000000", # "price": "34.50000", # "refid": "OKIVMP-5GVZN-Z2D2UA", # "starttm": "0.000000", # "status": "open", # "stopprice": "0.000000", # "userref": 0, # "vol": "10.00345345", # "vol_exec": "0.00000000" # } # }, # { # "OGTT3Y-C6I3P-XRI6HX": { # "cost": "0.00000", # "descr": { # "close": "", # "leverage": "0:1", # "order": "sell 0.00000010 XBT/EUR @ limit 5334.60000 with 0:1 leverage", # "ordertype": "limit", # "pair": "XBT/EUR", # "price": "5334.60000", # "price2": "0.00000", # "type": "sell" # }, # "expiretm": "0.000000", # "fee": "0.00000", # "limitprice": "5334.60000", # "misc": "", # "oflags": "fcib", # "opentm": "0.000000", # "price": "5334.60000", # "refid": "OKIVMP-5GVZN-Z2D2UA", # "starttm": "0.000000", # "status": "open", # "stopprice": "0.000000", # "userref": 0, # "vol": "0.00000010", # "vol_exec": "0.00000000" # } # }, # ], # "openOrders", # {"sequence": 234} # ] # # status-change # # [ # [ # {"OGTT3Y-C6I3P-XRI6HX": {"status": "closed"}}, # {"OGTT3Y-C6I3P-XRI6HX": {"status": "closed"}}, # ], # "openOrders", # {"sequence": 59342} # ] # allOrders = self.safe_value(message, 0, []) allOrdersLength = len(allOrders) if allOrdersLength > 0: limit = self.safe_integer(self.options, 'ordersLimit', 1000) if self.orders is None: self.orders = ArrayCacheBySymbolById(limit) stored = self.orders symbols = {} for i in range(0, len(allOrders)): orders = self.safe_value(allOrders, i, {}) ids = list(orders.keys()) for j in range(0, len(ids)): id = ids[j] order = orders[id] parsed = self.parse_ws_order(order) parsed['id'] = id symbol = None symbolsByOrderId = self.safe_value(self.options, 'symbolsByOrderId', {}) if parsed['symbol'] is not None: symbol = parsed['symbol'] symbolsByOrderId[id] = symbol self.options['symbolsByOrderId'] = symbolsByOrderId else: symbol = self.safe_string(symbolsByOrderId, id) previousOrders = self.safe_value(stored.hashmap, symbol) previousOrder = self.safe_value(previousOrders, id) newOrder = parsed if previousOrder is not None: newRawOrder = self.extend(previousOrder['info'], newOrder['info']) newOrder = self.parse_ws_order(newRawOrder) newOrder['id'] = id length = len(stored) if length == limit and (previousOrder is None): first = stored[0] if first['id'] in symbolsByOrderId: del symbolsByOrderId[first['id']] stored.append(newOrder) symbols[symbol] = True name = 'openOrders' client.resolve(self.orders, name) keys = list(symbols.keys()) for i in range(0, len(keys)): messageHash = name + ':' + keys[i] client.resolve(self.orders, messageHash) def parse_ws_order(self, order, market=None): # # createOrder # { # "avg_price": "0.00000", # "cost": "0.00000", # "descr": { # "close": null, # "leverage": null, # "order": "sell 0.01000000 ETH/USDT @ limit 1900.00000", # "ordertype": "limit", # "pair": "ETH/USDT", # "price": "1900.00000", # "price2": "0.00000", # "type": "sell" # }, # "expiretm": null, # "fee": "0.00000", # "limitprice": "0.00000", # "misc": '', # "oflags": "fciq", # "opentm": "1667522705.757622", # "refid": null, # "starttm": null, # "status": "open", # "stopprice": "0.00000", # "timeinforce": "GTC", # "userref": 0, # "vol": "0.01000000", # "vol_exec": "0.00000000" # } # description = self.safe_value(order, 'descr', {}) orderDescription = self.safe_string(description, 'order') side = None type = None wsName = None price = None amount = None if orderDescription is not None: parts = orderDescription.split(' ') side = self.safe_string(parts, 0) amount = self.safe_string(parts, 1) wsName = self.safe_string(parts, 2) type = self.safe_string(parts, 4) price = self.safe_string(parts, 5) side = self.safe_string(description, 'type', side) type = self.safe_string(description, 'ordertype', type) wsName = self.safe_string(description, 'pair', wsName) market = self.safe_value(self.options['marketsByWsName'], wsName, market) symbol = None timestamp = self.safe_timestamp(order, 'opentm') amount = self.safe_string(order, 'vol', amount) filled = self.safe_string(order, 'vol_exec') fee = None cost = self.safe_string(order, 'cost') price = self.safe_string(description, 'price', price) if (price is None) or (Precise.string_eq(price, '0.0')): price = self.safe_string(description, 'price2') if (price is None) or (Precise.string_eq(price, '0.0')): price = self.safe_string(order, 'price', price) average = self.safe_string_2(order, 'avg_price', 'price') if market is not None: symbol = market['symbol'] if 'fee' in order: flags = order['oflags'] feeCost = self.safe_string(order, 'fee') fee = { 'cost': feeCost, 'rate': None, } if flags.find('fciq') >= 0: fee['currency'] = market['quote'] elif flags.find('fcib') >= 0: fee['currency'] = market['base'] status = self.parse_order_status(self.safe_string(order, 'status')) id = self.safe_string(order, 'id') if id is None: txid = self.safe_value(order, 'txid') id = self.safe_string(txid, 0) clientOrderId = self.safe_string(order, 'userref') rawTrades = self.safe_value(order, 'trades') trades = None if rawTrades is not None: trades = self.parse_trades(rawTrades, market, None, None, {'order': id}) stopPrice = self.safe_number(order, 'stopprice') return self.safe_order({ 'id': id, 'clientOrderId': clientOrderId, 'info': order, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': None, 'status': status, 'symbol': symbol, 'type': type, 'timeInForce': None, 'postOnly': None, 'side': side, 'price': price, 'stopPrice': stopPrice, 'triggerPrice': stopPrice, 'cost': cost, 'amount': amount, 'filled': filled, 'average': average, 'remaining': None, 'fee': fee, 'trades': trades, }) async def watch_multi_helper(self, unifiedName: str, channelName: str, symbols: Strings = None, subscriptionArgs=None, params={}): await self.load_markets() # symbols are required symbols = self.market_symbols(symbols, None, False, True, False) messageHashes = [] for i in range(0, len(symbols)): messageHashes.append(self.get_message_hash(unifiedName, None, self.symbol(symbols[i]))) # for WS subscriptions, we can't use .market_ids(symbols), instead a custom is field needed markets = self.markets_for_symbols(symbols) wsMarketIds = [] for i in range(0, len(markets)): wsMarketId = self.safe_string(markets[i]['info'], 'wsname') wsMarketIds.append(wsMarketId) request = { 'event': 'subscribe', 'reqid': self.request_id(), 'pair': wsMarketIds, 'subscription': { 'name': channelName, }, } url = self.urls['api']['ws']['public'] return await self.watch_multiple(url, messageHashes, self.extend(request, params), messageHashes, subscriptionArgs) def get_message_hash(self, unifiedElementName: str, subChannelName: Str = None, symbol: Str = None): # unifiedElementName can be : orderbook, trade, ticker, bidask ... # subChannelName only applies to channel that needs specific variation(i.e. depth_50, depth_100..) to be selected withSymbol = symbol is not None messageHash = unifiedElementName if not withSymbol: messageHash += 's' else: messageHash += '@' + symbol if subChannelName is not None: messageHash += '#' + subChannelName return messageHash def handle_subscription_status(self, client: Client, message): # # public # # { # "channelID": 210, # "channelName": "book-10", # "event": "subscriptionStatus", # "reqid": 1574146735269, # "pair": "ETH/XBT", # "status": "subscribed", # "subscription": {depth: 10, name: "book"} # } # # private # # { # "channelName": "openOrders", # "event": "subscriptionStatus", # "reqid": 1, # "status": "subscribed", # "subscription": {maxratecount: 125, name: "openOrders"} # } # channelId = self.safe_string(message, 'channelID') if channelId is not None: client.subscriptions[channelId] = message # requestId = self.safe_string(message, "reqid") # if requestId in client.futures: # del client.futures[requestId] # } def handle_error_message(self, client: Client, message): # # { # "errorMessage": "Currency pair not in ISO 4217-A3 format foobar", # "event": "subscriptionStatus", # "pair": "foobar", # "reqid": 1574146735269, # "status": "error", # "subscription": {name: "ticker"} # } # errorMessage = self.safe_string(message, 'errorMessage') if errorMessage is not None: requestId = self.safe_value(message, 'reqid') if requestId is not None: broad = self.exceptions['ws']['broad'] broadKey = self.find_broadly_matched_key(broad, errorMessage) exception = None if broadKey is None: exception = ExchangeError(errorMessage) # c# requirement to convert the errorMessage to string else: exception = broad[broadKey](errorMessage) client.reject(exception, requestId) return False return True def handle_message(self, client: Client, message): if isinstance(message, list): channelId = self.safe_string(message, 0) subscription = self.safe_value(client.subscriptions, channelId, {}) info = self.safe_value(subscription, 'subscription', {}) messageLength = len(message) channelName = self.safe_string(message, messageLength - 2) name = self.safe_string(info, 'name') methods = { # public 'book': self.handle_order_book, 'ohlc': self.handle_ohlcv, 'ticker': self.handle_ticker, 'trade': self.handle_trades, # private 'openOrders': self.handle_orders, 'ownTrades': self.handle_my_trades, } method = self.safe_value_2(methods, name, channelName) if method is not None: method(client, message, subscription) else: if self.handle_error_message(client, message): event = self.safe_string(message, 'event') methods = { 'heartbeat': self.handle_heartbeat, 'systemStatus': self.handle_system_status, 'subscriptionStatus': self.handle_subscription_status, 'addOrderStatus': self.handle_create_edit_order, 'editOrderStatus': self.handle_create_edit_order, 'cancelOrderStatus': self.handle_cancel_order, 'cancelAllStatus': self.handle_cancel_all_orders, } method = self.safe_value(methods, event) if method is not None: method(client, message)