Ви запускаєте бота для 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 тижнів. Вартість розраховується індивідуально залежно від складності. Отримайте консультацію — ми оцінимо ваш проєкт і запропонуємо оптимальне рішення. Замовте розробку скринера для раннього виявлення токенів та зниження витрат на моніторинг.