refactors: fix bugs and improve code quality

README.md

@@ -21,6 +21,63 @@ Use `poetry run pyth-observer --help` for documentation on arguments and environ
 
 To run tests, use `poetry run pytest`.
 
+## Building CoinGecko Mapping
+
+The `scripts/build_coingecko_mapping.py` script automatically generates a CoinGecko mapping file by fetching all price feeds from the Pyth Hermes API and matching them with CoinGecko's coin list using fuzzy matching.
+
+### Basic Usage
+
+```sh
+# Generate a new mapping file
+poetry run python scripts/build_coingecko_mapping.py
+
+# Compare with existing mapping file
+poetry run python scripts/build_coingecko_mapping.py -e sample.coingecko.yaml
+
+# Specify custom output file
+poetry run python scripts/build_coingecko_mapping.py -o my_mapping.json
+
+# Skip price validation (faster, but less thorough)
+poetry run python scripts/build_coingecko_mapping.py --no-validate-prices
+
+# Adjust maximum price deviation threshold (default: 10.0%)
+poetry run python scripts/build_coingecko_mapping.py --max-price-deviation 5.0
+```
+
+### How It Works
+
+1. **Fetches Pyth Price Feeds**: Retrieves all price feeds from `https://hermes.pyth.network/v2/price_feeds`
+2. **Extracts Crypto Symbols**: Filters for Crypto asset types and extracts symbols (e.g., "Crypto.BTC/USD")
+3. **Matches with CoinGecko**: Uses multiple matching strategies:
+   - Exact symbol match (case-insensitive)
+   - Fuzzy symbol matching
+   - Fuzzy name matching based on Pyth description
+4. **Validates Mappings**: Compares generated mappings against known correct mappings
+5. **Validates Prices** (optional): Compares prices from Hermes and CoinGecko to detect mismatches
+6. **Generates Warnings**: Flags symbols that need manual review:
+   - Low-confidence fuzzy matches (shows similarity score)
+   - Symbols with no matches found
+   - Price deviations between sources
+
+### Output
+
+The script generates a JSON file in the format:
+```json
+{
+  "Crypto.BTC/USD": "bitcoin",
+  "Crypto.ETH/USD": "ethereum",
+  ...
+}
+```
+
+The script provides a summary showing:
+- Total symbols mapped
+- Exact matches (100% confidence)
+- Fuzzy matches (needs review)
+- No matches found
+
+Review the warnings output to manually verify and adjust any low-confidence matches before using the generated mapping file.
+
 ## Configuration
 
 See `sample.config.yaml` for configuration options.

pyproject.toml

@@ -4,7 +4,7 @@ ignore_missing_imports = true
 
 [tool.poetry]
 name = "pyth-observer"
-version = "2.1.4"
+version = "3.0.0"
 description = "Alerts and stuff"
 authors = []
 readme = "README.md"
@@ -50,4 +50,4 @@ requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.requires-plugins]
-poetry-plugin-export = ">=1.8"
+poetry-plugin-export = ">=1.8"

pyth_observer/__init__.py

@@ -2,10 +2,9 @@
 import os
 from typing import Any, Dict, List, Literal, Tuple
 
-from base58 import b58decode
 from loguru import logger
 from pythclient.market_schedule import MarketSchedule
-from pythclient.pythaccounts import PythPriceAccount, PythPriceType, PythProductAccount
+from pythclient.pythaccounts import PythProductAccount
 from pythclient.pythclient import PythClient
 from pythclient.solana import (
     SOLANA_DEVNET_HTTP_ENDPOINT,
@@ -21,9 +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.crosschain import CrosschainPrice
-from pyth_observer.crosschain import CrosschainPriceObserver as Crosschain
+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
@@ -57,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)
@@ -71,8 +68,6 @@ def __init__(
             rate_limit=int(config["network"]["request_rate_limit"]),
             period=float(config["network"]["request_rate_period"]),
         )
-        self.crosschain = Crosschain(self.config["network"]["crosschain_endpoint"])
-        self.crosschain_throttler = Throttler(rate_limit=1, period=1)
         self.coingecko_mapping = coingecko_mapping
 
         metrics.set_observer_info(
@@ -89,7 +84,9 @@ async def run(self) -> None:
 
                 products = await self.get_pyth_products()
                 coingecko_prices, coingecko_updates = await self.get_coingecko_prices()
-                crosschain_prices = await self.get_crosschain_prices()
+                await self.refresh_all_pyth_prices()
+
+                logger.info("Refreshed all state: products, coingecko, pyth")
 
                 health_server.observer_ready = True
 
@@ -98,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:
@@ -108,11 +105,7 @@ async def run(self) -> None:
                     # for each price account) and a list of publisher states (one
                     # for each publisher).
                     states: List[State] = []
-                    price_accounts = await self.get_pyth_prices(product)
-
-                    crosschain_price = crosschain_prices.get(
-                        b58decode(product.first_price_account_key.key).hex(), None
-                    )
+                    price_accounts = product.prices
 
                     for _, price_account in price_accounts.items():
                         # Handle potential None for min_publishers
@@ -146,11 +139,12 @@ 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"]
                             ),
-                            crosschain_price=crosschain_price,
                         )
 
                         states.append(price_feed_state)
@@ -231,21 +225,19 @@ async def get_pyth_products(self) -> List[PythProductAccount]:
             ).inc()
             raise
 
-    async def get_pyth_prices(
-        self, product: PythProductAccount
-    ) -> Dict[PythPriceType, PythPriceAccount]:
-        logger.debug("Fetching Pyth price accounts...")
+    async def refresh_all_pyth_prices(self) -> None:
+        """Refresh all Pyth prices once for all products."""
+        logger.debug("Refreshing all Pyth price accounts...")
 
         try:
             async with self.pyth_throttler:
                 with metrics.time_operation(
                     metrics.api_request_duration, service="pyth", endpoint="prices"
                 ):
-                    result = await product.refresh_prices()
+                    await self.pyth_client.refresh_all_prices()
                     metrics.api_request_total.labels(
                         service="pyth", endpoint="prices", status="success"
                     ).inc()
-                    return result
         except Exception:
             metrics.api_request_total.labels(
                 service="pyth", endpoint="prices", status="error"
@@ -285,19 +277,3 @@ async def get_coingecko_prices(
             updates[symbol] = data[symbol]["last_updated_at"]
 
         return (prices, updates)
-
-    async def get_crosschain_prices(self) -> Dict[str, CrosschainPrice]:
-        try:
-            with metrics.time_operation(
-                metrics.api_request_duration, service="crosschain", endpoint="prices"
-            ):
-                result = await self.crosschain.get_crosschain_prices()
-                metrics.api_request_total.labels(
-                    service="crosschain", endpoint="prices", status="success"
-                ).inc()
-                return result
-        except Exception:
-            metrics.api_request_total.labels(
-                service="crosschain", endpoint="prices", status="error"
-            ).inc()
-            raise

pyth_observer/alert_utils.py

@@ -0,0 +1,18 @@
+"""
+Utility functions for alert identification and management.
+"""
+
+from pyth_observer.check import Check
+from pyth_observer.check.publisher import PublisherState
+
+
+def generate_alert_identifier(check: Check) -> str:
+    """
+    Generate a unique alert identifier for a check.
+    This is a shared function to ensure consistency across the codebase.
+    """
+    alert_identifier = f"{check.__class__.__name__}-{check.state().symbol}"
+    state = check.state()
+    if isinstance(state, PublisherState):
+        alert_identifier += f"-{state.publisher_name}"
+    return alert_identifier