Разработка модели обнаружения инсайдерской торговли

Разработка модели детекции манипуляций ценой токенов

Price manipulation в DeFi — не абстрактная угроза. Flash loan атаки используют временное искажение цены токена для дренажа lending протоколов: займи миллиард, манипулируй ценой Oracle, возьми под залог несуществующей стоимости, верни займ, уйди с прибылью. Mango Markets (октябрь 2022) — $117M через манипуляцию ценой MNGO токена. CREAM Finance (октябрь 2021) — $130M через flash loan + oracle manipulation.

Модель детекции не предотвращает атаку, но позволяет: (1) остановить протокол до исполнения вредоносных транзакций, (2) изучать паттерны для улучшения Oracle защиты, (3) алертировать команду в реальном времени.

Типология манипуляций

Flash loan oracle manipulation

Классический вектор: AMM (Uniswap/Curve пул) используется как price oracle. Атакующий временно двигает цену в AMM через большой trade, использует искажённую цену в протоколе-жертве, возвращает AMM к норме.

Сигнатура атаки в on-chain данных:

• Flash loan транзакция (один block, один tx)
• Внутри tx: крупный swap → вызов протокола-жертвы → обратный swap
• Резкое отклонение spot price от TWAP в начале tx
• Возврат к нормальной цене в конце tx

Sandwich attack на oracle update

Менее известный вектор: атакующий знает что oracle обновляется при определённом условии, фронтраннит обновление, эксплуатирует период между старой и новой ценой.

Wash trading для price inflation

Серия координированных сделок между аффилированными кошельками создаёт искусственный объём и движение цены. Цель: поднять цену токена до исполнения крупной продажи или для манипуляции lending collateral value.

Сигнатура: высокая коррелированность между кошельками, чистые нулевые или близкие к нулю P&L циклы, необычно низкий slippage при высоком объёме.

Low-liquidity spot manipulation

Для токенов с низкой ликвидностью ($10K-$100K в пуле) небольшой trade ($50K-$200K) может двинуть цену на 50-200%. Если lending протокол принимает этот токен как collateral с spot price Oracle — exploit тривиален.

Статистические методы детекции

TWAP deviation detector

Самый простой и эффективный метод: сравнивать spot price с TWAP (Time-Weighted Average Price).

import numpy as np
from dataclasses import dataclass
from typing import List

@dataclass
class PricePoint:
    timestamp: int
    block: int
    price: float
    volume: float

def compute_twap(prices: List[PricePoint], window_seconds: int) -> float:
    """Compute time-weighted average price over window."""
    if not prices:
        return 0.0

    current_time = prices[-1].timestamp
    cutoff_time = current_time - window_seconds

    relevant = [p for p in prices if p.timestamp >= cutoff_time]
    if len(relevant) < 2:
        return prices[-1].price

    total_weighted = 0.0
    total_time = 0.0

    for i in range(1, len(relevant)):
        dt = relevant[i].timestamp - relevant[i-1].timestamp
        total_weighted += relevant[i-1].price * dt
        total_time += dt

    return total_weighted / total_time if total_time > 0 else relevant[-1].price

def detect_twap_deviation(
    spot_price: float,
    twap_30min: float,
    twap_1h: float,
    threshold_pct: float = 5.0
) -> dict:
    """
    Detect significant deviation between spot and TWAP prices.
    Returns risk assessment.
    """
    dev_30min = abs(spot_price - twap_30min) / twap_30min * 100
    dev_1h = abs(spot_price - twap_1h) / twap_1h * 100

    severity = 'normal'
    if dev_30min > threshold_pct * 3 or dev_1h > threshold_pct * 4:
        severity = 'critical'
    elif dev_30min > threshold_pct * 2 or dev_1h > threshold_pct * 2.5:
        severity = 'high'
    elif dev_30min > threshold_pct or dev_1h > threshold_pct * 1.5:
        severity = 'medium'

    return {
        'spot': spot_price,
        'twap_30min': twap_30min,
        'twap_1h': twap_1h,
        'deviation_30min_pct': dev_30min,
        'deviation_1h_pct': dev_1h,
        'severity': severity,
    }

Volume-price correlation anomaly detector

Нормальное движение цены сопровождается соответствующим объёмом. Резкое движение цены при аномально высоком объёме в одном направлении — сигнал манипуляции.

def detect_volume_price_anomaly(
    price_changes: List[float],    # % изменение цены за период
    volumes: List[float],           # объём торгов за период
    lookback: int = 100
) -> dict:
    """
    Детектирует аномальное соотношение объём/движение цены
    используя Z-score нормализацию.
    """
    if len(price_changes) < lookback:
        return {'anomaly': False, 'reason': 'insufficient data'}

    # Нормализуем по историческому распределению
    hist_prices = np.array(price_changes[-lookback:])
    hist_volumes = np.array(volumes[-lookback:])

    current_price_change = price_changes[-1]
    current_volume = volumes[-1]

    price_zscore = (current_price_change - hist_prices.mean()) / (hist_prices.std() + 1e-8)
    volume_zscore = (current_volume - hist_volumes.mean()) / (hist_volumes.std() + 1e-8)

    # Аномалия: высокий объём + большое движение в одну сторону
    is_anomaly = (
        abs(price_zscore) > 3.0 and     # цена > 3 std devs от нормы
        volume_zscore > 2.5 and          # объём > 2.5 std devs от нормы
        price_zscore * volume_zscore > 0 # оба в одном направлении
    )

    return {
        'anomaly': is_anomaly,
        'price_zscore': price_zscore,
        'volume_zscore': volume_zscore,
        'current_price_change_pct': current_price_change,
        'volume_vs_avg': current_volume / hist_volumes.mean(),
    }

Flash loan pattern detector (on-chain)

Flash loan атака имеет специфическую структуру транзакции. Детектируем через анализ call trace:

def analyze_transaction_for_flash_loan(tx_trace: dict) -> dict:
    """
    Анализирует call trace транзакции на признаки flash loan атаки.
    tx_trace — output от debug_traceTransaction или Tenderly API.
    """
    calls = flatten_calls(tx_trace)

    # Ищем flash loan паттерн: flashLoan → [вызовы] → flashLoan callback
    flash_loan_providers = {
        '0xba12222222228d8ba445958a75a0704d566bf2c8': 'Balancer',  # Balancer Vault
        '0x7d2768de32b0b80b7a3454c06bdac94a69ddc7a9': 'Aave V2',
        '0x87870bca3f3fd6335c3f4ce8392d69350b4fa4e2': 'Aave V3',
    }

    involved_flash_loans = []
    large_swaps = []
    target_protocol_calls = []

    for call in calls:
        if call['to'].lower() in flash_loan_providers:
            involved_flash_loans.append({
                'provider': flash_loan_providers[call['to'].lower()],
                'selector': call['input'][:10],
            })

        # Ищем крупные swap события
        if is_swap_call(call) and parse_swap_amount(call) > LARGE_SWAP_THRESHOLD:
            large_swaps.append(call)

    # Flash loan + крупный swap в одной транзакции = высокий риск
    risk_score = 0
    if involved_flash_loans:
        risk_score += 50
    if len(large_swaps) >= 2:  # swap туда и обратно
        risk_score += 30
    if has_target_protocol_interaction(calls):
        risk_score += 20

    return {
        'risk_score': risk_score,
        'is_high_risk': risk_score >= 80,
        'flash_loans': involved_flash_loans,
        'large_swaps': len(large_swaps),
        'recommendation': 'BLOCK' if risk_score >= 80 else 'MONITOR',
    }

Wash trading detector

def detect_wash_trading(
    trades: List[dict],       # {buyer, seller, amount, price, timestamp}
    lookback_hours: int = 24
) -> dict:
    """
    Детектирует wash trading паттерны.
    Признаки: один адрес одновременно buyer и seller,
    или кольцо взаимных сделок.
    """
    from collections import defaultdict

    # Строим граф торговых отношений
    trade_graph = defaultdict(lambda: defaultdict(float))

    cutoff = int(time.time()) - lookback_hours * 3600
    recent_trades = [t for t in trades if t['timestamp'] >= cutoff]

    for trade in recent_trades:
        trade_graph[trade['buyer']][trade['seller']] += trade['amount']
        trade_graph[trade['seller']][trade['buyer']] += trade['amount']

    wash_pairs = []
    addresses = list(trade_graph.keys())

    for i, addr_a in enumerate(addresses):
        for addr_b in addresses[i+1:]:
            flow_a_to_b = trade_graph[addr_a][addr_b]
            flow_b_to_a = trade_graph[addr_b][addr_a]

            if flow_a_to_b > 0 and flow_b_to_a > 0:
                symmetry_ratio = min(flow_a_to_b, flow_b_to_a) / max(flow_a_to_b, flow_b_to_a)

                # Высокая симметрия = подозрительно
                if symmetry_ratio > 0.8:
                    wash_pairs.append({
                        'address_a': addr_a,
                        'address_b': addr_b,
                        'flow_ab': flow_a_to_b,
                        'flow_ba': flow_b_to_a,
                        'symmetry': symmetry_ratio,
                    })

    return {
        'wash_pairs_found': len(wash_pairs),
        'suspicious_pairs': wash_pairs[:10],
        'is_suspicious': len(wash_pairs) > 0,
    }

On-chain Oracle защита

Модель детекции дополняется on-chain защитой в самом Oracle. Chainlink CCIP Price Feed имеет встроенные circuit breaker-ы, но кастомные Oracle (Uniswap V3 TWAP, Curve EMA) требуют ручной защиты:

contract ManipulationResistantOracle {
    IUniswapV3Pool public immutable pool;
    uint32 public constant TWAP_PERIOD = 1800;  // 30 минут
    uint256 public constant MAX_DEVIATION_BPS = 500; // 5%

    function getPrice() external view returns (uint256) {
        uint256 spotPrice = _getSpotPrice();
        uint256 twapPrice = _getTWAPPrice(TWAP_PERIOD);

        // Проверяем что spot не отклонился более чем на MAX_DEVIATION от TWAP
        uint256 deviation = spotPrice > twapPrice
            ? (spotPrice - twapPrice) * 10000 / twapPrice
            : (twapPrice - spotPrice) * 10000 / twapPrice;

        if (deviation > MAX_DEVIATION_BPS) {
            // В момент манипуляции — возвращаем TWAP, не spot
            return twapPrice;
        }

        return spotPrice;
    }

    function _getTWAPPrice(uint32 period) internal view returns (uint256) {
        uint32[] memory secondsAgos = new uint32[](2);
        secondsAgos[0] = period;
        secondsAgos[1] = 0;

        (int56[] memory tickCumulatives,) = pool.observe(secondsAgos);
        int56 tickDiff = tickCumulatives[1] - tickCumulatives[0];
        int24 avgTick = int24(tickDiff / int56(uint56(period)));

        return TickMath.getSqrtRatioAtTick(avgTick);
    }
}

Pipeline и инфраструктура

Блокчейн ноды (Alchemy/QuickNode)
        ↓ WebSocket streams
Event Processor (Node.js)
        ↓
Detection Models (Python FastAPI)
        ├── TWAP Deviation Checker
        ├── Volume Anomaly Detector
        ├── Flash Loan Analyzer
        └── Wash Trading Detector
        ↓
Risk Aggregator
        ├── Score < 40: log only
        ├── Score 40-70: alert команде
        └── Score > 70: auto-pause + alert
        ↓
Actions: Telegram/PagerDuty + Circuit Breaker

Latency критична: от появления подозрительной pending транзакции до выполнения pause должно пройти < 3 секунд.

Данные для обучения и тестирования

Исторические атаки — лучший источник ground truth. Публично документированные инциденты:

Каждая атака документирована в rekt.news и имеет on-chain данные (transaction hashes). Воспроизведение на Foundry fork: forge test --fork-url $ETH_RPC --fork-block-number [pre-attack block].

Процесс работы

Анализ угроз (1 неделя). Карта возможных векторов манипуляции для конкретного протокола. Какие Oracle используются, какова их ликвидность, какие функции зависят от цены.

Разработка detection моделей (2-3 недели). Python ML сервис с TWAP deviation + volume anomaly + flash loan pattern detectors. Backtest на исторических данных.

On-chain Oracle защита (1 неделя). Manipulation-resistant oracle wrapper, если текущий Oracle уязвим.

Интеграция с circuit breaker (1 неделя). Detection → auto-pause pipeline.

Аудит и тестирование (1 неделя). Воспроизведение известных атак на fork, проверка детекции.

Мониторинговая инфраструктура (1-2 недели). Event streaming, alerting, dashboards.

Полный цикл: 2-2,5 месяца. Стоимость — после оценки scope и поддерживаемых цепей.