Парсинг даних з блокчейн-експлорерів
Etherscan, BscScan, Polygonscan — це зручні інтерфейси поверх ноди, але їх API має жорсткі обмеження: 5 запитів на секунду на безплатному плані, немає streaming, пагінація обмежена 10000 записів. Якщо потрібно виги вузити історію 500K транзакцій конкретного контракту або відстежувати всі взаємодії з адресою — потрібно знати обходні шляхи та альтернативи.
Etherscan API: що умить та що ні
Що умить добре:
- Історія транзакцій адреси:
?module=account&action=txlist&address=0x... - ERC-20 transfers:
?module=account&action=tokentx&address=0x... - Верифікація контрактів та отримання ABI:
?module=contract&action=getabi - Source code контракту:
?module=contract&action=getsourcecode
Жорсткі обмеження:
- Максимум 10000 записів на запрос (обходиться через пагінацію по блокам)
-
startblock/endblockпараметри — єдиний спосіб пагінації - Rate limit: 5 req/sec (free), 10-20/sec (paid plans)
- Немає WebSocket/streaming
- Internal transactions (
action=txlistinternal) не завжди повні
import httpx
import asyncio
from typing import AsyncGenerator
async def get_all_transactions(
address: str,
api_key: str,
start_block: int = 0
) -> AsyncGenerator[dict, None]:
"""Виліщка ВСІХ транзакцій адреси через пагінацію по блокам"""
base_url = "https://api.etherscan.io/api"
current_block = start_block
while True:
async with httpx.AsyncClient() as client:
resp = await client.get(base_url, params={
"module": "account",
"action": "txlist",
"address": address,
"startblock": current_block,
"endblock": 99999999,
"sort": "asc",
"apikey": api_key,
"offset": 10000,
"page": 1,
})
data = resp.json()
if data["status"] != "1" or not data["result"]:
break
txs = data["result"]
for tx in txs:
yield tx
if len(txs) < 10000:
break # остання сторінка
# Наступний блок = останній отриманий + 1
current_block = int(txs[-1]["blockNumber"]) + 1
await asyncio.sleep(0.2) # rate limit 5 req/sec
Важлива нюанс: якщо в одному блоці > 10000 транзакцій на адресу (теоретично можливо для контрактів типу USDT) — цикл зависне. Рішення: додаткова пагінація всередину блоку через page параметр.
Альтернативи Etherscan API
Alchemy / Infura / QuickNode Enhanced APIs. Дозволяють запити типу "всі транзакції на адресу X" без обмеження 10000:
import { Alchemy, Network } from 'alchemy-sdk';
const alchemy = new Alchemy({ apiKey: process.env.ALCHEMY_KEY, network: Network.ETH_MAINNET });
// Отримати всі Asset Transfers (ERC-20, ERC-721, ETH)
const transfers = await alchemy.core.getAssetTransfers({
fromAddress: '0x...',
category: ['external', 'erc20', 'erc721', 'erc1155'],
withMetadata: true,
maxCount: 1000,
});
// Продовження через pageKey
if (transfers.pageKey) {
const more = await alchemy.core.getAssetTransfers({
pageKey: transfers.pageKey,
// ... ті ж параметри
});
}
Alchemy Asset Transfers API значно потужніший за Etherscan: немає ліміту в 10K, пагінація через pageKey, повертає ETH + всі токени в одному запиті.
Moralis Web3 API:
import Moralis from 'moralis';
await Moralis.start({ apiKey: process.env.MORALIS_KEY });
const response = await Moralis.EvmApi.transaction.getWalletTransactions({
chain: '0x1',
address: '0x...',
limit: 100,
cursor: undefined, // для пагінації
});
const { result, cursor } = response.toJSON();
Moralis також умить getWalletTokenTransfers, getNFTTransfers, cross-chain запити.
Парсинг HTML-сторінок експлорера (коли API не достатньо)
Іноді дані є тільки у веб-інтерфейсі, але не в API: список token holders на Etherscan, список верифікованих контрактів, деякі внутрішні виклики. У цьому випадку — scraping HTML.
import httpx
from bs4 import BeautifulSoup
import asyncio
async def get_token_holders(token_address: str, pages: int = 10) -> list[dict]:
"""Парсинг топ-держателів токена з Etherscan"""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html",
}
holders = []
async with httpx.AsyncClient(headers=headers) as client:
for page in range(1, pages + 1):
resp = await client.get(
f"https://etherscan.io/token/{token_address}",
params={"a": "#holders", "p": page}
)
soup = BeautifulSoup(resp.text, 'html.parser')
table = soup.find('table', {'id': 'transfersTable'})
if not table:
break
for row in table.find_all('tr')[1:]: # пропуск header
cols = row.find_all('td')
if len(cols) >= 3:
holders.append({
'rank': cols[0].text.strip(),
'address': cols[1].find('a')['href'].split('/')[-1],
'quantity': cols[2].text.strip(),
'percentage': cols[3].text.strip() if len(cols) > 3 else None,
})
await asyncio.sleep(2) # уважаємо сервер
return holders
Etherscan має bot protection (Cloudflare), агресивний парсинг може привести до тимчасової блокування IP. Краще використовувати residential прокси або офіційний API.
Прямо робота з нодою
Для максимальної повноти даних — власна нода + Erigon (--tracing для internal transactions) або власний Etherscan-like індексатор. Дорожче, але дає:
- Internal transactions без обмежень
- Дані по storage slots
- Trace calls для аналізу MEV
# eth_getBlockReceipts — всі receipts блоку одним запитом
curl -X POST $ETH_RPC_URL \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","method":"eth_getBlockReceipts","params":["0x1234567"],"id":1}'
eth_getBlockReceipts (EIP-1559 extension, підтримується Alchemy/Infura) повертає всі receipts блоку одним запитом — значно ефективніше ніж N окремих eth_getTransactionReceipt.
Хранение та дедупликация
При паралельному збору з кількох джерел з'являються дублі. Стратегія: INSERT ... ON CONFLICT (tx_hash) DO NOTHING для транзакцій, (tx_hash, log_index) unique constraint для для событій.
CREATE TABLE eth_transactions (
tx_hash CHAR(66) PRIMARY KEY,
block_number BIGINT NOT NULL,
from_address CHAR(42) NOT NULL,
to_address CHAR(42),
value NUMERIC(38) DEFAULT 0,
gas_used BIGINT,
status SMALLINT,
ts TIMESTAMPTZ
);
-- Safe upsert без помилок при дублях
INSERT INTO eth_transactions VALUES (...)
ON CONFLICT (tx_hash) DO NOTHING;