AI 辅助链上资金异常追踪：跨合约调用轨迹还原与可疑地址聚类分析

AI 辅助链上资金异常追踪跨合约调用轨迹还原与可疑地址聚类分析一、链上资金追踪当人工分析跟不上攻击者的速度链上资金异常事件的典型场景攻击合约在短时间内通过多合约交互提取大量资金然后通过一系列中间地址分散转移最终汇集到提现出口。人工分析师需要逐笔追踪交易调用链识别资金流向标记可疑地址——这个过程可能耗时数天而攻击者在这段时间内已经完成了资金转移和洗钱。更复杂的场景涉及跨合约调用轨迹攻击者不是直接调用目标合约的提取函数而是通过一系列中间合约的间接调用如闪电贷→价格操控→资产提取→跨桥转移完成攻击。每一层调用都发生在不同的合约上调用之间的因果关系需要通过交易内部轨迹internal transactions和事件日志event logs来还原。本文提出一种 AI 辅助的链上资金追踪方案自动提取跨合约调用轨迹构建资金流向图通过聚类分析识别可疑地址集群最终生成攻击路径报告。核心思路是利用 LLM 的语义推理能力处理模糊的关联线索如地址 A 和地址 B 在同一时间段活跃且交易模式相似补充纯规则引擎无法覆盖的关联判断。二、原理剖析跨合约调用轨迹还原与聚类算法2.1 跨合约调用轨迹提取以太坊交易的表层数据from、to、value只记录了交易的直接发起者和目标合约。但实际执行过程中目标合约可能调用其他合约形成多层调用栈。使用debug_traceTransactionAPI 可以获取完整的内部调用栈每一层调用记录了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: funknown_func_{func_sig}, params: {}} try: contract self.w3.eth.contract(addresscontract_addr, abiabi) # 解码函数调用数据 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, authneo4j_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, addrtrace[from], timetrace[tx_hash], typeunknown ) session.run( MERGE (a:Address {addr: $addr}) ON CREATE SET a.first_seen $time, a.type $type, addrtrace[to], timetrace[tx_hash], typeself._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, fromtrace[from], totrace[to], txtrace[tx_hash], tagtrace[semantic_tag], valuetrace[value_eth], depthtrace[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 unknown3.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 fcreator:{creator} clusters.setdefault(creator_key, []).append(addr) # 因子 2: 同一时间窗口内首次活跃的地址 first_active_block self._get_first_active_block(addr) window_key fwindow:{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_asciiFalse, indent2)} 规则聚类已有结果: {json.dumps(rule_clusters, ensure_asciiFalse, indent2)} 请以 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, limit50) 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_idrc[cluster_key], addressesrc[addresses], confidencerc[confidence], reasoningfRule-based: {rc[rule_factors][0]}, rule_factorsrc[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_idfai:{hash(frozenset(ac[addresses]))}, addressesac[addresses], confidenceadjusted_conf, reasoningac.get(reasoning, ), rule_factors[], ai_factors[pattern_similarity] )) return sorted(results, keylambda c: c.confidence, reverseTrue)四、边界分析追踪方案的局限与合规考量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便于分析师按优先级验证。下一步扩展多链追踪模块覆盖更多链引入实时告警机制缩短从事件发现到追踪启动的响应时间。