Add btc balance query

src/telliot_feeds/feeds/__init__.py

@@ -97,6 +97,8 @@
 from telliot_feeds.feeds.wsteth_feed import wsteth_usd_median_feed
 from telliot_feeds.feeds.xdai_usd_feed import xdai_usd_median_feed
 from telliot_feeds.feeds.yfi_usd_feed import yfi_usd_median_feed
+from telliot_feeds.feeds.btc_balance_feed import bitcoin_balance_feed
+from telliot_feeds.feeds.btc_balance_feed import btc_balance_feed_example
 
 
 CATALOG_FEEDS: Dict[str, DataFeed[Any]] = {
@@ -182,6 +184,7 @@
     "mnt-usd-spot": mnt_usd_median_feed,
     "usdy-usd-spot": usdy_usd_median_feed,
     "wmnt-usd-spot": wmnt_usd_median_feed,
+    "btc-bal-example": btc_balance_feed_example
 }
 
 DATAFEED_BUILDER_MAPPING: Dict[str, DataFeed[Any]] = {
@@ -203,4 +206,5 @@
     "MimicryMacroMarketMashup": mimicry_mashup_feed,
     "EVMCall": evm_call_feed,
     "CustomPrice": custom_price_manual_feed,
+    "BTCBalance": bitcoin_balance_feed,
 }

src/telliot_feeds/feeds/btc_balance_feed.py

@@ -0,0 +1,17 @@
+"""Datafeed for BTC balance of an address at a specific timestamp."""
+from telliot_feeds.datafeed import DataFeed
+from telliot_feeds.queries.btc_balance import BTCBalance
+from telliot_feeds.sources.btc_balance import BTCBalanceSource
+
+bitcoin_balance_feed = DataFeed(
+    source=BTCBalanceSource(),
+    query=BTCBalance(),
+)
+
+
+btcAddress = "bc1q06ywseed6sc3x2fafppchefqq8v9cqd0l6vx03"
+timestamp = 1706051389
+btc_balance_feed_example = DataFeed(
+    source=BTCBalanceSource(btcAddress=btcAddress, timestamp=timestamp),
+    query=BTCBalance(btcAddress == btcAddress, timestamp=timestamp),
+)

src/telliot_feeds/queries/btc_balance.py

@@ -0,0 +1,33 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from telliot_feeds.dtypes.value_type import ValueType
+from telliot_feeds.queries.abi_query import AbiQuery
+
+
+@dataclass
+class BTCBalance(AbiQuery):
+    """Returns the BTC balance of a given address at a specific timestamp.
+
+    More info: https://github.com/tellor-io/dataSpecs/blob/main/types/BTCBalance.md
+
+    Attributes:
+        btcAddress: the address of the bitcoin hodler
+        timestamp: timestamp which will be rounded down to the closest Bitcoin block
+    """
+
+    btcAddress: Optional[str] = None
+    timestamp: Optional[int] = None
+
+    #: ABI used for encoding/decoding parameters
+    abi = [{"name": "btcAddress", "type": "string"}, {"name": "timestamp", "type": "uint256"}]
+
+    @property
+    def value_type(self) -> ValueType:
+        """Data type returned for a BTCBalance query.
+
+        - 'uint256': balance in satoshis
+        - 'packed': false
+        """
+
+        return ValueType(abi_type="uint256", packed=False)

src/telliot_feeds/queries/query_catalog.py

@@ -5,6 +5,7 @@
 from telliot_feeds.integrations.diva_protocol import DIVA_DIAMOND_ADDRESS
 from telliot_feeds.queries.ampleforth.ampl_usd_vwap import AmpleforthCustomSpotPrice
 from telliot_feeds.queries.ampleforth.uspce import AmpleforthUSPCE
+from telliot_feeds.queries.btc_balance import BTCBalance
 from telliot_feeds.queries.catalog import Catalog
 from telliot_feeds.queries.custom_price import CustomPrice
 from telliot_feeds.queries.daily_volatility import DailyVolatility
@@ -513,3 +514,9 @@
     title="WMNT/USD spot price",
     q=SpotPrice(asset="wmnt", currency="usd"),
 )
+
+query_catalog.add_entry(
+    tag="btc-bal-example",
+    title="BTC balance example",
+    q=BTCBalance(btcAddress="bc1q06ywseed6sc3x2fafppchefqq8v9cqd0l6vx03", timestamp=1706051389),
+)

src/telliot_feeds/sources/btc_balance.py

@@ -0,0 +1,146 @@
+from dataclasses import dataclass
+from typing import Any
+from typing import Optional
+
+import requests
+from requests import JSONDecodeError
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
+
+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.utils.log import get_logger
+
+logger = get_logger(__name__)
+
+retry_strategy = Retry(
+    total=3,
+    backoff_factor=1,
+    status_forcelist=[429, 500, 502, 503, 504],
+    allowed_methods=["GET"],
+)
+adapter = HTTPAdapter(max_retries=retry_strategy)
+
+
+@dataclass
+class BTCBalanceSource(DataSource[Any]):
+    """DataSource for returning the balance of a BTC address at a given timestamp."""
+
+    btcAddress: Optional[str] = None
+    timestamp: Optional[int] = None
+
+    async def get_response(self) -> Optional[Any]:
+        """gets balance of address from https://blockchain.info/multiaddr?active=$address|$address"""
+        if not self.btcAddress:
+            raise ValueError("BTC address not provided")
+        if not self.timestamp:
+            raise ValueError("Timestamp not provided")
+
+        block_num = await self.block_num_from_timestamp(self.timestamp)
+        if block_num is None:
+            return None
+
+        with requests.Session() as s:
+            url = f"https://blockchain.info/multiaddr?active={self.btcAddress}|{self.btcAddress}"
+            try:
+                rsp = s.get(url)
+            except requests.exceptions.ConnectTimeout:
+                logger.error("Connection timeout getting BTC balance")
+                return None
+            except requests.exceptions.RequestException as e:
+                logger.error(f"Blockchain.info API error: {e}")
+                return None
+
+            try:
+                data = rsp.json()
+            except JSONDecodeError:
+                logger.error("Blockchain.info API returned invalid JSON")
+                return None
+
+            if "txs" not in data:
+                logger.warning("Blockchain.info response doesn't contain needed data")
+                return None
+
+            if "addresses" not in data:
+                logger.warning("Blockchain.info response doesn't contain needed data")
+                return None
+
+            if int(data["addresses"][0]["n_tx"]) == 0:
+                # No transactions for this address
+                logger.info("No transactions for this address")
+                return 0
+
+            # Sort transactions by time in ascending order
+            sorted_txs = sorted(data["txs"], key=lambda tx: (tx["block_height"], tx["tx_index"]))
+
+            #    Find the most recent transaction before the query's timestamp
+            last_tx = None
+            for tx in sorted_txs:
+                if tx["block_height"] > block_num:
+                    break
+                last_tx = tx
+            if last_tx is None:
+                # No transactions before the query's timestamp
+                logger.info("No transactions before the query's timestamp")
+                return 0
+
+            # Use the balance from the last transaction as the BTC balance
+            btc_balance = last_tx["balance"]
+            return btc_balance
+
+    async def block_num_from_timestamp(self, timestamp: int) -> Optional[int]:
+        """Fetches next Bitcoin blockhash after timestamp from API."""
+        with requests.Session() as s:
+            s.mount("https://", adapter)
+        ts = timestamp + 480 * 60
+
+        try:
+            rsp = s.get(f"https://blockchain.info/blocks/{ts * 1000}?format=json")
+        except requests.exceptions.ConnectTimeout:
+            logger.error("Connection timeout getting BTC block num from timestamp")
+            return None
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Blockchain.info API error: {e}")
+            return None
+
+        try:
+            blocks = rsp.json()
+        except JSONDecodeError:
+            logger.error("Blockchain.info API returned invalid JSON")
+            return None
+
+        if not isinstance(blocks, list):
+            logger.warning("Blockchain.info API response is not a list")
+            return None
+
+        if len(blocks) == 0:
+            logger.warning("Blockchain.info API returned no blocks")
+            return None
+
+        if "time" not in blocks[0]:
+            logger.warning("Blockchain.info response doesn't contain needed data")
+            return None
+
+        sorted_blocks = sorted(blocks, key=lambda block: block["time"])
+
+        block = sorted_blocks[0]
+        for b in sorted_blocks:
+            if b["time"] > timestamp:
+                break
+            block = b
+
+        if block["time"] > timestamp:
+            logger.warning("Blockchain.info API returned no blocks before or equal to timestamp")
+            return None
+        logger.info(f"Using BTC block number {block['height']}")
+        return int(block["height"])
+
+    async def fetch_new_datapoint(self) -> OptionalDataPoint[Any]:
+        """Fetches the balance of a BTC address."""
+        balance = await self.get_response()
+        if balance is None:
+            return None, None
+        datapoint = (balance, datetime_now_utc())
+        self.store_datapoint(datapoint)
+        return datapoint

tests/queries/test_query_btc_balance.py

@@ -0,0 +1,43 @@
+""" Unit tests for BTCBalance Query
+
+Copyright (c) 2024-, Tellor Development Community
+Distributed under the terms of the MIT License.
+"""
+from eth_abi import decode_abi
+
+from telliot_feeds.queries.btc_balance import BTCBalance
+
+
+def test_btc_balance_query():
+    """Validate btc balance query"""
+    q = BTCBalance(
+        btcAddress="bc1q06ywseed6sc3x2fafppchefqq8v9cqd0l6vx03",
+        timestamp=1706051389,
+    )
+    assert q.value_type.abi_type == "uint256"
+    assert q.value_type.packed is False
+
+    exp_abi = bytes.fromhex(
+        "0000000000000000000000000000000000000000000000000000000000000040000000"
+        + "0000000000000000000000000000000000000000000000000000000080000000000000"
+        + "000000000000000000000000000000000000000000000000000a42544342616c616e636"
+        + "50000000000000000000000000000000000000000000000000000000000000000000000"
+        + "000000000000000000000000000000000000a0000000000000000000000000000000000"
+        + "00000000000000000000000000000400000000000000000000000000000000000000000"
+        + "000000000000000065b0473d00000000000000000000000000000000000000000000000"
+        + "0000000000000002a626331713036797773656564367363337832666166707063686566"
+        + "7171387639637164306c367678303300000000000000000000000000000000000000000000"
+    )
+
+    assert q.query_data == exp_abi
+
+    query_type, encoded_param_vals = decode_abi(["string", "bytes"], q.query_data)
+    assert query_type == "BTCBalance"
+
+    (btcAddress, timestamp) = decode_abi(["string", "uint256"], encoded_param_vals)
+
+    assert btcAddress == "bc1q06ywseed6sc3x2fafppchefqq8v9cqd0l6vx03"
+    assert timestamp == 1706051389
+    assert isinstance(btcAddress, str)
+    assert isinstance(timestamp, int)
+    assert q.query_id.hex() == "8e0130642e4beec47a3c2e59daf498fb2ee4069ec58da4ddb34ebc0ed1c62626"

tests/sources/test_btc_balance_source.py

@@ -0,0 +1,30 @@
+from datetime import datetime
+from unittest import mock
+
+import pytest
+
+from telliot_feeds.sources.btc_balance import BTCBalanceSource
+
+
+@pytest.mark.asyncio
+async def test_btc_balance():
+    """Retrieve random number."""
+    # "1652075943"  # BCT block num: 731547
+    with mock.patch("telliot_feeds.utils.input_timeout.InputTimeout.__call__", side_effect=["1652075943", ""]):
+        btc_bal_source = BTCBalanceSource(address="bc1q06ywseed6sc3x2fafppchefqq8v9cqd0l6vx03", timestamp=1706051389)
+        v, t = await btc_bal_source.fetch_new_datapoint()
+
+        assert v == 151914
+
+        assert isinstance(v, int)
+        assert isinstance(t, datetime)
+
+
+@pytest.mark.asyncio
+async def test_no_value_from_api():
+    """Test that no value is returned if the API returns no value."""
+    with mock.patch("requests.Session.get", return_value=mock.Mock(json=lambda: {"status": "0", "result": None})):
+        btc_bal_source = BTCBalanceSource(address="bc1q06ywseed6sc3x2fafppchefqq8v9cqd0l6vx03", timestamp=1706051389)
+        v, t = await btc_bal_source.fetch_new_datapoint()
+        assert v is None
+        assert t is None