Merge pull request #54 from pyth-network/hermes

hermes support

examples/read_hermes.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+
+import asyncio
+
+from pythclient.hermes import HermesClient
+
+async def get_hermes_prices():
+    hermes_client = HermesClient([])
+    feed_ids = await hermes_client.get_price_feed_ids()
+    feed_ids_rel = feed_ids[:2]
+    version_http = 1
+    version_ws = 1
+
+    hermes_client.add_feed_ids(feed_ids_rel)
+
+    prices_latest = await hermes_client.get_all_prices(version=version_http)
+
+    for feed_id, price_feed in prices_latest.items():
+        print("Initial prices")
+        print(f"Feed ID: {feed_id}, Price: {price_feed['price'].price}, Confidence: {price_feed['price'].conf}, Time: {price_feed['price'].publish_time}")
+
+    print("Starting web socket...")
+    ws_call = hermes_client.ws_pyth_prices(version=version_ws)
+    ws_task = asyncio.create_task(ws_call)
+
+    while True:
+        await asyncio.sleep(5)
+        if ws_task.done():
+            break
+        print("Latest prices:")
+        for feed_id, price_feed in hermes_client.prices_dict.items():
+            print(f"Feed ID: {feed_id}, Price: {price_feed['price'].price}, Confidence: {price_feed['price'].conf}, Time: {price_feed['price'].publish_time}")
+
+asyncio.run(get_hermes_prices())

pythclient/hermes.py

@@ -0,0 +1,192 @@
+import asyncio
+from typing import TypedDict
+import httpx
+import os
+import json
+import websockets
+
+from .price_feeds import Price
+
+HERMES_ENDPOINT_HTTPS = "https://hermes.pyth.network/"
+HERMES_ENDPOINT_WSS = "wss://hermes.pyth.network/ws"
+
+
+class PriceFeed(TypedDict):
+    feed_id: str
+    price: Price
+    ema_price: Price
+    update_data: list[str]
+
+
+def parse_unsupported_version(version):
+    if isinstance(version, int):
+        raise ValueError("Version number {version} not supported")
+    else:
+        raise TypeError("Version must be an integer")
+
+
+class HermesClient:
+    def __init__(self, feed_ids: list[str], endpoint=HERMES_ENDPOINT_HTTPS, ws_endpoint=HERMES_ENDPOINT_WSS, feed_batch_size=100):
+        self.feed_ids = feed_ids
+        self.pending_feed_ids = feed_ids
+        self.prices_dict: dict[str, PriceFeed] = {}
+        self.endpoint = endpoint
+        self.ws_endpoint = ws_endpoint
+        self.feed_batch_size = feed_batch_size # max number of feed IDs to query at once in https requests
+
+    async def get_price_feed_ids(self) -> list[str]:
+        """
+        Queries the Hermes https endpoint for a list of the IDs of all Pyth price feeds.
+        """
+
+        url = os.path.join(self.endpoint, "api/price_feed_ids")
+
+        async with httpx.AsyncClient() as client:
+            data = (await client.get(url)).json()
+
+        return data
+
+    def add_feed_ids(self, feed_ids: list[str]):
+        self.feed_ids += feed_ids
+        self.feed_ids = list(set(self.feed_ids))
+        self.pending_feed_ids += feed_ids
+
+    @staticmethod
+    def extract_price_feed_v1(data: dict) -> PriceFeed:
+        """
+        Extracts PriceFeed object from the v1 JSON response (individual price feed) from Hermes.
+        """
+        price = Price.from_dict(data["price"])
+        ema_price = Price.from_dict(data["ema_price"])
+        update_data = data["vaa"]
+        price_feed = {
+            "feed_id": data["id"],
+            "price": price,
+            "ema_price": ema_price,
+            "update_data": [update_data],
+        }
+        return price_feed
+
+    @staticmethod
+    def extract_price_feed_v2(data: dict) -> list[PriceFeed]:
+        """
+        Extracts PriceFeed objects from the v2 JSON response (array of price feeds) from Hermes.
+        """
+        update_data = data["binary"]["data"]
+
+        price_feeds = []
+
+        for feed in data["parsed"]:
+            price = Price.from_dict(feed["price"])
+            ema_price = Price.from_dict(feed["ema_price"])
+            price_feed = {
+                "feed_id": feed["id"],
+                "price": price,
+                "ema_price": ema_price,
+                "update_data": update_data,
+            }
+            price_feeds.append(price_feed)
+
+        return price_feeds
+
+    async def get_pyth_prices_latest(self, feedIds: list[str], version=2) -> list[PriceFeed]:
+        """
+        Queries the Hermes https endpoint for the latest price feeds for a list of Pyth feed IDs.
+        """
+        if version==1:
+            url = os.path.join(self.endpoint, "api/latest_price_feeds")
+            params = {"ids[]": feedIds, "binary": "true"}
+        elif version==2:
+            url = os.path.join(self.endpoint, "v2/updates/price/latest")
+            params = {"ids[]": feedIds, "encoding": "base64", "parsed": "true"}
+        else:
+            parse_unsupported_version(version)
+
+        async with httpx.AsyncClient() as client:
+            data = (await client.get(url, params=params)).json()
+
+        if version==1:
+            results = []
+            for res in data:
+                price_feed = self.extract_price_feed_v1(res)
+                results.append(price_feed)
+        elif version==2:
+            results = self.extract_price_feed_v2(data)
+
+        return results
+
+    async def get_pyth_price_at_time(self, feed_id: str, timestamp: int, version=2) -> PriceFeed:
+        """
+        Queries the Hermes https endpoint for the price feed for a Pyth feed ID at a given timestamp.
+        """
+        if version==1:
+            url = os.path.join(self.endpoint, "api/get_price_feed")
+            params = {"id": feed_id, "publish_time": timestamp, "binary": "true"}
+        elif version==2:
+            url = os.path.join(self.endpoint, f"v2/updates/price/{timestamp}")
+            params = {"ids[]": [feed_id], "encoding": "base64", "parsed": "true"}
+        else:
+            parse_unsupported_version(version)
+
+        async with httpx.AsyncClient() as client:
+            data = (await client.get(url, params=params)).json()
+
+        if version==1:
+            price_feed = self.extract_price_feed_v1(data)
+        elif version==2:
+            price_feed = self.extract_price_feed_v2(data)[0]
+
+        return price_feed
+
+    async def get_all_prices(self, version=2) -> dict[str, PriceFeed]:
+        """
+        Queries the Hermes http endpoint for the latest price feeds for all feed IDs in the class object.
+
+        There is a limit on the number of feed IDs that can be queried at once, so this function queries the feed IDs in batches.
+        """
+        pyth_prices_latest = []
+        i = 0
+        while len(self.feed_ids[i : i + self.feed_batch_size]) > 0:
+            pyth_prices_latest += await self.get_pyth_prices_latest(
+                self.feed_ids[i : i + self.feed_batch_size],
+                version=version,
+            )
+            i += self.feed_batch_size
+
+        return dict([(feed['feed_id'], feed) for feed in pyth_prices_latest])
+
+    async def ws_pyth_prices(self, version=1):
+        """
+        Opens a websocket connection to Hermes for latest prices for all feed IDs in the class object.
+        """
+        if version != 1:
+            parse_unsupported_version(version)
+
+        async with websockets.connect(self.ws_endpoint) as ws:
+            while True:
+                # add new price feed ids to the ws subscription
+                if len(self.pending_feed_ids) > 0:
+                    json_subscribe = {
+                        "ids": self.pending_feed_ids,
+                        "type": "subscribe",
+                        "verbose": True,
+                        "binary": True,
+                    }
+                    await ws.send(json.dumps(json_subscribe))
+                    self.pending_feed_ids = []
+
+                msg = json.loads(await ws.recv())
+                if msg.get("type") == "response":
+                    if msg.get("status") != "success":
+                        raise Exception("Error in subscribing to websocket")
+                try:
+                    if msg["type"] != "price_update":
+                        continue
+
+                    feed_id = msg["price_feed"]["id"]
+                    new_feed = msg["price_feed"]
+
+                    self.prices_dict[feed_id] = self.extract_price_feed_v1(new_feed)
+
+                except Exception as e:
+                    raise Exception(f"Error in price_update message: {msg}") from e

pythclient/price_feeds.py

@@ -1,7 +1,7 @@
 import base64
 import binascii
 from struct import unpack
-from typing import List, Literal, Optional, Union, cast
+from typing import List, Literal, Optional, Union, cast, TypedDict
 
 from Crypto.Hash import keccak
 from loguru import logger
@@ -17,6 +17,11 @@
 
 MAX_MESSAGE_IN_SINGLE_UPDATE_DATA = 255
 
+class PriceDict(TypedDict):
+    conf: str
+    expo: int
+    price: str
+    publish_time: int
 
 class Price:
     def __init__(self, conf, expo, price, publish_time) -> None:
@@ -35,6 +40,16 @@ def to_dict(self):
             "price": self.price,
             "publish_time": self.publish_time,
         }
+
+    @staticmethod
+    def from_dict(price_dict: PriceDict):
+        return Price(
+            conf=int(price_dict["conf"]),
+            expo=price_dict["expo"],
+            price=int(price_dict["price"]),
+            publish_time=price_dict["publish_time"],
+        )
+
 
 
 class PriceUpdate:

setup.py

@@ -1,13 +1,13 @@
 from setuptools import setup
 
-requirements = ['aiodns', 'aiohttp>=3.7.4', 'backoff', 'base58', 'flake8', 'loguru', 'typing-extensions', 'pytz', 'pycryptodome']
+requirements = ['aiodns', 'aiohttp>=3.7.4', 'backoff', 'base58', 'flake8', 'loguru', 'typing-extensions', 'pytz', 'pycryptodome', 'httpx', 'websockets']
 
 with open('README.md', 'r', encoding='utf-8') as fh:
     long_description = fh.read()
 
 setup(
     name='pythclient',
-    version='0.1.22',
+    version='0.1.23',
     packages=['pythclient'],
     author='Pyth Developers',
     author_email='contact@pyth.network',