Граббінг даних нових токенів (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 перевірка потребує симуляції транзакцій покупки та продажу через eth_call — визначає buy/sell税 та можливість продати токен взагалі. Сервіси: 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, risk-аналізом та REST API: 3–5 тижнів розробки.