Merge pull request #227 from crypkit/2149-blockapi-btc-nft-api-replacement-after-simplehash-shutdown

feat: implement Unisat API for BTC NFTs; fetch NFTs, collections, offers, listings

Author: SimSef

blockapi/test/v2/api/data/unisat/collection_stats_v4.json

@@ -0,0 +1,23 @@
+{
+    "code": 0,
+    "msg": "ok",
+    "data": {
+        "collectionId": "pixel-pepes",
+        "name": "Pixel Pepes",
+        "desc": "The first ever airdrop on the Bitcoin network. 1563 rare Pixel Pepes airdropped to users who had made a transaction on ordinalswallet.com before block 777888.",
+        "icon": "47c1d21c508f6d49dfde64d958f14acd041244e1bb616f9b78114b8d9dc7b945i0",
+        "iconContentType": "image/png",
+        "btcValue": 39900000,
+        "btcValuePercent": 1,
+        "floorPrice": 990000,
+        "listed": 20,
+        "total": 1563,
+        "supply": null,
+        "attrs": [],
+        "twitter": "https://twitter.com/PepesPixel",
+        "discord": "https://discord.gg/ordinalswallet",
+        "website": "",
+        "pricePercent": 1,
+        "verification": false
+    }
+} 

blockapi/test/v2/api/nft/test_unisat.py

@@ -113,23 +113,14 @@ def test_parse_nfts_edge_cases(
     assert nft.asset_type == AssetType.AVAILABLE
 
 
-def test_fetch_collection(
-    requests_mock, unisat_client, collection_info, collection_items, collection_stats
-):
-    requests_mock.get(
-        f"{unisat_client.api_options.base_url}v1/collection-indexer/collection/{test_collection_id}/info",
-        text=collection_info,
-    )
-    requests_mock.get(
-        f"{unisat_client.api_options.base_url}v1/collection-indexer/collection/{test_collection_id}/items",
-        text=collection_items,
-    )
+def test_fetch_collection(requests_mock, unisat_client, collection_stats_v4):
     requests_mock.post(
-        f"{unisat_client.api_options.base_url}v3/market/collection/auction/collection_statistic",
-        text=collection_stats,
+        f"{unisat_client.api_options.base_url}market-v4/collection/auction/collection_statistic",
+        text=collection_stats_v4,
     )
 
-    fetch_result = unisat_client.fetch_collection(test_collection_id)
+    test_collection = "pixel-pepes"
+    fetch_result = unisat_client.fetch_collection(test_collection)
     assert not fetch_result.errors, f"Fetch errors: {fetch_result.errors}"
 
     parse_result = unisat_client.parse_collection(fetch_result)
@@ -138,16 +129,20 @@ def test_fetch_collection(
 
     collection = parse_result.data[0]
     assert isinstance(collection, NftCollection)
-    assert collection.ident == test_collection_id
-    assert collection.name == "Ordinal Punks"
+    assert collection.ident == "pixel-pepes"
+    assert collection.name == "Pixel Pepes"
     assert (
         collection.image
-        == "https://ordinals.com/content/6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0"
+        == "https://static.unisat.io/content/47c1d21c508f6d49dfde64d958f14acd041244e1bb616f9b78114b8d9dc7b945i0"
     )
     assert not collection.is_disabled
     assert not collection.is_nsfw
-    assert str(collection.total_stats.owners_count) == "150"
-    assert str(collection.total_stats.floor_price) == "0.1"
+    assert collection.blockchain == Blockchain.BITCOIN
+    assert str(collection.total_stats.floor_price) == "990000"
+    assert str(collection.total_stats.owners_count) == "1563"
+    assert str(collection.total_stats.sales_count) == "20"
+    assert str(collection.total_stats.volume) == "39900000"
+    assert str(collection.total_stats.market_cap) == str(990000 * 1563)
 
 
 def test_fetch_listings(requests_mock, unisat_client, listings_data):
@@ -252,21 +247,6 @@ def inscription_data():
     return read_file('data/unisat/inscription_data.json')
 
 
-@pytest.fixture
-def collection_info():
-    return read_file('data/unisat/collection_info.json')
-
-
-@pytest.fixture
-def collection_items():
-    return read_file('data/unisat/collection_items.json')
-
-
-@pytest.fixture
-def collection_stats():
-    return read_file('data/unisat/collection_stats.json')
-
-
 @pytest.fixture
 def inscription_data_edge_cases():
     return read_file('data/unisat/inscription_data_edge_cases.json')
@@ -285,3 +265,8 @@ def listings_data():
 @pytest.fixture
 def offers_data():
     return read_file('data/unisat/offers.json')
+
+
+@pytest.fixture
+def collection_stats_v4():
+    return read_file('data/unisat/collection_stats_v4.json')

blockapi/v2/api/nft/unisat.py

@@ -49,11 +49,9 @@ class UnisatApi(BlockchainApi, INftParser, INftProvider):
 
     supported_requests = {
         'get_nfts': 'v1/indexer/address/{address}/inscription-data',
-        'get_collection': 'v1/collection-indexer/collection/{collectionId}/info',
-        'get_collection_items': 'v1/collection-indexer/collection/{collectionId}/items',
-        'get_collection_stats': 'v3/market/collection/auction/collection_statistic',
         'get_listings': 'v3/market/collection/auction/list',
         'get_offers': 'v3/market/collection/auction/actions',
+        'get_collection_stats': 'market-v4/collection/auction/collection_statistic',
     }
 
     def __init__(self, api_key: str, sleep_provider: Optional[ISleepProvider] = None):
@@ -192,27 +190,19 @@ def _yield_parsed_nfts(self, data: Dict) -> Generator[NftToken, None, None]:
 
     def fetch_collection(self, collection: str) -> FetchResult:
         """Fetch collection data from Unisat API."""
-        info_response = self.get_data(
-            'get_collection',
-            headers=self.headers,
-            collectionId=collection,
-        )
-        items_response = self.get_data(
-            'get_collection_items',
-            headers=self.headers,
-            collectionId=collection,
-        )
-
-        stats_data = self.post(
-            'get_collection_stats',
-            json={'collectionId': collection},
-            headers=self.headers,
-        )
-        stats_response = FetchResult(data=stats_data)
-
-        return FetchResult.from_fetch_results(
-            info=info_response, items=items_response, stats=stats_response
-        )
+        try:
+            stats_data = self.post(
+                'get_collection_stats',
+                json={'collectionId': collection},
+                headers=self.headers,
+            )
+            return FetchResult(data=stats_data)
+        except (HTTPError, ValueError, TypeError) as e:
+            logger.error(f"Error fetching collection {collection}: {str(e)}")
+            return FetchResult(errors=[str(e)])
+        except Exception as e:
+            logger.error(f"Unexpected error fetching collection {collection}: {str(e)}")
+            return FetchResult(errors=[str(e)])
 
     def parse_collection(self, fetch_result: FetchResult) -> ParseResult:
         """
@@ -227,33 +217,45 @@ def parse_collection(self, fetch_result: FetchResult) -> ParseResult:
         if not fetch_result or not fetch_result.data:
             return ParseResult(errors=fetch_result.errors if fetch_result else None)
 
-        info = fetch_result.data.get("info", {}).get("data", {})
-        items = fetch_result.data.get("items", {}).get("data", {})
-        stats = fetch_result.data.get("stats", {}).get("data", {})
+        stats = fetch_result.data.get("data", {})
+        if not stats:
+            return ParseResult(errors=["No collection data found in response"])
 
         collection_id = stats.get("collectionId")
         if not collection_id:
             return ParseResult(errors=["No collection ID found in response"])
 
+        # Format the icon URL
+        icon = stats.get("icon")
+        icon_url = None
+        if icon:
+            icon_url = f"https://static.unisat.io/content/{icon}"
+
+        # Create NftCollectionTotalStats
+        floor_price = stats.get("floorPrice", 0)
+        total_nfts = stats.get("total", 0)
+        # Calculate market cap as floor price × total supply
+        market_cap = floor_price * total_nfts if floor_price and total_nfts else 0
+
         total_stats = NftCollectionTotalStats.from_api(
-            volume="",
-            sales_count="",
-            owners_count=str(info.get("holders", "")),
-            market_cap="",
-            floor_price=str(stats.get("floorPrice", "")),
-            average_price="",
+            volume=str(stats.get("btcValue", 0)),
+            sales_count=str(stats.get("listed", 0)),
+            owners_count=str(total_nfts),
+            market_cap=str(market_cap),
+            floor_price=str(floor_price),
+            average_price="0",
             coin=self.coin,
         )
 
         collection = NftCollection.from_api(
-            ident=collection_id,  # Matches test_collection_id
+            ident=collection_id,
             name=stats.get("name", f"Collection {collection_id}"),
             contracts=[
                 ContractInfo.from_api(
                     blockchain=Blockchain.BITCOIN, address=collection_id
                 )
             ],
-            image=stats.get("icon"),
+            image=icon_url,
             is_disabled=False,
             is_nsfw=False,
             blockchain=Blockchain.BITCOIN,
@@ -574,8 +576,6 @@ def _yield_parsed_offers(
             formatted_time = None
             if timestamp:
                 try:
-                    from datetime import datetime
-
                     timestamp_seconds = timestamp / 1000
                     formatted_time = datetime.fromtimestamp(
                         timestamp_seconds

blockapi/v2/models.py

@@ -839,7 +839,7 @@ def from_api(
             contract=contract,
             blockchain=blockchain,
             offerer=offerer,
-            start_time=parse_dt(start_time) if start_time else None,
+            start_time=parse_dt(start_time) if start_time else datetime.utcnow(),
             end_time=parse_dt(end_time) if end_time else None,
             offer_coin=offer_coin,
             offer_contract=offer_contract.lower() if offer_contract else None,