refactor!: improve coingecko mapping format

Author: ali-behjati

pyth_observer/__init__.py

@@ -20,7 +20,7 @@
 from pyth_observer.check import State
 from pyth_observer.check.price_feed import PriceFeedState
 from pyth_observer.check.publisher import PublisherState
-from pyth_observer.coingecko import Symbol, get_coingecko_prices
+from pyth_observer.coingecko import get_coingecko_prices
 from pyth_observer.dispatch import Dispatch
 from pyth_observer.metrics import metrics
 from pyth_observer.models import Publisher
@@ -54,7 +54,7 @@ def __init__(
         self,
         config: Dict[str, Any],
         publishers: Dict[str, Publisher],
-        coingecko_mapping: Dict[str, Symbol],
+        coingecko_mapping: Dict[str, str],
     ) -> None:
         self.config = config
         self.dispatch = Dispatch(config, publishers)
@@ -95,7 +95,7 @@ async def run(self) -> None:
 
                 for product in products:
                     # Skip tombstone accounts with blank metadata
-                    if "base" not in product.attrs:
+                    if "symbol" not in product.attrs:
                         continue
 
                     if not product.first_price_account_key:
@@ -139,9 +139,11 @@ async def run(self) -> None:
                             latest_trading_slot=price_account.last_slot,
                             price_aggregate=price_account.aggregate_price_info.price,
                             confidence_interval_aggregate=price_account.aggregate_price_info.confidence_interval,
-                            coingecko_price=coingecko_prices.get(product.attrs["base"]),
+                            coingecko_price=coingecko_prices.get(
+                                product.attrs["symbol"]
+                            ),
                             coingecko_update=coingecko_updates.get(
-                                product.attrs["base"]
+                                product.attrs["symbol"]
                             ),
                         )
 

pyth_observer/coingecko.py

@@ -1,25 +1,20 @@
-from typing import Any, Dict, TypedDict
+from typing import Any, Dict
 
 from loguru import logger
 from pycoingecko import CoinGeckoAPI
 from requests.exceptions import HTTPError
 from throttler import throttle
 
 
-class Symbol(TypedDict):
-    api: str
-    market: str
-
-
 # CoinGecko free API limit: 10-50 (varies) https://www.coingecko.com/en/api/pricing
 # However prices are updated every 1-10 minutes: https://www.coingecko.com/en/faq
 # Hence we only have to query once every minute.
-@throttle(rate_limit=1, period=60)
+@throttle(rate_limit=1, period=10)
 async def get_coingecko_prices(
-    mapping: Dict[str, Symbol],
+    symbol_to_ticker: Dict[str, str],
 ) -> Dict[str, Dict[str, Any]]:
-    inverted_mapping = {mapping[x]["api"]: x for x in mapping}
-    ids = [mapping[x]["api"] for x in mapping]
+    ticker_to_symbol = {v: k for k, v in symbol_to_ticker.items()}
+    ids = list(ticker_to_symbol.keys())
 
     try:
         prices = CoinGeckoAPI().get_price(
@@ -32,6 +27,4 @@ async def get_coingecko_prices(
         )
         prices = {}
 
-    # remap to symbol -> prices
-    prices_mapping = {inverted_mapping[x]: prices[x] for x in prices}
-    return prices_mapping
+    return {ticker_to_symbol[x]: prices[x] for x in prices}