Web3.py подключается к ноде Ethereum через WebSockets, подписывается на поток транзакций в мемпуле (pendingTransactions), декодирует вызовы смарт-контрактов «на лету» и мгновенно фильтрует переводы ETH и стейблкоинов (USDT/USDC) объемом выше заданного лимита.Исходный код
В сфере количественного анализа и алгоритмической торговли скорость получения информации определяет альфу. Мемпул (mempool) Ethereum — это «зал ожидания» блокчейна, где транзакции находятся перед тем, как валидаторы включат их в блок. Мониторинг мемпула позволяет крупным игрокам детектировать крупные перемещения средств (Whale Alerts) до их фактического исполнения. Это дает возможность прогнозировать локальные изменения ликвидности, выявлять крупные рыночные ордера на DEX и защищать свои алгоритмы от проскальзывания.
При разработке торговых систем, таких как Реализация Grid Bot для Bybit на Python: Гайд Quant-разработчика, скорость обработки рыночных событий является ключевым фактором успеха. Ниже представлен готовый к продакшену асинхронный скрипт. Он подключается к ноде Ethereum через WebSocket, подписывается на поток транзакций в мемпуле и в реальном времени фильтрует переводы ETH, а также стейблкоинов USDT и USDC, превышающие заданные лимиты.
import asyncio
import logging
from web3 import AsyncWeb3
from web3.providers import AsyncWebSocketProvider
# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Укажите ваш WSS URL (например, Alchemy или Infura)
WSS_NODE_URL = "wss://eth-mainnet.g.alchemy.com/v2/YOUR_API_KEY"
# Пороговые значения для фильтрации (Whale Alerts)
ETH_THRESHOLD = 50.0 # в ETH
USDT_THRESHOLD = 100000.0 # в USDT ($100k)
USDC_THRESHOLD = 100000.0 # в USDC ($100k)
# Адреса смарт-контрактов стейблкоинов
USDT_ADDRESS = "0xdAC17F958D2ee523a2206206994597C13D831ec7"
USDC_ADDRESS = "0xA0b86991c6218b36c1d19D4a2e9Eb0ce3606eB48"
# Селектор метода transfer(address,uint256) -> 0xa9059cbb
ERC20_TRANSFER_SIGNATURE = "0xa9059cbb"
def decode_erc20_transfer(input_data: str) -> tuple[str, float] | None:
"""Быстрое декодирование транзакции transfer без загрузки полного ABI"""
try:
if not input_data or len(input_data) < 138:
return None
method_id = input_data[:10]
if method_id != ERC20_TRANSFER_SIGNATURE:
return None
# Извлекаем адрес получателя (убираем лишние нули слева)
to_address = "0x" + input_data[34:74]
# Извлекаем сумму перевода в hex и переводим в int
value_hex = input_data[74:138]
raw_value = int(value_hex, 16)
return to_address, raw_value
except Exception:
return None
async def process_transaction(w3: AsyncWeb3, tx_hash: str):
"""Получение деталей транзакции и проверка лимитов"""
try:
tx = await w3.eth.get_transaction(tx_hash)
if tx is None:
return
# 1. Проверка транзакций с нативным ETH
if tx['value'] > 0:
eth_amount = w3.from_wei(tx['value'], 'ether')
if eth_amount >= ETH_THRESHOLD:
logger.info(f"[WHALE ETH] {eth_amount:.2f} ETH | From: {tx['from']} -> To: {tx['to']} | Hash: {tx_hash.hex()}")
return
# 2. Проверка транзакций со стейблкоинами (ERC-20)
if tx['to'] and tx['to'].lower() in [USDT_ADDRESS.lower(), USDC_ADDRESS.lower()]:
decoded = decode_erc20_transfer(tx['input'])
if decoded:
to_addr, raw_val = decoded
decimals = 6 # Для USDT и USDC
amount = raw_val / (10 ** decimals)
threshold = USDT_THRESHOLD if tx['to'].lower() == USDT_ADDRESS.lower() else USDC_THRESHOLD
if amount >= threshold:
token_symbol = "USDT" if tx['to'].lower() == USDT_ADDRESS.lower() else "USDC"
logger.info(f"[WHALE {token_symbol}] {amount:,.2f} {token_symbol} | From: {tx['from']} -> To: {to_addr} | Hash: {tx_hash.hex()}")
except Exception:
# Игнорируем ошибки (транзакция могла быть быстро отменена или уже смайнена)
pass
async def main():
async with AsyncWeb3(AsyncWebSocketProvider(WSS_NODE_URL)) as w3:
if not await w3.is_connected():
logger.error("Не удалось подключиться к ноде Ethereum")
return
logger.info("Успешное подключение к WebSocket. Подписка на мемпул...")
# Подписка на новые pending транзакции
subscription_id = await w3.eth.subscribe("pendingTransactions")
async for response in w3.socket.process_subscriptions():
tx_hash = response["result"]
# Запускаем обработку транзакции асинхронно, чтобы не блокировать поток мемпула
asyncio.create_task(process_transaction(w3, tx_hash))
if __name__ == "__main__":
asyncio.run(main())Разбор параметров
AsyncWebSocketProvider: Асинхронный провайдер для работы с WebSocket-соединением. Использование WebSockets критически важно для мемпула, так как HTTP-запросы (polling) создают слишком большую задержку и нагрузку на ноду.pendingTransactions: Тип подписки, который указывает ноде отправлять хэши всех транзакций, поступающих в мемпул, еще до их валидации.decode_erc20_transfer: Кастомная функция для быстрого парсинга сырых данных транзакции (input). Вместо тяжелой загрузки полного ABI контракта, мы напрямую проверяем сигнатуру методаtransfer(address,uint256), которая в шестнадцатеричном формате всегда равна0xa9059cbb. Это значительно ускоряет обработку.asyncio.create_task: Метод для неблокирующего запуска обработчика транзакции. Для параллельной обработки сотен транзакций в секунду мы используем асинхронность, аналогично тому, как реализуется Пакетная отмена ордеров (Batch Cancel) на Bybit V5 через asyncio.gather.
Как запустить
Для запуска скрипта вам понадобятся установленный Python версии 3.10+ и библиотека web3. Установите зависимость командой pip install web3.
Замените значение переменной WSS_NODE_URL на адрес вашей WebSocket-ноды (например, от Alchemy, Infura или собственной ноды Geth). Запустите скрипт в терминале. Он начнет выводить логи при обнаружении крупных транзакций.
Если в дальнейшем вы планируете масштабировать свои торговые инструменты на другие сети, например Solana, рекомендуем ознакомиться с руководством по Подпись запросов Ed25519 для Jupiter DEX на Python: Гайд.
