Розробка моделі виявлення інсайдерської торгівлі

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

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

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

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

Flash loan oracle маніпуляція

Класичний вектор: AMM (Uniswap/Curve пул) використовується як price oracle. Атакуючий тимчасово рухає ціну в AMM через великий swap, експлуатує викривлену ціну в протоколі-жертві, повертає AMM до норми.

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

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

Sandwich атака на oracle оновлення

Менш відома: атакуючий знає, коли oracle оновлюється при певній умові, фронтрани оновлення, експлуатує період між старою та новою ціною.

Wash trading для інфляції ціни

Серія скоординованих транзакцій між афільованими гаманцями створює штучний обсяг та рух ціни. Мета: підняти ціну токена перед великим продажем або маніпулювати вартістю collateral у lending.

Сигнатура: висока кореляція гаманців, нульові або близькі до нуля P&L цикли, необично низькі slippage при високому обсязі.

Маніпуляція spot з низькою ліквідністю

Для токенів з низькою ліквідністю ($10K-$100K в пулі) малий swap ($50K-$200K) може рухнути ціну на 50-200%. Якщо lending протокол приймає цей токен як collateral з spot price Oracle — експлойт тривіальний.

Статистичні методи детекції

TWAP deviation детектор

Найпростіший та найефективніший метод: порівняти 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:
    """Обчислити time-weighted середню ціну за вікно."""
    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:
    """
    Виявити значне відхилення між spot та TWAP цінами.
    Повертає оцінку ризику.
    """
    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 кореляції

Нормальний рух ціни супроводжується відповідним обсягом. Різкий рух ціни з аномально високим одностороннім обсягом — сигнал маніпуляції.

def detect_volume_price_anomaly(
    price_changes: List[float],    # % зміна ціни за період
    volumes: List[float],           # обсяг торгів за період
    lookback: int = 100
) -> dict:
    """
    Виявити аномальне співвідношення volume/price рух
    використовуючи 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 детектор (on-chain)

Flash loan атака має специфічну структуру транзакції. Виявляємо через аналіз call trace:

def analyze_transaction_for_flash_loan(tx_trace: dict) -> dict:
    """
    Аналізує call trace транзакції на ознаки flash loan атаки.
    tx_trace — вихід з 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

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 та інфраструктура

Blockchain ноди (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 детекторами. Backtest на історичних даних.

On-chain Oracle захист (1 тиждень). Manipulation-resistant oracle обгортка, якщо поточний Oracle уразливий.

Circuit breaker інтеграція (1 тиждень). Detection → auto-pause pipeline.

Аудит та тестування (1 тиждень). Відтворення відомих атак на fork, перевірка детекції.

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

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