Декодирование транзакций Uniswap V3 Mint/Burn из мемпула на Python

Суть: Извлечение транзакций добавления (Mint) и удаления (Burn/Decrease Liquidity) ликвидности Uniswap V3 из мемпула реализуется через подписку на поток pendingTransactions по протоколу WebSocket с последующим рекурсивным декодированием вызовов multicall с помощью библиотеки web3.py.

Содержание

Исходный код

Для успешного построения MEV-ботов и маркет-мейкинг стратегий критически важно уметь анализировать транзакции до их попадания в блок. Если вы только начинаете работать с мемпулом, рекомендуем сначала прочитать нашу базовую статью Мониторинг мемпула Ethereum на Python: Скрипт Whale Alerts через Web3.py.

Ниже представлен готовый асинхронный скрипт на Python, который подключается к мемпулу Ethereum через WebSocket, фильтрует транзакции к контракту NonfungiblePositionManager Uniswap V3 и декодирует их параметры, включая вложенные вызовы внутри multicall.

import asyncio
from web3 import AsyncWeb3
from web3.exceptions import TransactionNotFound

# Uniswap V3 NonfungiblePositionManager Address
NPM_ADDRESS = '0xC36442b4a4522E871399CD717aBDD847Ab11FE88'

# Minimal ABI for decoding Mint, Burn (decreaseLiquidity), and Multicall
NPM_ABI = [
    {
        'inputs': [
            {
                'components': [
                    {'name': 'token0', 'type': 'address'},
                    {'name': 'token1', 'type': 'address'},
                    {'name': 'fee', 'type': 'uint24'},
                    {'name': 'tickLower', 'type': 'int24'},
                    {'name': 'tickUpper', 'type': 'int24'},
                    {'name': 'amount0Desired', 'type': 'uint256'},
                    {'name': 'amount1Desired', 'type': 'uint256'},
                    {'name': 'amount0Min', 'type': 'uint256'},
                    {'name': 'amount1Min', 'type': 'uint256'},
                    {'name': 'recipient', 'type': 'address'},
                    {'name': 'deadline', 'type': 'uint256'}
                ],
                'name': 'params',
                'type': 'tuple'
            }
        ],
        'name': 'mint',
        'outputs': [],
        'stateMutability': 'payable',
        'type': 'function'
    },
    {
        'inputs': [
            {
                'components': [
                    {'name': 'tokenId', 'type': 'uint256'},
                    {'name': 'liquidity', 'type': 'uint128'},
                    {'name': 'amount0Min', 'type': 'uint256'},
                    {'name': 'amount1Min', 'type': 'uint256'},
                    {'name': 'deadline', 'type': 'uint256'}
                ],
                'name': 'params',
                'type': 'tuple'
            }
        ],
        'name': 'decreaseLiquidity',
        'outputs': [],
        'stateMutability': 'payable',
        'type': 'function'
    },
    {
        'inputs': [{'name': 'data', 'type': 'bytes[]'}],
        'name': 'multicall',
        'outputs': [],
        'stateMutability': 'payable',
        'type': 'function'
    }
]

def decode_tx_input(w3, contract, input_data):
    try:
        func_obj, decoded_params = contract.decode_function_input(input_data)
        if func_obj.fn_name == 'multicall':
            results = []
            for sub_data in decoded_params['data']:
                sub_decoded = decode_tx_input(w3, contract, sub_data)
                if sub_decoded:
                    results.append(sub_decoded)
            return {'method': 'multicall', 'calls': results}
        return {'method': func_obj.fn_name, 'params': decoded_params}
    except Exception:
        return None

async def monitor_mempool(ws_url):
    # Connect to Ethereum node via WebSocket
    async with AsyncWeb3.persistent_websocket(AsyncWeb3.WebSocketProvider(ws_url)) as w3:
        print('Connected to WebSocket. Subscribing to pending transactions...')
        
        # Subscribe to pending transactions
        subscription_id = await w3.eth.subscribe('pendingTransactions')
        contract = w3.eth.contract(address=AsyncWeb3.to_checksum_address(NPM_ADDRESS), abi=NPM_ABI)

        async for response in w3.ws.listen():
            try:
                tx_hash = response['result']
                tx = await w3.eth.get_transaction(tx_hash)
                
                # Check if transaction is sent to Uniswap V3 Position Manager
                if tx and tx.get('to') and tx['to'].lower() == NPM_ADDRESS.lower():
                    decoded = decode_tx_input(w3, contract, tx['input'])
                    if decoded:
                        print(f'New Uniswap V3 Tx: {tx_hash.hex()}')
                        print(f'Decoded Data: {decoded}\n')
            except TransactionNotFound:
                continue
            except Exception as e:
                print(f'Error processing transaction: {e}')

if __name__ == '__main__':
    INFURA_WS_URL = 'wss://mainnet.infura.io/ws/v3/YOUR_API_KEY'
    asyncio.run(monitor_mempool(INFURA_WS_URL))

Разбор параметров

NPM_ADDRESS: Адрес смарт-контракта NonfungiblePositionManager (0xC36442b4a4522E871399CD717aBDD847Ab11FE88) в сети Ethereum Mainnet, который управляет позициями ликвидности (NFT) для Uniswap V3.
NPM_ABI: Облегченный интерфейс ABI, содержащий только необходимые структуры для декодирования функций mint, decreaseLiquidity и multicall. Это снижает накладные расходы памяти.
decode_tx_input: Функция декодирования входящих данных транзакции. Если транзакция упакована в multicall (что является стандартом для Uniswap V3), функция рекурсивно разбирает каждый вложенный вызов.
pendingTransactions: Тип подписки в WebSocket-соединении, позволяющий получать хэши транзакций в режиме реального времени до их валидации майнерами.

Как запустить

Для запуска скрипта вам понадобятся установленные библиотеки web3 и asyncio. Установите их с помощью пакетного менеджера pip:

pip install web3 asyncio

Замените плейсхолдер YOUR_API_KEY в переменной INFURA_WS_URL на ваш реальный ключ доступа к ноде (Infura, Alchemy или QuickNode) и запустите скрипт:

python script.py

После запуска скрипт начнет выводить в консоль структурированные данные о транзакциях добавления и изъятия ликвидности Uniswap V3 прямо из мемпула.

После освоения мемпула Ethereum вы можете расширить свой стек на другие сети и торговые инструменты. Например, изучите Подпись запросов Ed25519 для Jupiter DEX на Python: Гайд для работы на Solana, или перейдите к классическому алготрейдингу с помощью руководства Реализация Grid Bot для Bybit на Python: Гайд Quant-разработчика.