Мониторинг баланса пулов Uniswap V2 с Web3.py: Руководство для Quant Developer

Суть: Данная статья-инструкция предоставляет готовый скрипт на Python с использованием библиотеки Web3.py для эффективного мониторинга текущих балансов токенов (резервов) в любом пуле ликвидности Uniswap V2. Скрипт взаимодействует напрямую со смарт-контрактом пула, динамически извлекая информацию о токенах и их количестве.

Содержание

Исходный код

Представленный ниже скрипт на Python позволяет подключиться к блокчейну Ethereum (или любой EVM-совместимой сети), используя RPC-провайдера, и получить актуальные данные о резервах токенов в указанном пуле ликвидности Uniswap V2. Он динамически определяет адреса токенов token0 и token1, а затем запрашивает их символы и количество десятичных знаков для корректного отображения балансов.


import os
from web3 import Web3

# --- Конфигурация --- 
# Замените на ваш RPC URL (например, Infura, Alchemy, или локальный узел Geth/OpenEthereum)
# Рекомендуется использовать переменные окружения для чувствительных данных
RPC_URL = os.getenv("WEB3_RPC_URL", "https://mainnet.infura.io/v3/ВАШ_INFURA_PROJECT_ID")
# Пример адреса пула WETH/USDC на Uniswap V2 Mainnet
# Вы можете найти адрес пула на Uniswap Info или Etherscan
PAIR_ADDRESS = "0xB4e16d0168e52d35f2444f8fA8Dcb07774fB7e8F" # WETH/USDC Pair on Ethereum Mainnet

# --- ABIs --- 
# Минимальный ABI для контракта Uniswap V2 Pair
# Содержит функции для получения адресов токенов и резервов
UNISWAP_V2_PAIR_ABI = [
    {
        "constant": True,
        "inputs": [],
        "name": "getReserves",
        "outputs": [
            {"internalType": "uint112", "name": "_reserve0", "type": "uint112"},
            {"internalType": "uint112", "name": "_reserve1", "type": "uint112"},
            {"internalType": "uint32", "name": "_blockTimestampLast", "type": "uint32"},
        ],
        "payable": False,
        "stateMutability": "view",
        "type": "function",
    },
    {
        "constant": True,
        "inputs": [],
        "name": "token0",
        "outputs": [{"internalType": "address", "name": "", "type": "address"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function",
    },
    {
        "constant": True,
        "inputs": [],
        "name": "token1",
        "outputs": [{"internalType": "address", "name": "", "type": "address"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function",
    },
]

# Минимальный ABI для ERC20 токена
# Содержит функции для получения символа и количества десятичных знаков
ERC20_ABI = [
    {
        "constant": True,
        "inputs": [],
        "name": "symbol",
        "outputs": [{"internalType": "string", "name": "", "type": "string"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function",
    },
    {
        "constant": True,
        "inputs": [],
        "name": "decimals",
        "outputs": [{"internalType": "uint8", "name": "", "type": "uint8"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function",
    },
]

def get_token_info(web3_instance: Web3, token_address: str):
    """Получает символ и количество десятичных знаков для ERC20 токена."""
    token_contract = web3_instance.eth.contract(address=token_address, abi=ERC20_ABI)
    try:
        symbol = token_contract.functions.symbol().call()
        decimals = token_contract.functions.decimals().call()
        return symbol, decimals
    except Exception as e:
        print(f"Ошибка при получении информации о токене {token_address}: {e}")
        return f"UNKNOWN_{token_address[:6]}", 18 # Fallback to default decimals

def monitor_uniswap_v2_pool(rpc_url: str, pair_address: str):
    """
    Подключается к блокчейну, получает информацию о пуле Uniswap V2
    и выводит текущие резервы токенов.
    """
    try:
        web3 = Web3(Web3.HTTPProvider(rpc_url))

        # Проверка подключения
        if not web3.is_connected():
            raise ConnectionError(f"Не удалось подключиться к RPC узлу: {rpc_url}")

        print(f"Успешно подключено к RPC узлу: {rpc_url}")
        print(f"Текущий блок: {web3.eth.block_number}")

        # Создаем экземпляр контракта пула Uniswap V2
        pair_contract = web3.eth.contract(address=web3.to_checksum_address(pair_address), abi=UNISWAP_V2_PAIR_ABI)

        # Получаем адреса токенов в пуле
        token0_address = pair_contract.functions.token0().call()
        token1_address = pair_contract.functions.token1().call()

        # Получаем информацию о токенах
        token0_symbol, token0_decimals = get_token_info(web3, token0_address)
        token1_symbol, token1_decimals = get_token_info(web3, token1_address)

        print(f"\nМониторинг пула: {pair_address}")
        print(f"Токен 0: {token0_symbol} ({token0_address})")
        print(f"Токен 1: {token1_symbol} ({token1_address})")

        # Получаем текущие резервы
        reserves = pair_contract.functions.getReserves().call()
        reserve0_raw = reserves[0]
        reserve1_raw = reserves[1]

        # Конвертируем резервы в читаемый формат
        reserve0_formatted = reserve0_raw / (10 ** token0_decimals)
        reserve1_formatted = reserve1_raw / (10 ** token1_decimals)

        print(f"\nТекущие резервы:")
        print(f"  {token0_symbol}: {reserve0_formatted:,.6f}")
        print(f"  {token1_symbol}: {reserve1_formatted:,.6f}")

    except ConnectionError as ce:
        print(f"Ошибка подключения: {ce}")
    except Exception as e:
        print(f"Произошла ошибка: {e}")

if __name__ == "__main__":
    # Пример использования:
    monitor_uniswap_v2_pool(RPC_URL, PAIR_ADDRESS)

    # Для сетей PoA (например, Polygon, BSC), возможно, потребуется добавить middleware:
    # from web3.middleware import geth_poa_middleware
    # web3.middleware_onion.inject(geth_poa_middleware, layer=0)
    # Это должно быть сделано после инициализации Web3(Web3.HTTPProvider(rpc_url))
    # и до любых вызовов контрактов.
    # В данном скрипте это не требуется, так как мы используем Ethereum Mainnet.

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

Для корректной работы скрипта необходимо настроить несколько ключевых параметров и понять назначение используемых функций и ABI.

RPC_URL:
Строка, содержащая URL-адрес узла блокчейна (например, Infura, Alchemy, QuickNode или ваш собственный Geth/OpenEthereum узел). Это точка входа для взаимодействия с сетью Ethereum. Для продакшн-среды рекомендуется использовать переменные окружения для хранения таких чувствительных данных, как ключи API.
PAIR_ADDRESS:
Адрес смарт-контракта пула ликвидности Uniswap V2, который вы хотите мониторить. Этот адрес можно найти на таких ресурсах, как Uniswap Info, Etherscan или других обозревателях блоков для соответствующей сети.
UNISWAP_V2_PAIR_ABI:
Минимальный Application Binary Interface (ABI) для контракта Uniswap V2 Pair. Он включает только те функции, которые необходимы для получения адресов токенов (token0(), token1()) и текущих резервов (getReserves()). ABI необходим Web3.py для правильного формирования вызовов к смарт-контракту.
ERC20_ABI:
Минимальный ABI для стандартного токена ERC20. Он содержит функции для получения символа токена (symbol()) и количества его десятичных знаков (decimals()). Эти данные критически важны для корректного форматирования и отображения балансов токенов.
get_token_info(web3_instance, token_address):
Вспомогательная функция, которая принимает экземпляр Web3 и адрес токена. Она создает экземпляр контракта ERC20 и вызывает функции symbol() и decimals() для получения соответствующей информации. В случае ошибки возвращает заглушку с 18 десятичными знаками по умолчанию.
monitor_uniswap_v2_pool(rpc_url, pair_address):
Основная функция скрипта. Она устанавливает соединение с блокчейном, создает экземпляр контракта пула Uniswap V2, извлекает адреса токенов, затем использует get_token_info() для получения их символов и десятичных знаков. В конечном итоге, она вызывает getReserves() для получения сырых значений резервов и форматирует их для вывода.

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

Для запуска скрипта выполните следующие шаги:

1. Установите необходимые библиотеки:

Откройте терминал или командную строку и выполните команду для установки web3.py:


pip install web3

2. Получите RPC URL:

Вам понадобится доступ к узлу Ethereum. Вы можете использовать бесплатные или платные сервисы, такие как Infura, Alchemy, QuickNode. Зарегистрируйтесь на одном из них, создайте проект и получите ваш RPC URL. Замените ВАШ_INFURA_PROJECT_ID в переменной RPC_URL на ваш реальный ID проекта.

3. Выберите адрес пула ликвидности:

Определите адрес пула Uniswap V2, который вы хотите мониторить. Например, для пула WETH/USDC на Ethereum Mainnet используется адрес 0xB4e16d0168e52d35f2444f8fA8Dcb07774fB7e8F. Замените значение переменной PAIR_ADDRESS на адрес интересующего вас пула.

4. Сохраните скрипт:

Скопируйте весь предоставленный исходный код и сохраните его в файл с расширением .py, например, monitor_uniswap.py.

5. Запустите скрипт:

Откройте терминал в директории, где вы сохранили файл, и выполните команду:


python monitor_uniswap.py

Ожидаемый вывод:

Скрипт выведет информацию о подключении к RPC узлу, текущем номере блока, адресах и символах токенов в пуле, а затем отобразит текущие резервы каждого токена в удобочитаемом формате.


Успешно подключено к RPC узлу: https://mainnet.infura.io/v3/ВАШ_INFURA_PROJECT_ID
Текущий блок: 18XXXXXX

Мониторинг пула: 0xB4e16d0168e52d35f2444f8fA8Dcb07774fB7e8F
Токен 0: WETH (0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2)
Токен 1: USDC (0xA0b86991c6218b36c1d19D4a2e9Eb0ce3606eB48)

Текущие резервы:
  WETH: 12,345.678901
  USDC: 23,456,789.012345

Теперь у вас есть рабочий скрипт для мониторинга балансов пулов ликвидности Uniswap V2, который можно адаптировать для более сложных торговых стратегий или систем оповещения.