Refactor and add Pyth replicator provider (#11)

* Add provider_engine to allow multiple providers

* Refactor and add Pythnet replicator

* Cleanup

* Update coingecko product

* Minor code refactor

* Improve logs

Author: ali-behjati

config/config.toml

@@ -1,10 +1,24 @@
+[publisher]
+
+# Set it to either 'coin_gecko' or 'pyth_replicator'. You need to provide
+# the configuration for the chosen engine as described below.
+provider_engine = 'pyth_replicator'
+
+product_update_interval_secs = 10
+
 [publisher.pythd]
 endpoint = 'ws://127.0.0.1:8910'
 
-[publisher.coin_gecko]
-update_interval_secs = 15
-confidence_ratio_bps = 10
+# [publisher.coin_gecko]
+# update_interval_secs = 15
+# confidence_ratio_bps = 10
+
+# [[publisher.coin_gecko.products]]
+# symbol = 'Crypto.BTC/USD'
+# coin_gecko_id = 'bitcoin'
 
-[[ publisher.products ]]
-pythd_symbol = 'Crypto.BTC/USD'
-coin_gecko_id = 'bitcoin'
+[publisher.pyth_replicator]
+http_endpoint = 'https://pythnet.rpcpool.com'
+ws_endpoint = 'wss://pythnet.rpcpool.com'
+first_mapping = 'AHtgzX45WTKfkPG53L6WYhGEXwQkN1BVknET3sVsLL8J'
+program_key = 'FsJ3A3u2vn5cTVofAjvy6y5kwABJAqYWpe4975bi2epH'

publisher/__main__.py

@@ -6,15 +6,15 @@
 import typed_settings as ts
 import click
 import logging
-from structlog import get_logger
+import structlog
 
 _DEFAULT_CONFIG_PATH = os.path.join("config", "config.toml")
 
-logging.basicConfig(
-    level=logging.DEBUG,
-)
 
-log = get_logger()
+log_level = logging._nameToLevel[os.environ.get("LOG_LEVEL", "DEBUG").upper()]
+structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(log_level))
+
+log = structlog.get_logger()
 
 
 @click.command()

publisher/coin_gecko.py

@@ -3,9 +3,9 @@
 
 
 @ts.settings
-class Product:
-    # The value of attr_dict["symbol"] for this product, which will be used to retrieve the price account
-    pythd_symbol: str
+class CoinGeckoProduct:
+    # Symbol name. e.g., Crypto.BTC/USD
+    symbol: str
     # The CoinGecko API ID for this product, used to query reference prices
     coin_gecko_id: str
 
@@ -17,31 +17,28 @@ class Pythd:
 
 
 @ts.settings
-class CoinGecko:
+class CoinGeckoConfig:
     # How often to poll CoinGecko for price information
     update_interval_secs: int
     # The confidence interval rate (to the price) in basis points to use for CoinGecko updates
     confidence_ratio_bps: int
+    products: List[CoinGeckoProduct]
 
 
 @ts.settings
-class AUST:
-    # The Terra RPC node to use to query contracts
-    terra_rpc_node: str
-    # The Chain ID to connect to
-    chain_id: str
-    # The address of the Anchor Money Market contract to query the AUST exchange rate from
-    anchor_money_market_contract_address: str
-    # How often to query the exchange rate from the Anchor Money Market contract
-    update_interval_secs: int
-    # The Pythd symbol
-    pythd_symbol: str
-    # The confidence interval in basis points
-    confidence_bps: int
+class PythReplicatorConfig:
+    http_endpoint: str
+    ws_endpoint: str
+    first_mapping: str
+    program_key: str
+    staleness_time_in_secs: int = ts.option(default=30)
+    account_update_interval_secs: int = ts.option(default=300)
 
 
 @ts.settings
 class Config:
+    provider_engine: str
     pythd: Pythd
-    products: List[Product]
-    coin_gecko: Optional[CoinGecko] = ts.option(default=None)
+    product_update_interval_secs: int = ts.option(default=60)
+    coin_gecko: Optional[CoinGeckoConfig] = ts.option(default=None)
+    pyth_replicator: Optional[PythReplicatorConfig] = ts.option(default=None)

publisher/config.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+from typing import List, Optional, Protocol
+
+Symbol = str
+
+
+@dataclass
+class Price:
+    price: float
+    conf: float
+
+
+class Provider(Protocol):
+    def upd_products(self, product_symbols: List[Symbol]):
+        ...
+
+    def start(self):
+        ...
+
+    def latest_price(self, symbol: Symbol) -> Optional[Price]:
+        ...

publisher/provider.py

@@ -0,0 +1,65 @@
+import asyncio
+from typing import Dict, List, Optional
+from pycoingecko import CoinGeckoAPI
+from structlog import get_logger
+
+from publisher.provider import Price, Provider, Symbol
+from ..config import CoinGeckoConfig
+
+log = get_logger()
+
+Id = str  # The "API id" of the CoinGecko price, listed on CoinGecko page for each coin.
+
+USD = "usd"
+
+
+class CoinGecko(Provider):
+    def __init__(self, config: CoinGeckoConfig) -> None:
+        self._api: CoinGeckoAPI = CoinGeckoAPI()
+        self._prices: Dict[Id, float] = {}
+        self._symbol_to_id: Dict[Symbol, Id] = {
+            product.symbol: product.coin_gecko_id for product in config.products
+        }
+        self._config = config
+
+    def start(self) -> None:
+        asyncio.create_task(self._update_loop())
+
+    def upd_products(self, product_symbols: List[Symbol]) -> None:
+        new_prices = {}
+        for coin_gecko_product in self._config.products:
+            if coin_gecko_product.symbol in product_symbols:
+                id = coin_gecko_product.coin_gecko_id
+                new_prices[id] = self._prices.get(id, None)
+            else:
+                raise ValueError(
+                    f"{coin_gecko_product.symbol} not found in available products"
+                )
+
+        self._prices = new_prices
+
+    async def _update_loop(self) -> None:
+        while True:
+            self._update_prices()
+            await asyncio.sleep(self._config.update_interval_secs)
+
+    def _update_prices(self) -> None:
+        result = self._api.get_price(
+            ids=list(self._prices.keys()), vs_currencies=USD, precision=18
+        )
+        for id_, prices in result.items():
+            self._prices[id_] = prices[USD]
+        log.info("updated prices from CoinGecko", prices=self._prices)
+
+    def _get_price(self, id: Id) -> float:
+        return self._prices.get(id, None)
+
+    def latest_price(self, symbol: Symbol) -> Optional[Price]:
+        id = self._symbol_to_id.get(symbol)
+        if not id:
+            return None
+
+        price = self._get_price(id)
+        if not price:
+            return None
+        return Price(price, price * self._config.confidence_ratio_bps / 10000)

publisher/providers/coin_gecko.py

@@ -0,0 +1,91 @@
+import asyncio
+from typing import Dict, List, Optional, Tuple
+from pythclient.pythclient import PythClient
+from pythclient.pythaccounts import PythPriceAccount
+import time
+
+
+from structlog import get_logger
+
+from publisher.provider import Price, Provider, Symbol
+
+from ..config import PythReplicatorConfig
+
+log = get_logger()
+
+UnixTimestamp = int
+
+
+class PythReplicator(Provider):
+    def __init__(self, config: PythReplicatorConfig) -> None:
+        self._config = config
+        self._client = PythClient(
+            solana_endpoint=config.http_endpoint,
+            solana_ws_endpoint=config.ws_endpoint,
+            first_mapping_account_key=config.first_mapping,
+            program_key=config.program_key,
+        )
+        self._prices: Dict[
+            str, Tuple[float | None, float | None, UnixTimestamp | None]
+        ] = {}
+
+    def start(self) -> None:
+        asyncio.create_task(self._ws_loop())
+
+    async def _ws_loop(self) -> None:
+        self._ws = self._client.create_watch_session()
+        log.info("Creating Pyth replicator WS")
+
+        await self._ws.connect()
+        await self._ws.program_subscribe(
+            self._config.program_key, await self._client.get_all_accounts()
+        )
+
+        asyncio.create_task(self._update_accounts_loop())
+
+        while True:
+            update = await self._ws.next_update()
+            log.debug("Received a WS update", account_key=update.key, slot=update.slot)
+            if isinstance(update, PythPriceAccount):
+                symbol = update.product.symbol
+
+                self._prices[symbol] = [
+                    update.aggregate_price,
+                    update.aggregate_price_confidence_interval,
+                    update.timestamp,
+                ]
+
+                log.info(
+                    "Received a price update", symbol=symbol, price=self._prices[symbol]
+                )
+
+    async def _update_accounts_loop(self) -> None:
+        while True:
+            log.info("Update Pyth accounts")
+            await self._client.refresh_products()
+            await self._client.refresh_all_prices()
+            self._ws.update_program_accounts(
+                self._config.program_key, await self._client.get_all_accounts()
+            )
+            log.info("Finished updating Pyth accounts")
+
+            await asyncio.sleep(self._config.account_update_interval_secs)
+
+    def upd_products(self, _: List[Symbol]) -> None:
+        # This provider stores all the possible feeds and
+        # does not care about the desired products as knowing
+        # them does not improve the performance of the replicator
+        # websocket. Although the websocket filters the given accounts
+        # but this filtering happens in the client-side and not on the server-side.
+        pass
+
+    def latest_price(self, symbol: Symbol) -> Optional[Price]:
+        price, conf, timestamp = self._prices.get(symbol, [None, None, None])
+
+        if not price or not conf or not timestamp:
+            return None
+
+        if time.time() - timestamp > self._config.staleness_time_in_secs:
+            return None
+
+        return Price(price, conf)