# -*- 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 from ccxt.base.types import Any, Int, Market, Order, OrderBook, Position, Str, Strings, Ticker, Trade from ccxt.async_support.base.ws.client import Client from typing import List from ccxt.base.errors import AuthenticationError from ccxt.base.errors import ArgumentsRequired from ccxt.base.errors import NotSupported from ccxt.base.precise import Precise class vertex(ccxt.async_support.vertex): def describe(self) -> Any: return self.deep_extend(super(vertex, self).describe(), { 'has': { 'ws': True, 'watchBalance': False, 'watchMyTrades': True, 'watchOHLCV': False, 'watchOrderBook': True, 'watchOrders': True, 'watchTicker': True, 'watchTickers': False, 'watchTrades': True, 'watchTradesForSymbols': False, 'watchPositions': True, }, 'urls': { 'api': { 'ws': 'wss://gateway.prod.vertexprotocol.com/v1/subscribe', }, 'test': { 'ws': 'wss://gateway.sepolia-test.vertexprotocol.com/v1/subscribe', }, }, 'requiredCredentials': { 'apiKey': False, 'secret': False, 'walletAddress': True, 'privateKey': True, }, 'options': { 'tradesLimit': 1000, 'ordersLimit': 1000, 'requestId': {}, 'watchPositions': { 'fetchPositionsSnapshot': True, # or False 'awaitPositionsSnapshot': True, # whether to wait for the positions snapshot before providing updates }, 'ws': { 'inflate': True, 'options': { 'headers': { 'Sec-WebSocket-Extensions': 'permessage-deflate', # requires permessage-deflate extension, maybe we can set self in client implementation when self.inflateis True }, }, }, }, 'streaming': { # 'ping': self.ping, 'keepAlive': 30000, }, 'exceptions': { 'ws': { 'exact': { 'Auth is needed.': AuthenticationError, }, }, }, }) def request_id(self, url): options = self.safe_dict(self.options, 'requestId', {}) previousValue = self.safe_integer(options, url, 0) newValue = self.sum(previousValue, 1) self.options['requestId'][url] = newValue return newValue async def watch_public(self, messageHash, message): url = self.urls['api']['ws'] requestId = self.request_id(url) subscribe = { 'id': requestId, } request = self.extend(subscribe, message) wsOptions = { 'headers': { 'Sec-WebSocket-Extensions': 'permessage-deflate', }, } self.options['ws'] = { 'options': wsOptions, } return await self.watch(url, messageHash, request, messageHash, subscribe) async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ watches information on multiple trades made in a market https://docs.vertexprotocol.com/developer-resources/api/subscriptions/streams :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 ` """ await self.load_markets() market = self.market(symbol) name = 'trade' topic = market['id'] + '@' + name request = { 'method': 'subscribe', 'stream': { 'type': name, 'product_id': self.parse_to_numeric(market['id']), }, } message = self.extend(request, params) trades = await self.watch_public(topic, message) if self.newUpdates: limit = trades.getLimit(market['symbol'], limit) return self.filter_by_symbol_since_limit(trades, symbol, since, limit, True) def handle_trade(self, client: Client, message): # # { # "type": "trade", # "timestamp": "1676151190656903000", # timestamp of the event in nanoseconds # "product_id": 1, # "price": "1000", # price the trade happened at, multiplied by 1e18 # # both taker_qty and maker_qty have the same value # # set to filled amount(min amount of taker and maker) when matching against book # # set to matched amm base amount when matching against amm # "taker_qty": "1000", # "maker_qty": "1000", # "is_taker_buyer": True, # "is_maker_amm": True # True when maker is amm # } # topic = self.safe_string(message, 'type') marketId = self.safe_string(message, 'product_id') trade = self.parse_ws_trade(message) symbol = trade['symbol'] if not (symbol in self.trades): limit = self.safe_integer(self.options, 'tradesLimit', 1000) stored = ArrayCache(limit) self.trades[symbol] = stored trades = self.trades[symbol] trades.append(trade) self.trades[symbol] = trades client.resolve(trades, marketId + '@' + topic) 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 https://docs.vertexprotocol.com/developer-resources/api/subscriptions/streams :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.user]: user address, will default to self.walletAddress if not provided :returns dict[]: a list of `order structures ` """ if symbol is None: raise ArgumentsRequired(self.id + ' watchMyTrades requires a symbol.') await self.load_markets() userAddress = None userAddress, params = self.handlePublicAddress('watchMyTrades', params) market = self.market(symbol) name = 'fill' topic = market['id'] + '@' + name request = { 'method': 'subscribe', 'stream': { 'type': name, 'product_id': self.parse_to_numeric(market['id']), 'subaccount': self.convertAddressToSender(userAddress), }, } message = self.extend(request, params) trades = await self.watch_public(topic, message) if self.newUpdates: limit = trades.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(trades, symbol, since, limit, True) def handle_my_trades(self, client: Client, message): # # { # "type": "fill", # "timestamp": "1676151190656903000", # timestamp of the event in nanoseconds # "product_id": 1, # # the subaccount that placed self order # "subaccount": "0x7a5ec2748e9065794491a8d29dcf3f9edb8d7c43746573743000000000000000", # # hash of the order that uniquely identifies it # "order_digest": "0xf4f7a8767faf0c7f72251a1f9e5da590f708fd9842bf8fcdeacbaa0237958fff", # # the amount filled, multiplied by 1e18 # "filled_qty": "1000", # # the amount outstanding unfilled, multiplied by 1e18 # "remaining_qty": "2000", # # the original order amount, multiplied by 1e18 # "original_qty": "3000", # # fill price # "price": "24991000000000000000000", # # True for `taker`, False for `maker` # "is_taker": True, # "is_bid": True, # # True when matching against amm # "is_against_amm": True, # # an optional `order id` that can be provided when placing an order # "id": 100 # } # topic = self.safe_string(message, 'type') marketId = self.safe_string(message, 'product_id') if self.myTrades is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) self.myTrades = ArrayCacheBySymbolById(limit) trades = self.myTrades parsed = self.parse_ws_trade(message) trades.append(parsed) client.resolve(trades, marketId + '@' + topic) def parse_ws_trade(self, trade, market=None): # # watchTrades # { # "type": "trade", # "timestamp": "1676151190656903000", # timestamp of the event in nanoseconds # "product_id": 1, # "price": "1000", # price the trade happened at, multiplied by 1e18 # # both taker_qty and maker_qty have the same value # # set to filled amount(min amount of taker and maker) when matching against book # # set to matched amm base amount when matching against amm # "taker_qty": "1000", # "maker_qty": "1000", # "is_taker_buyer": True, # "is_maker_amm": True # True when maker is amm # } # watchMyTrades # { # "type": "fill", # "timestamp": "1676151190656903000", # timestamp of the event in nanoseconds # "product_id": 1, # # the subaccount that placed self order # "subaccount": "0x7a5ec2748e9065794491a8d29dcf3f9edb8d7c43746573743000000000000000", # # hash of the order that uniquely identifies it # "order_digest": "0xf4f7a8767faf0c7f72251a1f9e5da590f708fd9842bf8fcdeacbaa0237958fff", # # the amount filled, multiplied by 1e18 # "filled_qty": "1000", # # the amount outstanding unfilled, multiplied by 1e18 # "remaining_qty": "2000", # # the original order amount, multiplied by 1e18 # "original_qty": "3000", # # fill price # "price": "24991000000000000000000", # # True for `taker`, False for `maker` # "is_taker": True, # "is_bid": True, # # True when matching against amm # "is_against_amm": True, # # an optional `order id` that can be provided when placing an order # "id": 100 # } # marketId = self.safe_string(trade, 'product_id') market = self.safe_market(marketId, market) symbol = market['symbol'] price = self.convertFromX18(self.safe_string(trade, 'price')) amount = self.convertFromX18(self.safe_string_2(trade, 'taker_qty', 'filled_qty')) cost = Precise.string_mul(price, amount) timestamp = self.safe_integer_product(trade, 'timestamp', 0.000001) takerOrMaker = None isTaker = self.safe_bool(trade, 'is_taker') if isTaker is not None: takerOrMaker = 'taker' if (isTaker) else 'maker' side = None isBid = self.safe_bool(trade, 'is_bid') if isBid is not None: side = 'buy' if (isBid) else 'sell' return self.safe_trade({ 'id': None, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'symbol': symbol, 'side': side, 'price': price, 'amount': amount, 'cost': cost, 'order': self.safe_string_2(trade, 'digest', 'id'), 'takerOrMaker': takerOrMaker, 'type': None, 'fee': None, 'info': trade, }, market) async def watch_ticker(self, symbol: str, params={}) -> Ticker: """ https://docs.vertexprotocol.com/developer-resources/api/subscriptions/streams 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() name = 'best_bid_offer' market = self.market(symbol) topic = market['id'] + '@' + name request = { 'method': 'subscribe', 'stream': { 'type': name, 'product_id': self.parse_to_numeric(market['id']), }, } message = self.extend(request, params) return await self.watch_public(topic, message) def parse_ws_ticker(self, ticker, market=None): # # { # "type": "best_bid_offer", # "timestamp": "1676151190656903000", # timestamp of the event in nanoseconds # "product_id": 1, # "bid_price": "1000", # the highest bid price, multiplied by 1e18 # "bid_qty": "1000", # quantity at the huighest bid, multiplied by 1e18. # # i.e. if self is USDC with 6 decimals, one USDC # # would be 1e12 # "ask_price": "1000", # lowest ask price # "ask_qty": "1000" # quantity at the lowest ask # } # timestamp = self.safe_integer_product(ticker, 'timestamp', 0.000001) return self.safe_ticker({ 'symbol': self.safe_symbol(None, market), 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'high': self.safe_string(ticker, 'high'), 'low': self.safe_string(ticker, 'low'), 'bid': self.convertFromX18(self.safe_string(ticker, 'bid_price')), 'bidVolume': self.convertFromX18(self.safe_string(ticker, 'bid_qty')), 'ask': self.convertFromX18(self.safe_string(ticker, 'ask_price')), 'askVolume': self.convertFromX18(self.safe_string(ticker, 'ask_qty')), 'vwap': None, 'open': None, 'close': None, 'last': None, 'previousClose': None, 'change': None, 'percentage': None, 'average': None, 'baseVolume': None, 'quoteVolume': None, 'info': ticker, }, market) def handle_ticker(self, client: Client, message): # # { # "type": "best_bid_offer", # "timestamp": "1676151190656903000", # timestamp of the event in nanoseconds # "product_id": 1, # "bid_price": "1000", # the highest bid price, multiplied by 1e18 # "bid_qty": "1000", # quantity at the huighest bid, multiplied by 1e18. # # i.e. if self is USDC with 6 decimals, one USDC # # would be 1e12 # "ask_price": "1000", # lowest ask price # "ask_qty": "1000" # quantity at the lowest ask # } # marketId = self.safe_string(message, 'product_id') market = self.safe_market(marketId) ticker = self.parse_ws_ticker(message, market) ticker['symbol'] = market['symbol'] self.tickers[market['symbol']] = ticker client.resolve(ticker, marketId + '@best_bid_offer') return message async def watch_order_book(self, symbol: str, limit: Int = None, params={}) -> OrderBook: """ https://docs.vertexprotocol.com/developer-resources/api/subscriptions/streams watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data :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() name = 'book_depth' market = self.market(symbol) messageHash = market['id'] + '@' + name url = self.urls['api']['ws'] requestId = self.request_id(url) request: dict = { 'id': requestId, 'method': 'subscribe', 'stream': { 'type': name, 'product_id': self.parse_to_numeric(market['id']), }, } subscription: dict = { 'id': str(requestId), 'name': name, 'symbol': symbol, 'method': self.handle_order_book_subscription, 'limit': limit, 'params': params, } message = self.extend(request, params) orderbook = await self.watch(url, messageHash, message, messageHash, subscription) return orderbook.limit() def handle_order_book_subscription(self, client: Client, message, subscription): defaultLimit = self.safe_integer(self.options, 'watchOrderBookLimit', 1000) limit = self.safe_integer(subscription, 'limit', defaultLimit) symbol = self.safe_string(subscription, 'symbol') # watchOrderBook if symbol in self.orderbooks: del self.orderbooks[symbol] self.orderbooks[symbol] = self.order_book({}, limit) self.spawn(self.fetch_order_book_snapshot, client, message, subscription) async def fetch_order_book_snapshot(self, client, message, subscription): symbol = self.safe_string(subscription, 'symbol') market = self.market(symbol) messageHash = market['id'] + '@book_depth' try: defaultLimit = self.safe_integer(self.options, 'watchOrderBookLimit', 1000) limit = self.safe_integer(subscription, 'limit', defaultLimit) params = self.safe_value(subscription, 'params') snapshot = await self.fetch_rest_order_book_safe(symbol, limit, params) if self.safe_value(self.orderbooks, symbol) is None: # if the orderbook is dropped before the snapshot is received return orderbook = self.orderbooks[symbol] orderbook.reset(snapshot) messages = orderbook.cache for i in range(0, len(messages)): messageItem = messages[i] lastTimestamp = self.parse_to_int(Precise.string_div(self.safe_string(messageItem, 'last_max_timestamp'), '1000000')) if lastTimestamp < orderbook['timestamp']: continue else: self.handle_order_book_message(client, messageItem, orderbook) self.orderbooks[symbol] = orderbook client.resolve(orderbook, messageHash) except Exception as e: del client.subscriptions[messageHash] client.reject(e, messageHash) def handle_order_book(self, client: Client, message): # # # the feed does not include a snapshot, just the deltas # # { # "type":"book_depth", # # book depth aggregates a number of events once every 50ms # # these are the minimum and maximum timestamps from # # events that contributed to self response # "min_timestamp": "1683805381879572835", # "max_timestamp": "1683805381879572835", # # the max_timestamp of the last book_depth event for self product # "last_max_timestamp": "1683805381771464799", # "product_id":1, # # changes to the bid side of the book in the form of [[price, new_qty]] # "bids":[["21594490000000000000000","51007390115411548"]], # # changes to the ask side of the book in the form of [[price, new_qty]] # "asks":[["21694490000000000000000","0"],["21695050000000000000000","0"]] # } # marketId = self.safe_string(message, 'product_id') market = self.safe_market(marketId) symbol = market['symbol'] if not (symbol in self.orderbooks): self.orderbooks[symbol] = self.order_book() orderbook = self.orderbooks[symbol] timestamp = self.safe_integer(orderbook, 'timestamp') if timestamp is None: # Buffer the events you receive from the stream. orderbook.cache.append(message) else: lastTimestamp = self.parse_to_int(Precise.string_div(self.safe_string(message, 'last_max_timestamp'), '1000000')) if lastTimestamp > timestamp: self.handle_order_book_message(client, message, orderbook) client.resolve(orderbook, marketId + '@book_depth') def handle_order_book_message(self, client: Client, message, orderbook): timestamp = self.parse_to_int(Precise.string_div(self.safe_string(message, 'last_max_timestamp'), '1000000')) # convert from X18 data = { 'bids': [], 'asks': [], } bids = self.safe_list(message, 'bids', []) for i in range(0, len(bids)): bid = bids[i] data['bids'].append([ self.convertFromX18(bid[0]), self.convertFromX18(bid[1]), ]) asks = self.safe_list(message, 'asks', []) for i in range(0, len(asks)): ask = asks[i] data['asks'].append([ self.convertFromX18(ask[0]), self.convertFromX18(ask[1]), ]) self.handle_deltas(orderbook['asks'], self.safe_list(data, 'asks', [])) self.handle_deltas(orderbook['bids'], self.safe_list(data, 'bids', [])) orderbook['timestamp'] = timestamp orderbook['datetime'] = self.iso8601(timestamp) return orderbook def handle_delta(self, bookside, delta): price = self.safe_float(delta, 0) amount = self.safe_float(delta, 1) bookside.store(price, amount) def handle_deltas(self, bookside, deltas): for i in range(0, len(deltas)): self.handle_delta(bookside, deltas[i]) def handle_subscription_status(self, client: Client, message): # # { # "result": null, # "id": 1574649734450 # } # id = self.safe_string(message, 'id') subscriptionsById = self.index_by(client.subscriptions, 'id') subscription = self.safe_value(subscriptionsById, id, {}) method = self.safe_value(subscription, 'method') if method is not None: method(client, message, subscription) return message async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]: """ https://docs.vertexprotocol.com/developer-resources/api/subscriptions/streams watch all open positions :param str[]|None symbols: list of unified market symbols @param since @param limit :param dict params: extra parameters specific to the exchange API endpoint :param str [params.user]: user address, will default to self.walletAddress if not provided :returns dict[]: a list of `position structure ` """ await self.load_markets() symbols = self.market_symbols(symbols) if not self.is_empty(symbols): if len(symbols) > 1: raise NotSupported(self.id + ' watchPositions require only one symbol.') else: raise ArgumentsRequired(self.id + ' watchPositions require one symbol.') userAddress = None userAddress, params = self.handlePublicAddress('watchPositions', params) url = self.urls['api']['ws'] client = self.client(url) self.set_positions_cache(client, symbols, params) fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', True) awaitPositionsSnapshot = self.handle_option('watchPositions', 'awaitPositionsSnapshot', True) if fetchPositionsSnapshot and awaitPositionsSnapshot and self.positions is None: snapshot = await client.future('fetchPositionsSnapshot') return self.filter_by_symbols_since_limit(snapshot, symbols, since, limit, True) name = 'position_change' market = self.market(symbols[0]) topic = market['id'] + '@' + name request = { 'method': 'subscribe', 'stream': { 'type': name, 'product_id': self.parse_to_numeric(market['id']), 'subaccount': self.convertAddressToSender(userAddress), }, } message = self.extend(request, params) newPositions = await self.watch_public(topic, message) if self.newUpdates: limit = newPositions.getLimit(symbols[0], limit) return self.filter_by_symbols_since_limit(self.positions, symbols, since, limit, True) def set_positions_cache(self, client: Client, symbols: Strings = None, params={}): fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', False) if fetchPositionsSnapshot: messageHash = 'fetchPositionsSnapshot' if not (messageHash in client.futures): client.future(messageHash) self.spawn(self.load_positions_snapshot, client, messageHash, symbols, params) else: self.positions = ArrayCacheBySymbolBySide() async def load_positions_snapshot(self, client, messageHash, symbols, params): positions = await self.fetch_positions(symbols, params) self.positions = ArrayCacheBySymbolBySide() cache = self.positions for i in range(0, len(positions)): position = positions[i] cache.append(position) # don't remove the future from the .futures cache future = client.futures[messageHash] future.resolve(cache) client.resolve(cache, 'positions') def handle_positions(self, client, message): # # { # "type":"position_change", # "timestamp": "1676151190656903000", # timestamp of event in nanoseconds # "product_id":1, # # whether self is a position change for the LP token for self product # "is_lp":false, # # subaccount who's position changed # "subaccount":"0x7a5ec2748e9065794491a8d29dcf3f9edb8d7c43706d00000000000000000000", # # new amount for self product # "amount":"51007390115411548", # # new quote balance for self product; zero for everything except non lp perps # # the negative of the entry cost of the perp # "v_quote_amount":"0" # } # if self.positions is None: self.positions = ArrayCacheBySymbolBySide() cache = self.positions topic = self.safe_string(message, 'type') marketId = self.safe_string(message, 'product_id') market = self.safe_market(marketId) position = self.parse_ws_position(message, market) cache.append(position) client.resolve(position, marketId + '@' + topic) def parse_ws_position(self, position, market=None): # # { # "type":"position_change", # "timestamp": "1676151190656903000", # timestamp of event in nanoseconds # "product_id":1, # # whether self is a position change for the LP token for self product # "is_lp":false, # # subaccount who's position changed # "subaccount":"0x7a5ec2748e9065794491a8d29dcf3f9edb8d7c43706d00000000000000000000", # # new amount for self product # "amount":"51007390115411548", # # new quote balance for self product; zero for everything except non lp perps # # the negative of the entry cost of the perp # "v_quote_amount":"0" # } # marketId = self.safe_string(position, 'product_id') market = self.safe_market(marketId) contractSize = self.convertFromX18(self.safe_string(position, 'amount')) side = 'buy' if Precise.string_lt(contractSize, '1'): side = 'sell' timestamp = self.parse_to_int(Precise.string_div(self.safe_string(position, 'timestamp'), '1000000')) return self.safe_position({ 'info': position, 'id': None, 'symbol': self.safe_string(market, 'symbol'), 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastUpdateTimestamp': None, 'initialMargin': None, 'initialMarginPercentage': None, 'maintenanceMargin': None, 'maintenanceMarginPercentage': None, 'entryPrice': None, 'notional': None, 'leverage': None, 'unrealizedPnl': None, 'contracts': None, 'contractSize': self.parse_number(contractSize), 'marginRatio': None, 'liquidationPrice': None, 'markPrice': None, 'lastPrice': None, 'collateral': None, 'marginMode': 'cross', 'marginType': None, 'side': side, 'percentage': None, 'hedged': None, 'stopLossPrice': None, 'takeProfitPrice': None, }) def handle_auth(self, client: Client, message): # # {result: null, id: 1} # messageHash = 'authenticated' error = self.safe_string(message, 'error') if error is None: # client.resolve(message, messageHash) future = self.safe_value(client.futures, 'authenticated') future.resolve(True) else: authError = AuthenticationError(self.json(message)) client.reject(authError, messageHash) # allows further authentication attempts if messageHash in client.subscriptions: del client.subscriptions['authenticated'] def build_ws_authentication_sig(self, message, chainId, verifyingContractAddress): messageTypes = { 'StreamAuthentication': [ {'name': 'sender', 'type': 'bytes32'}, {'name': 'expiration', 'type': 'uint64'}, ], } return self.buildSig(chainId, messageTypes, message, verifyingContractAddress) async def authenticate(self, params={}): self.check_required_credentials() url = self.urls['api']['ws'] client = self.client(url) messageHash = 'authenticated' future = client.future(messageHash) authenticated = self.safe_value(client.subscriptions, messageHash) if authenticated is None: requestId = self.request_id(url) contracts = await self.queryContracts() chainId = self.safe_string(contracts, 'chain_id') verifyingContractAddress = self.safe_string(contracts, 'endpoint_addr') now = self.nonce() nonce = now + 90000 authentication = { 'sender': self.convertAddressToSender(self.walletAddress), 'expiration': nonce, } request = { 'id': requestId, 'method': 'authenticate', 'tx': { 'sender': authentication['sender'], 'expiration': self.number_to_string(authentication['expiration']), }, 'signature': self.build_ws_authentication_sig(authentication, chainId, verifyingContractAddress), } message = self.extend(request, params) self.watch(url, messageHash, message, messageHash) return await future async def watch_private(self, messageHash, message, params={}): await self.authenticate(params) url = self.urls['api']['ws'] requestId = self.request_id(url) subscribe = { 'id': requestId, } request = self.extend(subscribe, message) return await self.watch(url, messageHash, request, messageHash, subscribe) 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://docs.vertexprotocol.com/developer-resources/api/subscriptions/streams :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 :returns dict[]: a list of `order structures ` """ if symbol is None: raise ArgumentsRequired(self.id + ' watchOrders requires a symbol.') self.check_required_credentials() await self.load_markets() name = 'order_update' market = self.market(symbol) topic = market['id'] + '@' + name request = { 'method': 'subscribe', 'stream': { 'type': name, 'subaccount': self.convertAddressToSender(self.walletAddress), 'product_id': self.parse_to_numeric(market['id']), }, } message = self.extend(request, params) orders = await self.watch_private(topic, message) if self.newUpdates: limit = orders.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True) def parse_ws_order_status(self, status): if status is not None: statuses = { 'filled': 'open', 'placed': 'open', 'cancelled': 'canceled', } return self.safe_string(statuses, status, status) return status def parse_ws_order(self, order, market: Market = None) -> Order: # # { # "type": "order_update", # # timestamp of the event in nanoseconds # "timestamp": "1695081920633151000", # "product_id": 1, # # order digest # "digest": "0xf7712b63ccf70358db8f201e9bf33977423e7a63f6a16f6dab180bdd580f7c6c", # # remaining amount to be filled. # # will be `0` if the order is either fully filled or cancelled. # "amount": "82000000000000000", # # any of: "filled", "cancelled", "placed" # "reason": "filled" # # an optional `order id` that can be provided when placing an order # "id": 100 # } # marketId = self.safe_string(order, 'product_id') timestamp = self.parse_to_int(Precise.string_div(self.safe_string(order, 'timestamp'), '1000000')) remainingString = self.convertFromX18(self.safe_string(order, 'amount')) remaining = self.parse_to_numeric(remainingString) status = self.parse_ws_order_status(self.safe_string(order, 'reason')) if Precise.string_eq(remainingString, '0') and status == 'open': status = 'closed' market = self.safe_market(marketId, market) symbol = market['symbol'] return self.safe_order({ 'info': order, 'id': self.safe_string_2(order, 'digest', 'id'), 'clientOrderId': None, 'timestamp': timestamp, 'datetime': self.iso8601(timestamp), 'lastTradeTimestamp': None, 'lastUpdateTimestamp': None, 'symbol': symbol, 'type': None, 'timeInForce': None, 'postOnly': None, 'reduceOnly': None, 'side': None, 'price': None, 'triggerPrice': None, 'amount': None, 'cost': None, 'average': None, 'filled': None, 'remaining': remaining, 'status': status, 'fee': None, 'trades': None, }, market) def handle_order_update(self, client: Client, message): # # { # "type": "order_update", # # timestamp of the event in nanoseconds # "timestamp": "1695081920633151000", # "product_id": 1, # # order digest # "digest": "0xf7712b63ccf70358db8f201e9bf33977423e7a63f6a16f6dab180bdd580f7c6c", # # remaining amount to be filled. # # will be `0` if the order is either fully filled or cancelled. # "amount": "82000000000000000", # # any of: "filled", "cancelled", "placed" # "reason": "filled" # # an optional `order id` that can be provided when placing an order # "id": 100 # } # topic = self.safe_string(message, 'type') marketId = self.safe_string(message, 'product_id') parsed = self.parse_ws_order(message) symbol = self.safe_string(parsed, 'symbol') orderId = self.safe_string(parsed, 'id') if symbol is not None: if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) self.orders = ArrayCacheBySymbolById(limit) cachedOrders = self.orders orders = self.safe_dict(cachedOrders.hashmap, symbol, {}) order = self.safe_dict(orders, orderId) if order is not None: parsed['timestamp'] = self.safe_integer(order, 'timestamp') parsed['datetime'] = self.safe_string(order, 'datetime') cachedOrders.append(parsed) client.resolve(self.orders, marketId + '@' + topic) def handle_error_message(self, client: Client, message): # # { # result: null, # error: 'error parsing request: missing field `expiration`', # id: 0 # } # errorMessage = self.safe_string(message, 'error') try: if errorMessage is not None: feedback = self.id + ' ' + self.json(message) self.throw_exactly_matched_exception(self.exceptions['exact'], errorMessage, feedback) return False except Exception as error: if isinstance(error, AuthenticationError): messageHash = 'authenticated' client.reject(error, messageHash) if messageHash in client.subscriptions: del client.subscriptions[messageHash] else: client.reject(error) return True def handle_message(self, client: Client, message): if self.handle_error_message(client, message): return methods = { 'trade': self.handle_trade, 'best_bid_offer': self.handle_ticker, 'book_depth': self.handle_order_book, 'fill': self.handle_my_trades, 'position_change': self.handle_positions, 'order_update': self.handle_order_update, } event = self.safe_string(message, 'type') method = self.safe_value(methods, event) if method is not None: method(client, message) return requestId = self.safe_string(message, 'id') if requestId is not None: self.handle_subscription_status(client, message) return # check whether it's authentication auth = self.safe_value(client.futures, 'authenticated') if auth is not None: self.handle_auth(client, message)