Разработка модели обнаружения pump-and-dump схем

Разработка модели обнаружения pump-and-dump схем

Pump-and-dump в crypto работает быстрее, чем в традиционных рынках: от начала координированной покупки до dump — часы или даже минуты. On-chain данные полностью публичны, что создаёт уникальную возможность для обнаружения: можно видеть движение кошельков, концентрацию объёма, синхронизацию транзакций в реальном времени.

Задача — построить систему, которая с достаточной достоверностью детектирует P&D схему в процессе pump-фазы, до dump, чтобы предупредить пользователей или автоматически защитить протокол.

Анатомия pump-and-dump схемы

Понимание механики критично для построения правильных признаков.

Фаза накопления (accum): организаторы постепенно скупают токен небольшими ордерами, стараясь не двигать цену. Признаки: рост числа уникальных hold addresses при стагнации цены, необычный buy volume в нерабочие часы, появление скоординированных кошельков (одновременное получение ETH с одного source).

Фаза pump: скоординированная покупка, обычно согласованная в Telegram/Discord. Цена растёт на 200-2000% за часы. Volume спайк в 10-100x от среднего. Social media спайк с шаблонными сообщениями.

Фаза dump: организаторы продают в пик. Розничные покупатели, привлечённые ростом, входят и оказываются holding bags. Цена падает до pre-pump уровня или ниже.

Признаки (features) для модели

On-chain метрики

Volume anomaly score:

VAS = current_volume / rolling_avg_volume_30d

Значения > 10 при отсутствии fundamental news — сильный сигнал.

Holder concentration delta: изменение HHI (Herfindahl-Hirschman Index) балансов за последние N часов:

HHI = Σ (balance_i / total_supply)²

Рост HHI = концентрация токенов у меньшего числа адресов = аккумуляция.

Transaction synchronization: коэффициент синхронности транзакций — насколько много независимых адресов совершили buy в одном временном окне (±5 минут). При органическом росте покупки распределены равномерно. При P&D — spike.

Wallet clustering: граф связей между адресами. Адреса, получавшие ETH с одного источника, куплю с одного EOA (Externally Owned Account), или совершавшие транзакции в схожем паттерне — вероятно контролируются одной сущностью. Если 60%+ volume от кластера — сигнал.

Price-volume divergence: в здоровом росте объём нарастает постепенно вместе с ценой. В P&D сначала объём, потом цена резко — или синхронно, без раскачки.

Кросс-рыночные метрики

DEX vs CEX price discrepancy: если цена на DEX значительно выше чем на CEX (где арбитраж медленнее) — возможна намеренная манипуляция DEX ценой.

Liquidity depth change: резкое уменьшение ликвидности в пуле (removal LP) накануне pump снижает resistance — классический preparation паттерн.

New wallet ratio: процент транзакций от кошельков, созданных < 7 дней назад. Высокий показатель = организаторы создают fresh адреса.

Social signals (опционально)

Telegram/Discord мониторинг через API на упоминания тикера. Резкий рост упоминаний + позитивный sentiment + шаблонные призывы = coordinated pump signal. Это существенно улучшает recall модели, но требует значительно более сложной инфраструктуры.

Архитектура системы обнаружения

Data pipeline

Blockchain RPC (гeth/erigon) 
    → Event streaming (WebSocket subscription)
    → Kafka / RabbitMQ (буфер)
    → Feature extractor (Python)
    → Feature store (Redis для realtime, PostgreSQL для исторических)
    → ML model inference
    → Alert engine

Real-time подключение к блокчейн ноде через WebSocket:

from web3 import Web3, AsyncWeb3
import asyncio

async def stream_swaps(token_address: str, callback):
    w3 = AsyncWeb3(AsyncWeb3.AsyncWebsocketProvider('wss://mainnet.infura.io/ws/v3/KEY'))

    # Подписка на Transfer события токена
    transfer_filter = await w3.eth.filter({
        'address': token_address,
        'topics': [Web3.keccak(text='Transfer(address,address,uint256)').hex()]
    })

    # Или подписка на DEX swap события через Uniswap V3 factory
    while True:
        events = await transfer_filter.get_new_entries()
        for event in events:
            await callback(event)
        await asyncio.sleep(0.1)

Feature extraction

import pandas as pd
import numpy as np
from dataclasses import dataclass

@dataclass
class TokenFeatures:
    token_address: str
    timestamp: float
    volume_anomaly_score: float
    new_wallet_ratio: float
    transaction_sync_score: float
    holder_hhi_delta: float
    liquidity_depth_change: float
    price_velocity: float  # скорость изменения цены

def compute_volume_anomaly(
    current_volume_usd: float,
    historical_volumes: list[float]
) -> float:
    if not historical_volumes:
        return 1.0
    rolling_avg = np.mean(historical_volumes[-30:])
    if rolling_avg == 0:
        return 1.0
    return current_volume_usd / rolling_avg

def compute_sync_score(
    transactions: pd.DataFrame,
    window_seconds: int = 300
) -> float:
    """Насколько синхронизированы независимые адреса в покупках"""
    tx_times = transactions['timestamp'].values
    unique_senders = transactions['from'].nunique()
    if unique_senders < 2:
        return 0.0

    # Гистограмма транзакций по временным окнам
    bins = np.arange(tx_times.min(), tx_times.max() + window_seconds, window_seconds)
    hist, _ = np.histogram(tx_times, bins=bins)

    # Нормализованная дисперсия: низкая дисперсия = высокая синхронизация
    if hist.mean() == 0:
        return 0.0
    cv = hist.std() / hist.mean()
    return max(0, 1 - cv / 2)  # преобразование в [0, 1]

ML модель

Для обнаружения P&D хорошо работают ансамблевые методы: XGBoost или LightGBM на tabular features. Они интерпретируемы (SHAP values для объяснения предсказания), быстро инференсируют, устойчивы к пропущенным данным.

import xgboost as xgb
from sklearn.model_selection import TimeSeriesSplit
import shap

# Разбивка по времени: нельзя использовать будущие данные для предсказания прошлого
tscv = TimeSeriesSplit(n_splits=5)

model = xgb.XGBClassifier(
    n_estimators=500,
    max_depth=6,
    learning_rate=0.01,
    subsample=0.8,
    colsample_bytree=0.8,
    scale_pos_weight=neg_count / pos_count,  # балансировка классов
    eval_metric='aucpr',  # PR-AUC важнее ROC-AUC при несбалансированных данных
    early_stopping_rounds=50
)

model.fit(
    X_train, y_train,
    eval_set=[(X_val, y_val)],
    verbose=100
)

# Объяснение предсказания через SHAP
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

Метрики оценки: precision-recall важнее accuracy из-за сильного class imbalance (P&D события редки). Цель: precision > 0.7 при recall > 0.6. False positives (ложные срабатывания) раздражают пользователей; false negatives (пропущенные P&D) — репутационный ущерб.

Обучающая выборка

Labeling исторических данных — самая трудоёмкая часть. Источники known P&D событий:

• CryptoManiac база данных pump-and-dump событий
• Ручная разметка через ретроспективный анализ: берём токены с аномальными price spikes + crash, верифицируем Telegram/Discord история
• Synthetic data augmentation: симуляция P&D паттернов на основе реальных характеристик

Минимальная обучающая выборка: 200-500 P&D событий + 5,000-10,000 non-P&D периодов. Без достаточной выборки — модель переобучается.

Реализация алертинга

Thresholds и confidence levels

Не бинарное «P&D / не P&D», а вероятность с порогами:

• > 0.8: высокая уверенность, немедленный алерт
• 0.6 - 0.8: средняя уверенность, предупреждение
• < 0.6: мониторинг, нет алерта

Интеграция с протоколом

Для протоколов, где важна защита: trading contract может читать оценку риска через oracle. Если риск высокий — повышенный slippage tolerance или временный pause конкретного торгового пула.

interface IPumpDetector {
    function getRiskScore(address token) external view returns (uint256); // 0-100
}

contract ProtectedDEX {
    IPumpDetector public detector;
    uint256 public constant HIGH_RISK_THRESHOLD = 75;

    function swap(address tokenIn, address tokenOut, uint256 amountIn) external {
        uint256 riskScore = detector.getRiskScore(tokenOut);
        if (riskScore >= HIGH_RISK_THRESHOLD) {
            // Требуем явного подтверждения или блокируем
            revert("High manipulation risk detected");
        }
        // ... swap logic
    }
}

Ограничения и оговорки

Система обнаружения не устраняет P&D — она предупреждает. Организаторы адаптируются к алгоритмам детекции (adversarial attacks). Качество модели деградирует со временем и требует переобучения на новых данных.

Юридическая сторона: автоматические блокировки на основе ML предсказаний несут правовые риски в зависимости от юрисдикции. Безопаснее — предупреждения пользователям, не автоматическое ограничение торговли.

Разработка системы: 8-14 недель. Сбор и разметка данных — 3-4 недели, модель — 2-3 недели, инфраструктура и алертинг — 3-5 недель, тестирование — 2 недели.