Парсинг данных из блокчейн-эксплореров
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 записей на запрос (обходится через pagination по блокам)
-
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, кросс-чейн запросы.
Парсинг HTML-страниц эксплорера (когда API не хватает)
Иногда данные есть только в веб-интерфейсе, но не в API: токен holders list на Etherscan, verified contract список, некоторые внутренние вызовы. В этом случае — 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;