Граббинг данных новых токенов (new token deployments)

Парсинг данных новых токенов (new token deployments)

Задача — детектировать новые токены на блокчейне в момент деплоя или вскоре после него. Это нужно для: систем мониторинга MEV-возможностей, скринеров новых токенов (как DEX Screener, Dextools), баз данных для аналитики, или защитных систем (сканирование на скам).

Как детектировать новый токен 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, abi=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 план или собственная нода.

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

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

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);

Мониторинг нескольких сетей

EVM-совместимые сети используют одинаковый механизм — один код с разными RPC endpoint'ами. Для non-EVM (Solana, TON) — отдельные адаптеры:

Solana: новые токены через InitializeMint инструкцию Token Program. Raydium/Orca pool creation через InitializePool.

BSC/Polygon: идентично Ethereum, но с другими адресами фабрик PancakeSwap/QuickSwap.

Полный скринер новых токенов для 3–4 EVM сетей с enrichment, риск-анализом и REST API: 3–5 недель разработки.