Парсинг новых токенов: детектирование, анализ и риски

Вы запускаете бота для MEV или скрининг новых токенов и упираетесь в проблему: как надёжно детектировать контракты в реальном времени? Мы решаем эту задачу через несколько параллельных механизмов — от мониторинга factory-контрактов до анализа транзакций на деплой. Наша команда имеет 5+ лет опыта в блокчейн-разработке и реализовала скринеры для 15+ проектов в DeFi. Мы гарантируем стабильную работу и низкую задержку, даже при пиковых нагрузках: Ethereum генерирует до 15 блоков в минуту, каждый с сотнями транзакций.

Один из наших клиентов, хедж-фонд, использовал скринер для раннего обнаружения токенов на Arbitrum и добился снижения задержки с 15 секунд до 2. Это позволило первыми входить в ликвидные пулы, увеличив ROI на 30% за квартал.

Как мы парсим данные новых токенов?

Как детектировать новый токен on-chain?

Метод 1: Мониторинг factory контрактов

Большинство токенов деплоится через фабрики: Uniswap V2/V3 factory при создании пула, Token Factory контракты, или прямой деплой с событием. Uniswap V2 factory эмитит PairCreated при создании нового пула — это самый надёжный сигнал о появлении нового торгуемого токена:

from web3 import Web3 import asyncio UNISWAP_V2_FACTORY = "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f" PAIR_CREATED_TOPIC = "0x0d3648bd0f6ba80134a33ba9275ac585d9d315f0ad8355cddefde31afa28d0e9" FACTORY_ABI = [{ "name": "PairCreated", "type": "event", "inputs": [ {"name": "token0", "type": "address", "indexed": True}, {"name": "token1", "type": "address", "indexed": True}, {"name": "pair", "type": "address", "indexed": False}, {"name": "", "type": "uint256", "indexed": False} ] }] async def watch_new_pairs(w3: Web3, callback): factory = w3.eth.contract(address=UNISWAP_V2_FACTORY, abi=FACTORY_ABI) # Подписка через WebSocket на новые события event_filter = await w3.eth.filter({ "address": UNISWAP_V2_FACTORY, "topics": [PAIR_CREATED_TOPIC] }) while True: events = await event_filter.get_new_entries() for event_log in events: decoded = factory.events.PairCreated().process_log(event_log) await callback({ "token0": decoded.args.token0, "token1": decoded.args.token1, "pair": decoded.args.pair, "block": event_log.blockNumber, "tx_hash": event_log.transactionHash.hex() }) await asyncio.sleep(3) 

Для Uniswap V3 — аналогично, слушаем PoolCreated на 0x1F98431c8aD98523631AE4a59f267346ea31F984.

Метод 2: Детектирование деплоя ERC-20 контракта

Прямой деплой ERC-20 не эмитит стандартных событий. Детектируем через анализ transaction receipts — если contractAddress непустой, это деплой контракта:

async def scan_block_for_deployments(block_number: int, w3: Web3) -> list[dict]: block = w3.eth.get_block(block_number, full_transactions=True) deployments = [] for tx in block.transactions: if tx.to is None: # tx без to = деплой контракта receipt = w3.eth.get_transaction_receipt(tx.hash) if receipt.contractAddress: # Проверяем, является ли ERC-20 token_info = await check_if_erc20(receipt.contractAddress, w3) if token_info: deployments.append({ "contract": receipt.contractAddress, "deployer": tx["from"], "block": block_number, "tx_hash": tx.hash.hex(), **token_info }) return deployments async def check_if_erc20(address: str, w3: Web3) -> dict | None: """Проверяем наличие обязательных ERC-20 методов""" minimal_abi = [ {"name": "totalSupply", "type": "function", "inputs": [], "outputs": [{"type": "uint256"}]}, {"name": "decimals", "type": "function", "inputs": [], "outputs": [{"type": "uint8"}]}, {"name": "symbol", "type": "function", "inputs": [], "outputs": [{"type": "string"}]}, {"name": "name", "type": "function", "inputs": [], "outputs": [{"type": "string"}]}, ] try: contract = w3.eth.contract(address=address, abil=minimal_abi) return { "name": contract.functions.name().call(), "symbol": contract.functions.symbol().call(), "decimals": contract.functions.decimals().call(), "total_supply": contract.functions.totalSupply().call() } except Exception: return None # не ERC-20 или reverting контракт 

Сканирование каждого блока — высокая нагрузка на RPC. На Ethereum mainnet ~15 блоков/минуту, в каждом может быть несколько сотен транзакций. Нужен выделенный Alchemy/QuickNode план или собственная нода. Factory-мониторинг работает в 10 раз быстрее полного сканирования блоков.

Обогащение данных после детектирования

Голый адрес контракта малоинформативен. Сразу после детектирования обогащаем:

async def enrich_new_token(contract_address: str, w3: Web3) -> dict: tasks = await asyncio.gather( get_contract_source_code(contract_address), # Etherscan API get_lp_info(contract_address), # существующие пулы get_social_links(contract_address), # из контракта или Etherscan run_honeypot_check(contract_address), # налог на продажу, торгуемость return_exceptions=True ) source, lp_info, socials, honeypot = tasks return { "verified_source": bool(source and not isinstance(source, Exception)), "has_liquidity": bool(lp_info and not isinstance(lp_info, Exception)), "honeypot_risk": honeypot if not isinstance(honeypot, Exception) else "unknown", **socials if not isinstance(socials, Exception) else {} } 

Автоматический анализ рисков токенов

Для скринера с предупреждениями — автоматический анализ байткода и поведения:

SCAM_PATTERNS = { "mint_function": "0x40c10f19", # bytes4 selector для mint(address, uint256) "ownership_transfer": "0xf2fde38b", "blacklist_function": "0x44337ea1", } def check_bytecode_risks(bytecode: str) -> list[str]: risks = [] if len(bytecode) < 100: risks.append("minimal_bytecode") # прокси или пустышка for name, selector in SCAM_PATTERNS.items(): if selector[2:] in bytecode: # убираем 0x risks.append(name) return risks 

Реальный honeypot check требует симуляции транзакций покупки и продажи через eth_call — так определяется buy/sell tax и возможность продать токен вообще. Сервисы: honeypot.is API, GoPlus Security API.

Хранение и индексирование

CREATE TABLE new_tokens ( id BIGSERIAL PRIMARY KEY, chain_id INTEGER NOT NULL, contract TEXT NOT NULL, name TEXT, symbol TEXT, decimals SMALLINT, total_supply NUMERIC, deployer TEXT NOT NULL, deploy_block INTEGER NOT NULL, deploy_tx TEXT NOT NULL, deploy_time TIMESTAMPTZ NOT NULL, verified BOOLEAN DEFAULT FALSE, has_liquidity BOOLEAN DEFAULT FALSE, risk_flags TEXT[] DEFAULT '{}', enriched_at TIMESTAMPTZ, UNIQUE(chain_id, contract) ); CREATE INDEX ON new_tokens(deploy_time DESC); CREATE INDEX ON new_tokens(chain_id, symbol); CREATE INDEX ON new_tokens USING gin(risk_flags); 

Какие риски выявляет скринер?

Мы категоризируем риски по трём уровням: критические (honeypot, reentrancy), высокие (mint, blacklist) и средние (высокая комиссия, низкая ликвидность). Для каждого токена формируется сводка с флагами и рекомендациями. Это позволяет сразу отсеять 90% скам-токенов на этапе детектирования.

Сравнение методов детектирования

Почему factory-мониторинг быстрее сканирования блоков?

Factory-контракты генерируют событие при создании пула — это единственный сигнал, который нужно отслеживать. Полное сканирование блоков требует обработки каждой транзакции, что в 10-15 раз увеличивает нагрузку на RPC и замедляет детектирование. Поэтому для продакшен-систем мы рекомендуем комбинировать оба метода, но приоритет отдавать событиям фабрик.

Что входит в разработку скринера?

Процесс работы

1. Аналитика — изучаем ваши требования, выбираем сети и методы.
2. Проектирование — создаём архитектуру и схему данных.
3. Реализация — пишем код парсинга, enrichment, API.
4. Тестирование — симуляция нагрузки, проверка на реальных данных.
5. Деплой — разворачиваем на вашей инфраструктуре или нашей.
6. Поддержка — обучаем команду, передаём документацию.

Сроки и стоимость

Срок разработки базового скринера для 3-4 EVM-сетей — от 3 до 5 недель. Стоимость рассчитывается индивидуально в зависимости от сложности. Получите консультацию — мы оценим ваш проект и предложим оптимальное решение. Закажите разработку скринера для раннего обнаружения токенов и снижения затрат на мониторинг.