hermes support

examples/read_hermes.py

@@ -0,0 +1,36 @@
+#!/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 = 2
+    version_ws = 1
+
+    hermes_client.add_feed_ids(feed_ids_rel)
+
+    prices_latest = await hermes_client.get_all_prices(version=version_http)
+
+    sd = list(prices_latest.keys())[0]
+
+    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='[email protected]',

tests/test_hermes.py

@@ -0,0 +1,106 @@
+import pytest
+
+from pytest_mock import MockerFixture
+
+from mock import AsyncMock
+
+from pythclient.hermes import HermesClient, PriceFeed
+
+@pytest.fixture
+def feed_ids():
+    return ["e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43"]
+
+@pytest.fixture
+def hermes_client(feed_ids):
+    return HermesClient(feed_ids)
+
+@pytest.fixture
+def data_v1():
+    return {
+        "ema_price": {
+            "conf": "509500001",
+            "expo": -8,
+            "price": "2920679499999",
+            "publish_time": 1708363256
+        },
+        "id": "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43",
+        "metadata": {
+            "emitter_chain": 26,
+            "prev_publish_time": 1708363256,
+            "price_service_receive_time": 1708363256,
+            "slot": 85480034
+        },
+        "price": {
+            "conf": "509500001",
+            "expo": -8,
+            "price": "2920679499999",
+            "publish_time": 1708363256
+        },
+        "vaa": "UE5BVQEAAAADuAEAAAADDQC1H7meY5fTed0FsykIb8dt+7nKpbuzfvU2DplDi+dcUl8MC+UIkS65+rkiq+zmNBxE2gaxkBkjdIicZ/fBo+X7AAEqp+WtlWb84np8jJfLpuQ2W+l5KXTigsdAhz5DyVgU3xs+EnaIZxBwcE7EKzjMam+V9rlRy0CGsiQ1kjqqLzfAAQLsoVO0Vu5gVmgc8XGQ7xYhoz36rsBgMjG+e3l/B01esQi/KzPuBf/Ar8Sg5aSEOvEU0muSDb+KIr6d8eEC+FtcAAPZEaBSt4ysXVL84LUcJemQD3SiG30kOfUpF8o7/wI2M2Jf/LyCsbKEQUyLtLbZqnJBSfZJR5AMsrnHDqngMLEGAAY4UDG9GCpRuPvg8hOlsrXuPP3zq7yVPqyG0SG+bNo8rEhP5b1vXlHdG4bZsutX47d5VZ6xnFROKudx3T3/fnWUAQgAU1+kUFc3e0ZZeX1dLRVEryNIVyxMQIcxWwdey+jlIAYowHRM0fJX3Scs80OnT/CERwh5LMlFyU1w578NqxW+AQl2E/9fxjgUTi8crOfDpwsUsmOWw0+Q5OUGhELv/2UZoHAjsaw9OinWUggKACo4SdpPlHYldoWF+J2yGWOW+F4iAQre4c+ocb6a9uSWOnTldFkioqhd9lhmV542+VonCvuy4Tu214NP+2UNd/4Kk3KJCf3iziQJrCBeLi1cLHdLUikgAQtvRFR/nepcF9legl+DywAkUHi5/1MNjlEQvlHyh2XbMiS85yu7/9LgM6Sr+0ukfZY5mSkOcvUkpHn+T+Nw/IrQAQ7lty5luvKUmBpI3ITxSmojJ1aJ0kj/dc0ZcQk+/qo0l0l3/eRLkYjw5j+MZKA8jEubrHzUCke98eSoj8l08+PGAA+DAKNtCwNZe4p6J1Ucod8Lo5RKFfA84CPLVyEzEPQFZ25U9grUK6ilF4GhEia/ndYXLBt3PGW3qa6CBBPM7rH3ABGAyYEtUwzB4CeVedA5o6cKpjRkIebqDNSOqltsr+w7kXdfFVtsK2FMGFZNt5rbpIR+ppztoJ6eOKHmKmi9nQ99ARKkTxRErOs9wJXNHaAuIRV38o1pxRrlQRzGsRuKBqxcQEpC8OPFpyKYcp6iD5l7cO/gRDTamLFyhiUBwKKMP07FAWTEJv8AAAAAABrhAfrtrFhR4yubI7X5QRqMK6xKrj7U3XuBHdGnLqSqcQAAAAAAGp0GAUFVV1YAAAAAAAUYUmIAACcQBsfKUtr4PgZbIXRxRESU79PjE4IBAFUA5i32yLSoX+GmfbRNwS3l2zMPesZrctxliv7fD0pBW0MAAAKqqMJFwAAAAAAqE/NX////+AAAAABkxCb7AAAAAGTEJvoAAAKqIcWxYAAAAAAlR5m4CP/mPsh1IezjYpDlJ4GRb5q4fTs2LjtyO6M0XgVimrIQ4kSh1qg7JKW4gbGkyRntVFR9JO/GNd3FPDit0BK6M+JzXh/h12YNCz9wxlZTvXrNtWNbzqT+91pvl5cphhSPMfAHyEzTPaGR9tKDy9KNu56pmhaY32d2vfEWQmKo22guegeR98oDxs67MmnUraco46a3zEnac2Bm80pasUgMO24="
+    }
+
+@pytest.fixture
+def data_v2():
+    return {
+        "binary": {
+            "encoding": "hex",
+            "data": [
+            "504e41550100000003b801000000030d014016474bab1868acfe943cdcd3cf7a8b7ccfaf6f2a31870694d11c441505d0552a42f57df50093df73eca16fad7ae3d768b0dd0e64dbaf71579fd5d05c46a5f20002098e46154c00ee17e878295edaca5decd18f7a1e9a1f0576ca090219f350118d1a4a0cc94b853c8ae1d5064439e719c953e61450745cf10086c37ec93d878b610003edf89d49fe5bb035d3cab5f3960ca5c7be37073b6680afb0f154ec684990923330f6db1fced4680dcfce8664c9d757fe2e8ca84aec8950004371ab794979db7101068a0231af6701f5fbfe55ac7dd31d640dd17f2fa92a10450d7a6e5db03c7c1f90131452ed1e3290fbbf00bc8528f616e81771460b2c307e02db811a84545180620107ab6ea34d72541f44cf34c8e919b9ef336eef9774ee4cf3d5c7cc71f5f90e49d23a05878e2e278402aff8217df84f9ce3ae782c389b3230d09e9e66fada355d6600084018b5993c68c4d616a570925e63a7c82c5444aee9a0f6153bd724e0755d3086374c9cf4e6ec2f08ab9c914b4cd3868e250ad4f946219cc2af0a31936cd38147000a079d8fb93db9c82263556dfd768b6173da40d35ea4691d21de59cf207381b5a05cb693fd4a75cb2b190c0270f2ddc14335adca66bcd5a634bf316a4385e97250010bf6dfa12e7820c58514c74ec21029d5c11f98a586743b2da9d2e20d8d78b44bd3730af5c6428c1ad865cb9d94ee795d059b3b51bb1e7bc8f81d52e5db18167648010c8558ac8aefd43cf489bce260afaee270d36fd1a34923439261fc8220cb33f30521cfefebfe0d7cf21d3aaa60c9149f8ab085c90b0509ad2850efe01fc618ccec010d6bc67036011a75277ca476ca1f4d962ca0d861805a94c6353ad0ff6ae17263bc5401e7d7ee3f3010f77c6349ff264c4185b167f32108c7de9743f7a258c62d03000e63f823f4b8f2cb1d158aac8f7ba0e45227b6d99106831a50729825bf8b97969503f55bc33778ef6c21e696a99d304b72c9e5ca3941dd178a7fc5367aed7d0e00010f22ccd76becc94aec99ff0bb1fce128cb25644268c65ac8f2bf5fe357910942381e184a62e8a768d5be200e21e40a34047a6e5cd981d2380de7eb4aa46a15ce0a00127957a07e69f8af6f8752a09f24dde0d43277c80d3bc24f09a281e5e68878d0ea0445b356257e25e80422ddff2f799bb732eafdeee43fc14c21d4eda349a547010165d38df800000000001ae101faedac5851e32b9b23b5f9411a8c2bac4aae3ed4dd7b811dd1a72ea4aa7100000000027a3abd0141555756000000000007823fd000002710b939e515de35dd00cf7feaba6be6aed77c83e09901005500e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43000004b868a1a543000000009ea4861cfffffff80000000065d38df80000000065d38df7000004bcd90bec4000000000c41abcc80ab559eef775bd945c821d89ceba075f3c60f2dba713f2f7ed0d210ea03ee4bead9c9b6ffd8fff45f0826e6950c44a8a7e0eac9b5bc1f2bdf276965107fc612f72a05bd37ca85017dc13b01fa5d434887f33527d87c34f1caf4ed69501a6972959e7faf96a6bc43c0d08e2b1a095c50ef6609bf81b7661102f69acb46430115e301f1ebda0f008438e31564240e1cbc9092db20b73bfc8dd832b6467fd242f0043a167ccafbc0ba479d38be012ad1d75f35e2681754e78e1f10096a55f65512fe381238a67ffce0970"
+            ]
+        },
+        "parsed": [
+            {
+                "id": "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43",
+                "price": {
+                    "price": "5190075917635",
+                    "conf": "2661582364",
+                    "expo": -8,
+                    "publish_time": 1708363256
+                },
+                "ema_price": {
+                    "price": "5209141800000",
+                    "conf": "3290086600",
+                    "expo": -8,
+                    "publish_time": 1708363256
+                },
+                "metadata": {
+                    "slot": 125976528,
+                    "proof_available_time": 1708363257,
+                    "prev_publish_time": 1708363255
+                }
+            }
+        ]
+    }
+
+@pytest.fixture
+def mock_get_price_feed_ids(mocker: MockerFixture):
+    async_mock = AsyncMock()
+    mocker.patch('pythclient.hermes.HermesClient.get_price_feed_ids', side_effect=async_mock)
+    return async_mock
+
+@pytest.mark.asyncio
+async def test_hermes_add_feed_ids(hermes_client: HermesClient, mock_get_price_feed_ids: AsyncMock):
+    mock_get_price_feed_ids.return_value = ["ff61491a931112ddf1bd8147cd1b641375f79f5825126d665480874634fd0ace"]
+
+    feed_ids = await hermes_client.get_price_feed_ids()
+
+    feed_ids_pre = hermes_client.feed_ids
+    pending_feed_ids_pre = hermes_client.pending_feed_ids
+
+    hermes_client.add_feed_ids(feed_ids)
+
+    assert set(hermes_client.feed_ids) == set(feed_ids_pre + feed_ids)
+    assert set(hermes_client.pending_feed_ids) == set(pending_feed_ids_pre + feed_ids)
+
+def test_hermes_extract_price_feed_v1(hermes_client: HermesClient, data_v1: dict):
+    price_feed = hermes_client.extract_price_feed_v1(data_v1)
+
+    assert isinstance(price_feed, dict)
+    assert set(price_feed.keys()) == set(PriceFeed.__annotations__.keys())        
+
+def test_hermes_extract_price_feed_v2(hermes_client: HermesClient, data_v2: dict):
+    price_feed = hermes_client.extract_price_feed_v2(data_v2)
+
+    assert isinstance(price_feed, list)
+    assert isinstance(price_feed[0], dict)
+    assert set(price_feed[0].keys()) == set(PriceFeed.__annotations__.keys())