מודל ML לזיהוי מסחר פיקטיבי (Wash Trading) בבלוקצ'יין

Chainalysis מדווחת שבכמה שווקי NFT, חלקו של הנפח המזויף עולה על 50%, ובחלקם עד 80%. מסחר פיקטיבי מעוות את נתוני השוק, מטעה משקיעים ומושך תשומת לב רגולטורית. שיטות סף מסורתיות (למשל, זיהוי עסקאות חוזרות בין אותן כתובות) מפספסות עד 60% מהמניפולציות. מודל המבוסס על ניתוח גרף בלוקצ'יין ו-Gradient Boosting עם פירוש SHAP מעלה את הדיוק ל-95% (ROC-AUC >0.95). אנו מפתחים מערכות כאלה במפתח מלא—מבניית גרף העסקאות ועד לפריסת ה-API והכשרת הצוות שלך. במהלך השנים, סיפקנו 20+ פרויקטים של ניתוח on-chain עבור פרוטוקולי DeFi, שווקי NFT ובורסות בלוקצ'יין.

אילו סוגי מסחר פיקטיבי קיימים ב-Web3?

הבנת המגוון קובעת את בחירת מאפייני המודל:

• מסחר עצמי: אותו ארנק קונה ומוכר לעצמו, או דרך שרשרת של כתובות קשורות.
• מסחר מעגלי: A מוכר ל-B, B מוכר ל-C, C מוכר ל-A. הנכס חוזר לבעלים המקורי.
• מסחר פיקטיבי מרובד: שרשראות מורכבות דרך 5–10 כתובות כדי להסתיר קישורים. משמש לניפוח NFTs לפני מכירה לקונים אמיתיים במחירים מנופחים.
• חקלאות איירדרופ: מסחר פיקטיבי לצבירת נפח מסחר עבור איירדרופ עתידי. זה היה נפוץ ב-Blur.
• ניצול לרעה של החזרי עמלות: קבלת החזרים מהבורסה דרך נפח מלאכותי.

בניית מודל ML למסחר פיקטיבי: תוכנית שלב-אחר-שלב

1. איסוף נתוני on-chain: דרך The Graph, Dune Analytics, או אינדקסר מותאם אישית. לניטור בזמן אמת, אנו משתמשים ב-WebSocket RPC (Infura, Alchemy). הנתונים כוללים: hash, שולח, מקבל, כמות, חותמת זמן, token_id.
2. בניית גרף העסקאות: באמצעות NetworkX, צור גרף מכוון ומשוקלל. משקל הקשת הוא הנפח המצטבר. חפש מעגלים באורך של עד 6 צמתים—סימן פשוט למסחר פיקטיבי.
3. קיבוץ כתובות קשורות: קבץ כתובות עם מקור מימון משותף ופעילות סינכרונית (מתאם >0.85). השתמש ב-Union-Find.
4. חילוץ מאפיינים: זמניים (סדירות, פעילות לילית), כלכליים (PNL, ריכוזיות עמיתים), ספציפיים ל-NFT (תדירות שינוי בעלות).
5. אימון Gradient Boosting: 200 עצים, max_depth=5, learning_rate=0.05. בצע אופטימיזציה ל-ROC-AUC עם חוסר איזון מחלקות (משקלי מחלקות).
6. פירוש עם SHAP: עבור כל חיזוי, קבל את 5 המאפיינים המובילים ואת תרומתם. האנליסט רואה מדוע כתובת מסומנת כחשודה.
7. פריסת API: נקודת קצה FastAPI 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 מחזירה הסתברות, רמת סיכון וגורמים תורמים.

מדוע ניתוח גרף הוא הכלי העיקרי?

ניתוח גרף מאפשר להמחיש זרימות כספים ולזהות דפוסים מחזוריים שאינם נראים בניתוח עסקאות בודדות. אנו בונים גרף מכוון שבו הקשתות משוקללות לפי נפח ומיישמים אלגוריתמים לזיהוי מעגלים וקיבוץ כתובות. זה מניב תוצאות ניתנות לפירוש ומזין את מודל ה-ML. ניתוח גרף עם NetworkX מעבד עד 100 אלף צמתים בשנייה—פי שניים מהר יותר מניתוח ידני.

בניית גרף העסקאות וזיהוי מעגלים

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

קיבוץ כתובות קשורות

כתובות השייכות לאותו אשכול (נשלטות על ידי ישות אחת) מזוהות דרך:

• מקור מימון זהה (קיבלו ETH מכתובת אחת)
• דפוסי פעילות מסונכרנים בזמן
• אסטרטגיות מחיר גז משותפות

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: מה מבדיל סוחר פיקטיבי

בנוסף לניתוח גרף, אנו בונים וקטור מאפיינים עבור כל זוג מסחר או כתובת. המאפיינים מתחלקים לשלוש קבוצות: זמניים, כלכליים וספציפיים ל-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 מספק יכולת פירוש. בניגוד לרשתות נוירונים, אנו מסבירים כל חיזוי: אילו מאפיינים וכיצד הם תרמו. זה קריטי לציות וקבלת החלטות. השוואה עם כללים: ספים ידניים מזהים רק 40% מהמסחר הפיקטיבי; המודל שלנו משיג 95% (ROC-AUC >0.95).

הערכת ביטחון ופירוש

המודל אינו מפיק תוצאה בינארית, אלא ציון עם הסבר. זה מאפשר לאנליסט לקבל החלטות מושכלות.

@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=[])

השוואת מקורות נתונים

למודל ייצור על DEX, אינדקסר מותאם אישית דרך WebSocket RPC מספק את זמן ההשהיה הנמוך ביותר ושליטה מלאה בנתונים. Dune Analytics טובה לפיתוח אך איטית מדי לניטור בזמן אמת.

פירוש ערכי SHAP

SHAP מציג את התרומה של כל מאפיין להסתברות הסופית. לדוגמה, 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] גבוה (>0.3) ו-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 נמוך (<5) מעידים לעתים קרובות על מסחר פיקטיבי. אנו מספקים לוח מחוונים עם גרפי SHAP עבור כל כתובת—האנליסט רואה מדוע המודל הגיע לפסק דינו.

מה כלול בפיתוח מודל במפתח מלא

• ניתוח דרישות ובחירת מקורות נתונים.
• פיתוח צינור איסוף ועיבוד נתוני on-chain.
• בניית מודל גרף וקיבוץ כתובות.
• פיתוח ואימון מודל ML (Gradient Boosting) עם כיול.
• שילוב SHAP לפירוש חיזויים.
• פריסת API להערכת כתובות.
• תיעוד ארכיטקטורה ומדריך משתמש.
• הכשרת צוות ומסירת קוד מקור.

עלות הפיתוח משתנה בהתאם למורכבות האינטגרציה ומספר הרשתות. החיסכון מזיהוי מניפולציות יכול להגיע ל-$300,000 בשנה על ידי מניעת הפסדים ממסחר פיקטיבי. הזמינו פיתוח מודל במפתח מלא—נריץ פיילוט על הנתונים שלכם תוך 2 ימי עסקים. קבלו ייעוץ: השאירו בקשה באתר. צרו קשר כדי לדון במקרה שלכם. אנו מבטיחים שקיפות ותמיכה לאחר הפריסה.