AI 辅助链上资金异常追踪：跨合约调用轨迹还原与可疑地址聚类分析
一、链上资金追踪：当人工分析跟不上攻击者的速度
链上资金异常事件的典型场景：攻击合约在短时间内通过多合约交互提取大量资金，然后通过一系列中间地址分散转移，最终汇集到提现出口。人工分析师需要逐笔追踪交易调用链，识别资金流向，标记可疑地址——这个过程可能耗时数天，而攻击者在这段时间内已经完成了资金转移和洗钱。
更复杂的场景涉及跨合约调用轨迹：攻击者不是直接调用目标合约的提取函数，而是通过一系列中间合约的间接调用（如闪电贷→价格操控→资产提取→跨桥转移）完成攻击。每一层调用都发生在不同的合约上，调用之间的因果关系需要通过交易内部轨迹（internal transactions）和事件日志（event logs）来还原。
本文提出一种 AI 辅助的链上资金追踪方案：自动提取跨合约调用轨迹，构建资金流向图，通过聚类分析识别可疑地址集群，最终生成攻击路径报告。核心思路是利用 LLM 的语义推理能力处理模糊的关联线索（如"地址 A 和地址 B 在同一时间段活跃且交易模式相似"），补充纯规则引擎无法覆盖的关联判断。
二、原理剖析：跨合约调用轨迹还原与聚类算法
2.1 跨合约调用轨迹提取
以太坊交易的表层数据（from、to、value）只记录了交易的直接发起者和目标合约。但实际执行过程中，目标合约可能调用其他合约，形成多层调用栈。使用 debug_traceTransaction API 可以获取完整的内部调用栈，每一层调用记录了 from、to、value、input（调用数据）和 output（返回数据）。
关键挑战：内部调用栈的深度可能达到数十层（尤其在闪电贷场景中），且每层调用的 input 数据是 ABI 编码的字节序列，需要解码才能理解调用意图。我们的方案是将 input 数据与已知合约的 ABI 进行匹配，自动解码函数名和参数，标注每层调用的语义标签（如"存款"、"提取"、"交换"）。
2.2 事件日志的多维度关联
合约事件日志是调用轨迹的重要补充。同一交易中的多个事件按执行顺序记录，形成"事件链"。通过事件链可以还原资金的精确流转路径：ERC20 Transfer 事件记录代币的每次转移，DEX Swap 事件记录交易对和交换量，Bridge 事件记录跨链转移的目标链和接收地址。
关联策略：将事件链与内部调用栈对齐——每个事件对应到触发它的那一层调用，形成"调用-事件"联合视图。这个视图既能回答"资金从哪个合约流出"（调用栈维度），又能回答"流出了多少资金到哪个地址"（事件维度）。
2.3 可疑地址聚类分析
可疑地址的识别需要两个层面的聚类：规则聚类处理确定性线索（如同一创建者部署的多个地址、同一时间段内密集交互的地址组），AI 聚类处理模糊线索（如交易模式相似的地址、Gas 消费特征接近的地址群）。
规则聚类的确定性因子：地址的 creator（部署交易中的 from）相同 → 强关联；地址在同一区块或相邻区块内首次活跃 → 中关联；地址之间有直接的资金转移 → 强关联。AI 聚类使用 LLM 处理更复杂的模式判断：地址 A 的交易频率、金额分布、交互合约集合与地址 B 高度相似，但两者没有直接的资金转移或创建者关联 → 弱-中关联，需要 LLM 的语义推理来判定是否属于同一操控者。
三、代码实践：AI 辅助追踪系统的核心实现
3.1 跨合约调用轨迹提取与解码
# trace_extractor.py — 内部调用轨迹提取与 ABI 解码
from web3 import Web3
from typing import Optional
class TraceExtractor:
"""提取交易的内部调用轨迹,解码函数调用意图"""
def __init__(self, w3: Web3, abi_registry: dict):
self.w3 = w3
# abi_registry: 已知合约地址 → ABI 的映射
self.abi_registry = abi_registry
async def extract_trace(self, tx_hash: str) -> list[dict]:
"""提取完整内部调用栈"""
trace_data = self.w3.provider.make_request(
"debug_traceTransaction",
[tx_hash, {"tracer": "callTracer"}]
)
call_tree = trace_data.get("result", {})
return self._flatten_and_decode(call_tree, tx_hash)
def _flatten_and_decode(self, node: dict, tx_hash: str, depth: int = 0) -> list[dict]:
"""递归展开调用树,解码每层调用的函数意图"""
entries = []
from_addr = node.get("from", "")
to_addr = node.get("to", "")
value = int(node.get("value", "0x0"), 16)
input_data = node.get("input", "0x")
output_data = node.get("output", "0x")
# 尝试 ABI 解码
decoded = self._decode_call(to_addr, input_data)
entry = {
"tx_hash": tx_hash,
"depth": depth,
"from": from_addr,
"to": to_addr,
"value_eth": value / 1e18,
"function_name": decoded.get("name", "unknown"),
"function_params": decoded.get("params", {}),
"semantic_tag": self._assign_tag(decoded.get("name", "")),
"error": node.get("error"),
}
entries.append(entry)
# 递归处理子调用
for subcall in node.get("calls", []):
entries.extend(self._flatten_and_decode(subcall, tx_hash, depth + 1))
return entries
def _decode_call(self, contract_addr: str, input_data: str) -> dict:
"""使用已知 ABI 解码函数调用"""
abi = self.abi_registry.get(contract_addr)
if not abi:
# 未知合约: 提取函数签名哈希作为标识
func_sig = input_data[:10] if len(input_data) >= 10 else "0x"
return {"name": f"unknown_func_{func_sig}", "params": {}}
try:
contract = self.w3.eth.contract(address=contract_addr, abi=abi)
# 解码函数调用数据
func, params = contract.decode_function_input(input_data)
return {"name": func.fn_name, "params": dict(params)}
except Exception:
return {"name": "decode_failed", "params": {}}
def _assign_tag(self, func_name: str) -> str:
"""为函数调用分配语义标签,标识资金流向意图"""
tag_map = {
"deposit": "funds_in", "withdraw": "funds_out",
"swap": "exchange", "transfer": "transfer",
"flashLoan": "borrow", "repay": "repay",
"bridge": "cross_chain",
}
return tag_map.get(func_name.lower(), "unknown_action")
3.2 资金流向图构建
# fund_flow_graph.py — 基于 Neo4j 的资金流向图构建
from neo4j import GraphDatabase
class FundFlowGraphBuilder:
"""将调用轨迹和事件日志构建为图数据库中的资金流向网络"""
def __init__(self, neo4j_uri: str, neo4j_auth: tuple):
self.driver = GraphDatabase.driver(neo4j_uri, auth=neo4j_auth)
def build_from_trace(self, traces: list[dict], events: list[dict]):
"""从调用轨迹和事件日志构建图节点和边"""
with self.driver.session() as session:
# 创建地址节点（标注类型: 攻击合约/中间地址/受害者合约/提现出口）
for trace in traces:
session.run(
"""MERGE (a:Address {addr: $addr})
ON CREATE SET a.first_seen = $time, a.type = $type
ON MATCH SET a.last_seen = $time""",
addr=trace["from"], time=trace["tx_hash"], type="unknown"
)
session.run(
"""MERGE (a:Address {addr: $addr})
ON CREATE SET a.first_seen = $time, a.type = $type""",
addr=trace["to"], time=trace["tx_hash"],
type=self._infer_addr_type(trace)
)
# 创建资金流转边（标注语义: deposit/withdraw/transfer/bridge）
for trace in traces:
if trace["value_eth"] > 0 or trace["semantic_tag"] != "unknown_action":
session.run(
"""MATCH (from:Address {addr: $from})
MATCH (to:Address {addr: $to})
MERGE (from)-[r:FLOW {tx: $tx, tag: $tag, value: $value}]->(to)
SET r.depth = $depth""",
from=trace["from"], to=trace["to"],
tx=trace["tx_hash"], tag=trace["semantic_tag"],
value=trace["value_eth"], depth=trace["depth"]
)
def _infer_addr_type(self, trace: dict) -> str:
"""根据调用特征推断地址类型"""
if trace["semantic_tag"] == "funds_out" and trace["depth"] == 0:
return "attacker_contract"
if trace["semantic_tag"] == "cross_chain":
return "bridge_exit"
if trace["value_eth"] > 100 and trace["depth"] > 3:
return "intermediate"
return "unknown"
3.3 AI 辅助可疑地址聚类
# address_cluster.py — 规则聚类 + AI 语义聚类的联合分析
import json
from dataclasses import dataclass
@dataclass
class ClusterResult:
cluster_id: str
addresses: list[str]
confidence: float # 0-1, 规则+AI联合置信度
reasoning: str # AI 生成的聚类推理说明
rule_factors: list[str] # 规则层面的确定性因子
ai_factors: list[str] # AI 层面的模糊性因子
class AddressClusterAnalyzer:
"""可疑地址聚类: 规则引擎 + LLM 联合推理"""
def __init__(self, w3, graph_builder, llm_client):
self.w3 = w3
self.graph = graph_builder
self.llm = llm_client
async def analyze(self, suspect_addresses: list[str]) -> list[ClusterResult]:
"""对可疑地址列表执行聚类分析"""
# Phase 1: 规则聚类 — 确定性因子
rule_clusters = self._rule_cluster(suspect_addresses)
# Phase 2: AI 聚类 — 模糊性因子
ai_clusters = await self._ai_cluster(suspect_addresses, rule_clusters)
# Phase 3: 合并 — 规则簇与 AI 簇的置信度融合
return self._merge_clusters(rule_clusters, ai_clusters)
def _rule_cluster(self, addresses: list[str]) -> list[dict]:
"""规则层面的确定性聚类"""
clusters: dict[str, list[str]] = {}
for addr in addresses:
# 因子 1: 同一创建者部署的地址属于同一簇
creator = self._get_creator(addr)
creator_key = f"creator:{creator}"
clusters.setdefault(creator_key, []).append(addr)
# 因子 2: 同一时间窗口内首次活跃的地址
first_active_block = self._get_first_active_block(addr)
window_key = f"window:{first_active_block // 100}" # 按100个区块分窗
clusters.setdefault(window_key, []).append(addr)
# 去重: 地址可能属于多个规则簇,取最强因子
result = []
for key, members in clusters.items():
if len(members) >= 2:
result.append({
"cluster_key": key,
"addresses": members,
"rule_factors": [key],
"confidence": 0.9 if "creator:" in key else 0.6,
})
return result
async def _ai_cluster(self, addresses: list[str], rule_clusters: list[dict]) -> list[dict]:
"""AI 层面的模糊聚类: 交易模式相似度"""
# 为每个地址生成交易行为摘要
profiles = {}
for addr in addresses:
profiles[addr] = self._build_tx_profile(addr)
# 将行为摘要交给 LLM 判断相似度
prompt = f"""分析以下地址的交易行为摘要,判断哪些地址可能由同一操控者控制。
考虑因素: 交易频率模式、金额分布、交互合约集合、Gas消费特征。
排除已在规则聚类中确认关联的地址组合,只分析规则未覆盖的模糊关联。
地址行为摘要:
{json.dumps(profiles, ensure_ascii=False, indent=2)}
规则聚类已有结果:
{json.dumps(rule_clusters, ensure_ascii=False, indent=2)}
请以 JSON 数组输出聚类结果,每个聚类包含: addresses列表, reasoning说明, confidence评分(0-1)。"""
response = await self.llm.generate(prompt)
return json.loads(response)
def _build_tx_profile(self, addr: str) -> dict:
"""为地址构建交易行为摘要"""
tx_count = self._get_tx_count(addr)
# 简化: 取最近50笔交易的统计特征
recent_txs = self._get_recent_txs(addr, limit=50)
return {
"address": addr,
"total_tx_count": tx_count,
"avg_tx_interval_s": self._calc_avg_interval(recent_txs),
"value_distribution": {
"median": self._calc_median_value(recent_txs),
"std_dev": self._calc_std_value(recent_txs),
},
"interacted_contracts": list(set(
tx["to"] for tx in recent_txs[:20]
)),
"gas_usage_pattern": {
"avg_gas": sum(tx.get("gas", 0) for tx in recent_txs) / len(recent_txs),
}
}
def _merge_clusters(self, rule_clusters: list, ai_clusters: list) -> list[ClusterResult]:
"""合并规则簇和 AI 簇,计算联合置信度"""
results = []
for rc in rule_clusters:
results.append(ClusterResult(
cluster_id=rc["cluster_key"],
addresses=rc["addresses"],
confidence=rc["confidence"],
reasoning=f"Rule-based: {rc['rule_factors'][0]}",
rule_factors=rc["rule_factors"],
ai_factors=[]
))
for ac in ai_clusters:
# 检查 AI 簇是否与已有规则簇重叠
overlap = any(
set(ac["addresses"]) & set(rc["addresses"])
for rc in rule_clusters
)
adjusted_conf = ac.get("confidence", 0.5) * (0.3 if overlap else 1.0)
results.append(ClusterResult(
cluster_id=f"ai:{hash(frozenset(ac['addresses']))}",
addresses=ac["addresses"],
confidence=adjusted_conf,
reasoning=ac.get("reasoning", ""),
rule_factors=[],
ai_factors=["pattern_similarity"]
))
return sorted(results, key=lambda c: c.confidence, reverse=True)
四、边界分析：追踪方案的局限与合规考量
4.1 ABI 解码的覆盖范围
已知合约的 ABI 覆盖率直接影响调用轨迹的解码质量。在我们的场景中，约 70% 的被调用合约有公开 ABI（主流 DeFi 协议），但剩余 30% 的合约（包括攻击者部署的合约和部分私有合约）无法解码。对于这些合约，只能提取函数签名哈希（4字节）作为标识，语义理解依赖 LLM 推理或社区维护的签名数据库（如 4byte.directory）。
4.2 跨链追踪的断点
当攻击者通过跨链桥将资金转移到其他链后，追踪链路出现断点——本链的追踪工具无法访问目标链的数据。解决方案是部署多链追踪模块，通过桥接合约的跨链消息记录建立两链之间的地址映射关系。但这需要目标链也有相应的追踪工具部署，目前覆盖范围仅限于以太坊、BSC 和 Polygon。
4.3 合规与隐私边界
链上数据虽然公开，但将地址与操控者身份关联的推理结果可能涉及合规风险。聚类分析的结果应标注为"技术推断"而非"身份确认"，避免被用于未经司法授权的身份追踪。生产部署需在报告输出中附加免责声明，并限制 AI 推理的粒度——只输出地址间的关联模式，不推断具体自然人身份。
五、总结
链上资金异常追踪的核心挑战在于跨合约调用轨迹的还原和可疑地址的聚类识别。本文方案的双引擎架构：规则引擎处理确定性线索（创建者关联、时间窗口关联、直接资金转移），AI 引擎（LLM）处理模糊线索（交易模式相似度、Gas 消费特征）。两引擎的结果通过置信度融合机制合并，生成带评分的可疑地址清单和攻击路径报告。
关键设计决策：使用 debug_traceTransaction 获取完整内部调用栈而非只看表层交易；ABI 解码覆盖率 70%，剩余 30% 通过函数签名哈希和 LLM 推理补位；聚类结果标注置信度等级（规则强关联 >0.9、AI 弱关联 0.3-0.7），便于分析师按优先级验证。下一步：扩展多链追踪模块覆盖更多链，引入实时告警机制缩短从事件发现到追踪启动的响应时间。
