Fix

Quorum/apis/price_feeds/__init__.py

@@ -1,5 +1,5 @@
 from .chainlink_api import ChainLinkAPI
 from .chronicle_api import ChronicleAPI
-from .price_feed_providers import PriceFeedProvider
+from .price_feed_utils import PriceFeedProvider, PriceFeedData
 
-all = [ChainLinkAPI, ChronicleAPI, PriceFeedProvider]
+all = [ChainLinkAPI, ChronicleAPI, PriceFeedProvider, PriceFeedData]

Quorum/apis/price_feeds/chainlink_api.py

@@ -1,51 +1,8 @@
 import requests
-from pydantic import BaseModel
-from typing import Optional
 
 from Quorum.utils.chain_enum import Chain
 from Quorum.utils.singleton import Singleton
-
-class Docs(BaseModel):
-    assetClass: Optional[str] = None 
-    assetName: Optional[str] = None
-    baseAsset: Optional[str] = None
-    baseAssetClic: Optional[str] = None
-    blockchainName: Optional[str] = None
-    clicProductName: Optional[str] = None
-    deliveryChannelCode: Optional[str] = None
-    feedType: Optional[str] = None
-    hidden: Optional[bool] = None
-    marketHours: Optional[str] = None
-    productSubType: Optional[str] = None
-    productType: Optional[str] = None
-    productTypeCode: Optional[str] = None
-    quoteAsset: Optional[str] = None
-    quoteAssetClic: Optional[str] = None
-
-
-class PriceFeedData(BaseModel):
-    compareOffchain: Optional[str] = None
-    contractAddress: str
-    contractType: Optional[str] = None
-    contractVersion: Optional[int] = None
-    decimalPlaces: Optional[int] = None
-    ens: Optional[str] = None
-    formatDecimalPlaces: Optional[int] = None
-    healthPrice: Optional[str] = None
-    heartbeat: Optional[int] = None
-    history: Optional[str | bool] = None
-    multiply: Optional[str] = None
-    name: Optional[str] = None
-    pair: Optional[list[Optional[str]]] = None
-    path: Optional[str] = None
-    proxyAddress: Optional[str] = None
-    threshold: Optional[float] = None
-    valuePrefix: Optional[str] = None
-    assetName: Optional[str] = None
-    feedCategory: Optional[str] = None
-    feedType: Optional[str] = None
-    docs: Optional[Docs] = None
-    decimals: Optional[int] = None
+from .price_feed_utils import PriceFeedData
 
 
 class ChainLinkAPI(metaclass=Singleton):
@@ -106,8 +63,8 @@ def get_price_feeds_info(self, chain: Chain) -> dict[str, PriceFeedData]:
             response = self.session.get(url)
             response.raise_for_status()
             chain_link_price_feeds = [PriceFeedData(**feed) for feed in response.json()]
-            chain_link_price_feeds = {feed.contractAddress: feed for feed in chain_link_price_feeds}
-            chain_link_price_feeds.update({feed.proxyAddress: feed for feed in chain_link_price_feeds if feed.proxyAddress})
+            chain_link_price_feeds = {feed.address: feed for feed in chain_link_price_feeds}
+            chain_link_price_feeds.update({feed.proxy_address: feed for feed in chain_link_price_feeds.values() if feed.proxy_address})
             self.memory[chain] = chain_link_price_feeds
 
         return self.memory[chain]

Quorum/apis/price_feeds/chronicle_api.py

@@ -3,6 +3,8 @@
 
 from Quorum.utils.chain_enum import Chain
 from Quorum.utils.singleton import Singleton
+from .price_feed_utils import PriceFeedData
+
 
 class ChronicleAPI(metaclass=Singleton):
     """
@@ -50,7 +52,7 @@ def get_price_feeds_info(self, chain: Chain) -> dict[str, dict]:
             if not pairs:
                 return {}
 
-            chronicle_price_feeds = []
+            chronicle_price_feeds: list[PriceFeedData] = []
             for pair in pairs:
                 response = self.session.get(
                     f"https://chroniclelabs.org/api/median/info/{pair}/{chain.value}/?testnet=false"
@@ -59,8 +61,8 @@ def get_price_feeds_info(self, chain: Chain) -> dict[str, dict]:
                 data = response.json()
 
                 for pair_info in data:
-                    chronicle_price_feeds.append(pair_info)
+                    chronicle_price_feeds.append(PriceFeedData(**pair_info))
 
-            self.memory[chain] = {feed.get("address"): feed for feed in chronicle_price_feeds}
+            self.memory[chain] = {feed.address: feed for feed in chronicle_price_feeds}
 
         return self.memory[chain]

Quorum/apis/price_feeds/price_feed_utils.py

@@ -0,0 +1,21 @@
+from enum import StrEnum
+from typing import Optional
+from pydantic import BaseModel, Field
+
+class PriceFeedProvider(StrEnum):
+    """
+    Enumeration for supported price feed providers.
+    """
+    CHAINLINK = 'Chainlink'
+    CHRONICLE = 'Chronicle'
+
+
+class PriceFeedData(BaseModel):
+    name: Optional[str]
+    pair: Optional[str | list]
+    address: str = Field(..., alias='contractAddress')
+    proxy_address: Optional[str] = Field(None, alias='proxyAddress')
+
+    class Config:
+        allow_population_by_field_name = True  # Allows population using field names
+        extra = 'ignore'  # Ignores extra fields not defined in the model

Quorum/checks/feed_price.py

@@ -2,7 +2,7 @@
 import re
 import json
 
-from Quorum.apis.price_feeds import ChainLinkAPI, ChronicleAPI, PriceFeedProvider
+from Quorum.apis.price_feeds import ChainLinkAPI, ChronicleAPI, PriceFeedProvider, PriceFeedData
 from Quorum.utils.chain_enum import Chain
 from Quorum.checks.check import Check
 from Quorum.apis.block_explorers.source_code import SourceCode
@@ -41,8 +41,12 @@ def __fetch_price_feed_providers(self) -> dict[PriceFeedProvider, dict]:
             dict[PriceFeedProvider, dict]: A dictionary mapping the price feed provider to the price feed data.
         """
         with open(config.GROUND_TRUTH_PATH) as f:
-            providers: list = json.load(f).get(self.customer).get("price_feed_providers", [])
+            providers: list = json.load(f).get(self.customer, {}).get("price_feed_providers", [])
 
+        if not providers:
+            pp.pretty_print(f"No price feed providers found for {self.customer}", pp.Colors.FAILURE)
+            return {}
+
         # Map providers to price feeds
         providers_to_price_feed = {}
         for provider in providers:
@@ -68,12 +72,12 @@ def __check_price_feed_address(self, address: str) -> dict | None:
         """
         for provider, price_feeds in self.providers_to_price_feed.items():
             if address in price_feeds:
-                feed = price_feeds[address]
+                feed: PriceFeedData = price_feeds[address]
                 pp.pretty_print(
-                    f"Found {address} on {provider.name}\nname:{feed.get('pair')}",
+                    f"Found {address} on {provider}\nname:{feed.name if feed.name else feed.pair}",
                     pp.Colors.SUCCESS
                 )
-                return feed
+                return feed.dict()
 
         pp.pretty_print(f"Address {address} not found in any price feed provider", pp.Colors.INFO)
         return None