From 6c0b63a82c02e07fd620a954bd547f01ecc4dd51 Mon Sep 17 00:00:00 2001 From: CoCoNuTeK Date: Tue, 8 Apr 2025 16:09:29 +0200 Subject: [PATCH] feat: implement Unisat API for BTC NFTs; fetch NFTs, collections, offers, listings --- .../api/data/unisat/collection_stats_v4.json | 23 + .../v2/api/data/unisat/inscription_data.json | 73 +++ .../unisat/inscription_data_edge_cases.json | 69 ++ .../test/v2/api/data/unisat/listings.json | 60 ++ blockapi/test/v2/api/data/unisat/offers.json | 107 +++ blockapi/test/v2/api/nft/test_unisat.py | 272 ++++++++ blockapi/v2/api/nft/__init__.py | 1 + blockapi/v2/api/nft/unisat.py | 609 ++++++++++++++++++ blockapi/v2/models.py | 12 +- 9 files changed, 1225 insertions(+), 1 deletion(-) create mode 100644 blockapi/test/v2/api/data/unisat/collection_stats_v4.json create mode 100644 blockapi/test/v2/api/data/unisat/inscription_data.json create mode 100644 blockapi/test/v2/api/data/unisat/inscription_data_edge_cases.json create mode 100644 blockapi/test/v2/api/data/unisat/listings.json create mode 100644 blockapi/test/v2/api/data/unisat/offers.json create mode 100644 blockapi/test/v2/api/nft/test_unisat.py create mode 100644 blockapi/v2/api/nft/unisat.py diff --git a/blockapi/test/v2/api/data/unisat/collection_stats_v4.json b/blockapi/test/v2/api/data/unisat/collection_stats_v4.json new file mode 100644 index 00000000..9d8ab159 --- /dev/null +++ b/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 + } +} \ No newline at end of file diff --git a/blockapi/test/v2/api/data/unisat/inscription_data.json b/blockapi/test/v2/api/data/unisat/inscription_data.json new file mode 100644 index 00000000..b94e4257 --- /dev/null +++ b/blockapi/test/v2/api/data/unisat/inscription_data.json @@ -0,0 +1,73 @@ +{ + "code": 0, + "msg": "OK", + "data": { + "cursor": 1, + "total": 2, + "totalConfirmed": 2, + "totalUnconfirmed": 0, + "totalUnconfirmedSpend": 0, + "inscription": [ + { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "inscriptionId": "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "inscriptionNumber": 12345, + "contentType": "image/png", + "contentLength": 1024, + "offset": 0, + "timestamp": 1672531200, + "utxo": { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "codeType": 0, + "height": 800000, + "idx": 0, + "inscriptions": [ + { + "inscriptionId": "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "inscriptionNumber": 12345, + "isBRC20": false, + "moved": false, + "offset": 0 + } + ], + "isOpInRBF": false, + "satoshi": 10000, + "scriptPk": "0014a2b4c6d8e0f2a4b6c8d0e2f4a6b8c0d2e4f6a8b0c2d4", + "scriptType": "p2wpkh", + "txid": "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5", + "vout": 0 + } + }, + { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "inscriptionId": "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "inscriptionNumber": 12346, + "contentType": "text/plain", + "contentLength": 512, + "offset": 0, + "timestamp": 1672531300, + "utxo": { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "codeType": 0, + "height": 800001, + "idx": 0, + "inscriptions": [ + { + "inscriptionId": "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "inscriptionNumber": 12346, + "isBRC20": false, + "moved": false, + "offset": 0 + } + ], + "isOpInRBF": false, + "satoshi": 10000, + "scriptPk": "0014a2b4c6d8e0f2a4b6c8d0e2f4a6b8c0d2e4f6a8b0c2d4", + "scriptType": "p2wpkh", + "txid": "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5", + "vout": 0 + } + } + ] + } +} \ No newline at end of file diff --git a/blockapi/test/v2/api/data/unisat/inscription_data_edge_cases.json b/blockapi/test/v2/api/data/unisat/inscription_data_edge_cases.json new file mode 100644 index 00000000..54cbe50d --- /dev/null +++ b/blockapi/test/v2/api/data/unisat/inscription_data_edge_cases.json @@ -0,0 +1,69 @@ +{ + "code": 0, + "msg": "OK", + "data": { + "cursor": 0, + "total": 4, + "totalConfirmed": 4, + "totalUnconfirmed": 0, + "totalUnconfirmedSpend": 0, + "inscription": [ + { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "contentType": "image/png", + "contentLength": 1000, + "offset": 0, + "timestamp": 1234567890, + "utxo": { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "txid": "txid1", + "vout": 0, + "satoshi": 10000 + } + }, + { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "inscriptionId": "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "contentType": "image/png", + "contentLength": 1000, + "offset": 0, + "timestamp": 1234567890, + "utxo": { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "txid": "txid2", + "vout": 0, + "satoshi": 10000 + } + }, + { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "inscriptionId": "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "contentType": "image/png", + "contentLength": 1000, + "offset": 0, + "timestamp": 1234567890, + "utxo": { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "txid": "txid3", + "vout": 0, + "satoshi": 10000 + } + }, + { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "inscriptionId": "8fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0", + "inscriptionNumber": 2, + "contentType": "image/png", + "contentLength": 1000, + "offset": 0, + "timestamp": 1234567890, + "utxo": { + "address": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh", + "txid": "8fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5", + "vout": 0, + "satoshi": 10000 + } + } + ] + } +} \ No newline at end of file diff --git a/blockapi/test/v2/api/data/unisat/listings.json b/blockapi/test/v2/api/data/unisat/listings.json new file mode 100644 index 00000000..f3dc1dff --- /dev/null +++ b/blockapi/test/v2/api/data/unisat/listings.json @@ -0,0 +1,60 @@ +{ + "code": 0, + "msg": "ok", + "data": { + "list": [ + { + "auctionId": "ye0io3o6x1wcoc9kbwe1xs5wue5ue4tw", + "amount": null, + "collectionId": "test-collection", + "collectionItemName": "Test 07", + "contentType": "image/png", + "domain": null, + "domainHex": null, + "domainType": null, + "inscriptionId": "3822c34e230b423f7092b4bf96b63cd3377fd02ac40fb025f5153461fe0a4b02i0", + "inscriptionNumber": 15388288, + "marketType": "fixedPrice", + "nftType": "collection", + "unitPrice": null, + "price": 50000, + "address": "bc1prx76vq7rv9cyhn2jrd5ygdsxwznsvpv58ccpfschl9zsp8vh4lws7qenhh" + }, + { + "auctionId": "5s7doqj4up5xb2ek5s0rhrwviav2cp0e", + "amount": null, + "collectionId": "test-collection", + "collectionItemName": "Test 08", + "contentType": "image/png", + "domain": null, + "domainHex": null, + "domainType": null, + "inscriptionId": "d5fe1825a1d6240442c67ddf0a312d6b2092a4d526b252935759c236c0bfd057i0", + "inscriptionNumber": 15388559, + "marketType": "fixedPrice", + "nftType": "collection", + "unitPrice": null, + "price": 50000, + "address": "bc1p7qsamzcjffpvg8ej9dqkf7gp2ygs0xdth3tn4f2a3xvl0jg43f7q25kx3a" + }, + { + "auctionId": "iegj06ozdwo01swy8u6dvesov7hbp60y", + "amount": null, + "collectionId": "unisat-og-pass", + "collectionItemName": "OG PASS #10010", + "contentType": "image/svg+xml", + "domain": null, + "domainHex": null, + "domainType": null, + "inscriptionId": "c4d40b8b09a92cc9272cb144c17d128bb0fbb63834c715f2b0a842d2b689cef7i0", + "inscriptionNumber": 837794, + "marketType": "fixedPrice", + "nftType": "collection", + "unitPrice": null, + "price": 50000000, + "address": "bc1pu7p5nk2ky6gfus5qjt9fguyxq5xsg7ehg4738k3fez07jdv42umqdxj9pr" + } + ], + "total": 87254 + } +} \ No newline at end of file diff --git a/blockapi/test/v2/api/data/unisat/offers.json b/blockapi/test/v2/api/data/unisat/offers.json new file mode 100644 index 00000000..40ea9124 --- /dev/null +++ b/blockapi/test/v2/api/data/unisat/offers.json @@ -0,0 +1,107 @@ +{ + "code": 0, + "msg": "ok", + "data": { + "list": [ + { + "auctionId": "dduc4mtjw32abdfti7e8mqqir43lyref", + "inscriptionId": "1e6f91f13fe9a4359e1b9d6e7723cacb815d250337dab921f7b90fca62913e73i648", + "inscriptionNumber": 64501580, + "event": "Updated", + "price": 173000, + "from": "bc1qdfluh306nkuhcw28xqhj3h29a4s3d6zrtfvmvt", + "to": null, + "timestamp": 1744117460264, + "nftType": "collection", + "endMsg": null, + "newest": true, + "txid": null, + "atomicalId": null, + "atomicalTxid": null, + "atomicalIndex": null, + "sellTxid": null, + "sellIndex": null, + "name": null, + "amount": 1, + "unitPrice": 239500, + "domain": null, + "domainType": null, + "domainCategorys": null, + "collectionId": "runestone", + "collectionItemName": "Runestone #101450", + "collectionHighResImgUrl": "https://next-cdn.unisat.io/collection/runestone.png", + "contentType": "model/gltf+json", + "contentBody": null, + "attributes": null, + "dbTag": "market-collection-tcc", + "nftConfirmNum": 56683 + }, + { + "auctionId": "g41k7qfrsdqkegz7wmsarr63ugc88den", + "event": "Cancel", + "amount": 1, + "atomicalId": null, + "atomicalIndex": null, + "atomicalTxid": null, + "attributes": null, + "collectionHighResImgUrl": null, + "collectionId": "bitcoinmiladys", + "collectionItemName": "Bitcoin Miladys #16", + "contentBody": null, + "contentType": "application/json", + "dbTag": "market-batch-tcc", + "domain": null, + "domainCategorys": null, + "domainType": null, + "endMsg": "handled by system", + "from": "bc1pdkqyxdmagq0up644kfq4xvj9drtv09qdx48wshgnj06fcwyulzqs7tlsst", + "inscriptionId": "89e305943e9fc5e5e846ae312c4c8efaf47a1bc143556e8d37fc483400e6951fi0", + "inscriptionNumber": 10128495, + "name": null, + "newest": true, + "nftType": "collection", + "price": 990000, + "sellIndex": null, + "sellTxid": null, + "timestamp": 1744115573205, + "to": null, + "txid": null, + "unitPrice": 990000 + }, + { + "auctionId": "kdw6octea9snqnmoatmxe87uav45a49l", + "event": "Listed", + "amount": 1, + "atomicalId": null, + "atomicalIndex": null, + "atomicalTxid": null, + "attributes": null, + "collectionHighResImgUrl": "https://next-cdn.unisat.io/collection/runestone.png", + "collectionId": "runestone", + "collectionItemName": "Runestone #96357", + "contentBody": null, + "contentType": "model/gltf+json", + "dbTag": "market-collection-tcc", + "domain": null, + "domainCategorys": null, + "domainType": null, + "endMsg": null, + "from": "bc1pme7y9me8z23426xvenrcjcnsdzfw5glu6yewc6vdqxryld8lt6rqzxpu4s", + "inscriptionId": "b37ca7758738d471a22522e2e9de789448991cd854d0020e481037e9df5ff710i1155", + "inscriptionNumber": 64491619, + "name": null, + "newest": false, + "nftType": "collection", + "price": 174000, + "sellIndex": null, + "sellTxid": null, + "timestamp": 1744115298769, + "to": null, + "txid": null, + "unitPrice": 174000, + "channel": "UniSat" + } + ], + "total": 176054 + } + } \ No newline at end of file diff --git a/blockapi/test/v2/api/nft/test_unisat.py b/blockapi/test/v2/api/nft/test_unisat.py new file mode 100644 index 00000000..c555be21 --- /dev/null +++ b/blockapi/test/v2/api/nft/test_unisat.py @@ -0,0 +1,272 @@ +import json +from pathlib import Path +from decimal import Decimal + +import pytest +import requests_mock + +from blockapi.test.v2.api.conftest import read_file +from blockapi.test.v2.api.fake_sleep_provider import FakeSleepProvider +from blockapi.v2.api.nft.unisat import UnisatApi +from blockapi.v2.models import ( + NftToken, + NftCollection, + NftOffer, + NftOfferDirection, + BtcNftType, +) +from blockapi.v2.models import Blockchain, AssetType + +nfts_test_address = 'bc1p3rwga6xsfal6f5d085scecg8lu4gsjl8drk5e07uqzk3cg9dq43s734vje' +test_collection_id = ( + '6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0' +) +test_nft_type = BtcNftType.COLLECTION + + +def test_parse_nfts(requests_mock, unisat_client, inscription_data): + """Test basic NFT parsing with valid data""" + requests_mock.get( + f"{unisat_client.api_options.base_url}v1/indexer/address/{nfts_test_address}/inscription-data", + text=inscription_data, + ) + + result = unisat_client.fetch_nfts(nfts_test_address) + assert not result.errors, f"Fetch errors: {result.errors}" + + parsed = unisat_client.parse_nfts(result) + assert not parsed.errors, f"Parse errors: {parsed.errors}" + assert len(parsed.data) == 2 + + # Test first NFT + nft1 = parsed.data[0] + assert ( + nft1.ident + == "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0" + ) + assert nft1.collection == "ordinals" + assert nft1.collection_name == "Bitcoin Ordinals" + assert ( + nft1.contract + == "6fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5" + ) + assert nft1.standard == "ordinals" + assert nft1.name == "Ordinal #12345" + assert nft1.amount == 1 + assert nft1.updated_time == 1672531200 + assert nft1.blockchain == Blockchain.BITCOIN + assert nft1.asset_type == AssetType.AVAILABLE + + # Test second NFT + nft2 = parsed.data[1] + assert ( + nft2.ident + == "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0" + ) + assert nft2.collection == "ordinals" + assert nft2.collection_name == "Bitcoin Ordinals" + assert ( + nft2.contract + == "7fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5" + ) + assert nft2.standard == "ordinals" + assert nft2.name == "Ordinal #12346" + assert nft2.amount == 1 + assert nft2.updated_time == 1672531300 + assert nft2.blockchain == Blockchain.BITCOIN + assert nft2.asset_type == AssetType.AVAILABLE + + +def test_parse_nfts_edge_cases( + requests_mock, unisat_client, inscription_data_edge_cases +): + """Test NFT parsing with various edge cases""" + requests_mock.get( + f"{unisat_client.api_options.base_url}v1/indexer/address/{nfts_test_address}/inscription-data", + text=inscription_data_edge_cases, + ) + + result = unisat_client.fetch_nfts(nfts_test_address) + assert not result.errors, f"Fetch errors: {result.errors}" + + parsed = unisat_client.parse_nfts(result) + assert not parsed.errors, f"Parse errors: {parsed.errors}" + # Should only parse the last inscription as it's the only one with all required fields + assert len(parsed.data) == 1 + + nft = parsed.data[0] + assert ( + nft.ident + == "8fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5i0" + ) + assert nft.collection == "ordinals" + assert nft.collection_name == "Bitcoin Ordinals" + assert ( + nft.contract + == "8fb976ab49dcec017f1e2015b625126c5c4d6b71174f5bc5af4f39b274a4b6b5" + ) + assert nft.standard == "ordinals" + assert nft.name == "Ordinal #2" + assert nft.amount == 1 + assert nft.updated_time == 1234567890 + assert nft.blockchain == Blockchain.BITCOIN + assert nft.asset_type == AssetType.AVAILABLE + + +def test_fetch_collection(requests_mock, unisat_client, collection_stats_v4): + requests_mock.post( + f"{unisat_client.api_options.base_url}market-v4/collection/auction/collection_statistic", + text=collection_stats_v4, + ) + + 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) + assert not parse_result.errors, f"Parse errors: {parse_result.errors}" + assert len(parse_result.data) == 1 + + collection = parse_result.data[0] + assert isinstance(collection, NftCollection) + assert collection.ident == "pixel-pepes" + assert collection.name == "Pixel Pepes" + assert ( + collection.image + == "https://static.unisat.io/content/47c1d21c508f6d49dfde64d958f14acd041244e1bb616f9b78114b8d9dc7b945i0" + ) + assert not collection.is_disabled + assert not collection.is_nsfw + 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): + """Test fetching and parsing NFT listings with focus on key attributes""" + requests_mock.post( + f"{unisat_client.api_options.base_url}v3/market/collection/auction/list", + text=listings_data, + ) + + fetch_result = unisat_client.fetch_listings(test_nft_type) + assert not fetch_result.errors, f"Fetch errors: {fetch_result.errors}" + + parsed = unisat_client.parse_listings(fetch_result) + assert not parsed.errors, f"Parse errors: {parsed.errors}" + assert len(parsed.data) == 3 + + listing1 = parsed.data[0] + assert isinstance(listing1, NftOffer) + assert listing1.offer_key == "ye0io3o6x1wcoc9kbwe1xs5wue5ue4tw" + assert listing1.direction == NftOfferDirection.LISTING + assert listing1.collection == "test-collection" + assert ( + listing1.offerer + == "bc1prx76vq7rv9cyhn2jrd5ygdsxwznsvpv58ccpfschl9zsp8vh4lws7qenhh" + ) + assert ( + listing1.offer_ident + == "3822c34e230b423f7092b4bf96b63cd3377fd02ac40fb025f5153461fe0a4b02i0" + ) + assert listing1.pay_amount == Decimal('0.0005') # 50000 satoshis = 0.0005 BTC + + listing2 = parsed.data[1] + assert listing2.offer_key == "5s7doqj4up5xb2ek5s0rhrwviav2cp0e" + assert listing2.collection == "test-collection" + assert ( + listing2.offerer + == "bc1p7qsamzcjffpvg8ej9dqkf7gp2ygs0xdth3tn4f2a3xvl0jg43f7q25kx3a" + ) + assert ( + listing2.offer_ident + == "d5fe1825a1d6240442c67ddf0a312d6b2092a4d526b252935759c236c0bfd057i0" + ) + assert listing2.pay_amount == Decimal('0.0005') # 50000 satoshis = 0.0005 BTC + + listing3 = parsed.data[2] + assert listing3.offer_key == "iegj06ozdwo01swy8u6dvesov7hbp60y" + assert listing3.collection == "unisat-og-pass" + assert ( + listing3.offerer + == "bc1pu7p5nk2ky6gfus5qjt9fguyxq5xsg7ehg4738k3fez07jdv42umqdxj9pr" + ) + assert ( + listing3.offer_ident + == "c4d40b8b09a92cc9272cb144c17d128bb0fbb63834c715f2b0a842d2b689cef7i0" + ) + assert listing3.pay_amount == Decimal('0.5') # 50000000 satoshis = 0.5 BTC + + +def test_fetch_offers(requests_mock, unisat_client, offers_data): + """Test fetching and parsing NFT offers with filtering by Listed status""" + requests_mock.post( + f"{unisat_client.api_options.base_url}v3/market/collection/auction/actions", + text=offers_data, + ) + + fetch_result = unisat_client.fetch_offers(test_nft_type, event="Listed") + assert not fetch_result.errors, f"Fetch errors: {fetch_result.errors}" + + parsed = unisat_client.parse_offers(fetch_result) + assert not parsed.errors, f"Parse errors: {parsed.errors}" + + assert len(parsed.data) == 1 + + offer = parsed.data[0] + assert isinstance(offer, NftOffer) + assert offer.offer_key == "kdw6octea9snqnmoatmxe87uav45a49l" + assert offer.direction == NftOfferDirection.OFFER + assert offer.collection == "runestone" + assert ( + offer.offerer + == "bc1pme7y9me8z23426xvenrcjcnsdzfw5glu6yewc6vdqxryld8lt6rqzxpu4s" + ) + assert ( + offer.offer_ident + == "b37ca7758738d471a22522e2e9de789448991cd854d0020e481037e9df5ff710i1155" + ) + assert offer.pay_amount == Decimal('0.00174') # 174000 satoshis = 0.00174 BTC + + +@pytest.fixture +def fake_sleep_provider(): + return FakeSleepProvider() + + +@pytest.fixture +def unisat_client(fake_sleep_provider): + return UnisatApi(api_key="test_key", sleep_provider=fake_sleep_provider) + + +@pytest.fixture +def inscription_data(): + return read_file('data/unisat/inscription_data.json') + + +@pytest.fixture +def inscription_data_edge_cases(): + return read_file('data/unisat/inscription_data_edge_cases.json') + + +@pytest.fixture +def collection_edge_cases(): + return read_file('data/unisat/collection_edge_cases.json') + + +@pytest.fixture +def listings_data(): + return read_file('data/unisat/listings.json') + + +@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') diff --git a/blockapi/v2/api/nft/__init__.py b/blockapi/v2/api/nft/__init__.py index 77f6c38f..1add232d 100644 --- a/blockapi/v2/api/nft/__init__.py +++ b/blockapi/v2/api/nft/__init__.py @@ -5,3 +5,4 @@ SimpleHashEthereumApi, SimpleHashSolanaApi, ) +from blockapi.v2.api.nft.unisat import UnisatApi diff --git a/blockapi/v2/api/nft/unisat.py b/blockapi/v2/api/nft/unisat.py new file mode 100644 index 00000000..ceaed11f --- /dev/null +++ b/blockapi/v2/api/nft/unisat.py @@ -0,0 +1,609 @@ +import logging +from typing import Optional, Dict, Generator +from enum import Enum +from datetime import datetime + +from blockapi.v2.base import BlockchainApi, INftParser, INftProvider, ISleepProvider +from blockapi.v2.coins import COIN_BTC +from blockapi.v2.models import ( + ApiOptions, + AssetType, + Blockchain, + Coin, + ContractInfo, + FetchResult, + NftToken, + ParseResult, + NftCollection, + NftCollectionTotalStats, + NftVolumes, + NftOffer, + NftOfferDirection, + BtcNftType, +) +from requests import HTTPError +import requests + +logger = logging.getLogger(__name__) + + +class NftSortBy(str, Enum): + UNIT_PRICE = 'unitPrice' + ON_SALE_TIME = 'onSaleTime' + INIT_PRICE = 'initPrice' + INSCRIPTION_NUMBER = 'inscriptionNumber' + + +class UnisatApi(BlockchainApi, INftParser, INftProvider): + """ + API docs: https://docs.unisat.io/ + """ + + coin = COIN_BTC + + api_options = ApiOptions( + blockchain=Blockchain.BITCOIN, + base_url='https://open-api.unisat.io/', + rate_limit=0.2, # 5 calls per second for free tier + ) + + supported_requests = { + 'get_nfts': 'v1/indexer/address/{address}/inscription-data', + '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): + """ + Initialize the Unisat API client + + Args: + api_key: Your Unisat API key. Required for all API calls. + sleep_provider: Optional sleep provider for rate limiting + """ + if not api_key: + raise ValueError("API key is required for Unisat API") + + super().__init__(sleep_provider=sleep_provider) + self.headers = { + 'Authorization': f'Bearer {api_key}', + 'Content-Type': 'application/json', + } + + def fetch_nfts( + self, address: str, cursor: Optional[int] = None, size: int = 16 + ) -> FetchResult: + """ + Fetch NFTs (inscriptions) owned by the address + + Args: + address: BTC address to fetch NFTs for + cursor: Pagination cursor (offset) + size: Number of items to return per page (default: 16) + + Returns: + FetchResult containing the NFT data + + Raises: + ValueError: If address is empty or invalid + """ + if not address: + raise ValueError("Address is required") + + params = {'size': size} + if cursor is not None: + params['cursor'] = cursor + + try: + return self.get_data( + 'get_nfts', + headers=self.headers, + params=params, + address=address, + extra=dict(address=address), + ) + except (HTTPError, ValueError, TypeError) as e: + logger.error(f"Error fetching NFTs for address {address}: {str(e)}") + return FetchResult(errors=[str(e)]) + except Exception as e: + logger.error( + f"Unexpected error fetching NFTs for address {address}: {str(e)}" + ) + return FetchResult(errors=[str(e)]) + + def parse_nfts(self, fetch_result: FetchResult) -> ParseResult: + """Parse NFT data from API response""" + errors = [] + data = [] + cursor = None + + if not fetch_result.data: + errors.append("No data in fetch result") + return ParseResult(data=[], errors=errors) + + inner_data = fetch_result.data.get("data", {}) + if not inner_data: + errors.append("No data in API response") + return ParseResult(data=[], errors=errors) + + cursor = ( + str(inner_data.get("cursor")) + if inner_data.get("cursor") is not None + else None + ) + + for nft in self._yield_parsed_nfts(inner_data): + data.append(nft) + + return ParseResult(data=data, errors=errors, cursor=cursor) + + def _yield_parsed_nfts(self, data: Dict) -> Generator[NftToken, None, None]: + """Yield parsed NFT tokens from API response data""" + if not data or "inscription" not in data: + return + + for item in data["inscription"]: + try: + if not all( + k in item + for k in [ + "inscriptionId", + "inscriptionNumber", + "timestamp", + "utxo", + ] + ): + logger.warning(f"Missing required fields in NFT data: {item}") + continue + + utxo = item["utxo"] + if not all(k in utxo for k in ["txid", "address"]): + logger.warning(f"Missing required fields in UTXO data: {utxo}") + continue + + inscription_number = str(item["inscriptionNumber"]) + timestamp = str(item["timestamp"]) + + yield NftToken( + ident=item["inscriptionId"], + collection="ordinals", + collection_name="Bitcoin Ordinals", + contract=utxo["txid"], + standard="ordinals", + name=f"Ordinal #{inscription_number}", + description="", + amount=1, + image_url="", + metadata_url=None, + metadata={}, + updated_time=int(timestamp), + is_disabled=False, + is_nsfw=False, + blockchain=Blockchain.BITCOIN, + asset_type=AssetType.AVAILABLE, + market_url=None, + ) + except Exception as e: + logger.warning(f"Error parsing NFT item {item}: {e}") + continue + + def fetch_collection(self, collection: str) -> FetchResult: + """Fetch collection data from Unisat API.""" + 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: + """ + Parse collection data from the API response + + Args: + fetch_result: Raw API response data + + Returns: + ParseResult containing parsed collection data + """ + if not fetch_result or not fetch_result.data: + return ParseResult(errors=fetch_result.errors if fetch_result else None) + + 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=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, + name=stats.get("name", f"Collection {collection_id}"), + contracts=[ + ContractInfo.from_api( + blockchain=Blockchain.BITCOIN, address=collection_id + ) + ], + image=icon_url, + is_disabled=False, + is_nsfw=False, + blockchain=Blockchain.BITCOIN, + total_stats=total_stats, + volumes=NftVolumes.from_api(coin=self.coin), + ) + + return ParseResult(data=[collection], errors=fetch_result.errors) + + def fetch_listings( + self, + nft_type: BtcNftType, + collection: Optional[str] = None, + cursor: Optional[str] = None, + limit: int = 100, + address: Optional[str] = None, + tick: Optional[str] = None, + min_price: Optional[int] = None, + max_price: Optional[int] = None, + nft_confirm: Optional[bool] = None, + is_end: Optional[bool] = None, + domain_type: Optional[str] = None, + domain_min_length: Optional[int] = None, + domain_max_length: Optional[int] = None, + domain_category: Optional[str] = None, + domain_fuzzy: Optional[str] = None, + collection_fuzzy: Optional[str] = None, + all_items: Optional[bool] = None, + sort_by: NftSortBy = NftSortBy.UNIT_PRICE, + sort_order: int = -1, + ) -> FetchResult: + """ + Fetch all current listings (sell offers) for a specific collection. + + Args: + nft_type: Type of NFT (brc20, domain, collection, arc20, runes) + collection: Collection ID (slug), optional + cursor: Pagination cursor (offset, 'start' parameter) + limit: Number of items per page + address: Filter by address + tick: Filter by tick (for BRC20) + min_price: Minimum price filter + max_price: Maximum price filter + nft_confirm: Filter by confirmation status + is_end: Filter by end status + domain_type: Filter by domain type (e.g., 'sats') + domain_min_length: Minimum domain length + domain_max_length: Maximum domain length + domain_category: Domain category filter + domain_fuzzy: Domain fuzzy search + collection_fuzzy: Collection fuzzy search + all_items: Whether to fetch all items + sort_by: Field to sort by (unitPrice, onSaleTime, initPrice, inscriptionNumber) + sort_order: Sort order (1 for ascending, -1 for descending) + + Returns: + FetchResult containing listing data + """ + if not nft_type: + raise ValueError("NFT type is required") + + # Ensure we get the string value if an enum is passed + nft_type_str = ( + nft_type.value if isinstance(nft_type, BtcNftType) else str(nft_type) + ) + + start = int(cursor) if cursor else 0 + + filter_dict = {"nftType": nft_type_str} + + if collection: + filter_dict["collectionId"] = collection + if address: + filter_dict["address"] = address + if tick: + filter_dict["tick"] = tick + if min_price is not None: + filter_dict["minPrice"] = min_price + if max_price is not None: + filter_dict["maxPrice"] = max_price + if nft_confirm is not None: + filter_dict["nftConfirm"] = nft_confirm + if is_end is not None: + filter_dict["isEnd"] = is_end + if domain_type: + filter_dict["domainType"] = domain_type + if domain_min_length is not None: + filter_dict["domainMinLength"] = domain_min_length + if domain_max_length is not None: + filter_dict["domainMaxLength"] = domain_max_length + if domain_category: + filter_dict["domainCategory"] = domain_category + if domain_fuzzy: + filter_dict["domainFuzzy"] = domain_fuzzy + if collection_fuzzy: + filter_dict["collectionFuzzy"] = collection_fuzzy + if all_items is not None: + filter_dict["all"] = all_items + + sort_dict = {} + sort_dict[sort_by] = sort_order + + request_body = { + "filter": filter_dict, + "sort": sort_dict, + "start": start, + "limit": limit, + } + + try: + response_data = self.post( + 'get_listings', json=request_body, headers=self.headers + ) + return FetchResult(data=response_data) + except (HTTPError, ValueError, TypeError) as e: + logger.error(f"Error fetching listings for nft_type {nft_type}: {str(e)}") + return FetchResult(errors=[str(e)]) + except Exception as e: + logger.error( + f"Unexpected error fetching listings for nft_type {nft_type}: {str(e)}" + ) + return FetchResult(errors=[str(e)]) + + def parse_listings(self, fetch_result: FetchResult) -> ParseResult: + """ + Parse listing data from API response. + + Args: + fetch_result: Raw API response data from fetch_listings + + Returns: + ParseResult containing parsed NftOffer objects + """ + if not fetch_result or not fetch_result.data: + return ParseResult(errors=fetch_result.errors if fetch_result else None) + + inner_data = fetch_result.data.get("data", {}) + if not inner_data: + return ParseResult(errors=fetch_result.errors) + + items = inner_data.get("list", []) + timestamp = inner_data.get("timestamp") + cursor = str(timestamp) if timestamp else None + + return ParseResult( + data=list(self._yield_parsed_listings(items)), + cursor=cursor, + errors=fetch_result.errors, + ) + + def _yield_parsed_listings( + self, items: list[Dict] + ) -> Generator[NftOffer, None, None]: + """Yield parsed NftOffer objects from listing items.""" + if not items: + return + + for item in items: + if not all( + k in item for k in ['auctionId', 'inscriptionId', 'address', 'price'] + ): + logger.warning( + f"Skipping listing item due to missing required fields: {item}" + ) + continue + + collection_id = item.get('collectionId', '') + + amount = item.get('amount') + if amount is None: + amount = 1 + + price = item.get('price') + if price is None: + price = 0 + + yield NftOffer.from_api( + offer_key=item["auctionId"], + direction=NftOfferDirection.LISTING, + collection=collection_id, + contract=collection_id, + blockchain=Blockchain.BITCOIN, + offerer=item["address"], + start_time=None, + end_time=None, + offer_coin=None, + offer_amount=amount, + offer_contract=collection_id, + offer_ident=item["inscriptionId"], + pay_contract=None, + pay_ident=None, + pay_amount=price, + pay_coin=self.coin, + ) + + def fetch_offers( + self, + nft_type: Optional[BtcNftType] = None, + address: Optional[str] = None, + inscription_id: Optional[str] = None, + event: Optional[str] = None, + tick: Optional[str] = None, + domain_type: Optional[str] = None, + collection_id: Optional[str] = None, + cursor: Optional[str] = None, + limit: int = 100, + ) -> FetchResult: + """ + Fetch listing events (historical or recent) in a collection. + + Args: + nft_type: Type of NFT (brc20, domain, collection, arc20, runes) + address: Filter by address + inscription_id: Filter by inscription ID + event: Filter by event type (Listed, Cancel, Buy) + tick: Filter by tick (for BRC20) + domain_type: Filter by domain type + collection_id: Collection ID to filter by + cursor: Pagination cursor (offset, 'start' parameter) + limit: Number of items per page + + Returns: + FetchResult containing the listing action data + """ + # Ensure we get the string value if an enum is passed + nft_type_str = ( + nft_type.value if isinstance(nft_type, BtcNftType) and nft_type else None + ) + + start = int(cursor) if cursor else 0 + + filter_dict = {} + if nft_type_str: + filter_dict["nftType"] = nft_type_str + if address: + filter_dict["address"] = address + if inscription_id: + filter_dict["inscriptionId"] = inscription_id + if event: + filter_dict["event"] = event + if tick: + filter_dict["tick"] = tick + if domain_type: + filter_dict["domainType"] = domain_type + if collection_id: + filter_dict["collectionId"] = collection_id + + request_body = { + "filter": filter_dict, + "start": start, + "limit": limit, + } + + try: + response_data = self.post( + 'get_offers', json=request_body, headers=self.headers + ) + return FetchResult(data=response_data) + except (HTTPError, ValueError, TypeError) as e: + logger.error(f"Error fetching listing actions: {str(e)}") + return FetchResult(errors=[str(e)]) + except Exception as e: + logger.error(f"Unexpected error fetching listing actions: {str(e)}") + return FetchResult(errors=[str(e)]) + + def parse_offers(self, fetch_result: FetchResult) -> ParseResult: + """ + Parse listing action data from API response. + + Args: + fetch_result: Raw API response data from fetch_listing_actions + + Returns: + ParseResult containing parsed NftOffer objects + """ + if not fetch_result or not fetch_result.data: + return ParseResult(errors=fetch_result.errors if fetch_result else None) + + inner_data = fetch_result.data.get("data", {}) + if not inner_data: + return ParseResult(errors=fetch_result.errors) + + items = inner_data.get("list", []) + + return ParseResult( + data=list(self._yield_parsed_offers(items)), + errors=fetch_result.errors, + ) + + def _yield_parsed_offers( + self, items: list[Dict] + ) -> Generator[NftOffer, None, None]: + """Yield parsed NftOffer objects from listing action items.""" + if not items: + return + + for item in items: + if not all(k in item for k in ['auctionId', 'inscriptionId', 'event']): + logger.warning( + f"Skipping listing action due to missing required fields: {item}" + ) + continue + + # Skip if not a listing event + event = item.get('event') + if event != 'Listed': + continue + + collection_id = item.get('collectionId', '') + + # Handle amount - use 1 as default for null values + amount = item.get('amount') + if amount is None: + amount = 1 + + # Convert from milliseconds to ISO format + timestamp = item.get('timestamp') + formatted_time = None + if timestamp: + try: + timestamp_seconds = timestamp / 1000 + formatted_time = datetime.fromtimestamp( + timestamp_seconds + ).isoformat() + except (ValueError, TypeError, OverflowError): + logger.warning( + f"Could not parse timestamp {timestamp} for item {item.get('auctionId')}" + ) + + price = item.get('price') + if price is None: + price = 0 + + yield NftOffer.from_api( + offer_key=item["auctionId"], + direction=NftOfferDirection.OFFER, + collection=collection_id, + contract=collection_id, + blockchain=Blockchain.BITCOIN, + offerer=item.get("from", ""), + start_time=formatted_time, + end_time=None, + offer_coin=None, + offer_amount=amount, + offer_contract=collection_id, + offer_ident=item["inscriptionId"], + pay_contract=None, + pay_ident=None, + pay_amount=price, + pay_coin=self.coin, + ) diff --git a/blockapi/v2/models.py b/blockapi/v2/models.py index 28106bb7..36958946 100644 --- a/blockapi/v2/models.py +++ b/blockapi/v2/models.py @@ -477,6 +477,16 @@ class OfferItemType(str, Enum): ERC1155_WITH_CRITERIA = 'erc-1155-limited' +class BtcNftType(str, Enum): + """Type of NFT of BTC chain""" + + BRC20 = "brc20" + DOMAIN = "domain" + COLLECTION = "collection" + ARC20 = "arc20" + RUNES = "runes" + + @attr.s(auto_attribs=True, slots=True) class ApiOptions: blockchain: Blockchain @@ -829,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,