Розробка моделі виявлення wash trading

Розробка моделі виявлення wash trading

Wash trading — штучне створення обсягу торгів через одночасне купівлю та продаж одного активу однією особою або групою координованих осіб. За даними Chainalysis, у 2023 році понад 50% обсягу торгів на кількох NFT маркетплейсах було wash trading. На централізованих біржах масштаб ще більший. Модель виявлення повинна працювати з on-chain даними, виявляти закономірності та надавати інтерпретовані результати.

Типи wash trading у Web3

Розуміння різновидів визначає вибір ознак моделі:

Self-trading: той самий гаманець купує і продає сам собі або через ланцюг афільованих адрес.

Circular trading: A продає B, B продає C, C продає A. Актив повертається до первісного власника.

Layered wash trading: складні ланцюги через 5-10 адрес для приховування зв'язків. Використовується для розкачування NFT перед продажем справжньому покупцю за завищеною ціною.

Airdrop farming: wash trading для накопичення обсягу торгів для майбутніх airdrop. Це було масовим на Blur у 2023 році.

Fee rebate abuse: отримання повернень від біржі через штучний обсяг.

Граф-аналіз: основний інструмент

Ключовий метод виявлення — побудова графа транзакцій та пошук циклів і кластерів:

Побудова графа транзакцій

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]]:
    """
    Знаходимо цикли в графі — ознака wash trading.
    max_length обмежує глибину пошуку для продуктивності.
    """
    cycles = []

    # simple_cycles з NetworkX — алгоритм Джонсона, O((n+e)(c+1))
    for cycle in nx.simple_cycles(G):
        if len(cycle) <= max_length:
            cycles.append(cycle)

    return cycles

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

Адреси з одного кластера (керовані однією особою) можна виявити через аналіз:

• Однаковий джерело фінансування (отримали ETH з однієї адреси)
• Закономірності синхронізації активності в часі
• Спільні стратегії gas price

def cluster_addresses(
    addresses: List[str],
    funding_map: Dict[str, str],  # address -> funding source
    time_correlations: Dict[Tuple[str, str], float]
) -> List[Set[str]]:
    """
    Union-Find для об'єднання афільованих адрес у кластери.
    """
    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 моделі

Крім граф-аналізу будуємо feature vector для кожної торгової пари або адреси:

Часові ознаки

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()
        # Малий 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

    # Частка торгів у позаробочий час (2-6 AM UTC)
    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 = {}

    # Чистий P&L: wash trader зазвичай має P&L близько 0 (тільки газ)
    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

NFT-специфічні ознаки

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

Модель класифікації

Збираємо ознаки та навчаємо 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  # 1 = wash trading, 0 = legitimate
):
    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)

    # SHAP для інтерпретовності
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X_test_scaled)

    # Precision-Recall важливіша за ROC для незбалансованих даних
    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

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

Модель видає не бінарний результат, а score з поясненням:

@dataclass
class WashTradingAssessment:
    address: str
    wash_probability: float    # 0.0 - 1.0
    risk_level: str           # LOW / MEDIUM / HIGH / CRITICAL
    contributing_factors: List[str]  # SHAP-based пояснення
    flagged_transactions: List[str]  # конкретні підозрілі tx

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

Модель не є абсолютним детектором. Високе значення score — сигнал для ручного аналізу, не автоматичної заборони. Особливо важливо це для NFT маркетплейсів, де market-maker-и та автоматизовані трейдери можуть давати хибні спрацьовування.