Разработка pipeline обработки on-chain данных для ML

Разработка pipeline обработки on-chain данных для ML

Задача формулируется примерно так: «хотим предсказывать whale-активность» или «нужна модель оценки on-chain кредитного риска». За этим стоит инженерная проблема, которую большинство команд недооценивает: сырые блокчейн-данные не пригодны для ML-моделей напрямую. Структура блока, raw hex-encoded calldata, адреса в bytes20 — это не фичи, это сырьё. Между RPC-нодой и обучающей выборкой лежит несколько недель инфраструктурной работы.

Откуда берутся данные и почему это сложно

Источники и их ограничения

Публичные RPC (eth_getLogs, eth_getBlockByNumber) — самый доступный, но наименее пригодный для ML источник. Ограничения: rate limits (Infura/Alchemy — 10-333 req/s на платных тарифах), отсутствие internal transactions без trace_ namespace, отсутствие pre/post state без архивной ноды.

Archive node с trace API — полная история включая internal calls, eth_getStorageAt на любом историческом блоке. Erigon — стандартный выбор для self-hosted: занимает ~2.5 TB для Ethereum mainnet, синхронизируется за 3-5 дней. Важно: Erigon и Geth/Besu имеют разные форматы trace_ ответов — парсер нужно адаптировать.

Firehose (StreamingFast/The Graph) — бинарный стриминговый протокол, экспортирует каждый блок с полным деревом вызовов, state diffs, receipt. Latency < 500ms от блока. Это самый производительный источник для исторической загрузки: 100k+ блоков в минуту против 1-5k через RPC. Используется в Substreams.

Специализированные поставщики — Nansen, Dune Analytics, Flipside Crypto, Allium. Готовые нормализованные таблицы, SQL-интерфейс. Минусы: задержка обновления 1-24 часа, ограниченный контроль над схемой, стоимость при большом объёме.

Почему raw данные не подходят для ML

Raw Transfer лог — это три bytes32 + data bytes. До ML-признаков нужно пройти:

1. Декодирование — ABI-декодинг topics и data
2. Нормализация адресов — uint256 → checksummed hex, label mapping (биржи, протоколы, MEV-боты)
3. Денежная нормализация — value / 10^decimals, конвертация в USD через исторический price feed
4. Entity resolution — один EOA может иметь сотни транзакций, но быть одним экономическим агентом; смарт-контракты — прокси, реализации, multisig

Пропуск любого из этих шагов приводит к мусорным признакам.

Архитектура production pipeline

Слой ингестии

Рекомендуемая архитектура — event-driven с разделением hot и cold path:

[Archive Node / Firehose]
         ↓
  [Kafka / Redpanda]          ← hot path: < 1s latency
         ↓
  [Stream Processor]          ← Flink или кастомный consumer
    /          \
[Raw Store]  [Feature Store]  ← cold: S3/Parquet, hot: Redis/Feast

Kafka topic per chain, ключ = block_number:log_index. Это гарантирует порядок и позволяет replay при ошибках обработки. Retention зависит от задачи: для real-time фичей — 7 дней, для переобучения — полный архив в S3.

Для Ethereum mainnet: ~6000 транзакций/блок × ~6500 блоков/день = ~39M транзакций/день. При среднем размере транзакции с trace ~2KB — ~75GB/день сырых данных. Планируйте хранилище.

Трансформация и feature engineering

Это самая трудоёмкая часть. Типовые on-chain признаки для различных ML-задач:

Wallet profiling (DeFi credit scoring, Sybil detection):

MEV detection:

• Sandwich attack pattern: три TX в одном блоке, один адрес, окружают target TX
• Arbitrage: циклические трансферы токенов возвращающиеся к sender в рамках одной TX
• Flashloan: FlashLoan event + position delta = 0 к концу блока

Whale activity prediction:

• Большие трансферы с exchange deposit addresses → вероятность sell pressure
• Accumulation pattern: множественные небольшие покупки с разных адресов → один получатель

# Пример feature engineering для wallet scoring
import polars as pl

def compute_wallet_features(txs: pl.DataFrame) -> pl.DataFrame:
    return txs.group_by("from_address").agg([
        pl.col("block_number").min().alias("first_seen_block"),
        pl.col("block_number").max().alias("last_seen_block"),
        pl.count("hash").alias("tx_count"),
        pl.col("to_address").n_unique().alias("unique_contracts"),
        pl.col("gas_price").quantile(0.5).alias("gas_price_median"),
        pl.col("value_usd").sum().alias("total_volume_usd"),
        pl.col("block_timestamp").diff().dt.total_seconds()
            .mean().alias("avg_interval_seconds"),
    ])

Polars вместо Pandas — разница в скорости обработки больших датасетов (миллионы строк) составляет 5-20x.

Temporal features и data leakage

Это главная методологическая ловушка on-chain ML. Признаки должны быть вычислены только из данных доступных до момента предсказания. Типичная ошибка: использование total_tx_count адреса вместо tx_count_at_time_T.

Паттерн: point-in-time correct features. Каждая строка в feature store имеет entity_id, feature_timestamp, feature_value. При генерации обучающей выборки джойн идёт по entity_id и feature_timestamp <= label_timestamp.

-- Point-in-time join
SELECT 
    l.wallet_address,
    l.label,
    l.label_timestamp,
    f.tx_count,
    f.unique_contracts,
    f.volume_usd_30d
FROM labels l
ASOF JOIN wallet_features f
    ON l.wallet_address = f.wallet_address
    AND f.feature_timestamp <= l.label_timestamp

ASOF JOIN — нативная операция в ClickHouse и TimescaleDB, в PostgreSQL эмулируется через LATERAL.

Feature Store

Для production ML на on-chain данных нужен feature store с двумя слоями:

Offline store — исторические фичи для обучения. ClickHouse или Parquet на S3 с Hive-partitioning по дате. Запросы по диапазону дат.

Online store — актуальные фичи для inference. Redis Hash structures: HGETALL wallet:{address}:features. Обновляется при каждом новом блоке для активных адресов.

Feast — популярный open-source feature store, поддерживает оба слоя. Но для on-chain специфики часто проще кастомное решение: слишком много особенностей (reorg handling, multi-chain entities, temporal correctness).

Обработка реорганизаций

Реорги на уровне ML-данных — серьёзная проблема. Если фичи вычислены из блока, который впоследствии стал orphaned, обучающая выборка содержит нереальные данные.

Решения:

• Confirmation lag — индексировать только блоки старше N блоков (обычно 12-32 для финальности на PoS Ethereum). Добавляет задержку, но устраняет проблему.
• Versioned features — хранить (entity, block_hash, features), при реорге помечать orphaned записи. Сложнее, но позволяет работать с малой задержкой.

MLOps интеграция

Pipeline должен дружить с существующим ML-стеком:

Feature generation → обучение: экспорт в Parquet/CSV для DVC или MLflow artifacts. Versioning датасетов критичен — модель обученная на данных за конкретный период должна быть воспроизводима.

Inference pipeline: новый блок → вычисление дельта-фичей → update в online store → триггер inference. Latency бюджет обычно 1-10 секунд от блока до предсказания.

Model drift monitoring: on-chain данные меняются структурно (merge, новые протоколы, изменения паттернов использования). Нужен мониторинг дистрибуции входных признаков — Evidently AI или кастомный.

Технологический стек

Типичные этапы проекта

Этап 1 — Data audit (1-2 недели). Определение нужных сигналов, их источников, доступности исторических данных. Прототип ингестора на небольшом блок-диапазоне.

Этап 2 — Historical backfill (2-4 недели). Загрузка исторических данных, нормализация, label mapping. Самый трудоёмкий этап.

Этап 3 — Feature pipeline (2-3 недели). Реализация feature engineering, point-in-time logic, хранилища.

Этап 4 — Real-time path (1-2 недели). Стриминг из ноды, online store, inference интеграция.

Этап 5 — MLOps (1-2 недели). Мониторинг дрейфа, версионирование датасетов, автоматизация переобучения.

Итого: 7-13 недель до production-ready pipeline. Оценка сильно зависит от количества цепей, глубины исторических данных и требований к latency inference.