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

Розробка ML-конвеєра обробки on-chain даних

Завдання зазвичай формулюється так: «хочемо передбачати 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.

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

Чому сирі дані не підходять для 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 конвеєра

Шар IngEstion

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

[Archive Node / Firehose]
         ↓
  [Kafka / Redpanda]          ← hot path: < 1s latency
         ↓
  [Stream Processor]          ← Flink або custom 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. При генерації навчальної вибірки join відбувається по 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 специфіки часто простіше custom рішення: занадто багато особливостей (reorg handling, multi-chain entities, temporal correctness).

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

Реорги на рівні ML-даних — серйозна проблема. Якщо ознаки обчислені з блоку, який згодом став orphaned, навчальна вибірка містить нереальні дані.

Рішення:

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

MLOps інтеграція

Конвеєр повинен дружити з існуючим ML-стеком:

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

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

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

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

Типові етапи проекту

Етап 1 — Data audit (1-2 тижні). Визначення потрібних сигналів, їх джерел, доступності історичних даних. Прототип IngEstor на невеликому блок-діапазоні.

Етап 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 конвеєра. Оцінка залежить від кількості ланцюгів, глибини історичних даних та вимог до latency inference.