Вы запускаете бота для 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% скам-токенов на этапе детектирования.
Сравнение методов детектирования
| Метод | Задержка | Надёжность | Нагрузка на RPC |
|---|---|---|---|
| Factory (Uniswap/PancakeSwap) | 1-2 блока | Высокая (гарантированное событие) | Низкая (фильтр по топику) |
| Прямой деплой ERC-20 | 1 блок | Средняя (не все контракты ERC-20) | Высокая (анализ каждой транзакции) |
| Non-EVM (Solana) | ~1 слот | Высокая (InitializeMint) | Средняя |
Почему factory-мониторинг быстрее сканирования блоков?
Factory-контракты генерируют событие при создании пула — это единственный сигнал, который нужно отслеживать. Полное сканирование блоков требует обработки каждой транзакции, что в 10-15 раз увеличивает нагрузку на RPC и замедляет детектирование. Поэтому для продакшен-систем мы рекомендуем комбинировать оба метода, но приоритет отдавать событиям фабрик.
Пример детального анализа байткода
При обнаружении функции `mint` с откзывами на продажу, токен маркируется как высокорисковый. Проверка через eth_call с разными суммами показывает скрытые комиссии до 99%.Что входит в разработку скринера?
| Компонент | Описание |
|---|---|
| Архитектура решения | Выбор стека, распределение нагрузки, схема БД |
| Код парсинга | Реализация мониторинга factory, прямых деплоев, enrichment |
| Риск-анализ | Байткод-анализ, honeypot check, интеграция с API |
| REST API | Эндпоинты для получения данных, фильтрация, WebSocket |
| Документация | README, описание эндпоинтов, примеры запросов |
| Поддержка | 3 месяца после деплоя, исправление багов, консультации |
Процесс работы
- Аналитика — изучаем ваши требования, выбираем сети и методы.
- Проектирование — создаём архитектуру и схему данных.
- Реализация — пишем код парсинга, enrichment, API.
- Тестирование — симуляция нагрузки, проверка на реальных данных.
- Деплой — разворачиваем на вашей инфраструктуре или нашей.
- Поддержка — обучаем команду, передаём документацию.
Сроки и стоимость
Срок разработки базового скринера для 3-4 EVM-сетей — от 3 до 5 недель. Стоимость рассчитывается индивидуально в зависимости от сложности. Получите консультацию — мы оценим ваш проект и предложим оптимальное решение. Закажите разработку скринера для раннего обнаружения токенов и снижения затрат на мониторинг.