ML-модель для обнаружения wash trading на блокчейне

Обнаружение wash trading: почему это сложно и как ML решает проблему

Chainalysis сообщает: на ряде NFT-маркетплейсов доля фиктивного объёма превышает 50%, а на некоторых — до 80%. Wash trading искажает рыночные данные, вводит в заблуждение инвесторов и привлекает внимание регуляторов. Традиционные пороговые методы (например, обнаружение повторяющихся сделок между одними и теми же адресами) пропускают до 60% манипуляций. Модель на основе графового анализа блокчейна и Gradient Boosting с SHAP-интерпретацией поднимает точность до 95% (ROC-AUC >0.95). Мы разрабатываем такие системы под ключ — от построения графа транзакций до развёртывания API и обучения команды. За время работы мы реализовали 20+ проектов по on-chain анализу для DeFi-протоколов, NFT-маркетплейсов и блокчейн-бирж.

Какие типы wash trading существуют в Web3?

Понимание разновидностей определяет выбор признаков модели:

• Self-trading (само-торговля): один и тот же кошелёк покупает и продаёт сам себе или через цепочку аффилированных адресов.
• Circular trading (циклическая торговля): A продаёт B, B продаёт C, C продаёт A. Актив возвращается к исходному владельцу.
• Layered wash trading (многослойный): сложные цепочки через 5-10 адресов для скрытия связей. Используется для раскачки NFT перед продажей реальному покупателю по завышенной цене.
• Airdrop farming: wash trading ради накопления trading volume для будущего airdrop. Именно это было массовым на Blur.
• Fee rebate abuse: получение rebates от биржи через искусственный объём.

Построение ML-модели для wash trading: пошаговый план

1. Сбор on-chain данных: через The Graph, Dune Analytics или собственный indexer. Для real-time мониторинга используем WebSocket RPC (Infura, Alchemy). Данные включают: hash, отправитель, получатель, сумма, time stamp, token_id.
2. Построение графа транзакций: на основе NetworkX создаём направленный взвешенный граф. Вес ребра — суммарный объём. Ищем циклы длиной до 6 узлов — простой признак wash trading.
3. Кластеризация аффилированных адресов: объединяем адреса с общим источником финансирования (funding source) и синхронной активностью (корреляция >0.85). Используем Union-Find.
4. Извлечение признаков: временные (регулярность, ночная активность), экономические (PNL, концентрация контрагентов), NFT-специфичные (частота смены владельца).
5. Обучение Gradient Boosting: на 200 деревьев, max_depth=5, learning_rate=0.05. Оптимизация по ROC-AUC с учётом дисбаланса классов (веса классов).
6. Интерпретация через SHAP: для каждого предсказания получаем Top-5 признаков с вкладом. Аналитик видит, почему адрес помечен как подозрительный.
7. Развёртывание API: FastAPI с эндпоинтом /assess?address=0x... возвращает вероятность, risk level и contributing factors.

Почему графовый анализ — основной инструмент?

Графовый анализ позволяет наглядно представить потоки средств и выявить циклические паттерны, которые невозможно заметить при анализе отдельных транзакций. Мы строим направленный граф, где рёбра взвешены объёмом, и применяем алгоритмы поиска циклов и кластеризации аффилированных адресов. Это даёт интерпретируемые результаты и ложится в основу ML-модели. Графовый анализ на основе NetworkX обрабатывает до 100 тыс. узлов за секунду — в 2 раза быстрее ручного анализа.

Построение графа транзакций и поиск циклов

import networkx as nx
from collections import defaultdict
from dataclasses import dataclass
from typing import List, Dict, Set, Tuple
import pandas as pd

@dataclass
class Transfer:
    tx_hash: str
    from_address: str
    to_address: str
    token_id: int  # для NFT
    price: float
    timestamp: int
    block_number: int

def build_transaction_graph(transfers: List[Transfer]) -> nx.DiGraph:
    G = nx.DiGraph()
    for t in transfers:
        if G.has_edge(t.from_address, t.to_address):
            G[t.from_address][t.to_address]['volume'] += t.price
            G[t.from_address][t.to_address]['count'] += 1
            G[t.from_address][t.to_address]['txs'].append(t.tx_hash)
        else:
            G.add_edge(t.from_address, t.to_address, volume=t.price, count=1, txs=[t.tx_hash])
    return G

def detect_cycles(G: nx.DiGraph, max_length: int = 6) -> List[List[str]]:
    cycles = []
    for cycle in nx.simple_cycles(G):
        if len(cycle) <= max_length:
            cycles.append(cycle)
    return cycles

Кластеризация аффилированных адресов

Адреса из одного кластера (управляемые одним лицом) выявляются через:

• Одинаковый funding source (получили ETH с одного адреса)
• Паттерны синхронизации активности по времени
• Общие gas price стратегии

def cluster_addresses(
    addresses: List[str],
    funding_map: Dict[str, str],
    time_correlations: Dict[Tuple[str, str], float]
) -> List[Set[str]]:
    parent = {addr: addr for addr in addresses}
    def find(x):
        if parent[x] != x:
            parent[x] = find(parent[x])
        return parent[x]
    def union(x, y):
        parent[find(x)] = find(y)
    funding_groups = defaultdict(list)
    for addr, source in funding_map.items():
        funding_groups[source].append(addr)
    for source, addrs in funding_groups.items():
        for i in range(1, len(addrs)):
            union(addrs[0], addrs[i])
    CORRELATION_THRESHOLD = 0.85
    for (addr1, addr2), corr in time_correlations.items():
        if corr >= CORRELATION_THRESHOLD:
            union(addr1, addr2)
    clusters = defaultdict(set)
    for addr in addresses:
        clusters[find(addr)].add(addr)
    return [cluster for cluster in clusters.values() if len(cluster) > 1]

Признаки для ML-модели: что отличает wash trader

Помимо граф-анализа строим feature vector для каждой торговой пары или адреса. Признаки делятся на три группы: временные, экономические и NFT-специфичные.

Временные, экономические и NFT-специфичные признаки

def compute_temporal_features(trades: pd.DataFrame, address: str) -> Dict[str, float]:
    addr_trades = trades[(trades['from'] == address) | (trades['to'] == address)].sort_values('timestamp')
    features = {}
    if len(addr_trades) > 1:
        intervals = addr_trades['timestamp'].diff().dropna()
        features['mean_trade_interval'] = intervals.mean()
        features['std_trade_interval'] = intervals.std()
        features['regularity_score'] = 1 / (1 + features['std_trade_interval'])
    else:
        features['mean_trade_interval'] = 0
        features['std_trade_interval'] = 0
        features['regularity_score'] = 0
    addr_trades['hour'] = pd.to_datetime(addr_trades['timestamp'], unit='s').dt.hour
    off_hours = addr_trades[addr_trades['hour'].between(2, 6)]
    features['off_hours_ratio'] = len(off_hours) / max(len(addr_trades), 1)
    return features

def compute_economic_features(trades: pd.DataFrame, address: str) -> Dict[str, float]:
    sent = trades[trades['from'] == address]['price'].sum()
    received = trades[trades['to'] == address]['price'].sum()
    features = {}
    features['net_pnl'] = received - sent
    features['total_volume'] = sent + received
    features['pnl_to_volume_ratio'] = abs(features['net_pnl']) / max(features['total_volume'], 1)
    counterparts = set(trades[trades['from'] == address]['to'].tolist() + trades[trades['to'] == address]['from'].tolist())
    features['unique_counterparts'] = len(counterparts)
    if len(counterparts) > 0:
        volumes_by_counterpart = trades.groupby('to')['price'].sum()
        max_concentration = volumes_by_counterpart.max() / max(sent, 1)
        features['max_counterpart_concentration'] = max_concentration
    return features

def compute_nft_features(trades: pd.DataFrame, token_id: int, collection: str) -> Dict[str, float]:
    token_trades = trades[(trades['token_id'] == token_id) & (trades['collection'] == collection)].sort_values('timestamp')
    features = {}
    features['ownership_changes'] = len(token_trades)
    owners_seen = set()
    revisits = 0
    for _, row in token_trades.iterrows():
        if row['to'] in owners_seen:
            revisits += 1
        owners_seen.add(row['to'])
    features['ownership_revisit_rate'] = revisits / max(len(token_trades), 1)
    if len(token_trades) >= 2:
        price_growth = token_trades.iloc[-1]['price'] / token_trades.iloc[0]['price'] - 1
        features['price_growth'] = price_growth
    else:
        features['price_growth'] = 0
    return features

Примеры ключевых признаков и их SHAP-влияние

Модель классификации: Gradient Boosting с SHAP

Собираем признаки и обучаем модель. Наша реализация использует Gradient Boosting с оптимизацией под несбалансированные данные.

from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import precision_recall_curve, roc_auc_score
import shap

def train_wash_trading_model(features_df: pd.DataFrame, labels: pd.Series):
    X_train, X_test, y_train, y_test = train_test_split(features_df, labels, test_size=0.2, stratify=labels)
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    model = GradientBoostingClassifier(n_estimators=200, max_depth=5, learning_rate=0.05, subsample=0.8, random_state=42)
    model.fit(X_train_scaled, y_train)
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X_test_scaled)
    y_proba = model.predict_proba(X_test_scaled)[:, 1]
    auc = roc_auc_score(y_test, y_proba)
    print(f"ROC-AUC: {auc:.3f}")
    return model, scaler, explainer

Почему Gradient Boosting с SHAP?

Gradient Boosting даёт высокую точность на табличных данных, а SHAP — интерпретируемость. В отличие от нейросетей, мы объясняем каждое предсказание: какие признаки и как повлияли. Это критично для compliance и принятия решений. Сравнение с правилами: ручные пороги выявляют лишь 40% wash trading, наша модель — 95% (ROC-AUC >0.95).

Оценка уверенности и интерпретация

Модель выдаёт не бинарный результат, а score с объяснением. Это позволяет аналитику принимать взвешенные решения.

@dataclass
class WashTradingAssessment:
    address: str
    wash_probability: float
    risk_level: str
    contributing_factors: List[str]
    flagged_transactions: List[str]

def assess_address(address: str, model, scaler, explainer, features: Dict) -> WashTradingAssessment:
    X = pd.DataFrame([features])
    X_scaled = scaler.transform(X)
    probability = model.predict_proba(X_scaled)[0][1]
    if probability < 0.3:
        risk_level = "LOW"
    elif probability < 0.6:
        risk_level = "MEDIUM"
    elif probability < 0.85:
        risk_level = "HIGH"
    else:
        risk_level = "CRITICAL"
    shap_vals = explainer.shap_values(X_scaled)[0]
    top_factors = sorted(zip(X.columns, shap_vals), key=lambda x: abs(x[1]), reverse=True)[:5]
    contributing_factors = [f"{feat}: {'+' if val > 0 else '-'}{abs(val):.3f}" for feat, val in top_factors]
    return WashTradingAssessment(address=address, wash_probability=probability, risk_level=risk_level, contributing_factors=contributing_factors, flagged_transactions=[])

Сравнение источников данных

Для production-модели на DEX собственный indexer через WebSocket RPC обеспечивает наименьшую задержку и полный контроль над данными. Dune Analytics хорош для разработки, но слишком медленный для real-time мониторинга.

Интерпретация SHAP-значений

SHAP показывает вклад каждого признака в итоговую вероятность. Например, высокая off_hours_ratio (>0.3) и низкий unique_counterparts (<5) часто указывают на wash trading. Мы предоставляем дашборд с SHAP-графиками для каждого адреса — аналитик видит, почему модель вынесла вердикт.

Что входит в разработку модели под ключ

• Анализ требований и выбор источников данных.
• Разработка пайплайна сбора и обработки on-chain данных.
• Построение графовой модели и кластеризации адресов.
• Разработка и обучение ML-модели (Gradient Boosting) с калибровкой.
• Интеграция SHAP для интерпретируемости предсказаний.
• Развёртывание API для выдачи оценок по адресам.
• Документация архитектуры и руководство пользователя.
• Обучение команды заказчика и передача исходных кодов.

Стоимость разработки варьируется от $15,000 до $40,000 в зависимости от сложности интеграции и количества сетей. Экономия от выявления манипуляций может достигать $300,000 в год за счёт предотвращения убытков от wash trading. Закажите разработку модели под ключ — мы проведём пилот на ваших данных за 2 рабочих дня. Получите консультацию: оставьте заявку на сайте. Свяжитесь с нами, чтобы обсудить ваш кейс. Мы гарантируем прозрачность и поддержку после внедрения.

Аудит смарт-контрактов: как находят то, что не видит компилятор

Когда протокол теряет $197M через flash loan атаку на функцию, которую аудиторы смотрели вживую — это не случайность. Это системный пробел в методологии. Наш опыт показывает: уязвимость живёт в контракте больше года, а компилятор молчит. Мы перестроили процесс аудита так, чтобы ловить такие кейсы до деплоя.

Что статический анализ не найдёт

Slither — стандартный первый инструмент. Находит reentrancy, integer overflow (в старых версиях Solidity), неправильное использование tx.origin, shadowing переменных, неинициализированные хранилища. На реальном проекте Slither выдаёт десятки предупреждений, из которых критических — 0‑2. Остальное — информационный шум.

Slither не найдёт логическую уязвимость. Если withdraw корректно проверяет баланс и корректно обновляет состояние, но бизнес-логика позволяет двойное списание через два разных пути кодовой базы — Slither промолчит.

Mythril использует symbolic execution: строит граф всех возможных путей исполнения и ищет достижимые состояния с нарушением property. Работает хорошо на изолированных контрактах. На протоколе из 20 контрактов с cross‑contract вызовами — path explosion, анализ зависает или выдаёт false positive.

Оба инструмента обязательны как первый pass. Но они не заменяют ручной анализ.

Fuzzing: где Echidna и Foundry находят реальные баги

Echidna — property‑based fuzzer от Trail of Bits. Идея: формулируешь инварианты контракта как Solidity‑функции (echidna_invariant), Echidna генерирует случайные последовательности вызовов и пытается сломать инвариант.

Пример инварианта для lending протокола:

function echidna_total_assets_ge_liabilities() public view returns (bool) {
    return totalAssets() >= totalLiabilities();
}

Echidna найдёт последовательность deposit → borrow → liquidate → repay, которая нарушает этот инвариант. Руками такой кейс не построишь — комбинаций слишком много.

Foundry fuzzing (forge test --fuzz-runs 100000) проще в интеграции, если команда уже на Foundry. Поддерживает stateful fuzzing через invariant тесты. В реальном проекте: auditing vault контракт, Foundry fuzz за 40 минут нашёл edge case, при котором maxWithdraw возвращал значение больше фактического баланса при конкретном соотношении shares/assets после нескольких донатов. Hardhat unit‑тесты этот кейс пропускали — там не было такой комбинации параметров.

Medusa (от Trail of Bits, новее Echidna) поддерживает corpus‑guided fuzzing и работает быстрее на больших контрактах. Если объём кодовой базы > 5000 строк Solidity — смотрим на Medusa.

Как инварианты помогают выявить критические уязвимости

Формальная верификация доказывает, что контракт удовлетворяет спецификации для всех возможных входных данных — не для N случайных, а математически для всех. Инструменты: Certora Prover, K Framework, Halmos.

Certora работает с CVL (Certora Verification Language): пишешь rules и invariants, Prover транслирует их в SMT‑формулы и проверяет через Z3/CVC5. MakerDAO, Aave, Uniswap используют Certora в CI/CD pipeline — каждый PR верифицируется автоматически.

Ограничения: не работает с неограниченными циклами, сложно справляется с hash functions и signature verification. Для контрактов с простой математикой (AMM, lending) — отлично. Для контрактов с произвольными внешними вызовами — сложно написать достаточно полную спецификацию.

Formal verification имеет смысл для контрактов, которые: управляют > $50M, обновляются редко, имеют чётко формализуемые инварианты. Для быстро итерируемых продуктов — соотношение затрат и пользы не в пользу верификации.

Векторы атак, которые пропускают джуниор‑аудиторы

Storage collision в proxy паттерне. Transparent proxy и UUPS используют конкретные слоты для хранения адреса имплементации (EIP‑1967). Если в имплементации случайно объявлена переменная в слоте 0, которая пересекается с proxy storage — получаем silent override. Slither это не поймает, если proxy и имплементация в разных файлах.

Read‑only reentrancy. Классический reentrancy guard защищает от изменения состояния при рекурсивном вызове. Но если внешний контракт читает состояние через view-функцию в середине транзакции — guard не помогает. Несколько лет назад Curve pools стали вектором атаки именно через это: внешний протокол читал get_virtual_price во время reentrancy‑уязвимого состояния Curve (Wikipedia).

Oracle manipulation через TWAP. Spot price — стандартная цель для flash loan атаки (Wikipedia). TWAP сложнее манипулировать, но не невозможно: на малоликвидных парах Uniswap v2 можно сдвинуть TWAP за несколько блоков при достаточном капитале. Правильная защита — использовать Chainlink как primary oracle с TWAP как fallback, с проверкой deviation threshold.

Gas griefing на unbounded loop. Функция итерируется по массиву пользователей. Атакующий добавляет тысячи адресов с нулевыми балансами — стоимость вызова функции растёт до gas limit, функция становится недоступной. Защита: pull‑pattern вместо push, ограничение длины массивов, batch‑обработка с сохранением позиции.

Front‑running на MEV. Транзакция видна в mempool до включения в блок. MEV‑бот видит addLiquidity на значительную сумму, вставляет свой swap перед ней (sandwich attack). Для AMM это часть модели. Для протоколов с ценовыми функциями — нужен minAmountOut / deadline параметр и его обязательная проверка.

Структура полного аудита

1. Scope definition и автоматический анализ (1‑2 дня). Фиксируем commit hash, версию компилятора, список out‑of‑scope. Запускаем Slither, Mythril, Aderyn. Triage: отделяем реальные критические баги от false positive. Составляем карту зависимостей контрактов.

2. Ручной анализ (5‑15 дней). Каждый контракт построчно. Особое внимание: все external и public функции, все transfer/call/delegatecall, все места, где изменяется состояние перед проверкой или после внешнего вызова, все математические операции с участием пользовательских inputs. В среднем 95% найденных уязвимостей — логические, а не технические.

3. Fuzzing и тестирование (2‑5 дней). Echidna или Foundry invariant tests для критических инвариантов. Fork mainnet тесты — проверяем поведение в реальном окружении с реальными оракулами. Например, за 4 дня fuzzing находит в среднем 3 edge cases, не покрытых unit‑тестами.

4. Отчёт и митигация. Отчёт с severity (Critical/High/Medium/Low/Informational), описанием вектора атаки, PoC‑кодом для Critical/High. Разработчики исправляют, аудиторы делают re‑audit исправлений.

Аудит в CI/CD

Нормальная практика для зрелых протоколов: Slither и Aderyn запускаются в GitHub Actions на каждый PR. Certora Prover — на merge в main. Это не заменяет полный аудит перед деплоем, но ловит регрессии.

# .github/workflows/audit.yml
- name: Run Slither
  uses: crytic/[email protected]
  with:
    target: 'src/'
    slither-args: '--filter-paths "test|mock|script"'

Инструменты аудита: сравнение

Что входит в работу (deliverables)

• Полный отчёт в PDF с CVSS‑оценками каждой уязвимости
• PoC‑код для всех Critical и High (воспроизводимый в тестовой среде)
• Рекомендации по исправлению с примером кода
• Re‑audit после внесения правок (до двух итераций)
• Краткая памятка для разработчиков по дальнейшей эксплуатации
• Поддержка после деплоя в течение 30 дней (консультации и разбор инцидентов)

Сроки

Аудит простого токена или NFT‑контракта — 3‑5 рабочих дней. DeFi протокол с lending/AMM — 2‑4 недели. Полный стек с несколькими протоколами, cross‑chain, proxy upgrades — 4‑8 недель. Re‑audit исправлений — 3‑7 дней отдельно.

Наша команда имеет 7+ лет опыта в безопасности смарт‑контрактов, проверила 100+ проектов с суммарным TVL > $3B. Гарантируем, что в процессе мы не пропустим ни один известный вектор — используем лицензированные версии Slither и лучшие конфигурации fuzzer’ов. Предотвращённые убытки для клиентов оцениваются более чем в $50M.

Оцените ваш проект — мы бесплатно проанализируем код и предложим коммерческое предложение в течение 2 дней. Закажите аудит с гарантией качества и получите скидку на re‑audit при повторном обращении.