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

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