config, readme, hyperliquid listener refactor

Author: merolish

apps/hip-3-pusher/README.md

@@ -0,0 +1,9 @@
+# HIP-3 Pusher
+
+`hip-3-pusher` is intended to serve as an oracle updater for
+[HIP-3 markets](https://hyperliquid.gitbook.io/hyperliquid-docs/hyperliquid-improvement-proposals-hips/hip-3-builder-deployed-perpetuals). 
+
+Currently it:
+- Sources market data from Hyperliquid, Pyth Lazer, and Pythnet
+- Supports KMS for signing oracle updates
+- Provides telemetry to Pyth's internal observability system

apps/hip-3-pusher/pyproject.toml

@@ -1,18 +1,17 @@
 [project]
 name = "hip-3-pusher"
-version = "0.1.3"
+version = "0.1.4"
 description = "Hyperliquid HIP-3 market oracle pusher"
 readme = "README.md"
-requires-python = ">=3.13"
+requires-python = "==3.13.*"
 dependencies = [
-    "asn1crypto>=1.5.1",
-    "boto3>=1.40.31",
-    "cryptography>=45.0.7",
-    "hyperliquid-python-sdk>=0.19.0",
-    "loguru>=0.7.3",
-    "opentelemetry-exporter-prometheus>=0.58b0",
-    "opentelemetry-sdk>=1.37.0",
-    "prometheus-client>=0.23.1",
-    "toml>=0.10.2",
-    "websockets>=15.0.1",
+    "asn1crypto~=1.5.1",
+    "boto3~=1.40.34",
+    "cryptography~=46.0.1",
+    "hyperliquid-python-sdk~=0.19.0",
+    "loguru~=0.7.3",
+    "opentelemetry-exporter-prometheus~=0.58b0",
+    "opentelemetry-sdk~=1.37.0",
+    "prometheus-client~=0.23.1",
+    "websockets~=15.0.1",
 ]

apps/hip-3-pusher/src/config.py

@@ -0,0 +1,45 @@
+from pydantic import BaseModel
+
+
+class KMSConfig(BaseModel):
+    enable_kms: bool
+    aws_region_name: str
+    key_path: str
+    access_key_id_path: str
+    secret_access_key_path: str
+
+
+class LazerConfig(BaseModel):
+    lazer_urls: list[str]
+    lazer_api_key: str
+    base_feed_id: int
+    base_feed_exponent: int
+    quote_feed_id: int
+    quote_feed_exponent: int
+
+
+class HermesConfig(BaseModel):
+    hermes_urls: list[str]
+    base_feed_id: str
+    base_feed_exponent: int
+    quote_feed_id: str
+    quote_feed_exponent: int
+
+
+class HyperliquidConfig(BaseModel):
+    hyperliquid_ws_urls: list[str]
+    market_name: str
+    market_symbol: str
+    use_testnet: bool
+    oracle_pusher_key_path: str
+    publish_interval: float
+    enable_publish: bool
+
+
+class Config(BaseModel):
+    stale_price_threshold_seconds: int
+    prometheus_port: int
+    hyperliquid: HyperliquidConfig
+    kms: KMSConfig
+    lazer: LazerConfig
+    hermes: HermesConfig

apps/hip-3-pusher/src/hermes_listener.py

@@ -4,17 +4,18 @@
 import time
 import websockets
 
+from config import Config
 from price_state import PriceState
 
 
 class HermesListener:
     """
     Subscribe to Hermes price updates for needed feeds.
     """
-    def __init__(self, config, price_state: PriceState):
-        self.hermes_urls = config["hermes"]["hermes_urls"]
-        self.base_feed_id = config["hermes"]["base_feed_id"]
-        self.quote_feed_id = config["hermes"]["quote_feed_id"]
+    def __init__(self, config: Config, price_state: PriceState):
+        self.hermes_urls = config.hermes.hermes_urls
+        self.base_feed_id = config.hermes.base_feed_id
+        self.quote_feed_id = config.hermes.quote_feed_id
         self.price_state = price_state
 
     def get_subscribe_request(self):

apps/hip-3-pusher/src/hyperliquid_listener.py

@@ -1,35 +1,71 @@
+import asyncio
+import json
+import websockets
 from loguru import logger
 import time
 
-from hyperliquid.info import Info
-from hyperliquid.utils.constants import TESTNET_API_URL, MAINNET_API_URL
-
+from config import Config
 from price_state import PriceState
 
+# This will be in config, but note here.
+# Other RPC providers exist but so far we've seen their support is incomplete.
+HYPERLIQUID_MAINNET_WS_URL = "wss://api.hyperliquid.xyz/ws"
+HYPERLIQUID_TESTNET_WS_URL = "wss://api.hyperliquid-testnet.xyz/ws"
+
 
 class HyperliquidListener:
     """
     Subscribe to any relevant Hyperliquid websocket streams
     See https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket
     """
-    def __init__(self, config: dict, price_state: PriceState):
-        self.market_symbol = config["hyperliquid"]["market_symbol"]
-        url = TESTNET_API_URL if config["hyperliquid"].get("use_testnet", True) else MAINNET_API_URL
-        self.info = Info(base_url=url)
+    def __init__(self, config: Config, price_state: PriceState):
+        self.hyperliquid_ws_urls = config.hyperliquid.hyperliquid_ws_urls
+        self.market_symbol = config.hyperliquid.market_symbol
         self.price_state = price_state
 
-    def subscribe(self):
-        self.info.subscribe({"type": "activeAssetCtx", "coin": self.market_symbol}, self.on_activeAssetCtx)
-
-    def on_activeAssetCtx(self, message):
-        """
-        Parse oraclePx and markPx from perp context update
-
-        :param message: activeAssetCtx websocket update message
-        :return: None
-        """
-        ctx = message["data"]["ctx"]
-        self.price_state.hl_oracle_price = ctx["oraclePx"]
-        self.price_state.hl_mark_price = ctx["markPx"]
-        logger.debug("on_activeAssetCtx: oraclePx: {} marketPx: {}", self.price_state.hl_oracle_price, self.price_state.hl_mark_price)
-        self.price_state.latest_hl_timestamp = time.time()
+    def get_subscribe_request(self, asset):
+        return {
+            "method": "subscribe",
+            "subscription": {"type": "activeAssetCtx", "coin": asset}
+        }
+
+    async def subscribe_all(self):
+        await asyncio.gather(*(self.subscribe_single(hyperliquid_ws_url) for hyperliquid_ws_url in self.hyperliquid_ws_urls))
+
+    async def subscribe_single(self, url):
+        while True:
+            try:
+                await self.subscribe_single_inner(url)
+            except websockets.ConnectionClosed:
+                logger.error("Connection to {} closed; retrying", url)
+            except Exception as e:
+                logger.exception("Error on {}: {}", url, e)
+
+    async def subscribe_single_inner(self, url):
+        async with websockets.connect(url) as ws:
+            subscribe_request = self.get_subscribe_request(self.market_symbol)
+            await ws.send(json.dumps(subscribe_request))
+            logger.info("Sent subscribe request to {}", url)
+
+            # listen for updates
+            async for message in ws:
+                try:
+                    data = json.loads(message)
+                    if "ctx" not in message:
+                        # TODO: Should check subscription status response
+                        logger.debug("HL update without ctx: {}", message)
+                        continue
+                    self.parse_hyperliquid_ws_message(data)
+                except json.JSONDecodeError as e:
+                    logger.error("Failed to decode JSON message: {}", e)
+
+    def parse_hyperliquid_ws_message(self, message):
+        try:
+            ctx = message["data"]["ctx"]
+            self.price_state.hl_oracle_price = ctx["oraclePx"]
+            self.price_state.hl_mark_price = ctx["markPx"]
+            logger.debug("on_activeAssetCtx: oraclePx: {} marketPx: {}", self.price_state.hl_oracle_price,
+                         self.price_state.hl_mark_price)
+            self.price_state.latest_hl_timestamp = time.time()
+        except Exception as e:
+            logger.error("parse_hyperliquid_ws_message error: message: {} e: {}", e)

apps/hip-3-pusher/src/kms_signer.py

@@ -10,18 +10,20 @@
 from hyperliquid.utils.signing import get_timestamp_ms, action_hash, construct_phantom_agent, l1_payload
 from loguru import logger
 
+from config import Config
+
 SECP256K1_N_HALF = SECP256K1_N // 2
 
 
 class KMSSigner:
-    def __init__(self, config):
-        use_testnet = config["hyperliquid"]["use_testnet"]
+    def __init__(self, config: Config):
+        use_testnet = config.hyperliquid.use_testnet
         url = TESTNET_API_URL if use_testnet else MAINNET_API_URL
         self.oracle_publisher_exchange: Exchange = Exchange(wallet=None, base_url=url)
         self.client = self._init_client(config)
 
         # Fetch public key once so we can derive address and check recovery id
-        key_path = config["kms"]["key_path"]
+        key_path = config.kms.key_path
         self.key_id = open(key_path, "r").read().strip()
         self.pubkey_der = self.client.get_public_key(KeyId=self.key_id)["PublicKey"]
         # Construct eth address to log
@@ -35,10 +37,10 @@ def __init__(self, config):
         logger.info("KMSSigner address: {}", self.address)
 
     def _init_client(self, config):
-        aws_region_name = config["kms"]["aws_region_name"]
-        access_key_id_path = config["kms"]["access_key_id_path"]
+        aws_region_name = config.kms.aws_region_name
+        access_key_id_path = config.kms.access_key_id_path
         access_key_id = open(access_key_id_path, "r").read().strip()
-        secret_access_key_path = config["kms"]["secret_access_key_path"]
+        secret_access_key_path = config.kms.secret_access_key_path
         secret_access_key = open(secret_access_key_path, "r").read().strip()
 
         return boto3.client(

apps/hip-3-pusher/src/lazer_listener.py

@@ -4,18 +4,19 @@
 import time
 import websockets
 
+from config import Config
 from price_state import PriceState
 
 
 class LazerListener:
     """
     Subscribe to Lazer price updates for needed feeds.
     """
-    def __init__(self, config, price_state: PriceState):
-        self.lazer_urls = config["lazer"]["lazer_urls"]
-        self.api_key = config["lazer"]["lazer_api_key"]
-        self.base_feed_id = config["lazer"]["base_feed_id"]
-        self.quote_feed_id = config["lazer"]["quote_feed_id"]
+    def __init__(self, config: Config, price_state: PriceState):
+        self.lazer_urls = config.lazer.lazer_urls
+        self.api_key = config.lazer.lazer_api_key
+        self.base_feed_id = config.lazer.base_feed_id
+        self.quote_feed_id = config.lazer.quote_feed_id
         self.price_state = price_state
 
     def get_subscribe_request(self, subscription_id: int):
@@ -52,7 +53,7 @@ async def subscribe_single_inner(self, router_url):
             subscribe_request = self.get_subscribe_request(1)
 
             await ws.send(json.dumps(subscribe_request))
-            logger.info("Sent Lazer subscribe request to {}", self.lazer_urls[0])
+            logger.info("Sent Lazer subscribe request to {}", router_url)
 
             # listen for updates
             async for message in ws:

apps/hip-3-pusher/src/main.py

@@ -3,8 +3,9 @@
 from loguru import logger
 import os
 import sys
-import toml
+import tomllib
 
+from config import Config
 from hyperliquid_listener import HyperliquidListener
 from lazer_listener import LazerListener
 from hermes_listener import HermesListener
@@ -21,8 +22,10 @@ def load_config():
         help="hip3-agent config file",
     )
     config_path = parser.parse_args().config
-    with open(config_path, "r") as config_file:
-        config = toml.load(config_file)
+    with open(config_path, "rb") as config_file:
+        config_toml = tomllib.load(config_file)
+        config = Config(**config_toml)
+        logger.debug("Config loaded: {}", config)
     return config
 
 
@@ -46,10 +49,9 @@ async def main():
     lazer_listener = LazerListener(config, price_state)
     hermes_listener = HermesListener(config, price_state)
 
-    # TODO: Probably pull this out of the sdk so we can handle reconnects.
-    hyperliquid_listener.subscribe()
     await asyncio.gather(
         publisher.run(),
+        hyperliquid_listener.subscribe_all(),
         lazer_listener.subscribe_all(),
         hermes_listener.subscribe_all(),
     )

apps/hip-3-pusher/src/metrics.py

@@ -3,14 +3,16 @@
 from opentelemetry.metrics import get_meter_provider, set_meter_provider
 from opentelemetry.sdk.metrics import MeterProvider
 
+from config import Config
+
 METER_NAME = "hip3pusher"
 
 
 class Metrics:
-    def __init__(self, config):
+    def __init__(self, config: Config):
         # Adapted from opentelemetry-exporter-prometheus example code.
         # Start Prometheus client
-        start_http_server(port=config["prometheus_port"])
+        start_http_server(port=config.prometheus_port)
         # Exporter to export metrics to Prometheus
         reader = PrometheusMetricReader()
         # Meter is responsible for creating and recording metrics

apps/hip-3-pusher/src/price_state.py

@@ -1,31 +1,33 @@
 from loguru import logger
 import time
 
+from config import Config
+
 DEFAULT_STALE_PRICE_THRESHOLD_SECONDS = 5
 
 
 class PriceState:
     """
     Maintain latest prices seen across listeners and publisher.
     """
-    def __init__(self, config):
-        self.stale_price_threshold_seconds = config.get("stale_price_threshold_seconds", DEFAULT_STALE_PRICE_THRESHOLD_SECONDS)
+    def __init__(self, config: Config):
+        self.stale_price_threshold_seconds = config.stale_price_threshold_seconds
         now = time.time()
 
         self.hl_oracle_price = None
         self.hl_mark_price = None
         self.latest_hl_timestamp = now
 
         self.lazer_base_price = None
-        self.lazer_base_exponent = config["lazer"]["base_feed_exponent"]
+        self.lazer_base_exponent = config.lazer.base_feed_exponent
         self.lazer_quote_price = None
-        self.lazer_quote_exponent = config["lazer"]["quote_feed_exponent"]
+        self.lazer_quote_exponent = config.lazer.quote_feed_exponent
         self.latest_lazer_timestamp = now
 
         self.hermes_base_price = None
-        self.hermes_base_exponent = config["hermes"]["base_feed_exponent"]
+        self.hermes_base_exponent = config.hermes.base_feed_exponent
         self.hermes_quote_price = None
-        self.hermes_quote_exponent = config["hermes"]["quote_feed_exponent"]
+        self.hermes_quote_exponent = config.hermes.quote_feed_exponent
         self.latest_hermes_timestamp = now
 
     def get_current_oracle_price(self):