链上数据 + AI 分析：Web3 链上行为的异常检测方案
一、钱包被盗 30 分钟后才发现的痛点
一个 DeFi 协议的用户钱包在凌晨被 drain，攻击者通过闪电贷操控价格预言机，3 笔交易卷走了 200 万美元。安全团队在 30 分钟后才发现——不是没有监控，而是传统的阈值告警（"转账金额 > 100 ETH"）在正常的大额交互中被反复触发，告警疲劳导致真正的事故被忽略。
链上数据的特点：公开透明、不可篡改、实时可查。但痛点也很明显——数据量大、行为模式复杂、攻击手段不断演化。传统的规则引擎跟不上，而 AI 恰好擅长从大量数据中发现异常模式。
二、链上异常检测的系统架构
异常检测方案的核心流程：链上数据采集 → 特征工程 → AI 模型推理 → 风险评分 → 分级告警。
flowchart TD subgraph Data[数据采集层] RPC[以太坊 RPC 节点] -->|WebSocket 推送| Stream[事件流处理] Stream --> Block[区块解析] Stream --> TX[交易解析] Stream --> Event[合约事件解析] end subgraph Feature[特征工程层] Block --> F1[地址历史行为特征] TX --> F2[交易图谱特征] Event --> F3[协议交互时序特征] F1 --> FV[特征向量] F2 --> FV F3 --> FV end subgraph AI[AI 分析层] FV --> EMB[Embedding 编码] EMB --> GNN[图神经网络: 地址关联分析] EMB --> AE[自编码器: 异常重构误差] EMB --> TS[时序异常检测: LSTM] GNN --> ENS[集成风险评分] AE --> ENS TS --> ENS end subgraph Action[响应层] ENS -->|评分 > 0.8| P0[P0: 暂停合约交互] ENS -->|评分 > 0.5| P1[P1: 人工审核] ENS -->|评分 > 0.3| P2[P2: 记录日志] end这个架构的设计要点是分层解耦：数据采集层关注实时性和完整性，特征工程层把原始链上数据转化成 AI 可处理的数值特征，AI 分析层做真正的智能判断，响应层根据风险评分触发不同级别的动作。
四、链上异常检测的边界与权衡
误报和漏报的平衡。AI 模型再强也有误判。DeFi 协议中，把正常的大户砸盘误判为攻击，可能导致无辜用户的资产被冻结。建议 AI 模型输出的不是"是/否"的二分类，而是 0-1 的风险评分，让响应层根据业务场景决定阈值。
实时性 vs 准确性的取舍。基于图神经网络的分析依赖多跳邻居特征，在大规模图上计算延迟较高。对于实时性要求高的场景（如闪电贷攻击检测），应优先使用轻量模型（如单地址行为分类器）；对于事后审计场景，可以使用更精准的重型模型。
数据时效性问题。链上数据不会消失，但"最新区块"在几秒后就不再是最新的了。特征工程需要考虑数据的时序窗口：检测闪电贷攻击可能只看最近 10 个区块，但检测钓鱼地址需要分析数月的交易历史。
链上隐私与合规。虽然链上数据是公开的，但将地址做异常标记并公开展示可能涉及隐私法规（如 GDPR 的被遗忘权）。地址标记系统应该在脱敏和透明度之间找到平衡。
三、Python 实现：基础版地址行为分析
import json import time from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Optional import numpy as np from web3 import Web3 from web3.middleware import geth_poa_middleware # ========== 数据结构 ========== @dataclass class AddressProfile: """地址画像""" address: str first_seen: datetime total_tx_count: int = 0 # 总交易数 unique_interactions: int = 0 # 不同合约交互数 failed_tx_ratio: float = 0.0 # 失败交易比例 avg_gas_price: float = 0.0 # 平均 Gas Price (Gwei) tx_time_span: timedelta = field(default_factory=timedelta) # 交易时间跨度 value_distribution: list[float] = field(default_factory=list) # 交易金额分布 risk_score: float = 0.0 # 风险评分 @dataclass class TransactionFeatures: """单笔交易的特征""" hash: str from_addr: str to_addr: str value_eth: float gas_price_gwei: float gas_used: int data_length: int # calldata 长度（合约交互复杂度） is_contract_creation: bool timestamp: datetime # ========== 特征提取 ========== class ChainFeatureExtractor: """链上数据特征提取器""" def __init__(self, rpc_url: str): self.w3 = Web3(Web3.HTTPProvider(rpc_url)) self.w3.middleware_onion.inject(geth_poa_middleware, layer=0) self.address_cache: dict[str, AddressProfile] = {} def extract_transaction_features(self, tx_hash: str) -> Optional[TransactionFeatures]: """从交易哈希提取特征""" try: tx = self.w3.eth.get_transaction(tx_hash) receipt = self.w3.eth.get_transaction_receipt(tx_hash) block = self.w3.eth.get_block(tx.blockNumber) return TransactionFeatures( hash=tx_hash, from_addr=tx["from"], to_addr=tx.get("to", "0x0"), # 合约创建的 to 为空 value_eth=self.w3.from_wei(tx["value"], "ether"), gas_price_gwei=tx["gasPrice"] / 1e9, gas_used=receipt["gasUsed"], data_length=len(tx.get("input", "0x")), is_contract_creation=tx.get("to") is None, timestamp=datetime.fromtimestamp(block["timestamp"]), ) except Exception as e: print(f"提取交易特征失败 {tx_hash}: {e}") return None def build_address_profile( self, address: str, tx_list: list[TransactionFeatures] ) -> AddressProfile: """根据交易历史构建地址画像""" if not tx_list: return AddressProfile( address=address, first_seen=datetime.now() ) timestamps = [tx.timestamp for tx in tx_list] values = [tx.value_eth for tx in tx_list] gas_prices = [tx.gas_price_gwei for tx in tx_list] unique_targets = set(tx.to_addr for tx in tx_list if tx.to_addr != "0x0") failed_count = sum( 1 for tx in tx_list if not tx.is_contract_creation and tx.gas_used > 200000 ) return AddressProfile( address=address, first_seen=min(timestamps), total_tx_count=len(tx_list), unique_interactions=len(unique_targets), failed_tx_ratio=failed_count / max(len(tx_list), 1), avg_gas_price=np.mean(gas_prices) if gas_prices else 0, tx_time_span=max(timestamps) - min(timestamps), value_distribution=values, ) # ========== AI 异常检测模型（简化版） ========== class ChainAnomalyDetector: """链上行为异常检测器""" def __init__(self): # 在实际项目中，这里会加载训练好的模型 # 这里用基于统计的方法做演示 self.normal_distributions: dict[str, dict] = {} def calculate_risk_score(self, profile: AddressProfile) -> tuple[float, list[str]]: """计算地址的风险评分和异常原因""" score = 0.0 reasons = [] # 特征 1：新创建的地址突然大额交互 if profile.total_tx_count < 10: max_value = max(profile.value_distribution) if profile.value_distribution else 0 if max_value > 10: # 新地址交易超过 10 ETH score += 0.3 reasons.append(f"新地址大额交易: {max_value:.2f} ETH") # 特征 2：交易失败比例过高（可能是攻击尝试） if profile.failed_tx_ratio > 0.5 and profile.total_tx_count > 5: score += 0.25 reasons.append(f"交易失败率异常: {profile.failed_tx_ratio:.1%}") # 特征 3：短时间内大量交易（女巫攻击特征） if profile.tx_time_span.total_seconds() > 0: tx_rate = profile.total_tx_count / max( profile.tx_time_span.total_seconds() / 3600, 0.01 ) if tx_rate > 100: # 每小时超过 100 笔 score += 0.2 reasons.append(f"交易频率异常: {tx_rate:.0f} 笔/小时") # 特征 4：Gas Price 异常（可能是 MEV 或抢跑攻击） if profile.avg_gas_price > 200: # 平均 Gas > 200 Gwei score += 0.15 reasons.append(f"Gas Price 异常: {profile.avg_gas_price:.0f} Gwei") # 特征 5：合约交互过于单一（可能是自动化脚本） if profile.unique_interactions == 1 and profile.total_tx_count > 20: score += 0.1 reasons.append(f"交互模式单一: 仅 {profile.unique_interactions} 个合约") return min(score, 1.0), reasons # ========== 使用示例 ========== def analyze_suspicious_address(rpc_url: str, address: str): """分析可疑地址的完整流程""" extractor = ChainFeatureExtractor(rpc_url) detector = ChainAnomalyDetector() # 实际项目中从区块浏览器或索引服务获取交易列表 sample_txs = [ TransactionFeatures( hash="0x...", from_addr=address, to_addr="0xdefi_contract", value_eth=50.0, gas_price_gwei=250.0, gas_used=150000, data_length=520, is_contract_creation=False, timestamp=datetime.now() - timedelta(minutes=5), ) ] profile = extractor.build_address_profile(address, sample_txs) risk_score, reasons = detector.calculate_risk_score(profile) return { "address": address, "risk_score": risk_score, "risk_level": "HIGH" if risk_score > 0.7 else "MEDIUM" if risk_score > 0.4 else "LOW", "reasons": reasons, "profile": profile, }五、总结
链上数据 + AI 的异常检测方案，核心价值不是替代传统规则引擎，而是补充它无法覆盖的"未知攻击模式"。实施路径建议：先用统计方法上线基础版的异常监控，收集标注数据，再逐步引入机器学习模型。不要一上来就追求图神经网络的酷炫效果——先让告警准确率跑赢纯规则方案，再谈模型升级。
