From b89db84c658cc91879aca17706c16ca932c4f3bc Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Thu, 1 Feb 2024 15:32:22 -0500 Subject: [PATCH 1/8] Hook Odyssey Perpetual connector --- hummingbot/connector/connector_status.py | 2 + .../hook_odyssey_perpetual/__init__.py | 0 .../hook_odyssey_perpetual/dummy.pxd | 2 + .../hook_odyssey_perpetual/dummy.pyx | 2 + ...ey_perpetual_api_order_book_data_source.py | 122 ++++ .../hook_odyssey_perpetual_constants.py | 47 ++ .../hook_odyssey_perpetual_data_source.py | 574 ++++++++++++++++++ .../hook_odyssey_perpetual_derivative.py | 565 +++++++++++++++++ ...hook_odyssey_perpetual_graphql_executor.py | 298 +++++++++ .../hook_odyssey_perpetual_signing.py | 104 ++++ .../hook_odyssey_perpetual_utils.py | 109 ++++ .../polkadex/polkadex_query_executor.py | 2 +- .../exchange/vertex/vertex_exchange.py | 2 +- 13 files changed, 1827 insertions(+), 2 deletions(-) create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/__init__.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pxd create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pyx create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_signing.py create mode 100644 hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py diff --git a/hummingbot/connector/connector_status.py b/hummingbot/connector/connector_status.py index 59ed850dfee..8f4921847c3 100644 --- a/hummingbot/connector/connector_status.py +++ b/hummingbot/connector/connector_status.py @@ -31,6 +31,8 @@ 'hitbtc': 'bronze', 'hyperliquid_perpetual_testnet': 'bronze', 'hyperliquid_perpetual': 'bronze', + 'hook_odyssey_perpetual': 'bronze', + 'hook_odyssey_perpetual_testnet': 'bronze', 'huobi': 'silver', 'kraken': 'bronze', 'kucoin': 'silver', diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/__init__.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pxd b/hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pxd new file mode 100644 index 00000000000..4b098d6f599 --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pxd @@ -0,0 +1,2 @@ +cdef class dummy(): + pass diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pyx b/hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pyx new file mode 100644 index 00000000000..4b098d6f599 --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/dummy.pyx @@ -0,0 +1,2 @@ +cdef class dummy(): + pass diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py new file mode 100644 index 00000000000..4efaffbf2ca --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py @@ -0,0 +1,122 @@ +import asyncio +from decimal import Decimal +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_data_source import ( + HookOdysseyPerpetualDataSource, +) +from hummingbot.core.data_type.funding_info import FundingInfo +from hummingbot.core.data_type.order_book_message import OrderBookMessage, OrderBookMessageType +from hummingbot.core.data_type.perpetual_api_order_book_data_source import PerpetualAPIOrderBookDataSource +from hummingbot.core.event.event_forwarder import EventForwarder +from hummingbot.core.event.events import OrderBookEvent +from hummingbot.core.web_assistant.ws_assistant import WSAssistant +from hummingbot.logger import HummingbotLogger + +if TYPE_CHECKING: + from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_derivative import ( + HookOdysseyPerpetualDerivative, + ) + + +class HookOdysseyPerpetualAPIOrderBookDataSource(PerpetualAPIOrderBookDataSource): + _hpopobds_logger: Optional[HummingbotLogger] = None + + def __init__( + self, + trading_pairs: List[str], + connector: "HookOdysseyPerpetualDerivative", + data_source: HookOdysseyPerpetualDataSource, + domain: str, + ): + super().__init__(trading_pairs) + self._connector = connector + self._domain = domain + self._data_source = data_source + self._forwarders = [] + self._configure_event_forwarders() + + async def get_last_traded_prices( + self, trading_pairs: List[str], domain: Optional[str] = None + ) -> Dict[str, float]: + return await self._connector.get_last_traded_prices(trading_pairs=trading_pairs) + + async def listen_for_subscriptions(self): + # The socket reconnection is managed by the data_source. This method should do nothing + pass + + async def _parse_order_book_snapshot_message( + self, raw_message: OrderBookMessage, message_queue: asyncio.Queue + ): + # In HookOdysseyPerpetual, 'raw_message' is the OrderBookMessage from the data source + message_queue.put_nowait(raw_message) + + async def _connected_websocket_assistant(self) -> WSAssistant: + # HookOdysseyPerpetual uses GraphQL websockets to consume stream events + raise NotImplementedError + + async def _subscribe_channels(self, ws: WSAssistant): + # HookOdysseyPerpetual uses GraphQL websockets to consume stream events + raise NotImplementedError + + async def _order_book_snapshot(self, trading_pair: str) -> OrderBookMessage: + return await self._data_source.get_order_book_snapshot(trading_pair) + + async def _parse_trade_message( + self, raw_message: OrderBookMessage, message_queue: asyncio.Queue + ): + # In HookOdysseyPerpetual, 'raw_message' is the OrderBookMessage from the data source + message_queue.put_nowait(raw_message) + + async def _parse_order_book_diff_message( + self, raw_message: OrderBookMessage, message_queue: asyncio.Queue + ): + # In HookOdysseyPerpetual, 'raw_message' is the OrderBookMessage from the data source + message_queue.put_nowait(raw_message) + + async def get_funding_info(self, trading_pair: str) -> FundingInfo: + index_price = await self._data_source.get_last_traded_price(trading_pair) + funding_info = await self._data_source.get_funding_info(trading_pair) + funding_info.index_price = Decimal(index_price) + return funding_info + + async def _parse_funding_info_message( + self, raw_message: Dict[str, Any], message_queue: asyncio.Queue + ): + """ + Placeholder method for parsing funding info messages. Actual implementation needed. + """ + pass + + def _configure_event_forwarders(self): + # Forward order book diffs + order_book_event_forwarder = EventForwarder( + to_function=self._process_order_book_event + ) + self._forwarders.append(order_book_event_forwarder) + self._data_source.add_listener( + event_tag=OrderBookEvent.OrderBookDataSourceUpdateEvent, + listener=order_book_event_forwarder, + ) + + # Forward trade updates + trade_event_forwarder = EventForwarder( + to_function=self._process_public_trade_event + ) + self._forwarders.append(trade_event_forwarder) + self._data_source.add_listener( + event_tag=OrderBookEvent.TradeEvent, listener=trade_event_forwarder + ) + + def _process_order_book_event(self, order_book_diff: OrderBookMessage): + if order_book_diff.type == OrderBookMessageType.SNAPSHOT: + self._message_queue[self._snapshot_messages_queue_key].put_nowait( + order_book_diff + ) + else: + self._message_queue[self._diff_messages_queue_key].put_nowait( + order_book_diff + ) + + def _process_public_trade_event(self, trade_update: OrderBookMessage): + self._message_queue[self._trade_messages_queue_key].put_nowait(trade_update) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py new file mode 100644 index 00000000000..a4001b5b1be --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py @@ -0,0 +1,47 @@ +from hummingbot.core.data_type.in_flight_order import OrderState + +EXCHANGE_NAME = "hook_odyssey_perpetual" +DOMAIN = EXCHANGE_NAME +TESTNET_DOMAIN = "hook_odyssey_perpetual_testnet" +SUPPORTED_COLLATERAL = ["ETH", "USDC"] + +BASE_URLS = { + DOMAIN: "https://goerli-api.hook.xyz/query", # TODO: Set to mainnet + TESTNET_DOMAIN: "https://api.hookdev.xyz/query", # TODO: Set to testnet +} + +WS_URLS = { + DOMAIN: "wss://goerli-api.hook.xyz/query", # TODO: Set to mainnet + TESTNET_DOMAIN: "wss://api.hookdev.xyz/query", # TODO: Set to testnet +} + +# Order Statuses +ORDER_STATE = { + "OPEN": OrderState.OPEN, + "PARTIALLY_MATCHED": OrderState.PARTIALLY_FILLED, + "MATCHED": OrderState.FILLED, + "PARTIALLY_FILLED": OrderState.PARTIALLY_FILLED, + "FILLED": OrderState.FILLED, + "CANCELED": OrderState.CANCELED, + "EXPIRED": OrderState.CANCELED, + "REJECTED": OrderState.FAILED, + "UNFILLABLE": OrderState.FAILED, +} + +RATE_LIMITS = [] + +# EIP 712 +EIP712_DOMAIN_NAME = "Hook" +EIP712_DOMAIN_VERSION = "1.0.0" + +CONTRACT_ADDRESSES = { + DOMAIN: "0x64247BeF0C0990aF63FCbdd21dc07aC2b251f500", # TODO: Set to mainnet + TESTNET_DOMAIN: "0xc6e7DF5E7b4f2A278906862b61205850344D4e7d", # TODO: Set to testnet +} + +CHAIN_IDS = { + DOMAIN: 46658378, # TODO: Set to mainnet + TESTNET_DOMAIN: 9999, # TODO: Set to testnet +} + +DEFAULT_TIMEOUT = 3 diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py new file mode 100644 index 00000000000..1940757f006 --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py @@ -0,0 +1,574 @@ +import asyncio +import logging +import time +from decimal import Decimal +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple + +from hummingbot.connector.derivative.hook_odyssey_perpetual import hook_odyssey_perpetual_constants as CONSTANTS +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_graphql_executor import ( + HookOdysseyPerpetualGrapQLExecutor, +) +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_signing import ( + HookOdysseyPerpetualSigner, + Order, +) +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_utils import ( + DEFAULT_FEES, + eth_to_wei, + wei_to_eth, +) +from hummingbot.connector.derivative.position import Position, PositionSide +from hummingbot.connector.utils import combine_to_hb_trading_pair +from hummingbot.core.data_type.common import OrderType, TradeType +from hummingbot.core.data_type.funding_info import FundingInfo +from hummingbot.core.data_type.in_flight_order import OrderUpdate, TradeUpdate +from hummingbot.core.data_type.order_book_message import OrderBookMessage, OrderBookMessageType +from hummingbot.core.data_type.trade_fee import TradeFeeBase +from hummingbot.core.event.event_listener import EventListener +from hummingbot.core.event.events import MarketEvent, OrderBookEvent +from hummingbot.core.network_iterator import NetworkStatus +from hummingbot.core.pubsub import PubSub +from hummingbot.logger import HummingbotLogger + +if TYPE_CHECKING: + from hummingbot.connector.exchange_py_base import ExchangePyBase + + +class HookOdysseyPerpetualDataSource: + _logger: Optional[HummingbotLogger] = None + + @classmethod + def logger(cls) -> HummingbotLogger: + if cls._logger is None: + cls._logger = logging.getLogger(HummingbotLogger.logger_name_for_class(cls)) + return cls._logger + + def __init__( + self, + hook_odyssey_perpetual_eth_address: str, + hook_odyssey_perpetual_private_key: str, + hook_odyssey_api_key: str, + connector: "ExchangePyBase", + domain: Optional[str] = CONSTANTS.DOMAIN, + trading_required: bool = True, + ): + self._connector = connector + self._domain = domain + self._trading_required = trading_required + self._hook_odyssey_perpetual_eth_address = hook_odyssey_perpetual_eth_address + self._graphql_executor = HookOdysseyPerpetualGrapQLExecutor( + hook_odyssey_api_key=hook_odyssey_api_key, domain=self._domain + ) + self._signer = HookOdysseyPerpetualSigner( + private_key=hook_odyssey_perpetual_private_key, domain=self._domain + ) + self._publisher = PubSub() + self._events_listening_tasks = [] + self._assets_map: Dict[str, str] = {} + + # Data source state - used to provide subscription data at any time + self._perpetual_pairs: Dict[ + str, Dict[str, Any] + ] = {} # Mapping of trading pair to perpetual pair + self._symbols: Dict[str, str] = {} # Mapping of symbol to trading pair + self._instrument_hashes: Dict[ + str, str + ] = {} # Mapping of instrument hash to trading pair + self._subaccounts: Dict[str, str] = {} # Mapping of subaccount to trading pair + self._last_traded_price: Dict[ + str, Decimal + ] = {} # Mapping of trading pair to last traded price + self._subaccount_balances: Dict[ + str, Decimal + ] = {} # Mapping of subaccount to balance + self._subaccount_positions: Dict[ + str, Position + ] = {} # Mapping of subaccount to position + self._funding_info: Dict[ + str, FundingInfo + ] = {} # Mapping of trading pair to funding info + self._orderbook_snapshots: Dict[ + str, OrderBookMessage + ] = {} # Mapping of trading pair to orderbook snapshot + self._fees: Tuple[Decimal, Decimal] = ( + DEFAULT_FEES.maker_percent_fee_decimal, + DEFAULT_FEES.taker_percent_fee_decimal, + ) + + self._initial_ticker_event = asyncio.Event() + self._initial_statistics_event = asyncio.Event() + self._initial_orderbook_event = asyncio.Event() + self._initial_subaccount_orders_event = asyncio.Event() + self._initial_subaccount_balances_event = asyncio.Event() + self._initial_subaccount_positions_event = asyncio.Event() + + async def start(self, trading_pairs: List[str]): + if len(self._events_listening_tasks) > 0: + raise AssertionError( + "Data source is already started and can't be started again" + ) + await self.get_supported_pairs() + + for trading_pair in trading_pairs: + perpetual_pair = self.get_perpetual_pair_for_trading_pair(trading_pair) + instrument_hash = perpetual_pair.get("instrumentHash", "") + symbol = perpetual_pair.get("symbol", "") + subaccount = perpetual_pair.get("subaccount", "") + + # Index price + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_ticker( + self._process_ticker_update, + symbol, + ) + ) + ) + # Statistics / Funding Info + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_statistics( + self._process_statistics_update, + symbol, + ) + ) + ) + # Orderbook + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_orderbook( + self._process_orderbook_update, + instrument_hash, + ) + ) + ) + + if self._trading_required: + # Private order updates + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_subaccount_orders( + self._process_subaccount_orders_update, + subaccount, + ) + ) + ) + + if self._trading_required: + # Private balances + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_subaccount_balances( + self._process_subaccount_balances, + self._hook_odyssey_perpetual_eth_address, + ) + ) + ) + # Private positions + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_subaccount_positions( + self._process_subaccount_positions, + self._hook_odyssey_perpetual_eth_address, + ) + ) + ) + + async def stop(self): + for task in self._events_listening_tasks: + task.cancel() + self._events_listening_tasks = [] + + def is_started(self) -> bool: + return len(self._events_listening_tasks) > 0 + + def add_listener(self, event_tag, listener: EventListener): + self._publisher.add_listener(event_tag=event_tag, listener=listener) + + def remove_listener(self, event_tag, listener: EventListener): + self._publisher.remove_listener(event_tag=event_tag, listener=listener) + + async def exchange_status(self): + resp = await self._graphql_executor.account_details() + if "accountDetails" not in resp: + return NetworkStatus.NOT_CONNECTED + if "makerFeeBips" not in resp["accountDetails"]: + return NetworkStatus.NOT_CONNECTED + return NetworkStatus.CONNECTED + + def get_fees(self) -> Tuple[Decimal, Decimal]: + return self._fees + + async def fetch_fees(self): + resp = await self._graphql_executor.account_details() + account_details = resp["accountDetails"] + maker_fee = Decimal(account_details["makerFeeBips"]) / Decimal(10000) + taker_fee = Decimal(account_details["takerFeeBips"]) / Decimal(10000) + self._fees = (maker_fee, taker_fee) + + async def get_supported_pairs(self) -> List[Dict[str, Any]]: + """Fetches and processes supported perpetual_pairs, updating the internal mapping.""" + resp = await self._graphql_executor.perpetual_pairs() + perpetual_pairs = resp["perpetualPairs"] + for perpetual_pair in perpetual_pairs: + trading_pair = combine_to_hb_trading_pair( + base=perpetual_pair["symbol"], quote=perpetual_pair["baseCurrency"] + ) + self._perpetual_pairs[trading_pair] = perpetual_pair + self._symbols[perpetual_pair["symbol"]] = trading_pair + self._instrument_hashes[perpetual_pair["instrumentHash"]] = trading_pair + self._subaccounts[perpetual_pair["subaccount"]] = trading_pair + return perpetual_pairs + + def get_perpetual_pair_for_trading_pair( + self, trading_pair: str + ) -> Optional[Dict[str, Any]]: + """Retrieves perpetual pair details for a specific trading pair.""" + return self._perpetual_pairs.get(trading_pair, None) + + async def get_last_traded_price(self, trading_pair: str) -> Decimal: + # Wait for the trade subscription snapshot to come in + await self._initial_ticker_event.wait() + return self._last_traded_price.get(trading_pair, 0.0) + + async def get_funding_info(self, trading_pair: str) -> FundingInfo: + # Wait for the funding info subscription snapshot to come in + await self._initial_statistics_event.wait() + if trading_pair in self._funding_info: + return self._funding_info[trading_pair] + else: + return FundingInfo( + trading_pair=trading_pair, + index_price=Decimal(0), + mark_price=Decimal(0), + next_funding_utc_timestamp=int(time.time()), + rate=Decimal(0), + ) + + async def get_order_book_snapshot(self, trading_pair: str) -> OrderBookMessage: + # Wait for the orderbook subscription snapshot to come in + await self._initial_orderbook_event.wait() + if trading_pair in self._orderbook_snapshots: + return self._orderbook_snapshots[trading_pair] + else: + content = { + "bids": [], + "asks": [], + "update_id": 0, + } + return OrderBookMessage( + message_type=OrderBookMessageType.SNAPSHOT, + content=content, + timestamp=int(time.time()), + ) + + async def get_balances(self) -> Dict[str, Decimal]: + # Wait for the subaccount balances subscription snapshot to come in + await self._initial_subaccount_balances_event.wait() + return self._subaccount_balances + + async def get_positions(self) -> Dict[str, Position]: + # Wait for the subaccount positions subscription snapshot to come in + await self._initial_subaccount_positions_event.wait() + return self._subaccount_positions + + async def _process_ticker_update(self, event: Dict[str, Any], symbol: str): + ticker_event = event["ticker"] + + if not self._initial_ticker_event.is_set(): + self._initial_ticker_event.set() + + if symbol not in self._symbols: + return + trading_pair = self._symbols[symbol] + self._last_traded_price[trading_pair] = wei_to_eth(ticker_event["price"]) + return self._last_traded_price[trading_pair] + + async def _process_statistics_update(self, event: Dict[str, Any], symbol: str): + statistics_event = event["statistics"] + + if ( + statistics_event["eventType"] == "SNAPSHOT" + and not self._initial_statistics_event.is_set() + ): + self._initial_statistics_event.set() + + if symbol not in self._symbols: + return + trading_pair = self._symbols[symbol] + self._funding_info[trading_pair] = FundingInfo( + trading_pair=trading_pair, + index_price=Decimal(0), + mark_price=Decimal(wei_to_eth(statistics_event["markPrice"])), + next_funding_utc_timestamp=int(statistics_event["nextFundingEpoch"]), + rate=Decimal(statistics_event["fundingRateBips"]) / Decimal(10000), + ) + + async def _process_orderbook_update( + self, event: Dict[str, Any], instrument_hash: str + ): + if instrument_hash not in self._instrument_hashes: + return + trading_pair = self._instrument_hashes[instrument_hash] + + orderbook_event = event["orderbook"] + message_type = OrderBookMessageType.DIFF + if orderbook_event["eventType"] == "SNAPSHOT": + message_type = OrderBookMessageType.SNAPSHOT + + if not self._initial_orderbook_event.is_set(): + self._initial_orderbook_event.set() + + bids = [ + [ + wei_to_eth(bl["price"]), + wei_to_eth(bl["size"]), + ] + for bl in orderbook_event["bidLevels"] + ] + asks = [ + [ + wei_to_eth(al["price"]), + wei_to_eth(al["size"]), + ] + for al in orderbook_event["askLevels"] + ] + + order_book_message: OrderBookMessage = OrderBookMessage( + message_type=message_type, + content={ + "trading_pair": trading_pair, + "update_id": orderbook_event["timestamp"], + "bids": bids, + "asks": asks, + }, + timestamp=float(orderbook_event["timestamp"]), + ) + + self._apply_local_orderbook_update(order_book_message) + + self._publisher.trigger_event( + event_tag=OrderBookEvent.OrderBookDataSourceUpdateEvent, + message=order_book_message, + ) + + async def _process_subaccount_orders_update( + self, event: Dict[str, Any], subaccount: str + ): + orders_event = event["subaccountOrders"] + + if orders_event["eventType"] == "SNAPSHOT": + if not self._initial_subaccount_orders_event.is_set(): + self._initial_subaccount_orders_event.set() + return # Ignore snapshots + orders = orders_event["orders"] + + if subaccount not in self._subaccounts: + return + trading_pair = self._subaccounts[subaccount] + + for order in orders: + trading_pair = trading_pair + # Update order state + order_state = CONSTANTS.ORDER_STATE[order["status"]] + order_update = OrderUpdate( + trading_pair=trading_pair, + update_timestamp=int(time.time()), + new_state=order_state, + client_order_id=order["orderHash"], + exchange_order_id=order["orderHash"], + ) + self._publisher.trigger_event( + event_tag=MarketEvent.OrderUpdate, message=order_update + ) + + # Update trade + if order_state == "FILLED" or order_state == "PARTIALLY_FILLED": + size = wei_to_eth(order["size"]) + remaining_size = wei_to_eth(order["remainingSize"]) + fill_amount = size - remaining_size + fill_price = wei_to_eth(order["limitPrice"]) + trade_type = ( + TradeType.BUY if order["direction"] == "BUY" else TradeType.SELL + ) + + maker_fee, taker_fee = self.get_fees() + fee_rate = maker_fee if order["orderType"] == "MARKET" else taker_fee + fee = TradeFeeBase.new_perpetual_fee( + trade_type=trade_type, percent=fee_rate + ) + + trade_update = TradeUpdate( + trade_id=f"{order['orderHash']}-{order['timestamp']}", + client_order_id=order["orderHash"], + exchange_order_id=order["orderHash"], + trading_pair=trading_pair, + fill_timestamp=self._time(), + fill_price=fill_price, + fill_base_amount=fill_amount, + fill_quote_amount=fill_price * fill_amount, + fee=fee, + ) + self._publisher.trigger_event( + event_tag=MarketEvent.TradeUpdate, message=trade_update + ) + + async def _process_subaccount_balances(self, event: Dict[str, Any], address: str): + balances_event = event["subaccountBalances"] + + if ( + balances_event["eventType"] == "SNAPSHOT" + and not self._initial_subaccount_balances_event.is_set() + ): + self._initial_subaccount_balances_event.set() + + for balance in balances_event["balances"]: + # Ignore primary account balances + if balance["subaccountID"] != 0: + self._subaccount_balances[balance["subaccount"]] = wei_to_eth( + balance["balance"] + ) + + async def _process_subaccount_positions(self, event: Dict[str, Any], address: str): + positions_event = event["subaccountPositions"] + + if ( + positions_event["eventType"] == "SNAPSHOT" + and not self._initial_subaccount_positions_event.is_set() + ): + self._initial_subaccount_positions_event.set() + + for position in positions_event["positions"]: + size = wei_to_eth(position["size"]) + side = PositionSide.LONG if size > 0 else PositionSide.SHORT + size = abs(size) + instrument_hash = position["instrument"]["id"] + if instrument_hash not in self._instrument_hashes: + continue + trading_pair = self._instrument_hashes[instrument_hash] + + if size == 0: + self._subaccount_positions.pop(instrument_hash) + else: + self._subaccount_positions[instrument_hash] = Position( + trading_pair=trading_pair, + position_side=side, + unrealized_pnl=Decimal(0), + entry_price=Decimal(0), + amount=size, + leverage=Decimal(1), + ) + + def order_hash( + self, + market_hash: str, + instrument_hash: str, + subaccount: str, + amount: Decimal, + price: Decimal, + trade_type: TradeType, + order_type: OrderType, + nonce: int, + ) -> str: + amount = eth_to_wei(amount) + price = eth_to_wei(price) + trade_type = "BUY" if trade_type == TradeType.BUY else "SELL" + order_type = "MARKET" if order_type == OrderType.MARKET else "LIMIT" + order_input = { + "marketHash": market_hash, + "instrumentHash": instrument_hash, + "subaccount": subaccount, + "orderType": order_type, + "direction": trade_type, + "size": str(amount), + "limitPrice": str(price), + "timeInForce": "GTC", # Good till cancelled + "nonce": str(nonce), + } + + order = Order( + market=order_input["marketHash"], + instrumentType=2, + instrumentId=order_input["instrumentHash"], + direction=0 if order_input["direction"] == "BUY" else 1, + maker=int(order_input["subaccount"]), + taker=0, # Taker is not specified + amount=int(order_input["size"]), + limitPrice=int(order_input["limitPrice"]), + expiration=0, # Time in force is GTC + nonce=nonce, + counter=0, + postOnly=False, + reduceOnly=False, + allOrNothing=False, + ) + return self._signer.compute_order_hash(order) + + async def place_order( + self, + market_hash: str, + instrument_hash: str, + subaccount: str, + amount: Decimal, + price: Decimal, + trade_type: TradeType, + order_type: OrderType, + nonce: int, + ) -> Tuple[str, float]: + amount = eth_to_wei(amount) + price = eth_to_wei(price) + trade_type = "BUY" if trade_type == TradeType.BUY else "SELL" + order_type = "MARKET" if order_type == OrderType.MARKET else "LIMIT" + order_input = { + "marketHash": market_hash, + "instrumentHash": instrument_hash, + "subaccount": subaccount, + "orderType": order_type, + "direction": trade_type, + "size": str(amount), + "limitPrice": str(price), + "timeInForce": "GTC", # Good till cancelled + "nonce": str(nonce), + } + + order = Order( + market=order_input["marketHash"], + instrumentType=2, + instrumentId=order_input["instrumentHash"], + direction=0 if order_input["direction"] == "BUY" else 1, + maker=int(order_input["subaccount"]), + taker=0, # Taker is not specified + amount=int(order_input["size"]), + limitPrice=int(order_input["limitPrice"]), + expiration=0, # Time in force is GTC + nonce=nonce, + counter=0, + postOnly=False, + reduceOnly=False, + allOrNothing=False, + ) + signature, order_hash = self._signer.sign_order(order) + + signature_input = { + "signatureType": "DIRECT", + "signature": signature, + } + success = await self._graphql_executor.place_order(order_input, signature_input) + if not success: + raise AssertionError("Failed to place order") + return order_hash, int(time.time()) + + async def cancel_order(self, order_id: str) -> bool: + return await self._graphql_executor.cancel_order(order_id) + + def _apply_local_orderbook_update(self, order_book_message: OrderBookMessage): + trading_pair = order_book_message.trading_pair + if order_book_message.type == OrderBookMessageType.SNAPSHOT: + self._orderbook_snapshots[trading_pair] = order_book_message + else: + # Apply diff to snapshot + curr_snapshot = self._orderbook_snapshots.get(trading_pair) + if "bids" in order_book_message.content: + curr_snapshot.content["bids"] = order_book_message.content["bids"] + if "asks" in order_book_message.content: + curr_snapshot.content["asks"] = order_book_message.content["asks"] diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py new file mode 100644 index 00000000000..282dc3420b7 --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py @@ -0,0 +1,565 @@ +import asyncio +from typing import TYPE_CHECKING, Dict, List, Optional, Tuple + +from _decimal import Decimal +from bidict import bidict + +from hummingbot.connector.constants import s_decimal_NaN +from hummingbot.connector.derivative.hook_odyssey_perpetual import hook_odyssey_perpetual_constants as CONSTANTS +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_api_order_book_data_source import ( + HookOdysseyPerpetualAPIOrderBookDataSource, +) +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_data_source import ( + HookOdysseyPerpetualDataSource, +) +from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_utils import ( + get_tracking_nonce, + wei_to_eth, +) +from hummingbot.connector.perpetual_derivative_py_base import PerpetualDerivativePyBase +from hummingbot.connector.trading_rule import TradingRule +from hummingbot.connector.utils import combine_to_hb_trading_pair +from hummingbot.core.api_throttler.data_types import RateLimit +from hummingbot.core.data_type.common import OrderType, PositionAction, PositionMode, TradeType +from hummingbot.core.data_type.in_flight_order import InFlightOrder, OrderUpdate, TradeUpdate +from hummingbot.core.data_type.order_book_tracker_data_source import OrderBookTrackerDataSource +from hummingbot.core.data_type.trade_fee import AddedToCostTradeFee, TradeFeeBase +from hummingbot.core.data_type.user_stream_tracker_data_source import UserStreamTrackerDataSource +from hummingbot.core.event.event_forwarder import EventForwarder +from hummingbot.core.event.events import AccountEvent, BalanceUpdateEvent, MarketEvent +from hummingbot.core.network_iterator import NetworkStatus +from hummingbot.core.utils.async_utils import safe_ensure_future, safe_gather +from hummingbot.core.web_assistant.auth import AuthBase +from hummingbot.core.web_assistant.web_assistants_factory import WebAssistantsFactory + +if TYPE_CHECKING: + from hummingbot.client.config.config_helpers import ClientConfigAdapter + + +class HookOdysseyPerpetualDerivative(PerpetualDerivativePyBase): + def __init__( + self, + client_config_map: "ClientConfigAdapter", + hook_odyssey_perpetual_eth_address: str, + hook_odyssey_perpetual_private_key: str, + hook_odyssey_api_key: str, + trading_pairs: Optional[List[str]] = None, + trading_required: bool = True, + domain: str = CONSTANTS.DOMAIN, + ): + self.hook_odyssey_perpetual_eth_address = hook_odyssey_perpetual_eth_address + self.hook_odyssey_perpetual_private_key = hook_odyssey_perpetual_private_key + self._trading_required = trading_required + self._trading_pairs = trading_pairs + self._domain = domain + self._data_source = HookOdysseyPerpetualDataSource( + hook_odyssey_perpetual_eth_address=hook_odyssey_perpetual_eth_address, + hook_odyssey_perpetual_private_key=hook_odyssey_perpetual_private_key, + hook_odyssey_api_key=hook_odyssey_api_key, + connector=self, + domain=self.domain, + trading_required=self._trading_required, + ) + super().__init__(client_config_map) + self._forwarders = [] + self._configure_event_forwarders() + + @property + def name(self) -> str: + return self._domain + + @property + def authenticator(self) -> AuthBase: + return None + + @property + def rate_limits_rules(self) -> List[RateLimit]: + return CONSTANTS.RATE_LIMITS + + @property + def domain(self) -> str: + return self._domain + + @property + def client_order_id_max_length(self) -> int: + return None + + @property + def client_order_id_prefix(self) -> str: + return CONSTANTS.BROKER_ID + + @property + def trading_rules_request_path(self) -> str: + raise NotImplementedError + + @property + def trading_pairs_request_path(self) -> str: + raise NotImplementedError + + @property + def check_network_request_path(self) -> str: + raise NotImplementedError + + @property + def trading_pairs(self): + return self._trading_pairs + + @property + def is_cancel_request_in_exchange_synchronous(self) -> bool: + return False + + @property + def is_trading_required(self) -> bool: + return self._trading_required + + @property + def funding_fee_poll_interval(self) -> int: + return 60 + + @property + def status_dict(self) -> Dict[str, bool]: + return { + "symbols_mapping_initialized": self.trading_pair_symbol_map_ready(), + "order_books_initialized": self.order_book_tracker.ready, + "account_balance": not self.is_trading_required + or len(self._account_balances) > 0, + "trading_rule_initialized": len(self._trading_rules) > 0 + if self.is_trading_required + else True, + } + + async def start_network(self): + await super().start_network() + await self._data_source.start(self._trading_pairs) + await self._initialize_trading_pair_symbol_map() + await self._update_trading_rules() + # self._ready = True # Mark connector as ready + + async def stop_network(self): + """ + This function is executed when the connector is stopped. It performs a general cleanup and stops all background + tasks that require the connection with the exchange to work. + """ + await super().stop_network() + await self._data_source.stop() + + async def check_network(self) -> NetworkStatus: + """ + Checks connectivity with the exchange using the API + """ + try: + status = await self._data_source.exchange_status() + except asyncio.CancelledError: + raise + except Exception: + status = NetworkStatus.NOT_CONNECTED + return status + + def supported_order_types(self) -> List[OrderType]: + """ + :return a list of OrderType supported by this connector + """ + return [OrderType.LIMIT, OrderType.MARKET] + + def supported_position_modes(self): + """ + This method needs to be overridden to provide the accurate information depending on the exchange. + """ + return [PositionMode.ONEWAY] + + def get_buy_collateral_token(self, trading_pair: str) -> str: + trading_rule: TradingRule = self._trading_rules[trading_pair] + return trading_rule.buy_order_collateral_token + + def get_sell_collateral_token(self, trading_pair: str) -> str: + trading_rule: TradingRule = self._trading_rules[trading_pair] + return trading_rule.sell_order_collateral_token + + def _is_request_exception_related_to_time_synchronizer( + self, request_exception: Exception + ): + return False + + def _is_order_not_found_during_status_update_error(self, status_update_exception: Exception) -> bool: + return False + + def _is_order_not_found_during_cancelation_error(self, cancelation_exception: Exception) -> bool: + return False + + def _create_web_assistants_factory(self) -> WebAssistantsFactory: + return WebAssistantsFactory(throttler=self._throttler) + + async def _initialize_trading_pair_symbol_map(self): + markets = await self._data_source.get_supported_pairs() + mapping = bidict() + for market in markets: + symbol = market["symbol"] + baseCurrency = market["baseCurrency"] + mapping[symbol] = combine_to_hb_trading_pair(symbol, baseCurrency) + self._set_trading_pair_symbol_map(mapping) + + async def _update_trading_rules(self): + self._trading_rules = dict() + for trading_pair in self._trading_pairs: + perpetual_pair = self._data_source.get_perpetual_pair_for_trading_pair(trading_pair) + if perpetual_pair is None: + continue + self._trading_rules[trading_pair] = TradingRule( + trading_pair=trading_pair, + min_order_size=Decimal(wei_to_eth(perpetual_pair["minOrderSize"])), + max_order_size=Decimal(wei_to_eth(perpetual_pair["maxOrderSize"])), + min_base_amount_increment=Decimal(wei_to_eth(perpetual_pair["minOrderSizeIncrement"])), + min_price_increment=Decimal(wei_to_eth(perpetual_pair["minPriceIncrement"])), + buy_order_collateral_token=perpetual_pair["baseCurrency"], + sell_order_collateral_token=perpetual_pair["baseCurrency"], + ) + + def _create_order_book_data_source(self) -> OrderBookTrackerDataSource: + return HookOdysseyPerpetualAPIOrderBookDataSource( + trading_pairs=self._trading_pairs, + connector=self, + data_source=self._data_source, + domain=self.domain, + ) + + def _create_user_stream_data_source(self) -> UserStreamTrackerDataSource: + # Not used in HookOdysseyPerpetual + raise NotImplementedError + + def _is_user_stream_initialized(self): + # HookOdysseyPerpetual does not have private websocket endpoints + return self._data_source.is_started() + + def _create_user_stream_tracker(self): + # HookOdysseyPerpetual does not use a tracker for the private streams + return None + + def _create_user_stream_tracker_task(self): + # HookOdysseyPerpetual does not use a tracker for the private streams + return None + + def _get_fee( + self, + base_currency: str, + quote_currency: str, + order_type: OrderType, + order_side: TradeType, + amount: Decimal, + price: Decimal = s_decimal_NaN, + is_maker: Optional[bool] = None, + ) -> TradeFeeBase: + maker_fee, taker_fee = self._data_source._fees + fee_rate = maker_fee if is_maker else taker_fee + return AddedToCostTradeFee(percent=fee_rate) + + async def _update_trading_fees(self): + """ + Update fees information from the exchange + """ + self._fees = await self._data_source.fetch_fees() + + async def _status_polling_loop_fetch_updates(self): + await safe_gather( + self._update_balances(), + self._update_positions(), + ) + pass + + async def _place_cancel(self, order_id: str, tracked_order: InFlightOrder): + hook_order_hash = tracked_order.exchange_order_id + return await self._data_source.cancel_order(hook_order_hash) + + async def _place_order( + self, + order_id: str, + trading_pair: str, + amount: Decimal, + trade_type: TradeType, + order_type: OrderType, + price: Decimal, + position_action: PositionAction = PositionAction.NIL, + **kwargs, + ) -> Tuple[str, float]: + # Get nonce from kwargs + nonce = kwargs.get("nonce") + market_hash = kwargs.get("market_hash") + instrument_hash = kwargs.get("instrument_hash") + subaccount = kwargs.get("subaccount") + return await self._data_source.place_order( + market_hash=market_hash, + instrument_hash=instrument_hash, + subaccount=subaccount, + amount=amount, + price=price, + trade_type=trade_type, + order_type=order_type, + nonce=nonce, + ) + + def buy( + self, + trading_pair: str, + amount: Decimal, + order_type=OrderType.LIMIT, + price: Decimal = s_decimal_NaN, + **kwargs, + ) -> str: + """ + Creates a promise to create a buy order using the parameters + + :param trading_pair: the token pair to operate with + :param amount: the order amount + :param order_type: the type of order to create (MARKET, LIMIT, LIMIT_MAKER) + :param price: the order price + + :return: the id assigned by the connector to the order (the client id) + """ + perpetual_pair = self._data_source.get_perpetual_pair_for_trading_pair(trading_pair) + market_hash = perpetual_pair["marketHash"] + instrument_hash = perpetual_pair["instrumentHash"] + subaccount = perpetual_pair["subaccount"] + + if order_type in [OrderType.LIMIT, OrderType.LIMIT_MAKER]: + price = self.quantize_order_price(trading_pair, price) + amount = self.quantize_order_amount(trading_pair=trading_pair, amount=amount) + nonce = get_tracking_nonce() + order_id = self._data_source.order_hash( + market_hash=market_hash, + instrument_hash=instrument_hash, + subaccount=subaccount, + amount=amount, + price=price, + order_type=order_type, + trade_type=TradeType.BUY, + nonce=nonce, + ) + safe_ensure_future( + self._create_order( + trade_type=TradeType.BUY, + order_id=order_id, + trading_pair=trading_pair, + amount=amount, + order_type=order_type, + price=price, + nonce=nonce, + market_hash=market_hash, + instrument_hash=instrument_hash, + subaccount=subaccount, + **kwargs, + ) + ) + return order_id + + def sell( + self, + trading_pair: str, + amount: Decimal, + order_type: OrderType = OrderType.LIMIT, + price: Decimal = s_decimal_NaN, + **kwargs, + ) -> str: + """ + Creates a promise to create a sell order using the parameters. + :param trading_pair: the token pair to operate with + :param amount: the order amount + :param order_type: the type of order to create (MARKET, LIMIT, LIMIT_MAKER) + :param price: the order price + :return: the id assigned by the connector to the order (the client id) + """ + perpetual_pair = self._data_source.get_perpetual_pair_for_trading_pair(trading_pair) + market_hash = perpetual_pair["marketHash"] + instrument_hash = perpetual_pair["instrumentHash"] + subaccount = perpetual_pair["subaccount"] + + if order_type in [OrderType.LIMIT, OrderType.LIMIT_MAKER]: + price = self.quantize_order_price(trading_pair, price) + amount = self.quantize_order_amount(trading_pair=trading_pair, amount=amount) + nonce = get_tracking_nonce() + order_id = self._data_source.order_hash( + market_hash=market_hash, + instrument_hash=instrument_hash, + subaccount=subaccount, + amount=amount, + price=price, + order_type=order_type, + trade_type=TradeType.SELL, + nonce=nonce, + ) + safe_ensure_future( + self._create_order( + trade_type=TradeType.SELL, + order_id=order_id, + trading_pair=trading_pair, + amount=amount, + order_type=order_type, + price=price, + nonce=nonce, + market_hash=market_hash, + instrument_hash=instrument_hash, + subaccount=subaccount, + **kwargs, + ) + ) + return order_id + + def _initialize_trading_pair_symbols_from_exchange_info(self, exchange_info: List): + raise NotImplementedError + + async def _get_last_traded_price(self, trading_pair: str) -> float: + ltp = await self._data_source.get_last_traded_price(trading_pair) + return float(ltp) + + async def _update_balances(self): + if not self._is_user_stream_initialized() or len(self._trading_pairs) == 0: + # We are unable to get the balance snapshot without the user stream running + # Additionally without trading pairs we cannot map the balances to the correct trading pairs + for collateral in CONSTANTS.SUPPORTED_COLLATERAL: + self._account_balances[collateral] = Decimal(0) + self._account_available_balances[collateral] = Decimal(0) + return + + balances = await self._data_source.get_balances() + for trading_pair in self._trading_pairs: + # Use the subaccount balance for the first found trading pair + perpetual_pair = self._data_source.get_perpetual_pair_for_trading_pair(trading_pair) + if perpetual_pair is not None: + subaccount = perpetual_pair["subaccount"] + self._account_balances[perpetual_pair["baseCurrency"]] = balances[subaccount] + self._account_available_balances[perpetual_pair["baseCurrency"]] = balances[subaccount] + break + + async def _update_positions(self): + positions = await self._data_source.get_positions() + for position in positions.values(): + if position.amount != 0: + self._perpetual_trading.set_position(position.trading_pair, position) + else: + self._perpetual_trading.remove_position(position.trading_pair) + + async def _get_position_mode(self) -> Optional[PositionMode]: + return PositionMode.ONEWAY + + async def _trading_pair_position_mode_set( + self, mode: PositionMode, trading_pair: str + ) -> Tuple[bool, str]: + msg = "" + success = True + initial_mode = await self._get_position_mode() + if initial_mode != mode: + msg = "hook odyssey only supports the ONEWAY position mode." + success = False + return success, msg + + async def _set_trading_pair_leverage( + self, trading_pair: str, leverage: int + ) -> Tuple[bool, str]: + """ + Leverage is set on a per order basis + """ + return True, "" + + async def _fetch_last_fee_payment( + self, trading_pair: str + ) -> Tuple[int, Decimal, Decimal]: + return 0, Decimal(0), Decimal(0) + + async def _all_trade_updates_for_order( + self, order: InFlightOrder + ) -> List[TradeUpdate]: + # Hook Odyssey Perpetual does not support fetching all trade updates for an order + pass + + async def _format_trading_rules( + self, exchange_info_dict: List + ) -> List[TradingRule]: + return [] + + async def _request_order_status(self, tracked_order: InFlightOrder) -> OrderUpdate: + return None + + async def _user_stream_event_listener(self): + # User stream events are managed through HookOdysseyPerpetualDataSource + pass + + async def _update_time_synchronizer( + self, pass_on_non_cancelled_error: bool = False + ): + pass + + def _configure_event_forwarders(self): + event_forwarder = EventForwarder(to_function=self._process_user_trade_update) + self._forwarders.append(event_forwarder) + self._data_source.add_listener( + event_tag=MarketEvent.TradeUpdate, listener=event_forwarder + ) + + event_forwarder = EventForwarder(to_function=self._process_user_order_update) + self._forwarders.append(event_forwarder) + self._data_source.add_listener( + event_tag=MarketEvent.OrderUpdate, listener=event_forwarder + ) + + event_forwarder = EventForwarder(to_function=self._process_balance_event) + self._forwarders.append(event_forwarder) + self._data_source.add_listener( + event_tag=AccountEvent.BalanceEvent, listener=event_forwarder + ) + + def _process_balance_event(self, event: BalanceUpdateEvent): + self._account_balances[event.asset_name] = event.total_balance + self._account_available_balances[event.asset_name] = event.available_balance + + def _process_user_order_update(self, order_update: OrderUpdate): + tracked_order = self._order_tracker.all_updatable_orders.get( + order_update.client_order_id + ) + + if tracked_order is not None: + self.logger().debug( + f"Processing order update {order_update}\nUpdatable order {tracked_order.to_json()}" + ) + order_update_to_process = OrderUpdate( + trading_pair=tracked_order.trading_pair, + update_timestamp=order_update.update_timestamp, + new_state=order_update.new_state, + client_order_id=tracked_order.client_order_id, + exchange_order_id=order_update.exchange_order_id, + ) + self._order_tracker.process_order_update( + order_update=order_update_to_process + ) + + def _process_user_trade_update(self, trade_update: TradeUpdate): + tracked_order = ( + self._order_tracker.all_fillable_orders_by_exchange_order_id.get( + trade_update.exchange_order_id + ) + ) + + if tracked_order is not None: + self.logger().debug( + f"Processing trade update {trade_update}\nFillable order {tracked_order.to_json()}" + ) + + # Fetch the current fee rates + maker_fee, taker_fee = self._data_source.get_fees() + fee_rate = maker_fee if tracked_order.is_maker else taker_fee + fee = TradeFeeBase.new_perpetual_fee( + trade_type=tracked_order.trade_type, percent=fee_rate + ) + + fill_amount = ( + trade_update.fill_base_amount - tracked_order.executed_amount_base + ) + trade_update: TradeUpdate = TradeUpdate( + trade_id=trade_update.trade_id, + client_order_id=tracked_order.client_order_id, + exchange_order_id=trade_update.exchange_order_id, + trading_pair=tracked_order.trading_pair, + fill_timestamp=trade_update.fill_timestamp, + fill_price=trade_update.fill_price, + fill_base_amount=fill_amount, + fill_quote_amount=fill_amount * trade_update.fill_price, + fee=fee, + ) + self._order_tracker.process_trade_update(trade_update) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py new file mode 100644 index 00000000000..9abb063a0bd --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py @@ -0,0 +1,298 @@ +import logging +from abc import ABC, abstractmethod +from typing import Any, Callable, Dict, Optional + +from gql import Client, gql +from gql.transport.aiohttp import AIOHTTPTransport +from gql.transport.websockets import WebsocketsTransport + +from hummingbot.connector.derivative.hook_odyssey_perpetual import hook_odyssey_perpetual_constants as CONSTANTS +from hummingbot.logger import HummingbotLogger + + +class BaseGraphQLExecutor(ABC): + @abstractmethod + async def perpetual_pairs(self) -> Dict[str, Any]: + raise NotImplementedError + + @abstractmethod + async def account_details(self) -> Dict[str, Any]: + raise NotImplementedError + + @abstractmethod + async def subscribe_ticker(self, handler: Callable, symbol: str) -> None: + raise NotImplementedError + + async def subscribe_statistics(self, handler: Callable, symbol: str) -> None: + raise NotImplementedError + + @abstractmethod + async def subscribe_orderbook( + self, handler: Callable, instrument_hash: str + ) -> None: + raise NotImplementedError + + @abstractmethod + async def subscribe_subaccount_orders( + self, handler: Callable, subaccount: str + ) -> None: + raise NotImplementedError + + @abstractmethod + async def subscribe_subaccount_balances( + self, handler: Callable, address: str + ) -> None: + raise NotImplementedError + + @abstractmethod + async def subscribe_subaccount_positions( + self, handler: Callable, address: str + ) -> None: + raise NotImplementedError + + @abstractmethod + async def place_order( + self, + order_input: Dict[str, Any], + signature: Dict[str, Any], + ) -> bool: + raise NotImplementedError + + @abstractmethod + async def cancel_order(self, order_hash: str) -> bool: + raise NotImplementedError + + +class HookOdysseyPerpetualGrapQLExecutor(BaseGraphQLExecutor): + _logger: Optional[HummingbotLogger] = None + + @classmethod + def logger(cls) -> HummingbotLogger: + if cls._logger is None: + cls._logger = logging.getLogger(HummingbotLogger.logger_name_for_class(cls)) + return cls._logger + + def __init__( + self, hook_odyssey_api_key: str, domain: Optional[str] = CONSTANTS.DOMAIN + ): + super().__init__() + self._hook_odyssey_api_key = hook_odyssey_api_key + self._domain = domain + self._client = None + + async def _execute( + self, + query: str, + variables: Optional[Dict[str, Any]] = None, + ): + headers = {"X-HOOK-API-KEY": self._hook_odyssey_api_key} + url = CONSTANTS.BASE_URLS[self._domain] + transport = AIOHTTPTransport(url=url, headers=headers) + async with Client( + transport=transport, + fetch_schema_from_transport=False, + execute_timeout=CONSTANTS.DEFAULT_TIMEOUT, + ) as session: + result = await session.execute(gql(query), variable_values=variables) + + return result + + async def _execute_subscription( + self, subscription_query: str, variables: dict[str, Any] | None = None + ): + headers = {"X-HOOK-API-KEY": self._hook_odyssey_api_key} + url = CONSTANTS.WS_URLS[self._domain] + transport = WebsocketsTransport(url=url, headers=headers) + async with Client( + transport=transport, execute_timeout=CONSTANTS.DEFAULT_TIMEOUT + ) as session: + subscription = gql(subscription_query) + async for result in session.subscribe( + subscription, variable_values=variables + ): + yield result + + async def perpetual_pairs(self) -> Dict[str, Any]: + query = """ + query PerpetualPairs { + perpetualPairs { + marketHash + instrumentHash + symbol + baseCurrency + minOrderSize + maxOrderSize + minOrderSizeIncrement + minPriceIncrement + initialMarginBips + subaccount + } + } + """ + result = await self._execute(query=query, variables={}) + return result + + async def account_details(self) -> Dict[str, Any]: + query = """ + query AccountDetails { + accountDetails { + makerFeeBips + takerFeeBips + } + } + """ + result = await self._execute(query=query, variables={}) + return result + + async def subscribe_ticker(self, handler: Callable, symbol: str): + query = """ + subscription onTickerEvent($symbol: String!) { + ticker(symbol: $symbol) { + price + timestamp + } + } + """ + variables = {"symbol": symbol} + async for event in self._execute_subscription( + subscription_query=query, variables=variables + ): + await handler(event=event, symbol=symbol) + + async def subscribe_statistics(self, handler: Callable, symbol: str): + query = """ + subscription onStatisticsEvent($symbol: String!) { + statistics(symbol: $symbol) { + eventType + timestamp + markPrice + fundingRateBips + nextFundingEpoch + } + } + """ + variables = {"symbol": symbol} + async for event in self._execute_subscription( + subscription_query=query, variables=variables + ): + await handler(event=event, symbol=symbol) + + async def subscribe_orderbook(self, handler: Callable, instrument_hash: str): + query = """ + subscription onOrderbookEvent($instrumentHash: ID!) { + orderbook(instrumentHash: $instrumentHash) { + eventType + timestamp + bidLevels { + direction + size + price + } + askLevels { + direction + size + price + } + } + } + """ + variables = {"instrumentHash": instrument_hash} + async for event in self._execute_subscription( + subscription_query=query, variables=variables + ): + await handler(event=event, instrument_hash=instrument_hash) + + async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str): + query = """ + subscription onSubaccountOrderEvent($subaccount: BigInt!) { + subaccountOrders(subaccount: $subaccount) { + eventType + orders { + instrument { + id + } + direction + size + remainingSize + orderHash + status + } + } + } + """ + variables = {"subaccount": subaccount} + async for event in self._execute_subscription( + subscription_query=query, variables=variables + ): + await handler(event=event, subaccount=subaccount) + + async def subscribe_subaccount_balances(self, handler: Callable, address: str): + query = """ + subscription onSubaccountBalanceEvent($address: Address!) { + subaccountBalances(address: $address) { + eventType + balances { + subaccount + subaccountID + balance + } + } + } + """ + variables = {"address": address} + async for event in self._execute_subscription( + subscription_query=query, variables=variables + ): + await handler(event=event, address=address) + + async def subscribe_subaccount_positions(self, handler: Callable, address: str): + query = """ + subscription onSubaccountPositionEvent($address: Address!) { + subaccountPositions(address: $address) { + eventType + positions { + instrument { + id + } + subaccount + marketHash + size + } + } + } + """ + variables = {"address": address} + async for event in self._execute_subscription( + subscription_query=query, variables=variables + ): + await handler(event=event, address=address) + + async def place_order( + self, + order_input: Dict[str, Any], + signature: Dict[str, Any], + ) -> bool: + mutation = """ + mutation PlaceOrder( + $orderInput: PlaceOrderInput! + $signature: SignatureInput! + ) { + placeOrderV2( + orderInput: $orderInput + signature: $signature + ) + } + """ + variables = {"orderInput": order_input, "signature": signature} + result = await self._execute(query=mutation, variables=variables) + return result.get("placeOrderV2", False) + + async def cancel_order(self, order_hash: str) -> bool: + mutation = """ + mutation CancelOrder($orderHash: String!) { + cancelOrderV2(orderHash: $orderHash) + } + """ + variables = {"orderHash": order_hash} + result = await self._execute(query=mutation, variables=variables) + return result.get("cancelOrderV2", False) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_signing.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_signing.py new file mode 100644 index 00000000000..ad82d336eec --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_signing.py @@ -0,0 +1,104 @@ +from typing import Tuple + +import sha3 +from coincurve import PrivateKey +from eip712_structs import Address, Boolean, Bytes, EIP712Struct, String, Uint, make_domain +from eth_utils import big_endian_to_int + +import hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_constants as CONSTANTS + + +class EIP712Domain(EIP712Struct): + name = String() + version = String() + chainId = Uint(256) + verifyingContract = Address() + + +class Order(EIP712Struct): + """ + Order typehash + "Order(bytes32 market,uint8 instrumentType,bytes32 instrumentId,uint8 direction,uint256 maker,uint256 taker,uint256 amount,uint256 limitPrice,uint256 expiration,uint256 nonce,uint256 counter,bool postOnly,bool reduceOnly,bool allOrNothing)" + """ + + market = Bytes(32) + instrumentType = Uint(8) # 0 = Call, 1 = Put, 2 = Perpetual + instrumentId = Bytes(32) + direction = Uint(8) # 0 = Buy, 1 = Sell + maker = Uint(256) + taker = Uint(256) + amount = Uint(256) + limitPrice = Uint(256) + expiration = Uint(256) + nonce = Uint(256) + counter = Uint(256) + postOnly = Boolean() + reduceOnly = Boolean() + allOrNothing = Boolean() + + +def keccak_hash(x): + return sha3.keccak_256(x).digest() + + +class HookOdysseyPerpetualSigner: + def __init__(self, private_key: str, domain: str = CONSTANTS.DOMAIN): + self.private_key = private_key + self.domain = domain + self.contract_address = CONSTANTS.CONTRACT_ADDRESSES[self.domain] + self.chain_id = CONSTANTS.CHAIN_IDS[self.domain] + + def compute_order_hash(self, order: Order) -> str: + """ + Gets the eip 712 hash of an Order for use as a client order id + + :param order: the order to hash + + :return: a string hex of the keccak_256 of the signable_bytes of the order + """ + domain = make_domain( + name=CONSTANTS.EIP712_DOMAIN_NAME, + version=CONSTANTS.EIP712_DOMAIN_VERSION, + chainId=self.chain_id, + verifyingContract=self.contract_address, + ) + signable_bytes = order.signable_bytes(domain) + return self.generate_digest(signable_bytes) + + def sign_order(self, order: Order) -> Tuple[str, str]: + """ + Signs the order using the private key and returns the signature and order hash (digest) + + :param order: the order using EIP712 structure for signing + + :return: a tuple with both a string hex of the signature of the order payload and order hash (digest) + """ + domain = make_domain( + name=CONSTANTS.EIP712_DOMAIN_NAME, + version=CONSTANTS.EIP712_DOMAIN_VERSION, + chainId=self.chain_id, + verifyingContract=self.contract_address, + ) + signable_bytes = order.signable_bytes(domain) + # Digest for order tracking in Hummingbot + digest = self.generate_digest(signable_bytes) + + pk = PrivateKey.from_hex(self.private_key) + signature = pk.sign_recoverable(signable_bytes, hasher=keccak_hash) + + v = signature[64] + 27 + r = big_endian_to_int(signature[0:32]) + s = big_endian_to_int(signature[32:64]) + + final_sig = r.to_bytes(32, "big") + s.to_bytes(32, "big") + v.to_bytes(1, "big") + return f"0x{final_sig.hex()}", digest + + def generate_digest(self, signable_bytes: bytearray) -> str: + """ + Generates the digest of the payload for use across Vetext lookups + + :param signable_bytes: the bytes of the payload + + :return: a string hex of the keccak_256 of the signable_bytes of the payload + """ + return f"0x{keccak_hash(signable_bytes).hex()}" diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py new file mode 100644 index 00000000000..a5d176bd502 --- /dev/null +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py @@ -0,0 +1,109 @@ +from decimal import Decimal + +from pydantic import Field, SecretStr + +from hummingbot.client.config.config_data_types import BaseConnectorConfigMap, ClientFieldData +from hummingbot.core.data_type.trade_fee import TradeFeeSchema +from hummingbot.core.utils.tracking_nonce import NonceCreator + +CENTRALIZED = True +EXAMPLE_PAIR = "MILADY-ETH" + +DEFAULT_FEES = TradeFeeSchema( + maker_percent_fee_decimal=Decimal("0.001"), # 10 bps + taker_percent_fee_decimal=Decimal("0.0025"), # 25 bps + buy_percent_fee_deducted_from_returns=True, +) + +_nonce_provider = NonceCreator.for_microseconds() + + +class HookOdysseyPerpetualConfigMap(BaseConnectorConfigMap): + connector: str = Field( + default="hook_odyssey_perpetual", const=True, client_data=None + ) + hook_odyssey_perpetual_eth_address: str = Field( + default=..., + client_data=ClientFieldData( + prompt=lambda cm: "Enter your Ethereum address", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_perpetual_private_key: SecretStr = Field( + default=..., + client_data=ClientFieldData( + prompt=lambda cm: "Enter your Ethereum private key", + is_secure=True, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_api_key: SecretStr = Field( + default=..., + client_data=ClientFieldData( + prompt=lambda cm: "Enter your Hook Odyssey API key", + is_secure=True, + is_connect_key=True, + prompt_on_new=True, + ), + ) + + class Config: + title = "hook_odyssey_perpetual" + + +KEYS = HookOdysseyPerpetualConfigMap.construct() + +OTHER_DOMAINS = ["hook_odyssey_perpetual_testnet"] +OTHER_DOMAINS_PARAMETER = { + "hook_odyssey_perpetual_testnet": "hook_odyssey_perpetual_testnet" +} +OTHER_DOMAINS_EXAMPLE_PAIR = {"hook_odyssey_perpetual_testnet": EXAMPLE_PAIR} +OTHER_DOMAINS_DEFAULT_FEES = {"hook_odyssey_perpetual_testnet": DEFAULT_FEES} + + +class HookOdysseyPerpetualTestnetConfigMap(BaseConnectorConfigMap): + connector: str = Field( + default="hook_odyssey_perpetual_testnet", const=True, client_data=None + ) + hook_odyssey_perpetual_testnet_public_key: SecretStr = Field( + default=..., + client_data=ClientFieldData( + prompt=lambda cm: "Enter your Ethereum wallet public key", + is_secure=True, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_perpetual_testnet_private_key: SecretStr = Field( + default=..., + client_data=ClientFieldData( + prompt=lambda cm: "Enter your Ethereum wallet private key", + is_secure=True, + is_connect_key=True, + prompt_on_new=True, + ), + ) + + class Config: + title = "hook_odyssey_perpetual_testnet" + + +OTHER_DOMAINS_KEYS = { + "hook_odyssey_perpetual_testnet": HookOdysseyPerpetualTestnetConfigMap.construct() +} + + +def eth_to_wei(eth_amount: Decimal) -> int: + return int(eth_amount * Decimal("1e18")) + + +def wei_to_eth(wei_amount: str | int | Decimal) -> Decimal: + return Decimal(wei_amount) / Decimal("1e18") + + +def get_tracking_nonce() -> int: + nonce = _nonce_provider.get_tracking_nonce() + return nonce diff --git a/hummingbot/connector/exchange/polkadex/polkadex_query_executor.py b/hummingbot/connector/exchange/polkadex/polkadex_query_executor.py index 39be3d7e3ef..3a2f3a1f1cf 100644 --- a/hummingbot/connector/exchange/polkadex/polkadex_query_executor.py +++ b/hummingbot/connector/exchange/polkadex/polkadex_query_executor.py @@ -97,7 +97,7 @@ def logger(cls) -> HummingbotLogger: cls._logger = logging.getLogger(HummingbotLogger.logger_name_for_class(cls)) return cls._logger - def __init__(self, auth: AppSyncAuthentication, domain: Optional[str] = CONSTANTS.DEFAULT_DOMAIN): + def __init__(self, auth: AppSyncAuthentication, domain: Optional[str] = CONSTANTS.DOMAIN): super().__init__() self._auth = auth self._domain = domain diff --git a/hummingbot/connector/exchange/vertex/vertex_exchange.py b/hummingbot/connector/exchange/vertex/vertex_exchange.py index 4250eab3c59..05bdd14db35 100644 --- a/hummingbot/connector/exchange/vertex/vertex_exchange.py +++ b/hummingbot/connector/exchange/vertex/vertex_exchange.py @@ -41,7 +41,7 @@ def __init__( vertex_arbitrum_private_key: str, trading_pairs: Optional[List[str]] = None, trading_required: bool = True, - domain: str = CONSTANTS.DEFAULT_DOMAIN, + domain: str = CONSTANTS.DOMAIN, ): self.sender_address = utils.convert_address_to_sender(vertex_arbitrum_address) self.private_key = vertex_arbitrum_private_key From 3f207e069cb55ebabc70b818b667075309bda4b5 Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Sun, 4 Feb 2024 21:02:59 -0500 Subject: [PATCH 2/8] Get mark price from bbo subscription --- ...ey_perpetual_api_order_book_data_source.py | 2 +- .../hook_odyssey_perpetual_data_source.py | 129 ++++++++---------- .../hook_odyssey_perpetual_derivative.py | 2 +- ...hook_odyssey_perpetual_graphql_executor.py | 78 +++++------ 4 files changed, 97 insertions(+), 114 deletions(-) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py index 4efaffbf2ca..8cd372d2d81 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py @@ -75,7 +75,7 @@ async def _parse_order_book_diff_message( message_queue.put_nowait(raw_message) async def get_funding_info(self, trading_pair: str) -> FundingInfo: - index_price = await self._data_source.get_last_traded_price(trading_pair) + index_price = await self._data_source.get_index_price(trading_pair) funding_info = await self._data_source.get_funding_info(trading_pair) funding_info.index_price = Decimal(index_price) return funding_info diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py index 1940757f006..ca3a1969109 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py @@ -59,37 +59,22 @@ def __init__( self._graphql_executor = HookOdysseyPerpetualGrapQLExecutor( hook_odyssey_api_key=hook_odyssey_api_key, domain=self._domain ) - self._signer = HookOdysseyPerpetualSigner( - private_key=hook_odyssey_perpetual_private_key, domain=self._domain - ) + self._signer = HookOdysseyPerpetualSigner(private_key=hook_odyssey_perpetual_private_key, domain=self._domain) self._publisher = PubSub() self._events_listening_tasks = [] self._assets_map: Dict[str, str] = {} # Data source state - used to provide subscription data at any time - self._perpetual_pairs: Dict[ - str, Dict[str, Any] - ] = {} # Mapping of trading pair to perpetual pair + self._perpetual_pairs: Dict[str, Dict[str, Any]] = {} # Mapping of trading pair to perpetual pair self._symbols: Dict[str, str] = {} # Mapping of symbol to trading pair - self._instrument_hashes: Dict[ - str, str - ] = {} # Mapping of instrument hash to trading pair + self._instrument_hashes: Dict[str, str] = {} # Mapping of instrument hash to trading pair self._subaccounts: Dict[str, str] = {} # Mapping of subaccount to trading pair - self._last_traded_price: Dict[ - str, Decimal - ] = {} # Mapping of trading pair to last traded price - self._subaccount_balances: Dict[ - str, Decimal - ] = {} # Mapping of subaccount to balance - self._subaccount_positions: Dict[ - str, Position - ] = {} # Mapping of subaccount to position - self._funding_info: Dict[ - str, FundingInfo - ] = {} # Mapping of trading pair to funding info - self._orderbook_snapshots: Dict[ - str, OrderBookMessage - ] = {} # Mapping of trading pair to orderbook snapshot + self._index_prices: Dict[str, Decimal] = {} # Mapping of trading pair to index price + self._mark_prices: Dict[str, Decimal] = {} # Mapping of trading pair to mark price + self._subaccount_balances: Dict[str, Decimal] = {} # Mapping of subaccount to balance + self._subaccount_positions: Dict[str, Position] = {} # Mapping of subaccount to position + self._funding_info: Dict[str, FundingInfo] = {} # Mapping of trading pair to funding info + self._orderbook_snapshots: Dict[str, OrderBookMessage] = {} # Mapping of trading pair to orderbook snapshot self._fees: Tuple[Decimal, Decimal] = ( DEFAULT_FEES.maker_percent_fee_decimal, DEFAULT_FEES.taker_percent_fee_decimal, @@ -97,6 +82,7 @@ def __init__( self._initial_ticker_event = asyncio.Event() self._initial_statistics_event = asyncio.Event() + self._initial_bbo_event = asyncio.Event() self._initial_orderbook_event = asyncio.Event() self._initial_subaccount_orders_event = asyncio.Event() self._initial_subaccount_balances_event = asyncio.Event() @@ -104,9 +90,7 @@ def __init__( async def start(self, trading_pairs: List[str]): if len(self._events_listening_tasks) > 0: - raise AssertionError( - "Data source is already started and can't be started again" - ) + raise AssertionError("Data source is already started and can't be started again") await self.get_supported_pairs() for trading_pair in trading_pairs: @@ -133,6 +117,15 @@ async def start(self, trading_pairs: List[str]): ) ) ) + # BBO - Top of book + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_bbo( + self._process_bbo_update, + symbol, + ) + ) + ) # Orderbook self._events_listening_tasks.append( asyncio.create_task( @@ -220,16 +213,19 @@ async def get_supported_pairs(self) -> List[Dict[str, Any]]: self._subaccounts[perpetual_pair["subaccount"]] = trading_pair return perpetual_pairs - def get_perpetual_pair_for_trading_pair( - self, trading_pair: str - ) -> Optional[Dict[str, Any]]: + def get_perpetual_pair_for_trading_pair(self, trading_pair: str) -> Optional[Dict[str, Any]]: """Retrieves perpetual pair details for a specific trading pair.""" return self._perpetual_pairs.get(trading_pair, None) - async def get_last_traded_price(self, trading_pair: str) -> Decimal: + async def get_index_price(self, trading_pair: str) -> Decimal: # Wait for the trade subscription snapshot to come in await self._initial_ticker_event.wait() - return self._last_traded_price.get(trading_pair, 0.0) + return self._index_prices.get(trading_pair, 0.0) + + async def get_mark_price(self, trading_pair: str) -> Decimal: + # Wait for the bbo subscription snapshot to come in + await self._initial_bbo_event.wait() + return self._mark_prices.get(trading_pair, 0.0) async def get_funding_info(self, trading_pair: str) -> FundingInfo: # Wait for the funding info subscription snapshot to come in @@ -281,32 +277,45 @@ async def _process_ticker_update(self, event: Dict[str, Any], symbol: str): if symbol not in self._symbols: return trading_pair = self._symbols[symbol] - self._last_traded_price[trading_pair] = wei_to_eth(ticker_event["price"]) - return self._last_traded_price[trading_pair] + self._index_prices[trading_pair] = wei_to_eth(ticker_event["price"]) + return self._index_prices[trading_pair] async def _process_statistics_update(self, event: Dict[str, Any], symbol: str): statistics_event = event["statistics"] - if ( - statistics_event["eventType"] == "SNAPSHOT" - and not self._initial_statistics_event.is_set() - ): + if statistics_event["eventType"] == "SNAPSHOT" and not self._initial_statistics_event.is_set(): self._initial_statistics_event.set() if symbol not in self._symbols: return trading_pair = self._symbols[symbol] + + # Get index price + index_price = await self.get_index_price(trading_pair) + mark_price = await self.get_mark_price(trading_pair) self._funding_info[trading_pair] = FundingInfo( trading_pair=trading_pair, - index_price=Decimal(0), - mark_price=Decimal(wei_to_eth(statistics_event["markPrice"])), + index_price=index_price, + mark_price=mark_price, next_funding_utc_timestamp=int(statistics_event["nextFundingEpoch"]), rate=Decimal(statistics_event["fundingRateBips"]) / Decimal(10000), ) - async def _process_orderbook_update( - self, event: Dict[str, Any], instrument_hash: str - ): + async def _process_bbo_update(self, event: Dict[str, Any], symbol: str): + bbo_event = event["bbo"] + + if bbo_event["eventType"] == "SNAPSHOT" and not self._initial_bbo_event.is_set(): + self._initial_bbo_event.set() + + if symbol not in self._symbols: + return + + for instrument in bbo_event["instruments"]: + instrument_hash = instrument["id"] + trading_pair = self._instrument_hashes[instrument_hash] + self._mark_prices[trading_pair] = wei_to_eth(instrument["markPrice"]) + + async def _process_orderbook_update(self, event: Dict[str, Any], instrument_hash: str): if instrument_hash not in self._instrument_hashes: return trading_pair = self._instrument_hashes[instrument_hash] @@ -352,9 +361,7 @@ async def _process_orderbook_update( message=order_book_message, ) - async def _process_subaccount_orders_update( - self, event: Dict[str, Any], subaccount: str - ): + async def _process_subaccount_orders_update(self, event: Dict[str, Any], subaccount: str): orders_event = event["subaccountOrders"] if orders_event["eventType"] == "SNAPSHOT": @@ -378,9 +385,7 @@ async def _process_subaccount_orders_update( client_order_id=order["orderHash"], exchange_order_id=order["orderHash"], ) - self._publisher.trigger_event( - event_tag=MarketEvent.OrderUpdate, message=order_update - ) + self._publisher.trigger_event(event_tag=MarketEvent.OrderUpdate, message=order_update) # Update trade if order_state == "FILLED" or order_state == "PARTIALLY_FILLED": @@ -388,15 +393,11 @@ async def _process_subaccount_orders_update( remaining_size = wei_to_eth(order["remainingSize"]) fill_amount = size - remaining_size fill_price = wei_to_eth(order["limitPrice"]) - trade_type = ( - TradeType.BUY if order["direction"] == "BUY" else TradeType.SELL - ) + trade_type = TradeType.BUY if order["direction"] == "BUY" else TradeType.SELL maker_fee, taker_fee = self.get_fees() fee_rate = maker_fee if order["orderType"] == "MARKET" else taker_fee - fee = TradeFeeBase.new_perpetual_fee( - trade_type=trade_type, percent=fee_rate - ) + fee = TradeFeeBase.new_perpetual_fee(trade_type=trade_type, percent=fee_rate) trade_update = TradeUpdate( trade_id=f"{order['orderHash']}-{order['timestamp']}", @@ -409,33 +410,23 @@ async def _process_subaccount_orders_update( fill_quote_amount=fill_price * fill_amount, fee=fee, ) - self._publisher.trigger_event( - event_tag=MarketEvent.TradeUpdate, message=trade_update - ) + self._publisher.trigger_event(event_tag=MarketEvent.TradeUpdate, message=trade_update) async def _process_subaccount_balances(self, event: Dict[str, Any], address: str): balances_event = event["subaccountBalances"] - if ( - balances_event["eventType"] == "SNAPSHOT" - and not self._initial_subaccount_balances_event.is_set() - ): + if balances_event["eventType"] == "SNAPSHOT" and not self._initial_subaccount_balances_event.is_set(): self._initial_subaccount_balances_event.set() for balance in balances_event["balances"]: # Ignore primary account balances if balance["subaccountID"] != 0: - self._subaccount_balances[balance["subaccount"]] = wei_to_eth( - balance["balance"] - ) + self._subaccount_balances[balance["subaccount"]] = wei_to_eth(balance["balance"]) async def _process_subaccount_positions(self, event: Dict[str, Any], address: str): positions_event = event["subaccountPositions"] - if ( - positions_event["eventType"] == "SNAPSHOT" - and not self._initial_subaccount_positions_event.is_set() - ): + if positions_event["eventType"] == "SNAPSHOT" and not self._initial_subaccount_positions_event.is_set(): self._initial_subaccount_positions_event.set() for position in positions_event["positions"]: diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py index 282dc3420b7..9a1a4d1bb8e 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py @@ -406,7 +406,7 @@ def _initialize_trading_pair_symbols_from_exchange_info(self, exchange_info: Lis raise NotImplementedError async def _get_last_traded_price(self, trading_pair: str) -> float: - ltp = await self._data_source.get_last_traded_price(trading_pair) + ltp = await self._data_source.get_index_price(trading_pair) return float(ltp) async def _update_balances(self): diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py index 9abb063a0bd..4468447eec4 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py @@ -27,27 +27,23 @@ async def subscribe_statistics(self, handler: Callable, symbol: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_orderbook( - self, handler: Callable, instrument_hash: str - ) -> None: + async def subscribe_bbo(self, handler: Callable, symbol: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_subaccount_orders( - self, handler: Callable, subaccount: str - ) -> None: + async def subscribe_orderbook(self, handler: Callable, instrument_hash: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_subaccount_balances( - self, handler: Callable, address: str - ) -> None: + async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_subaccount_positions( - self, handler: Callable, address: str - ) -> None: + async def subscribe_subaccount_balances(self, handler: Callable, address: str) -> None: + raise NotImplementedError + + @abstractmethod + async def subscribe_subaccount_positions(self, handler: Callable, address: str) -> None: raise NotImplementedError @abstractmethod @@ -72,9 +68,7 @@ def logger(cls) -> HummingbotLogger: cls._logger = logging.getLogger(HummingbotLogger.logger_name_for_class(cls)) return cls._logger - def __init__( - self, hook_odyssey_api_key: str, domain: Optional[str] = CONSTANTS.DOMAIN - ): + def __init__(self, hook_odyssey_api_key: str, domain: Optional[str] = CONSTANTS.DOMAIN): super().__init__() self._hook_odyssey_api_key = hook_odyssey_api_key self._domain = domain @@ -97,19 +91,13 @@ async def _execute( return result - async def _execute_subscription( - self, subscription_query: str, variables: dict[str, Any] | None = None - ): + async def _execute_subscription(self, subscription_query: str, variables: dict[str, Any] | None = None): headers = {"X-HOOK-API-KEY": self._hook_odyssey_api_key} url = CONSTANTS.WS_URLS[self._domain] transport = WebsocketsTransport(url=url, headers=headers) - async with Client( - transport=transport, execute_timeout=CONSTANTS.DEFAULT_TIMEOUT - ) as session: + async with Client(transport=transport, execute_timeout=CONSTANTS.DEFAULT_TIMEOUT) as session: subscription = gql(subscription_query) - async for result in session.subscribe( - subscription, variable_values=variables - ): + async for result in session.subscribe(subscription, variable_values=variables): yield result async def perpetual_pairs(self) -> Dict[str, Any]: @@ -154,9 +142,7 @@ async def subscribe_ticker(self, handler: Callable, symbol: str): } """ variables = {"symbol": symbol} - async for event in self._execute_subscription( - subscription_query=query, variables=variables - ): + async for event in self._execute_subscription(subscription_query=query, variables=variables): await handler(event=event, symbol=symbol) async def subscribe_statistics(self, handler: Callable, symbol: str): @@ -165,16 +151,30 @@ async def subscribe_statistics(self, handler: Callable, symbol: str): statistics(symbol: $symbol) { eventType timestamp - markPrice fundingRateBips nextFundingEpoch } } """ variables = {"symbol": symbol} - async for event in self._execute_subscription( - subscription_query=query, variables=variables - ): + async for event in self._execute_subscription(subscription_query=query, variables=variables): + await handler(event=event, symbol=symbol) + + async def subscribe_bbo(self, handler: Callable, symbol: str) -> None: + query = """ + subscription onBboEvent($symbol: String!, $instrumentType: InstrumentType!) { + bbo(symbol: $symbol, instrumentType: $instrumentType) { + eventType + timestamp + instruments { + id + markPrice + } + } + } + """ + variables = {"symbol": symbol, "instrumentType": "PERPETUAL"} + async for event in self._execute_subscription(subscription_query=query, variables=variables): await handler(event=event, symbol=symbol) async def subscribe_orderbook(self, handler: Callable, instrument_hash: str): @@ -197,9 +197,7 @@ async def subscribe_orderbook(self, handler: Callable, instrument_hash: str): } """ variables = {"instrumentHash": instrument_hash} - async for event in self._execute_subscription( - subscription_query=query, variables=variables - ): + async for event in self._execute_subscription(subscription_query=query, variables=variables): await handler(event=event, instrument_hash=instrument_hash) async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str): @@ -221,9 +219,7 @@ async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str): } """ variables = {"subaccount": subaccount} - async for event in self._execute_subscription( - subscription_query=query, variables=variables - ): + async for event in self._execute_subscription(subscription_query=query, variables=variables): await handler(event=event, subaccount=subaccount) async def subscribe_subaccount_balances(self, handler: Callable, address: str): @@ -240,9 +236,7 @@ async def subscribe_subaccount_balances(self, handler: Callable, address: str): } """ variables = {"address": address} - async for event in self._execute_subscription( - subscription_query=query, variables=variables - ): + async for event in self._execute_subscription(subscription_query=query, variables=variables): await handler(event=event, address=address) async def subscribe_subaccount_positions(self, handler: Callable, address: str): @@ -262,9 +256,7 @@ async def subscribe_subaccount_positions(self, handler: Callable, address: str): } """ variables = {"address": address} - async for event in self._execute_subscription( - subscription_query=query, variables=variables - ): + async for event in self._execute_subscription(subscription_query=query, variables=variables): await handler(event=event, address=address) async def place_order( From 33a5fc3a5c59dddb61423e1b17bca27643cd342c Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Sun, 4 Feb 2024 21:04:03 -0500 Subject: [PATCH 3/8] Format with 120 char line len --- ...ey_perpetual_api_order_book_data_source.py | 40 +++------- .../hook_odyssey_perpetual_derivative.py | 77 +++++-------------- .../hook_odyssey_perpetual_utils.py | 16 +--- 3 files changed, 33 insertions(+), 100 deletions(-) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py index 8cd372d2d81..f3153ed2fb6 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py @@ -36,18 +36,14 @@ def __init__( self._forwarders = [] self._configure_event_forwarders() - async def get_last_traded_prices( - self, trading_pairs: List[str], domain: Optional[str] = None - ) -> Dict[str, float]: + async def get_last_traded_prices(self, trading_pairs: List[str], domain: Optional[str] = None) -> Dict[str, float]: return await self._connector.get_last_traded_prices(trading_pairs=trading_pairs) async def listen_for_subscriptions(self): # The socket reconnection is managed by the data_source. This method should do nothing pass - async def _parse_order_book_snapshot_message( - self, raw_message: OrderBookMessage, message_queue: asyncio.Queue - ): + async def _parse_order_book_snapshot_message(self, raw_message: OrderBookMessage, message_queue: asyncio.Queue): # In HookOdysseyPerpetual, 'raw_message' is the OrderBookMessage from the data source message_queue.put_nowait(raw_message) @@ -62,15 +58,11 @@ async def _subscribe_channels(self, ws: WSAssistant): async def _order_book_snapshot(self, trading_pair: str) -> OrderBookMessage: return await self._data_source.get_order_book_snapshot(trading_pair) - async def _parse_trade_message( - self, raw_message: OrderBookMessage, message_queue: asyncio.Queue - ): + async def _parse_trade_message(self, raw_message: OrderBookMessage, message_queue: asyncio.Queue): # In HookOdysseyPerpetual, 'raw_message' is the OrderBookMessage from the data source message_queue.put_nowait(raw_message) - async def _parse_order_book_diff_message( - self, raw_message: OrderBookMessage, message_queue: asyncio.Queue - ): + async def _parse_order_book_diff_message(self, raw_message: OrderBookMessage, message_queue: asyncio.Queue): # In HookOdysseyPerpetual, 'raw_message' is the OrderBookMessage from the data source message_queue.put_nowait(raw_message) @@ -80,9 +72,7 @@ async def get_funding_info(self, trading_pair: str) -> FundingInfo: funding_info.index_price = Decimal(index_price) return funding_info - async def _parse_funding_info_message( - self, raw_message: Dict[str, Any], message_queue: asyncio.Queue - ): + async def _parse_funding_info_message(self, raw_message: Dict[str, Any], message_queue: asyncio.Queue): """ Placeholder method for parsing funding info messages. Actual implementation needed. """ @@ -90,9 +80,7 @@ async def _parse_funding_info_message( def _configure_event_forwarders(self): # Forward order book diffs - order_book_event_forwarder = EventForwarder( - to_function=self._process_order_book_event - ) + order_book_event_forwarder = EventForwarder(to_function=self._process_order_book_event) self._forwarders.append(order_book_event_forwarder) self._data_source.add_listener( event_tag=OrderBookEvent.OrderBookDataSourceUpdateEvent, @@ -100,23 +88,15 @@ def _configure_event_forwarders(self): ) # Forward trade updates - trade_event_forwarder = EventForwarder( - to_function=self._process_public_trade_event - ) + trade_event_forwarder = EventForwarder(to_function=self._process_public_trade_event) self._forwarders.append(trade_event_forwarder) - self._data_source.add_listener( - event_tag=OrderBookEvent.TradeEvent, listener=trade_event_forwarder - ) + self._data_source.add_listener(event_tag=OrderBookEvent.TradeEvent, listener=trade_event_forwarder) def _process_order_book_event(self, order_book_diff: OrderBookMessage): if order_book_diff.type == OrderBookMessageType.SNAPSHOT: - self._message_queue[self._snapshot_messages_queue_key].put_nowait( - order_book_diff - ) + self._message_queue[self._snapshot_messages_queue_key].put_nowait(order_book_diff) else: - self._message_queue[self._diff_messages_queue_key].put_nowait( - order_book_diff - ) + self._message_queue[self._diff_messages_queue_key].put_nowait(order_book_diff) def _process_public_trade_event(self, trade_update: OrderBookMessage): self._message_queue[self._trade_messages_queue_key].put_nowait(trade_update) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py index 9a1a4d1bb8e..0c4ea93633a 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py @@ -121,11 +121,8 @@ def status_dict(self) -> Dict[str, bool]: return { "symbols_mapping_initialized": self.trading_pair_symbol_map_ready(), "order_books_initialized": self.order_book_tracker.ready, - "account_balance": not self.is_trading_required - or len(self._account_balances) > 0, - "trading_rule_initialized": len(self._trading_rules) > 0 - if self.is_trading_required - else True, + "account_balance": not self.is_trading_required or len(self._account_balances) > 0, + "trading_rule_initialized": len(self._trading_rules) > 0 if self.is_trading_required else True, } async def start_network(self): @@ -175,9 +172,7 @@ def get_sell_collateral_token(self, trading_pair: str) -> str: trading_rule: TradingRule = self._trading_rules[trading_pair] return trading_rule.sell_order_collateral_token - def _is_request_exception_related_to_time_synchronizer( - self, request_exception: Exception - ): + def _is_request_exception_related_to_time_synchronizer(self, request_exception: Exception): return False def _is_order_not_found_during_status_update_error(self, status_update_exception: Exception) -> bool: @@ -439,9 +434,7 @@ async def _update_positions(self): async def _get_position_mode(self) -> Optional[PositionMode]: return PositionMode.ONEWAY - async def _trading_pair_position_mode_set( - self, mode: PositionMode, trading_pair: str - ) -> Tuple[bool, str]: + async def _trading_pair_position_mode_set(self, mode: PositionMode, trading_pair: str) -> Tuple[bool, str]: msg = "" success = True initial_mode = await self._get_position_mode() @@ -450,28 +443,20 @@ async def _trading_pair_position_mode_set( success = False return success, msg - async def _set_trading_pair_leverage( - self, trading_pair: str, leverage: int - ) -> Tuple[bool, str]: + async def _set_trading_pair_leverage(self, trading_pair: str, leverage: int) -> Tuple[bool, str]: """ Leverage is set on a per order basis """ return True, "" - async def _fetch_last_fee_payment( - self, trading_pair: str - ) -> Tuple[int, Decimal, Decimal]: + async def _fetch_last_fee_payment(self, trading_pair: str) -> Tuple[int, Decimal, Decimal]: return 0, Decimal(0), Decimal(0) - async def _all_trade_updates_for_order( - self, order: InFlightOrder - ) -> List[TradeUpdate]: + async def _all_trade_updates_for_order(self, order: InFlightOrder) -> List[TradeUpdate]: # Hook Odyssey Perpetual does not support fetching all trade updates for an order pass - async def _format_trading_rules( - self, exchange_info_dict: List - ) -> List[TradingRule]: + async def _format_trading_rules(self, exchange_info_dict: List) -> List[TradingRule]: return [] async def _request_order_status(self, tracked_order: InFlightOrder) -> OrderUpdate: @@ -481,43 +466,31 @@ async def _user_stream_event_listener(self): # User stream events are managed through HookOdysseyPerpetualDataSource pass - async def _update_time_synchronizer( - self, pass_on_non_cancelled_error: bool = False - ): + async def _update_time_synchronizer(self, pass_on_non_cancelled_error: bool = False): pass def _configure_event_forwarders(self): event_forwarder = EventForwarder(to_function=self._process_user_trade_update) self._forwarders.append(event_forwarder) - self._data_source.add_listener( - event_tag=MarketEvent.TradeUpdate, listener=event_forwarder - ) + self._data_source.add_listener(event_tag=MarketEvent.TradeUpdate, listener=event_forwarder) event_forwarder = EventForwarder(to_function=self._process_user_order_update) self._forwarders.append(event_forwarder) - self._data_source.add_listener( - event_tag=MarketEvent.OrderUpdate, listener=event_forwarder - ) + self._data_source.add_listener(event_tag=MarketEvent.OrderUpdate, listener=event_forwarder) event_forwarder = EventForwarder(to_function=self._process_balance_event) self._forwarders.append(event_forwarder) - self._data_source.add_listener( - event_tag=AccountEvent.BalanceEvent, listener=event_forwarder - ) + self._data_source.add_listener(event_tag=AccountEvent.BalanceEvent, listener=event_forwarder) def _process_balance_event(self, event: BalanceUpdateEvent): self._account_balances[event.asset_name] = event.total_balance self._account_available_balances[event.asset_name] = event.available_balance def _process_user_order_update(self, order_update: OrderUpdate): - tracked_order = self._order_tracker.all_updatable_orders.get( - order_update.client_order_id - ) + tracked_order = self._order_tracker.all_updatable_orders.get(order_update.client_order_id) if tracked_order is not None: - self.logger().debug( - f"Processing order update {order_update}\nUpdatable order {tracked_order.to_json()}" - ) + self.logger().debug(f"Processing order update {order_update}\nUpdatable order {tracked_order.to_json()}") order_update_to_process = OrderUpdate( trading_pair=tracked_order.trading_pair, update_timestamp=order_update.update_timestamp, @@ -525,32 +498,20 @@ def _process_user_order_update(self, order_update: OrderUpdate): client_order_id=tracked_order.client_order_id, exchange_order_id=order_update.exchange_order_id, ) - self._order_tracker.process_order_update( - order_update=order_update_to_process - ) + self._order_tracker.process_order_update(order_update=order_update_to_process) def _process_user_trade_update(self, trade_update: TradeUpdate): - tracked_order = ( - self._order_tracker.all_fillable_orders_by_exchange_order_id.get( - trade_update.exchange_order_id - ) - ) + tracked_order = self._order_tracker.all_fillable_orders_by_exchange_order_id.get(trade_update.exchange_order_id) if tracked_order is not None: - self.logger().debug( - f"Processing trade update {trade_update}\nFillable order {tracked_order.to_json()}" - ) + self.logger().debug(f"Processing trade update {trade_update}\nFillable order {tracked_order.to_json()}") # Fetch the current fee rates maker_fee, taker_fee = self._data_source.get_fees() fee_rate = maker_fee if tracked_order.is_maker else taker_fee - fee = TradeFeeBase.new_perpetual_fee( - trade_type=tracked_order.trade_type, percent=fee_rate - ) + fee = TradeFeeBase.new_perpetual_fee(trade_type=tracked_order.trade_type, percent=fee_rate) - fill_amount = ( - trade_update.fill_base_amount - tracked_order.executed_amount_base - ) + fill_amount = trade_update.fill_base_amount - tracked_order.executed_amount_base trade_update: TradeUpdate = TradeUpdate( trade_id=trade_update.trade_id, client_order_id=tracked_order.client_order_id, diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py index a5d176bd502..8dfdcc36aac 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py @@ -19,9 +19,7 @@ class HookOdysseyPerpetualConfigMap(BaseConnectorConfigMap): - connector: str = Field( - default="hook_odyssey_perpetual", const=True, client_data=None - ) + connector: str = Field(default="hook_odyssey_perpetual", const=True, client_data=None) hook_odyssey_perpetual_eth_address: str = Field( default=..., client_data=ClientFieldData( @@ -57,17 +55,13 @@ class Config: KEYS = HookOdysseyPerpetualConfigMap.construct() OTHER_DOMAINS = ["hook_odyssey_perpetual_testnet"] -OTHER_DOMAINS_PARAMETER = { - "hook_odyssey_perpetual_testnet": "hook_odyssey_perpetual_testnet" -} +OTHER_DOMAINS_PARAMETER = {"hook_odyssey_perpetual_testnet": "hook_odyssey_perpetual_testnet"} OTHER_DOMAINS_EXAMPLE_PAIR = {"hook_odyssey_perpetual_testnet": EXAMPLE_PAIR} OTHER_DOMAINS_DEFAULT_FEES = {"hook_odyssey_perpetual_testnet": DEFAULT_FEES} class HookOdysseyPerpetualTestnetConfigMap(BaseConnectorConfigMap): - connector: str = Field( - default="hook_odyssey_perpetual_testnet", const=True, client_data=None - ) + connector: str = Field(default="hook_odyssey_perpetual_testnet", const=True, client_data=None) hook_odyssey_perpetual_testnet_public_key: SecretStr = Field( default=..., client_data=ClientFieldData( @@ -91,9 +85,7 @@ class Config: title = "hook_odyssey_perpetual_testnet" -OTHER_DOMAINS_KEYS = { - "hook_odyssey_perpetual_testnet": HookOdysseyPerpetualTestnetConfigMap.construct() -} +OTHER_DOMAINS_KEYS = {"hook_odyssey_perpetual_testnet": HookOdysseyPerpetualTestnetConfigMap.construct()} def eth_to_wei(eth_amount: Decimal) -> int: From 507c5692a11aaab353dde0d67ee2b87372fed66b Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Sat, 10 Feb 2024 18:53:54 -0500 Subject: [PATCH 4/8] Support position sub changes --- .../hook_odyssey_perpetual_data_source.py | 23 ++++++++++++++----- ...hook_odyssey_perpetual_graphql_executor.py | 4 +++- 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py index ca3a1969109..0750531e996 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py @@ -430,13 +430,24 @@ async def _process_subaccount_positions(self, event: Dict[str, Any], address: st self._initial_subaccount_positions_event.set() for position in positions_event["positions"]: - size = wei_to_eth(position["size"]) - side = PositionSide.LONG if size > 0 else PositionSide.SHORT - size = abs(size) + size = wei_to_eth(position["sizeHeld"]) + if position["isLong"]: + side = PositionSide.LONG + else: + side = PositionSide.SHORT + size = Decimal(-1) * size instrument_hash = position["instrument"]["id"] + entry_price = wei_to_eth(position["averageCost"]) + if instrument_hash not in self._instrument_hashes: continue trading_pair = self._instrument_hashes[instrument_hash] + mark_price = await self.get_mark_price(trading_pair) + + if side == PositionSide.LONG: + unrealized_pnl = (mark_price - entry_price) * abs(size) + else: + unrealized_pnl = (entry_price - mark_price) * abs(size) if size == 0: self._subaccount_positions.pop(instrument_hash) @@ -444,9 +455,9 @@ async def _process_subaccount_positions(self, event: Dict[str, Any], address: st self._subaccount_positions[instrument_hash] = Position( trading_pair=trading_pair, position_side=side, - unrealized_pnl=Decimal(0), - entry_price=Decimal(0), - amount=size, + unrealized_pnl=unrealized_pnl.normalize(), + entry_price=entry_price.normalize(), + amount=size.normalize(), leverage=Decimal(1), ) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py index 4468447eec4..77d4fa498f5 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py @@ -250,7 +250,9 @@ async def subscribe_subaccount_positions(self, handler: Callable, address: str): } subaccount marketHash - size + sizeHeld + isLong + averageCost } } } From 30385fdcebd79567c9d6dfb0ee1ef3532ab42b0a Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Mon, 12 Feb 2024 17:41:05 -0500 Subject: [PATCH 5/8] Additional script for deployment --- Dockerfile | 5 +++++ setup_connector.sh | 19 +++++++++++++++++++ 2 files changed, 24 insertions(+) create mode 100755 setup_connector.sh diff --git a/Dockerfile b/Dockerfile index bb2924402d4..c2c0451f81c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -24,6 +24,11 @@ COPY LICENSE . COPY README.md . COPY DATA_COLLECTION.md . +# ARG CONNECTOR_FILE_NAME +# ARG CONNECTOR_DEST_FILE_NAME +# COPY setup_connector.sh /home/hummingbot/ +# RUN chmod +x /home/hummingbot/setup_connector.sh && /home/hummingbot/setup_connector.sh + # activate hummingbot env when entering the CT SHELL [ "/bin/bash", "-lc" ] RUN echo "conda activate hummingbot" >> ~/.bashrc diff --git a/setup_connector.sh b/setup_connector.sh new file mode 100755 index 00000000000..26ff2b8ad14 --- /dev/null +++ b/setup_connector.sh @@ -0,0 +1,19 @@ +#!/bin/bash + +echo "Setting up connector..." +if [ -z "$CONNECTOR_FILE_NAME" ]; then + exit 1 +fi + +if [ -z "$CONNECTOR_DEST_FILE_NAME" ]; then + exit 1 +fi + +SOURCE_FILE="conf/connectors/$CONNECTOR_FILE_NAME" +DEST_FILE="conf/connectors/$CONNECTOR_DEST_FILE_NAME" + +if [ ! -f "$SOURCE_FILE" ]; then + exit 1 +fi + +mv "$SOURCE_FILE" "$DEST_FILE" From 843f3fcfdedd0fd68ccecf3ca72549fc845b84d8 Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Fri, 15 Mar 2024 12:57:57 -0400 Subject: [PATCH 6/8] Restart subs on err; support pools --- ...ey_perpetual_api_order_book_data_source.py | 3 - .../hook_odyssey_perpetual_data_source.py | 417 ++++++++++++------ .../hook_odyssey_perpetual_utils.py | 79 +++- 3 files changed, 359 insertions(+), 140 deletions(-) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py index f3153ed2fb6..fc39a1fa9e7 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_api_order_book_data_source.py @@ -1,5 +1,4 @@ import asyncio -from decimal import Decimal from typing import TYPE_CHECKING, Any, Dict, List, Optional from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_data_source import ( @@ -67,9 +66,7 @@ async def _parse_order_book_diff_message(self, raw_message: OrderBookMessage, me message_queue.put_nowait(raw_message) async def get_funding_info(self, trading_pair: str) -> FundingInfo: - index_price = await self._data_source.get_index_price(trading_pair) funding_info = await self._data_source.get_funding_info(trading_pair) - funding_info.index_price = Decimal(index_price) return funding_info async def _parse_funding_info_message(self, raw_message: Dict[str, Any], message_queue: asyncio.Queue): diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py index 0750531e996..074fbfdd7b6 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py @@ -4,6 +4,8 @@ from decimal import Decimal from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple +from websockets.exceptions import ConnectionClosed + from hummingbot.connector.derivative.hook_odyssey_perpetual import hook_odyssey_perpetual_constants as CONSTANTS from hummingbot.connector.derivative.hook_odyssey_perpetual.hook_odyssey_perpetual_graphql_executor import ( HookOdysseyPerpetualGrapQLExecutor, @@ -19,13 +21,19 @@ ) from hummingbot.connector.derivative.position import Position, PositionSide from hummingbot.connector.utils import combine_to_hb_trading_pair -from hummingbot.core.data_type.common import OrderType, TradeType -from hummingbot.core.data_type.funding_info import FundingInfo +from hummingbot.core.data_type.common import OrderType, PositionAction, TradeType +from hummingbot.core.data_type.funding_info import FundingInfo, FundingInfoUpdate from hummingbot.core.data_type.in_flight_order import OrderUpdate, TradeUpdate from hummingbot.core.data_type.order_book_message import OrderBookMessage, OrderBookMessageType -from hummingbot.core.data_type.trade_fee import TradeFeeBase +from hummingbot.core.data_type.trade_fee import TradeFeeBase, TradeFeeSchema from hummingbot.core.event.event_listener import EventListener -from hummingbot.core.event.events import MarketEvent, OrderBookEvent +from hummingbot.core.event.events import ( + AccountEvent, + BalanceUpdateEvent, + MarketEvent, + OrderBookEvent, + PositionUpdateEvent, +) from hummingbot.core.network_iterator import NetworkStatus from hummingbot.core.pubsub import PubSub from hummingbot.logger import HummingbotLogger @@ -45,24 +53,33 @@ def logger(cls) -> HummingbotLogger: def __init__( self, - hook_odyssey_perpetual_eth_address: str, - hook_odyssey_perpetual_private_key: str, + hook_odyssey_perpetual_signer_address: str, + hook_odyssey_perpetual_signer_pkey: str, hook_odyssey_api_key: str, + hook_odyssey_pool_address: str, + hook_odyssey_pool_subaccount: str, + hook_odyssey_pool_trading_pair: str, connector: "ExchangePyBase", + trade_fee_schema: TradeFeeSchema, domain: Optional[str] = CONSTANTS.DOMAIN, trading_required: bool = True, ): self._connector = connector self._domain = domain self._trading_required = trading_required - self._hook_odyssey_perpetual_eth_address = hook_odyssey_perpetual_eth_address + self._hook_odyssey_perpetual_signer_address = hook_odyssey_perpetual_signer_address + self._hook_odyssey_perpetual_pool_address = hook_odyssey_pool_address + self._hook_odyssey_perpetual_pool_subaccount = hook_odyssey_pool_subaccount + self._hook_odyssey_perpetual_pool_trading_pair = hook_odyssey_pool_trading_pair self._graphql_executor = HookOdysseyPerpetualGrapQLExecutor( hook_odyssey_api_key=hook_odyssey_api_key, domain=self._domain ) - self._signer = HookOdysseyPerpetualSigner(private_key=hook_odyssey_perpetual_private_key, domain=self._domain) + self._signer = HookOdysseyPerpetualSigner(private_key=hook_odyssey_perpetual_signer_pkey, domain=self._domain) self._publisher = PubSub() + self._connection_task: Optional[asyncio.Task] = None self._events_listening_tasks = [] self._assets_map: Dict[str, str] = {} + self._trade_fee_schema = trade_fee_schema # Data source state - used to provide subscription data at any time self._perpetual_pairs: Dict[str, Dict[str, Any]] = {} # Mapping of trading pair to perpetual pair @@ -71,6 +88,7 @@ def __init__( self._subaccounts: Dict[str, str] = {} # Mapping of subaccount to trading pair self._index_prices: Dict[str, Decimal] = {} # Mapping of trading pair to index price self._mark_prices: Dict[str, Decimal] = {} # Mapping of trading pair to mark price + self._account_equity: Dict[str, Decimal] = {} # Mapping of subaccount to equity self._subaccount_balances: Dict[str, Decimal] = {} # Mapping of subaccount to balance self._subaccount_positions: Dict[str, Position] = {} # Mapping of subaccount to position self._funding_info: Dict[str, FundingInfo] = {} # Mapping of trading pair to funding info @@ -88,89 +106,45 @@ def __init__( self._initial_subaccount_balances_event = asyncio.Event() self._initial_subaccount_positions_event = asyncio.Event() + def pools_configured(self) -> bool: + return all( + [ + self._hook_odyssey_perpetual_pool_address, + self._hook_odyssey_perpetual_pool_subaccount, + self._hook_odyssey_perpetual_pool_trading_pair, + ] + ) + async def start(self, trading_pairs: List[str]): if len(self._events_listening_tasks) > 0: raise AssertionError("Data source is already started and can't be started again") await self.get_supported_pairs() - - for trading_pair in trading_pairs: - perpetual_pair = self.get_perpetual_pair_for_trading_pair(trading_pair) - instrument_hash = perpetual_pair.get("instrumentHash", "") - symbol = perpetual_pair.get("symbol", "") - subaccount = perpetual_pair.get("subaccount", "") - - # Index price - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_ticker( - self._process_ticker_update, - symbol, - ) - ) - ) - # Statistics / Funding Info - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_statistics( - self._process_statistics_update, - symbol, - ) - ) - ) - # BBO - Top of book - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_bbo( - self._process_bbo_update, - symbol, - ) - ) - ) - # Orderbook - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_orderbook( - self._process_orderbook_update, - instrument_hash, - ) - ) - ) - - if self._trading_required: - # Private order updates - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_subaccount_orders( - self._process_subaccount_orders_update, - subaccount, - ) - ) - ) - - if self._trading_required: - # Private balances - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_subaccount_balances( - self._process_subaccount_balances, - self._hook_odyssey_perpetual_eth_address, - ) - ) - ) - # Private positions - self._events_listening_tasks.append( - asyncio.create_task( - self._graphql_executor.subscribe_subaccount_positions( - self._process_subaccount_positions, - self._hook_odyssey_perpetual_eth_address, - ) - ) - ) + self._connection_task = asyncio.create_task(self.subscription_connection(trading_pairs)) async def stop(self): for task in self._events_listening_tasks: task.cancel() - self._events_listening_tasks = [] + try: + # Wait for the task cancellation to complete, with timeout + await asyncio.wait_for(task, timeout=3) + except asyncio.TimeoutError: + self.logger().error(f"Task cancellation timed out for {task}") + except asyncio.CancelledError: + # Task cancellation has completed + pass + self._events_listening_tasks.clear() + # Cancel the main connection task + if self._connection_task is not None: + self._connection_task.cancel() + try: + # Wait for the task cancellation to complete, with timeout + await asyncio.wait_for(self._connection_task, timeout=3) + except asyncio.TimeoutError: + self.logger().error(f"Task cancellation timed out for {self._connection_task}") + except asyncio.CancelledError: + # Task cancellation has completed + pass + self._connection_task = None def is_started(self) -> bool: return len(self._events_listening_tasks) > 0 @@ -181,6 +155,102 @@ def add_listener(self, event_tag, listener: EventListener): def remove_listener(self, event_tag, listener: EventListener): self._publisher.remove_listener(event_tag=event_tag, listener=listener) + async def subscription_connection(self, trading_pairs: List[str]): + address = self._hook_odyssey_perpetual_signer_address + if self.pools_configured(): + address = self._hook_odyssey_perpetual_pool_address + + while True: + try: + client = self._graphql_executor.ws_client() + async with client as session: + for trading_pair in trading_pairs: + perpetual_pair = self.get_perpetual_pair_for_trading_pair(trading_pair) + instrument_hash = perpetual_pair.get("instrumentHash", "") + symbol = perpetual_pair.get("symbol", "") + subaccount = perpetual_pair.get("subaccount", "") + + # Index price + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_ticker( + session, + self._process_ticker_update, + symbol, + ) + ) + ) + # Statistics / Funding Info + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_statistics( + session, + self._process_statistics_update, + symbol, + ) + ) + ) + # BBO - Top of book + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_bbo( + session, + self._process_bbo_update, + symbol, + ) + ) + ) + # Orderbook + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_orderbook( + session, + self._process_orderbook_update, + instrument_hash, + ) + ) + ) + + if self._trading_required: + # Private order updates + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_subaccount_orders( + session, + self._process_subaccount_orders_update, + subaccount, + ) + ) + ) + if self._trading_required: + # Private balances + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_subaccount_balances( + session, + self._process_subaccount_balances, + address, + ) + ) + ) + # Private positions + self._events_listening_tasks.append( + asyncio.create_task( + self._graphql_executor.subscribe_subaccount_positions( + session, + self._process_subaccount_positions, + address, + ) + ) + ) + await asyncio.gather(*self._events_listening_tasks) + except (ConnectionClosed, asyncio.TimeoutError, Exception) as e: + self.logger().error(f"Subscription connection error: {str(e)}, attempting to reconnect...") + for task in self._events_listening_tasks: + task.cancel() + self._events_listening_tasks.clear() + await asyncio.sleep(3) # A short delay before attempting to reconnect + async def exchange_status(self): resp = await self._graphql_executor.account_details() if "accountDetails" not in resp: @@ -207,10 +277,17 @@ async def get_supported_pairs(self) -> List[Dict[str, Any]]: trading_pair = combine_to_hb_trading_pair( base=perpetual_pair["symbol"], quote=perpetual_pair["baseCurrency"] ) + if self.pools_configured() and trading_pair != self._hook_odyssey_perpetual_pool_trading_pair: + continue self._perpetual_pairs[trading_pair] = perpetual_pair self._symbols[perpetual_pair["symbol"]] = trading_pair self._instrument_hashes[perpetual_pair["instrumentHash"]] = trading_pair - self._subaccounts[perpetual_pair["subaccount"]] = trading_pair + subaccount = perpetual_pair["subaccount"] + if self.pools_configured(): + # Use the configured subaccount + subaccount = self._hook_odyssey_perpetual_pool_subaccount + perpetual_pair["subaccount"] = subaccount + self._subaccounts[subaccount] = trading_pair return perpetual_pairs def get_perpetual_pair_for_trading_pair(self, trading_pair: str) -> Optional[Dict[str, Any]]: @@ -220,12 +297,12 @@ def get_perpetual_pair_for_trading_pair(self, trading_pair: str) -> Optional[Dic async def get_index_price(self, trading_pair: str) -> Decimal: # Wait for the trade subscription snapshot to come in await self._initial_ticker_event.wait() - return self._index_prices.get(trading_pair, 0.0) + return self._index_prices.get(trading_pair, Decimal(0)) async def get_mark_price(self, trading_pair: str) -> Decimal: # Wait for the bbo subscription snapshot to come in await self._initial_bbo_event.wait() - return self._mark_prices.get(trading_pair, 0.0) + return self._mark_prices.get(trading_pair, Decimal(0)) async def get_funding_info(self, trading_pair: str) -> FundingInfo: # Wait for the funding info subscription snapshot to come in @@ -263,29 +340,63 @@ async def get_balances(self) -> Dict[str, Decimal]: await self._initial_subaccount_balances_event.wait() return self._subaccount_balances + async def get_account_equity(self) -> Dict[str, Decimal]: + # Wait for the subaccount balances and bbo subscription snapshots to come in + await self._initial_subaccount_balances_event.wait() + await self._initial_bbo_event.wait() + return self._account_equity + async def get_positions(self) -> Dict[str, Position]: # Wait for the subaccount positions subscription snapshot to come in await self._initial_subaccount_positions_event.wait() return self._subaccount_positions + async def trigger_balance_updates(self): + """ + Compute account equity for each subaccount and trigger balance update events. + """ + for subaccount, trading_pair in self._subaccounts.items(): + # Compute account equity + # equity = position_size * mark_price + balance + perpetual_pair = self.get_perpetual_pair_for_trading_pair(trading_pair) + positions = await self.get_positions() + position = positions.get(perpetual_pair["instrumentHash"], None) + mark_price = await self.get_mark_price(trading_pair) + balances = await self.get_balances() + balance = balances.get(subaccount, Decimal(0)) + if position is not None: + position_size = position.amount + if position.position_side == PositionSide.SHORT: + position_size = Decimal(-1) * position_size + equity = ((position_size * mark_price)) + balance + else: + equity = balance + self._account_equity[subaccount] = equity + + # Trigger balance update event + balance_msg = BalanceUpdateEvent( + timestamp=time.time(), + asset_name=perpetual_pair["baseCurrency"], + total_balance=equity, + available_balance=balance, + ) + self._publisher.trigger_event(event_tag=AccountEvent.BalanceEvent, message=balance_msg) + async def _process_ticker_update(self, event: Dict[str, Any], symbol: str): ticker_event = event["ticker"] - if not self._initial_ticker_event.is_set(): - self._initial_ticker_event.set() - if symbol not in self._symbols: return trading_pair = self._symbols[symbol] - self._index_prices[trading_pair] = wei_to_eth(ticker_event["price"]) + price = wei_to_eth(ticker_event["price"]) + self.logger().info(f"Ticker: {symbol} - price: {price.normalize()}") + self._index_prices[trading_pair] = price + self._initial_ticker_event.set() return self._index_prices[trading_pair] async def _process_statistics_update(self, event: Dict[str, Any], symbol: str): statistics_event = event["statistics"] - if statistics_event["eventType"] == "SNAPSHOT" and not self._initial_statistics_event.is_set(): - self._initial_statistics_event.set() - if symbol not in self._symbols: return trading_pair = self._symbols[symbol] @@ -293,20 +404,30 @@ async def _process_statistics_update(self, event: Dict[str, Any], symbol: str): # Get index price index_price = await self.get_index_price(trading_pair) mark_price = await self.get_mark_price(trading_pair) + next_funding_timestamp = int(statistics_event["nextFundingEpoch"]) + rate = Decimal(statistics_event["fundingRateBips"]) / Decimal(10000) self._funding_info[trading_pair] = FundingInfo( trading_pair=trading_pair, index_price=index_price, mark_price=mark_price, - next_funding_utc_timestamp=int(statistics_event["nextFundingEpoch"]), - rate=Decimal(statistics_event["fundingRateBips"]) / Decimal(10000), + next_funding_utc_timestamp=next_funding_timestamp, + rate=rate, ) + if statistics_event["eventType"] == "SNAPSHOT": + self._initial_statistics_event.set() + elif statistics_event["eventType"] == "UPDATE": + funding_info_update_msg = FundingInfoUpdate( + trading_pair=trading_pair, + index_price=index_price, + mark_price=mark_price, + next_funding_utc_timestamp=next_funding_timestamp, + rate=rate, + ) + self.publisher.trigger_event(event_tag=MarketEvent.FundingInfo, message=funding_info_update_msg) async def _process_bbo_update(self, event: Dict[str, Any], symbol: str): bbo_event = event["bbo"] - if bbo_event["eventType"] == "SNAPSHOT" and not self._initial_bbo_event.is_set(): - self._initial_bbo_event.set() - if symbol not in self._symbols: return @@ -315,6 +436,10 @@ async def _process_bbo_update(self, event: Dict[str, Any], symbol: str): trading_pair = self._instrument_hashes[instrument_hash] self._mark_prices[trading_pair] = wei_to_eth(instrument["markPrice"]) + if bbo_event["eventType"] == "SNAPSHOT": + self._initial_bbo_event.set() + await self.trigger_balance_updates() + async def _process_orderbook_update(self, event: Dict[str, Any], instrument_hash: str): if instrument_hash not in self._instrument_hashes: return @@ -325,9 +450,6 @@ async def _process_orderbook_update(self, event: Dict[str, Any], instrument_hash if orderbook_event["eventType"] == "SNAPSHOT": message_type = OrderBookMessageType.SNAPSHOT - if not self._initial_orderbook_event.is_set(): - self._initial_orderbook_event.set() - bids = [ [ wei_to_eth(bl["price"]), @@ -355,56 +477,80 @@ async def _process_orderbook_update(self, event: Dict[str, Any], instrument_hash ) self._apply_local_orderbook_update(order_book_message) - self._publisher.trigger_event( event_tag=OrderBookEvent.OrderBookDataSourceUpdateEvent, message=order_book_message, ) + self._initial_orderbook_event.set() async def _process_subaccount_orders_update(self, event: Dict[str, Any], subaccount: str): orders_event = event["subaccountOrders"] - - if orders_event["eventType"] == "SNAPSHOT": - if not self._initial_subaccount_orders_event.is_set(): - self._initial_subaccount_orders_event.set() - return # Ignore snapshots + if orders_event["eventType"] == "SNAPSHOT" and not self._initial_subaccount_orders_event.is_set(): + self._initial_subaccount_orders_event.set() orders = orders_event["orders"] - if subaccount not in self._subaccounts: return trading_pair = self._subaccounts[subaccount] for order in orders: - trading_pair = trading_pair + if order["status"] in ["PARTIALLY_MATCHED", "MATCHED"]: + # Ignore partially matched and matched orders + continue + if order["orderType"] != "LIMIT": + # Ignore market orders + continue # Update order state order_state = CONSTANTS.ORDER_STATE[order["status"]] + trade_type = TradeType.BUY if order["direction"] == "BUY" else TradeType.SELL + misc_updates = { + "order_type": OrderType.LIMIT, + "trade_type": trade_type, + "amount": wei_to_eth(order["size"]), + "price": wei_to_eth(order["limitPrice"]), + } order_update = OrderUpdate( trading_pair=trading_pair, update_timestamp=int(time.time()), new_state=order_state, client_order_id=order["orderHash"], exchange_order_id=order["orderHash"], + misc_updates=misc_updates, ) self._publisher.trigger_event(event_tag=MarketEvent.OrderUpdate, message=order_update) - # Update trade - if order_state == "FILLED" or order_state == "PARTIALLY_FILLED": + if order["status"] in ["PARTIALLY_FILLED", "FILLED"]: size = wei_to_eth(order["size"]) remaining_size = wei_to_eth(order["remainingSize"]) fill_amount = size - remaining_size fill_price = wei_to_eth(order["limitPrice"]) - trade_type = TradeType.BUY if order["direction"] == "BUY" else TradeType.SELL - - maker_fee, taker_fee = self.get_fees() - fee_rate = maker_fee if order["orderType"] == "MARKET" else taker_fee - fee = TradeFeeBase.new_perpetual_fee(trade_type=trade_type, percent=fee_rate) + perpetual_pair = self.get_perpetual_pair_for_trading_pair(trading_pair) + + # Determine if the trade is the same side as the position + position = self._subaccount_positions.get(perpetual_pair["instrumentHash"], None) + position_action = PositionAction.OPEN + if position is not None: + if position.position_side == PositionSide.LONG and trade_type == TradeType.BUY: + position_action = PositionAction.OPEN + elif position.position_side == PositionSide.SHORT and trade_type == TradeType.SELL: + position_action = PositionAction.OPEN + elif position.position_side == PositionSide.LONG and trade_type == TradeType.SELL: + position_action = PositionAction.CLOSE + elif position.position_side == PositionSide.SHORT and trade_type == TradeType.BUY: + position_action = PositionAction.CLOSE + + fee_asset = perpetual_pair["baseCurrency"] + fee = TradeFeeBase.new_perpetual_fee( + fee_schema=self._trade_fee_schema, + position_action=position_action, + percent_token=fee_asset, + ) trade_update = TradeUpdate( - trade_id=f"{order['orderHash']}-{order['timestamp']}", + trade_id=f"{order['orderHash']}-{order['remainingSize']}", client_order_id=order["orderHash"], exchange_order_id=order["orderHash"], trading_pair=trading_pair, - fill_timestamp=self._time(), + fill_timestamp=time.time(), fill_price=fill_price, fill_base_amount=fill_amount, fill_quote_amount=fill_price * fill_amount, @@ -415,27 +561,25 @@ async def _process_subaccount_orders_update(self, event: Dict[str, Any], subacco async def _process_subaccount_balances(self, event: Dict[str, Any], address: str): balances_event = event["subaccountBalances"] - if balances_event["eventType"] == "SNAPSHOT" and not self._initial_subaccount_balances_event.is_set(): - self._initial_subaccount_balances_event.set() - for balance in balances_event["balances"]: # Ignore primary account balances if balance["subaccountID"] != 0: - self._subaccount_balances[balance["subaccount"]] = wei_to_eth(balance["balance"]) + bal = wei_to_eth(balance["balance"]) + self._subaccount_balances[balance["subaccount"]] = bal + + if balances_event["eventType"] == "SNAPSHOT": + self._initial_subaccount_balances_event.set() + await self.trigger_balance_updates() async def _process_subaccount_positions(self, event: Dict[str, Any], address: str): positions_event = event["subaccountPositions"] - if positions_event["eventType"] == "SNAPSHOT" and not self._initial_subaccount_positions_event.is_set(): - self._initial_subaccount_positions_event.set() - for position in positions_event["positions"]: size = wei_to_eth(position["sizeHeld"]) if position["isLong"]: side = PositionSide.LONG else: side = PositionSide.SHORT - size = Decimal(-1) * size instrument_hash = position["instrument"]["id"] entry_price = wei_to_eth(position["averageCost"]) @@ -450,7 +594,7 @@ async def _process_subaccount_positions(self, event: Dict[str, Any], address: st unrealized_pnl = (entry_price - mark_price) * abs(size) if size == 0: - self._subaccount_positions.pop(instrument_hash) + self._subaccount_positions.pop(instrument_hash, None) else: self._subaccount_positions[instrument_hash] = Position( trading_pair=trading_pair, @@ -460,6 +604,21 @@ async def _process_subaccount_positions(self, event: Dict[str, Any], address: st amount=size.normalize(), leverage=Decimal(1), ) + if size == 0: + side = None + position_msg = PositionUpdateEvent( + timestamp=time.time(), + trading_pair=trading_pair, + position_side=side, + unrealized_pnl=unrealized_pnl.normalize(), + entry_price=entry_price.normalize(), + amount=size.normalize(), + leverage=Decimal(1), + ) + self._publisher.trigger_event(event_tag=AccountEvent.PositionUpdate, message=position_msg) + if positions_event["eventType"] == "SNAPSHOT": + self._initial_subaccount_positions_event.set() + await self.trigger_balance_updates() def order_hash( self, @@ -550,9 +709,9 @@ async def place_order( allOrNothing=False, ) signature, order_hash = self._signer.sign_order(order) - + signature_type = "EIP1271" if self._hook_odyssey_perpetual_signer_address != self._hook_odyssey_perpetual_pool_address else "DIRECT" signature_input = { - "signatureType": "DIRECT", + "signatureType": signature_type, "signature": signature, } success = await self._graphql_executor.place_order(order_input, signature_input) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py index 8dfdcc36aac..fe5d33d775b 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_utils.py @@ -20,19 +20,19 @@ class HookOdysseyPerpetualConfigMap(BaseConnectorConfigMap): connector: str = Field(default="hook_odyssey_perpetual", const=True, client_data=None) - hook_odyssey_perpetual_eth_address: str = Field( + hook_odyssey_perpetual_signer_address: str = Field( default=..., client_data=ClientFieldData( - prompt=lambda cm: "Enter your Ethereum address", + prompt=lambda cm: "Enter the order signer address", is_secure=False, is_connect_key=True, prompt_on_new=True, ), ) - hook_odyssey_perpetual_private_key: SecretStr = Field( + hook_odyssey_perpetual_signer_pkey: SecretStr = Field( default=..., client_data=ClientFieldData( - prompt=lambda cm: "Enter your Ethereum private key", + prompt=lambda cm: "Enter the order signer private key", is_secure=True, is_connect_key=True, prompt_on_new=True, @@ -47,6 +47,33 @@ class HookOdysseyPerpetualConfigMap(BaseConnectorConfigMap): prompt_on_new=True, ), ) + hook_odyssey_pool_address: str = Field( + default='', + client_data=ClientFieldData( + prompt=lambda cm: "Enter the pool address to trade on behalf of", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_pool_subaccount: str = Field( + default='', + client_data=ClientFieldData( + prompt=lambda cm: "Enter the pool subaccount to trade on behalf of", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_pool_trading_pair: str = Field( + default='', + client_data=ClientFieldData( + prompt=lambda cm: "Enter the trading pair configured for the pool", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) class Config: title = "hook_odyssey_perpetual" @@ -62,24 +89,60 @@ class Config: class HookOdysseyPerpetualTestnetConfigMap(BaseConnectorConfigMap): connector: str = Field(default="hook_odyssey_perpetual_testnet", const=True, client_data=None) - hook_odyssey_perpetual_testnet_public_key: SecretStr = Field( + hook_odyssey_perpetual_signer_address: str = Field( default=..., client_data=ClientFieldData( - prompt=lambda cm: "Enter your Ethereum wallet public key", + prompt=lambda cm: "Enter your Ethereum address", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_perpetual_signer_pkey: SecretStr = Field( + default=..., + client_data=ClientFieldData( + prompt=lambda cm: "Enter your Ethereum private key", is_secure=True, is_connect_key=True, prompt_on_new=True, ), ) - hook_odyssey_perpetual_testnet_private_key: SecretStr = Field( + hook_odyssey_api_key: SecretStr = Field( default=..., client_data=ClientFieldData( - prompt=lambda cm: "Enter your Ethereum wallet private key", + prompt=lambda cm: "Enter your Hook Odyssey API key", is_secure=True, is_connect_key=True, prompt_on_new=True, ), ) + hook_odyssey_pool_address: str = Field( + default='', + client_data=ClientFieldData( + prompt=lambda cm: "Enter the pool address to trade on behalf of", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_pool_subaccount: str = Field( + default='', + client_data=ClientFieldData( + prompt=lambda cm: "Enter the pool subaccount to trade on behalf of", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) + hook_odyssey_pool_trading_pair: str = Field( + default=None, + client_data=ClientFieldData( + prompt=lambda cm: "Enter the trading pair configured for the pool", + is_secure=False, + is_connect_key=True, + prompt_on_new=True, + ), + ) class Config: title = "hook_odyssey_perpetual_testnet" From e2a4cefbd8a3b535d3932d08f8ce1ca84625c3b0 Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Fri, 15 Mar 2024 13:02:03 -0400 Subject: [PATCH 7/8] Forgot some files --- .../hook_odyssey_perpetual_constants.py | 20 ++-- .../hook_odyssey_perpetual_derivative.py | 96 ++++++++++++------- ...hook_odyssey_perpetual_graphql_executor.py | 66 +++++++------ 3 files changed, 107 insertions(+), 75 deletions(-) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py index a4001b5b1be..f51821285e3 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_constants.py @@ -6,26 +6,26 @@ SUPPORTED_COLLATERAL = ["ETH", "USDC"] BASE_URLS = { - DOMAIN: "https://goerli-api.hook.xyz/query", # TODO: Set to mainnet - TESTNET_DOMAIN: "https://api.hookdev.xyz/query", # TODO: Set to testnet + DOMAIN: "https://api-prod.hook.xyz/query", + TESTNET_DOMAIN: "https://goerli-api.hook.xyz/query", } WS_URLS = { - DOMAIN: "wss://goerli-api.hook.xyz/query", # TODO: Set to mainnet - TESTNET_DOMAIN: "wss://api.hookdev.xyz/query", # TODO: Set to testnet + DOMAIN: "wss://api-prod.hook.xyz/query", + TESTNET_DOMAIN: "wss://goerli-api.hook.xyz/query", } # Order Statuses +# Note: "Partially Matched" and "Matched" are ignored because they are not final states ORDER_STATE = { "OPEN": OrderState.OPEN, - "PARTIALLY_MATCHED": OrderState.PARTIALLY_FILLED, - "MATCHED": OrderState.FILLED, "PARTIALLY_FILLED": OrderState.PARTIALLY_FILLED, "FILLED": OrderState.FILLED, "CANCELED": OrderState.CANCELED, "EXPIRED": OrderState.CANCELED, "REJECTED": OrderState.FAILED, "UNFILLABLE": OrderState.FAILED, + "FAILED": OrderState.FAILED, } RATE_LIMITS = [] @@ -35,13 +35,13 @@ EIP712_DOMAIN_VERSION = "1.0.0" CONTRACT_ADDRESSES = { - DOMAIN: "0x64247BeF0C0990aF63FCbdd21dc07aC2b251f500", # TODO: Set to mainnet - TESTNET_DOMAIN: "0xc6e7DF5E7b4f2A278906862b61205850344D4e7d", # TODO: Set to testnet + DOMAIN: "0xF9Bd1BaB25442A3b6888f2086736C6aC76A4Cf4B", + TESTNET_DOMAIN: "0x64247BeF0C0990aF63FCbdd21dc07aC2b251f500", } CHAIN_IDS = { - DOMAIN: 46658378, # TODO: Set to mainnet - TESTNET_DOMAIN: 9999, # TODO: Set to testnet + DOMAIN: 4665, + TESTNET_DOMAIN: 46658378, } DEFAULT_TIMEOUT = 3 diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py index 0c4ea93633a..265c5900da0 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_derivative.py @@ -1,4 +1,5 @@ import asyncio +import time from typing import TYPE_CHECKING, Dict, List, Optional, Tuple from _decimal import Decimal @@ -21,12 +22,12 @@ from hummingbot.connector.utils import combine_to_hb_trading_pair from hummingbot.core.api_throttler.data_types import RateLimit from hummingbot.core.data_type.common import OrderType, PositionAction, PositionMode, TradeType -from hummingbot.core.data_type.in_flight_order import InFlightOrder, OrderUpdate, TradeUpdate +from hummingbot.core.data_type.in_flight_order import InFlightOrder, OrderState, OrderUpdate, TradeUpdate from hummingbot.core.data_type.order_book_tracker_data_source import OrderBookTrackerDataSource from hummingbot.core.data_type.trade_fee import AddedToCostTradeFee, TradeFeeBase from hummingbot.core.data_type.user_stream_tracker_data_source import UserStreamTrackerDataSource from hummingbot.core.event.event_forwarder import EventForwarder -from hummingbot.core.event.events import AccountEvent, BalanceUpdateEvent, MarketEvent +from hummingbot.core.event.events import AccountEvent, BalanceUpdateEvent, MarketEvent, PositionUpdateEvent from hummingbot.core.network_iterator import NetworkStatus from hummingbot.core.utils.async_utils import safe_ensure_future, safe_gather from hummingbot.core.web_assistant.auth import AuthBase @@ -40,23 +41,28 @@ class HookOdysseyPerpetualDerivative(PerpetualDerivativePyBase): def __init__( self, client_config_map: "ClientConfigAdapter", - hook_odyssey_perpetual_eth_address: str, - hook_odyssey_perpetual_private_key: str, + hook_odyssey_perpetual_signer_address: str, + hook_odyssey_perpetual_signer_pkey: str, hook_odyssey_api_key: str, + hook_odyssey_pool_address: str, + hook_odyssey_pool_subaccount: str, + hook_odyssey_pool_trading_pair: str, trading_pairs: Optional[List[str]] = None, trading_required: bool = True, domain: str = CONSTANTS.DOMAIN, ): - self.hook_odyssey_perpetual_eth_address = hook_odyssey_perpetual_eth_address - self.hook_odyssey_perpetual_private_key = hook_odyssey_perpetual_private_key self._trading_required = trading_required self._trading_pairs = trading_pairs self._domain = domain self._data_source = HookOdysseyPerpetualDataSource( - hook_odyssey_perpetual_eth_address=hook_odyssey_perpetual_eth_address, - hook_odyssey_perpetual_private_key=hook_odyssey_perpetual_private_key, + hook_odyssey_perpetual_signer_address=hook_odyssey_perpetual_signer_address, + hook_odyssey_perpetual_signer_pkey=hook_odyssey_perpetual_signer_pkey, hook_odyssey_api_key=hook_odyssey_api_key, + hook_odyssey_pool_address=hook_odyssey_pool_address, + hook_odyssey_pool_subaccount=hook_odyssey_pool_subaccount, + hook_odyssey_pool_trading_pair=hook_odyssey_pool_trading_pair, connector=self, + trade_fee_schema=self.trade_fee_schema(), domain=self.domain, trading_required=self._trading_required, ) @@ -262,6 +268,8 @@ async def _status_polling_loop_fetch_updates(self): async def _place_cancel(self, order_id: str, tracked_order: InFlightOrder): hook_order_hash = tracked_order.exchange_order_id + if hook_order_hash is None or hook_order_hash == 'None': + hook_order_hash = tracked_order.client_order_id return await self._data_source.cancel_order(hook_order_hash) async def _place_order( @@ -314,9 +322,6 @@ def buy( instrument_hash = perpetual_pair["instrumentHash"] subaccount = perpetual_pair["subaccount"] - if order_type in [OrderType.LIMIT, OrderType.LIMIT_MAKER]: - price = self.quantize_order_price(trading_pair, price) - amount = self.quantize_order_amount(trading_pair=trading_pair, amount=amount) nonce = get_tracking_nonce() order_id = self._data_source.order_hash( market_hash=market_hash, @@ -366,9 +371,6 @@ def sell( instrument_hash = perpetual_pair["instrumentHash"] subaccount = perpetual_pair["subaccount"] - if order_type in [OrderType.LIMIT, OrderType.LIMIT_MAKER]: - price = self.quantize_order_price(trading_pair, price) - amount = self.quantize_order_amount(trading_pair=trading_pair, amount=amount) nonce = get_tracking_nonce() order_id = self._data_source.order_hash( market_hash=market_hash, @@ -414,12 +416,13 @@ async def _update_balances(self): return balances = await self._data_source.get_balances() + equity = await self._data_source.get_account_equity() for trading_pair in self._trading_pairs: # Use the subaccount balance for the first found trading pair perpetual_pair = self._data_source.get_perpetual_pair_for_trading_pair(trading_pair) if perpetual_pair is not None: subaccount = perpetual_pair["subaccount"] - self._account_balances[perpetual_pair["baseCurrency"]] = balances[subaccount] + self._account_balances[perpetual_pair["baseCurrency"]] = equity[subaccount] self._account_available_balances[perpetual_pair["baseCurrency"]] = balances[subaccount] break @@ -482,14 +485,57 @@ def _configure_event_forwarders(self): self._forwarders.append(event_forwarder) self._data_source.add_listener(event_tag=AccountEvent.BalanceEvent, listener=event_forwarder) + event_forwarder = EventForwarder(to_function=self._process_position_event) + self._forwarders.append(event_forwarder) + self._data_source.add_listener(event_tag=AccountEvent.PositionUpdate, listener=event_forwarder) + def _process_balance_event(self, event: BalanceUpdateEvent): self._account_balances[event.asset_name] = event.total_balance self._account_available_balances[event.asset_name] = event.available_balance + def _process_position_event(self, position_update_event: PositionUpdateEvent): + self._last_received_message_timestamp = time.time() + position_key = self._perpetual_trading.position_key( + position_update_event.trading_pair, position_update_event.position_side + ) + if position_update_event.amount != Decimal("0"): + position = self._perpetual_trading.get_position( + trading_pair=position_update_event.trading_pair, side=position_update_event.position_side + ) + if position is not None: + position.update_position( + position_side=position_update_event.position_side, + unrealized_pnl=position_update_event.unrealized_pnl, + entry_price=position_update_event.entry_price, + amount=position_update_event.amount, + leverage=position_update_event.leverage, + ) + else: + safe_ensure_future(coro=self._update_positions()) + else: + self._perpetual_trading.remove_position(post_key=position_key) + def _process_user_order_update(self, order_update: OrderUpdate): tracked_order = self._order_tracker.all_updatable_orders.get(order_update.client_order_id) - if tracked_order is not None: + if tracked_order is None: + # Add the order update to the order tracker if active + if order_update.new_state in [OrderState.FILLED, OrderState.CANCELED, OrderState.FAILED]: + return + order_type = order_update.misc_updates.get("order_type") + trade_type = order_update.misc_updates.get("trade_type") + price = order_update.misc_updates.get("price") + amount = order_update.misc_updates.get("amount") + self.start_tracking_order( + order_id=order_update.client_order_id, + exchange_order_id=order_update.exchange_order_id, + trading_pair=order_update.trading_pair, + order_type=order_type, + trade_type=trade_type, + price=price, + amount=amount, + ) + else: self.logger().debug(f"Processing order update {order_update}\nUpdatable order {tracked_order.to_json()}") order_update_to_process = OrderUpdate( trading_pair=tracked_order.trading_pair, @@ -505,22 +551,4 @@ def _process_user_trade_update(self, trade_update: TradeUpdate): if tracked_order is not None: self.logger().debug(f"Processing trade update {trade_update}\nFillable order {tracked_order.to_json()}") - - # Fetch the current fee rates - maker_fee, taker_fee = self._data_source.get_fees() - fee_rate = maker_fee if tracked_order.is_maker else taker_fee - fee = TradeFeeBase.new_perpetual_fee(trade_type=tracked_order.trade_type, percent=fee_rate) - - fill_amount = trade_update.fill_base_amount - tracked_order.executed_amount_base - trade_update: TradeUpdate = TradeUpdate( - trade_id=trade_update.trade_id, - client_order_id=tracked_order.client_order_id, - exchange_order_id=trade_update.exchange_order_id, - trading_pair=tracked_order.trading_pair, - fill_timestamp=trade_update.fill_timestamp, - fill_price=trade_update.fill_price, - fill_base_amount=fill_amount, - fill_quote_amount=fill_amount * trade_update.fill_price, - fee=fee, - ) self._order_tracker.process_trade_update(trade_update) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py index 77d4fa498f5..d676cff5b78 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_graphql_executor.py @@ -3,6 +3,7 @@ from typing import Any, Callable, Dict, Optional from gql import Client, gql +from gql.client import AsyncClientSession from gql.transport.aiohttp import AIOHTTPTransport from gql.transport.websockets import WebsocketsTransport @@ -11,6 +12,10 @@ class BaseGraphQLExecutor(ABC): + @abstractmethod + def ws_client(self) -> Client: + raise NotImplementedError + @abstractmethod async def perpetual_pairs(self) -> Dict[str, Any]: raise NotImplementedError @@ -20,30 +25,30 @@ async def account_details(self) -> Dict[str, Any]: raise NotImplementedError @abstractmethod - async def subscribe_ticker(self, handler: Callable, symbol: str) -> None: + async def subscribe_ticker(self, session: AsyncClientSession, handler: Callable, symbol: str) -> None: raise NotImplementedError - async def subscribe_statistics(self, handler: Callable, symbol: str) -> None: + async def subscribe_statistics(self, session: AsyncClientSession, handler: Callable, symbol: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_bbo(self, handler: Callable, symbol: str) -> None: + async def subscribe_bbo(self, session: AsyncClientSession, handler: Callable, symbol: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_orderbook(self, handler: Callable, instrument_hash: str) -> None: + async def subscribe_orderbook(self, session: AsyncClientSession, handler: Callable, instrument_hash: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str) -> None: + async def subscribe_subaccount_orders(self, session: AsyncClientSession, handler: Callable, subaccount: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_subaccount_balances(self, handler: Callable, address: str) -> None: + async def subscribe_subaccount_balances(self, session: AsyncClientSession, handler: Callable, address: str) -> None: raise NotImplementedError @abstractmethod - async def subscribe_subaccount_positions(self, handler: Callable, address: str) -> None: + async def subscribe_subaccount_positions(self, session: AsyncClientSession, handler: Callable, address: str) -> None: raise NotImplementedError @abstractmethod @@ -72,7 +77,12 @@ def __init__(self, hook_odyssey_api_key: str, domain: Optional[str] = CONSTANTS. super().__init__() self._hook_odyssey_api_key = hook_odyssey_api_key self._domain = domain - self._client = None + + def ws_client(self) -> Client: + headers = {"X-HOOK-API-KEY": self._hook_odyssey_api_key} + url = CONSTANTS.WS_URLS[self._domain] + transport = WebsocketsTransport(url=url, headers=headers) + return Client(transport=transport, execute_timeout=CONSTANTS.DEFAULT_TIMEOUT) async def _execute( self, @@ -91,15 +101,6 @@ async def _execute( return result - async def _execute_subscription(self, subscription_query: str, variables: dict[str, Any] | None = None): - headers = {"X-HOOK-API-KEY": self._hook_odyssey_api_key} - url = CONSTANTS.WS_URLS[self._domain] - transport = WebsocketsTransport(url=url, headers=headers) - async with Client(transport=transport, execute_timeout=CONSTANTS.DEFAULT_TIMEOUT) as session: - subscription = gql(subscription_query) - async for result in session.subscribe(subscription, variable_values=variables): - yield result - async def perpetual_pairs(self) -> Dict[str, Any]: query = """ query PerpetualPairs { @@ -132,7 +133,7 @@ async def account_details(self) -> Dict[str, Any]: result = await self._execute(query=query, variables={}) return result - async def subscribe_ticker(self, handler: Callable, symbol: str): + async def subscribe_ticker(self, session: AsyncClientSession, handler: Callable, symbol: str): query = """ subscription onTickerEvent($symbol: String!) { ticker(symbol: $symbol) { @@ -142,10 +143,10 @@ async def subscribe_ticker(self, handler: Callable, symbol: str): } """ variables = {"symbol": symbol} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, symbol=symbol) - async def subscribe_statistics(self, handler: Callable, symbol: str): + async def subscribe_statistics(self, session: AsyncClientSession, handler: Callable, symbol: str): query = """ subscription onStatisticsEvent($symbol: String!) { statistics(symbol: $symbol) { @@ -157,10 +158,10 @@ async def subscribe_statistics(self, handler: Callable, symbol: str): } """ variables = {"symbol": symbol} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, symbol=symbol) - async def subscribe_bbo(self, handler: Callable, symbol: str) -> None: + async def subscribe_bbo(self, session: AsyncClientSession, handler: Callable, symbol: str) -> None: query = """ subscription onBboEvent($symbol: String!, $instrumentType: InstrumentType!) { bbo(symbol: $symbol, instrumentType: $instrumentType) { @@ -174,10 +175,10 @@ async def subscribe_bbo(self, handler: Callable, symbol: str) -> None: } """ variables = {"symbol": symbol, "instrumentType": "PERPETUAL"} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, symbol=symbol) - async def subscribe_orderbook(self, handler: Callable, instrument_hash: str): + async def subscribe_orderbook(self, session: AsyncClientSession, handler: Callable, instrument_hash: str): query = """ subscription onOrderbookEvent($instrumentHash: ID!) { orderbook(instrumentHash: $instrumentHash) { @@ -197,10 +198,10 @@ async def subscribe_orderbook(self, handler: Callable, instrument_hash: str): } """ variables = {"instrumentHash": instrument_hash} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, instrument_hash=instrument_hash) - async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str): + async def subscribe_subaccount_orders(self, session: AsyncClientSession, handler: Callable, subaccount: str): query = """ subscription onSubaccountOrderEvent($subaccount: BigInt!) { subaccountOrders(subaccount: $subaccount) { @@ -214,15 +215,17 @@ async def subscribe_subaccount_orders(self, handler: Callable, subaccount: str): remainingSize orderHash status + orderType + limitPrice } } } """ variables = {"subaccount": subaccount} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, subaccount=subaccount) - async def subscribe_subaccount_balances(self, handler: Callable, address: str): + async def subscribe_subaccount_balances(self, session: AsyncClientSession, handler: Callable, address: str): query = """ subscription onSubaccountBalanceEvent($address: Address!) { subaccountBalances(address: $address) { @@ -231,15 +234,16 @@ async def subscribe_subaccount_balances(self, handler: Callable, address: str): subaccount subaccountID balance + assetName } } } """ variables = {"address": address} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, address=address) - async def subscribe_subaccount_positions(self, handler: Callable, address: str): + async def subscribe_subaccount_positions(self, session: AsyncClientSession, handler: Callable, address: str): query = """ subscription onSubaccountPositionEvent($address: Address!) { subaccountPositions(address: $address) { @@ -258,7 +262,7 @@ async def subscribe_subaccount_positions(self, handler: Callable, address: str): } """ variables = {"address": address} - async for event in self._execute_subscription(subscription_query=query, variables=variables): + async for event in session.subscribe(gql(query), variable_values=variables): await handler(event=event, address=address) async def place_order( From b2f6508f28be008736de1f8de23089a071b72b15 Mon Sep 17 00:00:00 2001 From: Regynald Augustin Date: Sat, 6 Apr 2024 14:32:37 -0400 Subject: [PATCH 8/8] Reconnect to ws if no snapshots are received --- .../hook_odyssey_perpetual_data_source.py | 29 +++++++++++++++++-- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py index 074fbfdd7b6..0dbddca2421 100644 --- a/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py +++ b/hummingbot/connector/derivative/hook_odyssey_perpetual/hook_odyssey_perpetual_data_source.py @@ -106,6 +106,9 @@ def __init__( self._initial_subaccount_balances_event = asyncio.Event() self._initial_subaccount_positions_event = asyncio.Event() + # Track the timestamp of the last snapshot + self._last_orderbook_msg = time.time() + def pools_configured(self) -> bool: return all( [ @@ -155,13 +158,22 @@ def add_listener(self, event_tag, listener: EventListener): def remove_listener(self, event_tag, listener: EventListener): self._publisher.remove_listener(event_tag=event_tag, listener=listener) + async def orderbook_event_watcher(self): + while True: + if time.time() - self._last_orderbook_msg > 60: + self.logger().warning("Snapshot timeout detected, raising exception") + raise asyncio.TimeoutError("Snapshot timeout detected") + await asyncio.sleep(1) + async def subscription_connection(self, trading_pairs: List[str]): address = self._hook_odyssey_perpetual_signer_address if self.pools_configured(): address = self._hook_odyssey_perpetual_pool_address while True: + self._events_listening_tasks.clear() try: + self._events_listening_tasks.append(asyncio.create_task(self.orderbook_event_watcher())) client = self._graphql_executor.ws_client() async with client as session: for trading_pair in trading_pairs: @@ -243,7 +255,13 @@ async def subscription_connection(self, trading_pairs: List[str]): ) ) ) - await asyncio.gather(*self._events_listening_tasks) + while True: + # Check if any task has unexpectedly completed + if any(task.done() for task in self._events_listening_tasks): + raise Exception("A task unexpectedly completed") + + # A short delay before checking if the tasks are still running + await asyncio.sleep(0.5) except (ConnectionClosed, asyncio.TimeoutError, Exception) as e: self.logger().error(f"Subscription connection error: {str(e)}, attempting to reconnect...") for task in self._events_listening_tasks: @@ -427,6 +445,7 @@ async def _process_statistics_update(self, event: Dict[str, Any], symbol: str): async def _process_bbo_update(self, event: Dict[str, Any], symbol: str): bbo_event = event["bbo"] + self._last_orderbook_msg = time.time() if symbol not in self._symbols: return @@ -446,6 +465,7 @@ async def _process_orderbook_update(self, event: Dict[str, Any], instrument_hash trading_pair = self._instrument_hashes[instrument_hash] orderbook_event = event["orderbook"] + message_type = OrderBookMessageType.DIFF if orderbook_event["eventType"] == "SNAPSHOT": message_type = OrderBookMessageType.SNAPSHOT @@ -485,12 +505,15 @@ async def _process_orderbook_update(self, event: Dict[str, Any], instrument_hash async def _process_subaccount_orders_update(self, event: Dict[str, Any], subaccount: str): orders_event = event["subaccountOrders"] - if orders_event["eventType"] == "SNAPSHOT" and not self._initial_subaccount_orders_event.is_set(): - self._initial_subaccount_orders_event.set() + self._last_orderbook_msg = time.time() + orders = orders_event["orders"] if subaccount not in self._subaccounts: return trading_pair = self._subaccounts[subaccount] + if orders_event["eventType"] == "SNAPSHOT": + self._initial_subaccount_orders_event.set() + self.logger().info(f"SubaccountOrders snapshot: {trading_pair}") for order in orders: if order["status"] in ["PARTIALLY_MATCHED", "MATCHED"]: