AI 链上数据分析：Ethereum 交易模式识别与异常检测的工程实战
____simple_html_dom__voku__html_wrapper____>
AI 链上数据分析：Ethereum 交易模式识别与异常检测的工程实战
一、链上数据洪流与人工审计的极限：Ethereum 监控的工程痛点
以太坊主网每天产生约 120 万笔交易，每笔交易包含发送方、接收方、金额、Gas 消耗和调用数据等多个维度。对于 DeFi 协议运营方和安全团队而言，从海量交易中识别异常模式（如闪电贷攻击、三明治攻击、洗钱路径）是核心需求，但传统的人工审计方式已无法应对数据规模。
链上异常检测的工程痛点集中在三个方面。第一，交易关联的图结构复杂。一笔攻击交易通常涉及数十个合约调用和资金转移路径，人工追踪需要逐笔展开内部交易，耗时且容易遗漏。第二，攻击模式持续演化。闪电贷攻击从最初的简单套利演变为跨协议组合攻击，基于规则的检测系统需要频繁更新规则库，维护成本高。第三，实时性要求与计算成本的矛盾。链上监控需要在交易确认后的秒级时间内发出告警，但复杂的模式分析需要大量计算资源。
本文将构建一套基于 AI 的链上交易分析系统，覆盖交易数据索引、模式识别和实时告警三个核心环节。
二、交易图谱与特征工程：链上异常检测的数据基础
链上异常检测的本质是图数据上的模式匹配问题。每笔交易可以建模为有向图中的一个边，节点是地址（EOA 或合约），边是价值转移或合约调用。异常模式（如循环转账、闪电贷路径）在这个图上表现为特定的子图结构。
flowchart TB
subgraph 数据索引层
A[区块数据] --> B[交易解析]
B --> C[内部交易追踪]
C --> D[事件日志提取]
D --> E[地址实体标注]
end
subgraph 特征工程层
E --> F[地址级特征]
E --> G[交易级特征]
E --> H[图结构特征]
F --> I[活跃时段分布/交易频率/合约交互数]
G --> J[Gas 异常度/金额偏离度/调用深度]
H --> K[入度出度比/聚类系数/中心度]
end
subgraph AI 检测层
I --> L[异常评分模型]
J --> L
K --> L
L --> M{评分阈值判定}
M -->|高于阈值| N[实时告警]
M -->|低于阈值| O[正常交易归档]
end
subgraph 反馈优化层
N --> P[人工标注]
P --> Q[模型增量训练]
Q --> L
end
上图展示了链上异常检测系统的四层架构。数据索引层负责从区块中提取结构化数据；特征工程层从原始数据中提取多维度特征；AI 检测层基于特征进行异常评分；反馈优化层通过人工标注持续优化模型。
特征工程是检测效果的关键。地址级特征刻画地址的行为模式，交易级特征刻画单笔交易的异常程度，图结构特征刻画地址在交易网络中的位置特征。三类特征的组合可以有效区分正常交易和异常交易。
以闪电贷攻击为例，其特征模式为：同一地址在同一区块内完成借款、套利操作和还款，交易级特征表现为极高的调用深度和 Gas 消耗，图结构特征表现为从 DEX 合约出发的星形调用模式。这些特征组合可以高效识别闪电贷攻击，无需逐笔分析内部交易。
三、生产级代码实现：链上交易分析与异常检测
3.1 链上数据索引与特征提取
# services/chain_indexer.py
import asyncio
from dataclasses import dataclass, field
from typing import Optional
from web3 import Web3
from web3.types import TxData, TxReceipt, LogReceipt
@dataclass
class AddressFeatures:
"""地址级特征——刻画地址的行为模式"""
address: str
tx_count_24h: int = 0 # 24小时交易数
unique_counterparts: int = 0 # 唯一交易对手数
contract_call_ratio: float = 0 # 合约调用占比
avg_gas_price: float = 0 # 平均 Gas 价格
active_hours: set[int] = field(default_factory=set) # 活跃时段
first_seen: int = 0 # 首次出现区块号
label: str = "unknown" # 实体标签（DEX/借贷/用户等）
@dataclass
class TransactionFeatures:
"""交易级特征——刻画单笔交易的异常程度"""
tx_hash: str
from_addr: str
to_addr: str
value_eth: float
gas_price_gwei: float
gas_used: int
# 异常特征指标
call_depth: int = 0 # 内部调用深度
internal_tx_count: int = 0 # 内部交易数
token_transfer_count: int = 0 # ERC20 转账数
value_deviation: float = 0 # 金额偏离度（与历史均值的标准差倍数）
gas_deviation: float = 0 # Gas 偏离度
class ChainIndexer:
"""链上数据索引器——从区块中提取结构化特征"""
def __init__(self, rpc_url: str):
self.w3 = Web3(Web3.HTTPProvider(rpc_url))
self.address_features: dict[str, AddressFeatures] = {}
self.token_abi = self._load_erc20_abi()
async def index_block(self, block_number: int) -> list[TransactionFeatures]:
"""索引单个区块的所有交易并提取特征"""
block = self.w3.eth.get_block(block_number, full_transactions=True)
tx_features = []
for tx in block.transactions:
try:
features = await self._extract_tx_features(tx)
if features:
tx_features.append(features)
# 更新地址级特征
self._update_address_features(tx, features)
except Exception as e:
# 单笔交易解析失败不应阻塞整个区块的索引
print(f"交易 {tx.hash.hex()} 解析失败: {e}")
continue
return tx_features
async def _extract_tx_features(self, tx: TxData) -> Optional[TransactionFeatures]:
"""提取单笔交易的特征"""
if tx.to is None:
# 合约创建交易——跳过，通常不涉及攻击模式
return None
receipt = self.w3.eth.get_transaction_receipt(tx.hash)
# 计算内部调用深度——通过 trace API 或 Gas 消耗推算
# trace API 需要归档节点支持，这里用 Gas 消耗作为近似
call_depth = self._estimate_call_depth(receipt)
# 统计 ERC20 转账事件——Transfer 事件的主题哈希
transfer_topic = Web3.keccak(text="Transfer(address,address,uint256)").hex()
token_transfers = [
log for log in receipt.logs
if log.topics and log.topics[0].hex() == transfer_topic
]
# 计算金额偏离度——与该地址历史交易的均值比较
value_eth = float(Web3.from_wei(tx.value, 'ether'))
gas_price_gwei = float(Web3.from_wei(tx.gasPrice, 'gwei'))
value_deviation = self._calc_deviation(
value_eth, tx['from'], 'value'
)
gas_deviation = self._calc_deviation(
gas_price_gwei, tx['from'], 'gas'
)
return TransactionFeatures(
tx_hash=tx.hash.hex(),
from_addr=tx['from'],
to_addr=tx.to,
value_eth=value_eth,
gas_price_gwei=gas_price_gwei,
gas_used=receipt.gasUsed,
call_depth=call_depth,
internal_tx_count=max(0, len(receipt.logs) - len(token_transfers)),
token_transfer_count=len(token_transfers),
value_deviation=value_deviation,
gas_deviation=gas_deviation,
)
def _estimate_call_depth(self, receipt: TxReceipt) -> int:
"""通过 Gas 消耗和日志数推算调用深度"""
# 经验公式：调用深度与 Gas 消耗和日志数正相关
# 这不是精确计算，但足以用于异常检测的特征输入
if receipt.gasUsed < 50000:
return 1
elif receipt.gasUsed < 200000:
return 2
elif receipt.gasUsed < 500000:
return 3
else:
return min(receipt.gasUsed // 150000, 10)
def _calc_deviation(self, value: float, address: str, metric: str) -> float:
"""计算偏离度——与地址历史均值的标准差倍数"""
features = self.address_features.get(address)
if not features or features.tx_count_24h < 5:
return 0.0
# 简化计算：使用历史均值作为基准
# 生产环境应维护完整的统计量（均值、标准差）
if metric == 'value':
baseline = 0.1 # 默认基准值
else:
baseline = 20.0 # Gas 价格基准
if baseline == 0:
return 0.0
return abs(value - baseline) / baseline
def _update_address_features(self, tx: TxData, features: TransactionFeatures):
"""更新地址级特征统计"""
for addr in [tx['from'], tx.to]:
if addr not in self.address_features:
self.address_features[addr] = AddressFeatures(address=addr)
af = self.address_features[addr]
af.tx_count_24h += 1
def _load_erc20_abi(self) -> list:
"""加载 ERC20 标准 ABI——用于解析 Transfer 事件"""
return [{"anonymous": False, "inputs": [
{"indexed": True, "name": "from", "type": "address"},
{"indexed": True, "name": "to", "type": "address"},
{"indexed": False, "name": "value", "type": "uint256"},
], "name": "Transfer", "type": "event"}]
3.2 AI 异常检测模型——基于规则的快速筛选 + ML 精排
# services/anomaly_detector.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional
class AlertLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class AnomalyAlert:
"""异常告警——包含风险评分和检测依据"""
tx_hash: str
alert_level: AlertLevel
risk_score: float # 0.0 ~ 1.0
detected_patterns: list[str] # 检测到的异常模式
description: str
related_addresses: list[str]
class AnomalyDetector:
"""链上异常检测器——规则引擎快速筛选 + ML 模型精排"""
# 规则引擎阈值——基于历史攻击案例的统计特征
FLASH_LOAN_GAS_THRESHOLD = 500000 # 闪电贷交易 Gas 下限
SANDWICH_GAS_PRICE_MULTIPLE = 3.0 # 三明治攻击 Gas 价格倍数
WASH_TRADE_COUNTERPART_LIMIT = 3 # 洗钱地址交易对手数上限
def __init__(self, ml_model=None):
self.ml_model = ml_model # 可选的 ML 精排模型
def detect(self, tx: TransactionFeatures,
addr_features: dict[str, AddressFeatures]) -> Optional[AnomalyAlert]:
"""对单笔交易执行异常检测"""
patterns = []
risk_score = 0.0
# 规则1：闪电贷攻击检测
# 特征：高 Gas 消耗 + 多 ERC20 转账 + 高调用深度
if (tx.gas_used > self.FLASH_LOAN_GAS_THRESHOLD
and tx.token_transfer_count > 5
and tx.call_depth >= 4):
patterns.append("flash_loan_attack")
risk_score += 0.4
# 规则2：三明治攻击检测
# 特征：Gas 价格显著高于网络均值 + 与 DEX 合约交互
if tx.gas_deviation > self.SANDWICH_GAS_PRICE_MULTIPLE:
from_af = addr_features.get(tx.from_addr)
if from_af and from_af.contract_call_ratio > 0.8:
patterns.append("sandwich_attack")
risk_score += 0.3
# 规则3：洗钱模式检测
# 特征：交易对手极少 + 金额偏离度高 + 活跃时段集中
from_af = addr_features.get(tx.from_addr)
if from_af and from_af.unique_counterparts <= self.WASH_TRADE_COUNTERPART_LIMIT:
if tx.value_deviation > 2.0 and len(from_af.active_hours) <= 4:
patterns.append("wash_trading")
risk_score += 0.25
# 规则4：异常大额转账
# 特征：金额偏离历史均值超过 5 个标准差
if tx.value_deviation > 5.0:
patterns.append("abnormal_value_transfer")
risk_score += 0.2
# 无异常模式时返回 None——避免产生过多低质量告警
if not patterns:
return None
# ML 精排——如果模型可用，用模型评分替代规则评分
if self.ml_model:
try:
ml_score = self.ml_model.predict(self._extract_ml_features(tx, addr_features))
# 融合规则评分和 ML 评分——规则评分作为先验
risk_score = 0.4 * risk_score + 0.6 * ml_score
except Exception as e:
# ML 模型推理失败时回退到纯规则评分
print(f"ML 模型推理失败，回退到规则评分: {e}")
# 确定告警等级
alert_level = self._determine_alert_level(risk_score)
return AnomalyAlert(
tx_hash=tx.tx_hash,
alert_level=alert_level,
risk_score=round(risk_score, 3),
detected_patterns=patterns,
description=self._generate_description(patterns, tx),
related_addresses=[tx.from_addr, tx.to_addr],
)
def _determine_alert_level(self, score: float) -> AlertLevel:
"""根据风险评分确定告警等级"""
if score >= 0.8:
return AlertLevel.CRITICAL
elif score >= 0.6:
return AlertLevel.HIGH
elif score >= 0.4:
return AlertLevel.MEDIUM
else:
return AlertLevel.LOW
def _generate_description(self, patterns: list[str], tx: TransactionFeatures) -> str:
"""生成告警描述——便于安全团队快速理解风险"""
pattern_names = {
"flash_loan_attack": "闪电贷攻击",
"sandwich_attack": "三明治攻击",
"wash_trading": "洗钱交易",
"abnormal_value_transfer": "异常大额转账",
}
detected = "、".join(pattern_names.get(p, p) for p in patterns)
return (f"检测到 {detected} 模式。"
f"交易 {tx.tx_hash[:10]}...，"
f"Gas 消耗 {tx.gas_used}，"
f"ERC20 转账 {tx.token_transfer_count} 次")
def _extract_ml_features(self, tx: TransactionFeatures,
addr_features: dict[str, AddressFeatures]) -> list[float]:
"""提取 ML 模型输入特征向量"""
from_af = addr_features.get(tx.from_addr, AddressFeatures(address=tx.from_addr))
return [
tx.value_eth,
tx.gas_price_gwei,
float(tx.gas_used),
float(tx.call_depth),
float(tx.internal_tx_count),
float(tx.token_transfer_count),
tx.value_deviation,
tx.gas_deviation,
float(from_af.tx_count_24h),
float(from_af.unique_counterparts),
from_af.contract_call_ratio,
]
四、AI 链上检测的边界：误报率与实时性的权衡
链上异常检测系统的核心权衡在于误报率与检测覆盖率的平衡。
规则引擎的误报问题。 基于阈值的规则引擎会产生大量误报。例如，正常的大额 DeFi 交易也可能触发闪电贷检测规则（高 Gas + 多转账）。降低阈值可以提高检测覆盖率，但会显著增加误报率，导致安全团队对告警产生疲劳。
ML 模型的冷启动问题。 链上攻击样本稀少，且攻击模式持续演化。模型在训练数据覆盖的攻击类型上表现良好，但对新型攻击可能完全失效。需要持续收集标注数据并定期重训练，但标注数据的获取成本高。
实时检测的延迟约束。 完整的特征提取（包括内部交易追踪和图结构计算）可能需要数秒，而实时告警要求在交易确认后 1-2 秒内发出。解决方案是将检测分为两层：轻量级规则引擎在秒级内完成初筛，ML 模型在后台异步完成精排。
适用边界。 AI 链上检测适用于 DeFi 协议的实时风控、交易所的合规监控和安全团队的威胁狩猎。对于普通用户的交易监控，基于规则的简单告警（如大额转账通知）更为实用。
五、总结
本文构建了一套从链上数据索引、特征工程到 AI 异常检测的完整工程方案。关键要点如下：
第一，链上异常检测的本质是图数据上的模式匹配，特征工程需要从地址级、交易级和图结构级三个维度提取特征。
第二，规则引擎 + ML 精排的两层架构是当前最务实的方案，规则引擎保证实时性，ML 模型提升准确率。
第三，误报率管理是系统可用性的关键，过高的误报率会导致安全团队忽略告警。
落地路线建议：先部署规则引擎并积累告警数据，通过人工标注建立训练集后，再引入 ML 精排模型。实时告警仅依赖规则引擎，ML 结果作为补充分析。建议从监控单一协议（如 Uniswap）开始，验证检测效果后再扩展到多协议覆盖。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
