Added Custom Price Aggregator & ORDI/USD Spot

src/telliot_feeds/feeds/__init__.py

@@ -9,6 +9,7 @@
 from telliot_feeds.feeds.badger_usd_feed import badger_usd_median_feed
 from telliot_feeds.feeds.bch_usd_feed import bch_usd_median_feed
 from telliot_feeds.feeds.bct_usd_feed import bct_usd_median_feed
+from telliot_feeds.feeds.brc20_ordi_usd_feed import ordi_usd_median_feed
 from telliot_feeds.feeds.brl_usd_feed import brl_usd_median_feed
 from telliot_feeds.feeds.btc_usd_feed import btc_usd_median_feed
 from telliot_feeds.feeds.cbeth_usd_feed import cbeth_usd_median_feed
@@ -170,6 +171,7 @@
     "oeth-usd-spot": oeth_usd_median_feed,
     "pyth-usd-spot": pyth_usd_median_feed,
     "ogv-eth-spot": ogv_eth_median_feed,
+    "brc20-ordi-usd-spot": ordi_usd_median_feed,
 }
 
 DATAFEED_BUILDER_MAPPING: Dict[str, DataFeed[Any]] = {

src/telliot_feeds/feeds/brc20_ordi_usd_feed.py

@@ -0,0 +1,18 @@
+from telliot_feeds.datafeed import DataFeed
+from telliot_feeds.queries.custom_price import CustomPrice
+from telliot_feeds.sources.custom_price_aggregator import CustomPriceAggregator
+from telliot_feeds.sources.price.spot.coingecko import CoinGeckoBRC20SpotPriceSource
+
+ordi_usd_median_feed = DataFeed(
+    query=CustomPrice(identifier="brc20", asset="ordi", currency="btc", unit=""),
+    source=CustomPriceAggregator(
+        identifier="brc20",
+        asset="ordi",
+        currency="usd",
+        unit="",
+        algorithm="median",
+        sources=[
+            CoinGeckoBRC20SpotPriceSource(identifier="brc20", asset="ordi", currency="usd", unit=""),
+        ],
+    ),
+)

src/telliot_feeds/queries/price/spot_price.py

@@ -73,6 +73,7 @@
     "OETH/USD",
     "PYTH/USD",
     "OGV/ETH",
+    "ORDI/USD",
 ]
 
 

src/telliot_feeds/queries/query_catalog.py

@@ -477,3 +477,9 @@
     title="OGV/ETH spot price",
     q=SpotPrice(asset="ogv", currency="eth"),
 )
+
+query_catalog.add_entry(
+    tag="brc20-ordi-usd-spot",
+    title="ORDI/USD spot price",
+    q=CustomPrice(identifier="brc20", asset="ordi", currency="usd", unit=""),
+)

src/telliot_feeds/sources/custom_price_aggregator.py

@@ -0,0 +1,105 @@
+import asyncio
+import statistics
+from dataclasses import dataclass
+from dataclasses import field
+from typing import Callable
+from typing import List
+from typing import Literal
+
+from telliot_feeds.datasource import DataSource
+from telliot_feeds.dtypes.datapoint import datetime_now_utc
+from telliot_feeds.dtypes.datapoint import OptionalDataPoint
+from telliot_feeds.pricing.price_source import PriceSource
+from telliot_feeds.utils.log import get_logger
+
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class CustomPriceAggregator(DataSource[float]):
+
+    # identifier
+    identifier: str = ""
+
+    #: Asset
+    asset: str = ""
+
+    # : Currency of returned price
+    currency: str = ""
+
+    # unit
+    unit: str = ""
+
+    #: Callable algorithm that accepts an iterable of floats
+    algorithm: Literal["median", "mean"] = "median"
+
+    #: Private storage for actual algorithm function
+    _algorithm: Callable[..., float] = field(default=statistics.median, init=False, repr=False)
+
+    #: Data feed sources
+    sources: List[PriceSource] = field(default_factory=list)
+
+    def __post_init__(self) -> None:
+        if self.algorithm == "median":
+            self._algorithm = statistics.median
+        elif self.algorithm == "mean":
+            self._algorithm = statistics.mean
+
+    def __str__(self) -> str:
+        """Human-readable representation."""
+        asset = self.asset.upper()
+        currency = self.currency.upper()
+        symbol = asset + "/" + currency
+        return f"PriceAggregator {symbol} {self.algorithm}"
+
+    async def update_sources(self) -> List[OptionalDataPoint[float]]:
+        """Update data feed sources
+
+        Returns:
+            Dictionary of updated source values, mapping data source UID
+            to the time-stamped answer for that data source
+        """
+
+        async def gather_inputs() -> List[OptionalDataPoint[float]]:
+            sources = self.sources
+            datapoints = await asyncio.gather(*[source.fetch_new_datapoint() for source in sources])
+            return datapoints
+
+        inputs = await gather_inputs()
+
+        return inputs
+
+    async def fetch_new_datapoint(self) -> OptionalDataPoint[float]:
+        """Update current value with time-stamped value fetched from source
+
+        Args:
+            store:  If true and applicable, updated value will be stored
+                    to the database
+
+        Returns:
+            Current time-stamped value
+        """
+        datapoints = await self.update_sources()
+
+        prices = []
+        for datapoint in datapoints:
+            v, _ = datapoint  # Ignore input timestamps
+            # Check for valid answers
+            if v is not None and isinstance(v, float):
+                prices.append(v)
+
+        if not prices:
+            logger.warning(f"No prices retrieved for {self}.")
+            return None, None
+
+        # Run the algorithm on all valid prices
+        logger.info(f"Running {self.algorithm} on {prices}")
+        result = self._algorithm(prices)
+        datapoint = (result, datetime_now_utc())
+        self.store_datapoint(datapoint)
+
+        logger.info("Feed Price: {} reported at time {}".format(datapoint[0], datapoint[1]))
+        logger.info("Number of sources used in aggregate: {}".format(len(prices)))
+
+        return datapoint

src/telliot_feeds/sources/price/spot/coingecko.py

@@ -63,6 +63,7 @@
     "wbeth": "wrapped-beacon-eth",
     "pyth": "pyth-network",
     "ogv": "origin-defi-governance",
+    "ordi": "ordinals",
 }
 
 
@@ -123,3 +124,64 @@ class CoinGeckoSpotPriceSource(PriceSource):
     asset: str = ""
     currency: str = ""
     service: CoinGeckoSpotPriceService = field(default_factory=CoinGeckoSpotPriceService, init=False)
+
+
+class CoinGeckoBRC20SpotPriceService(WebPriceService):
+    """CoinGecko Price Service"""
+
+    def __init__(self, **kwargs: Any) -> None:
+        kwargs["name"] = "CoinGecko Price Service"
+        kwargs["url"] = "https://api.coingecko.com"
+        super().__init__(**kwargs)
+
+    async def get_price(self, asset: str, currency: str) -> OptionalDataPoint[float]:
+        """Implement PriceServiceInterface
+
+        This implementation gets the price from the Coingecko API
+
+        Note that coingecko does not return a timestamp so one is
+        locally generated.
+        """
+
+        asset = asset.lower()
+        currency = currency.lower()
+
+        coin_id = coingecko_coin_id.get(asset, None)
+        if not coin_id:
+            raise Exception("Asset not supported: {}".format(asset))
+
+        url_params = urlencode({"ids": coin_id, "vs_currencies": currency})
+        request_url = "/api/v3/simple/price?{}".format(url_params)
+
+        d = self.get_url(request_url)
+
+        if "error" in d:
+            if "api.coingecko.com used Cloudflare to restrict access" in str(d["exception"]):
+                logger.warning("CoinGecko API rate limit exceeded")
+            else:
+                logger.error(d)
+            return None, None
+        elif "response" in d:
+            response = d["response"]
+
+            try:
+                price = float(response[coin_id][currency])
+                return price, datetime_now_utc()
+            except KeyError as e:
+                msg = "Error parsing Coingecko API response: KeyError: {}".format(e)
+                logger.error(msg)
+                return None, None
+
+        else:
+            msg = "Invalid response from get_url"
+            logger.error(msg)
+            return None, None
+
+
+@dataclass
+class CoinGeckoBRC20SpotPriceSource(PriceSource):
+    identifier: str = ""
+    asset: str = ""
+    currency: str = ""
+    unit: str = ""
+    service: CoinGeckoBRC20SpotPriceService = field(default_factory=CoinGeckoBRC20SpotPriceService, init=False)