Как правильно отслеживать транзакции для нескольких кошельков Ethereum
Мониторинг блокчейн-активности - важная задача для трейдеров, разработчиков и аналитиков. Часто возникает необходимость следить за несколькими адресами одновременно, получая уведомления о новых транзакциях. В этой статье мы разберём, как настроить фильтр транзакций в Web3.py, почему ваш код может не работать, и предложим надёжное решение.
Почему стандартный фильтр может не срабатывать
При использовании w3.eth.filter с параметром 'fromBlock': 'latest' вы создаёте фильтр, который отслеживает только самые свежие блоки. Однако есть несколько причин, по которым транзакции могут быть пропущены:
- Неподтверждённые транзакции: фильтр возвращает только те транзакции, которые уже включены в блок. Если транзакция ещё в мемпуле, она не будет отображена.
- Задержка сети: блоки добавляются не мгновенно, и ваш код может проверять фильтр слишком редко.
- Ошибка в адресе: даже небольшая опечатка в адресе кошелька приведёт к отсутствию результатов.
Альтернативный подход: сканирование последних блоков
Вместо фильтров, которые могут быть ненадёжны на публичных нодах (например, Infura), лучше использовать прямой парсинг блоков. Вот рабочий пример на Python:
from web3 import Web3
import time
w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/MY_TOKEN'))
interesting_wallets = [
'0x0c9779a8aff3ea077b07935c141369d95ffedfb1',
'0xae2fc483527b8ef99eb5d9b44875f005ba1fae13'
]
# Приводим адреса к checksum
checksummed = [w3.to_checksum_address(w) for w in interesting_wallets]
last_checked_block = w3.eth.block_number
while True:
current_block = w3.eth.block_number
if current_block > last_checked_block:
for block_num in range(last_checked_block + 1, current_block + 1):
block = w3.eth.get_block(block_num, full_transactions=True)
for tx in block.transactions:
if tx['from'] in checksummed or tx['to'] in checksummed:
print(f'Транзакция найдена: {tx.hash.hex()}')
last_checked_block = current_block
time.sleep(5)Объяснение кода
В этом решении мы каждые 5 секунд проверяем номер последнего блока. Если появились новые блоки, мы загружаем их с полной информацией о транзакциях. Затем проверяем, участвует ли интересующий нас адрес в качестве отправителя (from) или получателя (to). Такой подход гарантирует, что ни одна подтверждённая транзакция не будет упущена.
Оптимизация для больших объёмов данных
Если вы отслеживаете десятки кошельков или сеть с высокой нагрузкой, рассмотрите использование WebSocket-провайдера и подписку на события. Это снизит нагрузку на API и ускорит получение данных. Для этого замените HTTPProvider на WebSocketProvider:
w3 = Web3(Web3.WebsocketProvider('wss://mainnet.infura.io/ws/v3/MY_TOKEN'))Также можно использовать сервисы вроде Alchemy, которые предоставляют более продвинутые фильтры и вебхуки.
Распространённые ошибки и их решение
- Транзакции не видны: убедитесь, что адреса корректны и приведены к формату checksum. Используйте
w3.to_checksum_address(). - Код зависает: добавьте обработку исключений, чтобы при временных ошибках сети скрипт продолжал работу.
- Пропуск транзакций: не используйте
'fromBlock': 'latest'- это может привести к потере данных. Всегда парьте блоки последовательно.
Заключение
Мониторинг нескольких кошельков Ethereum - задача, требующая внимания к деталям. Используя прямой парсинг блоков вместо фильтров, вы добьётесь стабильной работы и точности. Начните с приведённого выше кода, а затем адаптируйте его под свои нужды, добавляя логирование или отправку уведомлений через Telegram или email.