《万里追凶》与链上追踪：交易追踪作为破案工具
在1998年的悬疑片《万里追凶》（A Simple Plan）中，一笔意外发现的四百万美元现金引发了一场横跨整个冬天的追踪与杀戮。如果用区块链的视角重新审视这个故事，那笔现金的序列号就是链上的一串哈希值，每一次转移都是一笔不可篡改的交易记录，而主角们试图掩盖的每一次行动，都只会让链上证据更加确凿。
第一幕：链上追踪的镜头语言
《万里追凶》的叙事核心在于"追踪"——主角Hank和Jacob兄弟在雪地中发现了一架失事飞机和四百万美元现金，他们决定私吞这笔钱，却发现自己被不断追踪。从电影语言的角度看，导演Sam Raimi用了一系列特写镜头来表现钞票的序列号——这些数字在电影中反复出现，成为追踪的关键线索。
在区块链的世界里，每一笔交易都有一个唯一的交易哈希（TxHash），就像电影中那叠钞票的序列号一样。但不同的是，区块链上的"序列号"是不可伪造、不可销毁的。当我们把这个概念映射到链上追踪时，一个全新的叙事维度就此展开。
区块链浏览器（如Etherscan、Solscan）就是链上世界的"警用监控系统"。每一笔交易的时间戳、发送方地址、接收方地址、金额和交易数据都被永久记录。就像电影中FBI通过银行记录追踪钞票流向一样，链上调查员通过地址分析和交易图谱追踪资金流向。
第二幕：链上追踪的技术架构
从广播电视编导的视角来看，链上追踪就像是一场精心编排的纪录片拍摄——你需要确定拍摄对象（目标地址）、选择拍摄角度（分析工具）、建立叙事线索（资金流向图），最终呈现一个完整的故事。
让我们来看第一个代码示例——一个用于追踪以太坊交易历史的Solidity事件日志分析器：
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;
contract TransactionTracker {
struct Transaction {
address from;
address to;
uint256 amount;
uint256 timestamp;
string memo;
}
event TrackedTransaction(
address indexed from,
address indexed to,
uint256 indexed amount,
uint256 timestamp,
string memo
);
mapping(bytes32 => Transaction) public transactions;
mapping(address => bytes32[]) public addressTransactions;
bytes32[] public allTransactions;
function recordTransaction(
address _to,
uint256 _amount,
string memory _memo
) public returns (bytes32) {
bytes32 txHash = keccak256(
abi.encodePacked(
msg.sender,
_to,
_amount,
block.timestamp,
_memo
)
);
transactions[txHash] = Transaction({
from: msg.sender,
to: _to,
amount: _amount,
timestamp: block.timestamp,
memo: _memo
});
addressTransactions[msg.sender].push(txHash);
addressTransactions[_to].push(txHash);
allTransactions.push(txHash);
emit TrackedTransaction(
msg.sender, _to, _amount,
block.timestamp, _memo
);
return txHash;
}
function getAddressHistory(address _addr)
public view returns (bytes32[] memory)
{
return addressTransactions[_addr];
}
function getTransaction(bytes32 _txHash)
public view returns (Transaction memory)
{
return transactions[_txHash];
}
function traceFlow(address _from, address _to)
public view returns (Transaction[] memory)
{
uint256 count;
for (uint256 i = 0; i < allTransactions.length; i++) {
Transaction storage tx = transactions[allTransactions[i]];
if (tx.from == _from && tx.to == _to) {
count++;
}
}
Transaction[] memory result = new Transaction[](count);
uint256 index;
for (uint256 i = 0; i < allTransactions.length; i++) {
Transaction storage tx = transactions[allTransactions[i]];
if (tx.from == _from && tx.to == _to) {
result[index] = tx;
index++;
}
}
return result;
}
}
这个智能合约就像一个链上"侦探笔记"，记录每一笔交易的"镜头"——谁给了谁、多少钱、什么时候、为什么。在《万里追凶》中，如果主角们知道每一笔交易都被不可篡改地记录，他们可能从一开始就不会动那笔钱。
第三幕：Python驱动的链上追踪分析
在真实的链上调查中，Solang和Python是最常用的工具组合。就像电影中的侦探用放大镜检查证据一样，链上调查员用Python脚本扫描区块链数据。
import requests
import json
from datetime import datetime
from typing import List, Dict, Optional
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt
class ChainTracker:
def __init__(self, api_key: str, chain: str = "ethereum"):
self.api_key = api_key
self.chain = chain
self.base_url = f"https://api.etherscan.io/api"
self.transaction_graph = nx.DiGraph()
def get_address_transactions(
self, address: str, start_block: int = 0, end_block: int = 99999999
) -> List[Dict]:
"""获取地址的交易历史，就像电影中追踪钞票的序列号"""
params = {
'module': 'account',
'action': 'txlist',
'address': address,
'startblock': start_block,
'endblock': end_block,
'sort': 'asc',
'apikey': self.api_key
}
response = requests.get(self.base_url, params=params)
data = response.json()
if data['status'] == '1':
return data['result']
return []
def build_flow_graph(
self, source_address: str, depth: int = 3
) -> nx.DiGraph:
"""构建资金流向图，就像电影中的追踪地图"""
visited = set()
queue = [(source_address, 0)]
while queue:
current_addr, current_depth = queue.pop(0)
if current_depth >= depth or current_addr in visited:
continue
visited.add(current_addr)
txs = self.get_address_transactions(current_addr)
for tx in txs:
from_addr = tx['from']
to_addr = tx['to']
value = float(tx['value']) / 1e18
self.transaction_graph.add_edge(
from_addr, to_addr,
value=value,
hash=tx['hash'],
timestamp=datetime.fromtimestamp(int(tx['timeStamp']))
)
if to_addr not in visited:
queue.append((to_addr, current_depth + 1))
if from_addr not in visited and current_depth < depth:
queue.append((from_addr, current_depth + 1))
return self.transaction_graph
def find_common_ancestors(
self, address_a: str, address_b: str, max_hops: int = 5
) -> List[str]:
"""寻找两个地址的共同资金来源，就像电影中寻找共同上线"""
ancestors_a = self._get_all_ancestors(address_a, max_hops)
ancestors_b = self._get_all_ancestors(address_b, max_hops)
common = set(ancestors_a) & set(ancestors_b)
return list(common)
def _get_all_ancestors(
self, address: str, max_hops: int
) -> List[str]:
ancestors = []
current = address
for _ in range(max_hops):
txs = self.get_address_transactions(current)
if not txs:
break
for tx in txs[:1]: # 取最早一笔交易
ancestors.append(tx['from'])
current = tx['from']
break
return ancestors
def detect_mixing_activity(
self, address: str, time_window_hours: int = 24
) -> Dict:
"""检测混币活动，就像电影中识别洗钱行为"""
txs = self.get_address_transactions(address)
# 统计时间窗口内的交易频率
now = datetime.now()
recent_txs = []
for tx in txs:
tx_time = datetime.fromtimestamp(int(tx['timeStamp']))
if (now - tx_time).total_seconds() / 3600 <= time_window_hours:
recent_txs.append(tx)
# 分析交易模式
unique_from = set(tx['from'] for tx in recent_txs)
unique_to = set(tx['to'] for tx in recent_txs)
return {
'address': address,
'total_transactions': len(txs),
'recent_transactions': len(recent_txs),
'unique_senders': len(unique_from),
'unique_receivers': len(unique_to),
'suspicious_score': len(recent_txs) / (len(unique_from) + 1)
}
def visualize_flow(self, highlight_address: Optional[str] = None):
"""可视化资金流向，就像电影中的追踪地图"""
pos = nx.spring_layout(self.transaction_graph, k=3, iterations=50)
plt.figure(figsize=(16, 12))
# 绘制节点
node_colors = []
for node in self.transaction_graph.nodes():
if node == highlight_address:
node_colors.append('red')
else:
node_colors.append('lightblue')
nx.draw_networkx_nodes(
self.transaction_graph, pos,
node_color=node_colors, node_size=500,
alpha=0.8
)
# 绘制边
edge_labels = {}
for u, v, data in self.transaction_graph.edges(data=True):
edge_labels[(u, v)] = f"{data['value']:.2f} ETH"
nx.draw_networkx_edges(
self.transaction_graph, pos,
edge_color='gray', arrows=True,
arrowsize=20, alpha=0.6
)
nx.draw_networkx_labels(
self.transaction_graph, pos,
font_size=8, font_color='black'
)
plt.title(f"资金流向追踪图 - 源地址: {highlight_address or 'N/A'}")
plt.axis('off')
plt.tight_layout()
return plt
# 使用示例
if __name__ == "__main__":
tracker = ChainTracker(api_key="YOUR_API_KEY")
# 追踪一个可疑地址
suspicious_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18"
# 构建资金流向图
flow_graph = tracker.build_flow_graph(suspicious_address, depth=3)
# 检测混币活动
activity = tracker.detect_mixing_activity(suspicious_address)
print(f"可疑地址: {suspicious_address}")
print(f"交易总数: {activity['total_transactions']}")
print(f"可疑评分: {activity['suspicious_score']:.2f}")
# 可视化
plt = tracker.visualize_flow(highlight_address=suspicious_address)
plt.savefig("flow_analysis.png", dpi=150, bbox_inches='tight')
这个Python脚本就像电影中的"侦探工作室"——它不仅仅是记录数据，更重要的是通过分析发现模式。就像《万里追凶》中主角们的一举一动都在FBI的监控之下，链上的每一笔交易都在分析工具的"镜头"之下。
第四幕：JavaScript构建的链上追踪看板
对于影视编导而言，数据可视化就像电影剪辑——你需要把零散的素材（交易数据）剪辑成一个连贯的故事（资金流向）。JavaScript可以帮助我们构建实时的链上追踪看板。
// 链上追踪实时看板
const Web3 = require('web3');
const axios = require('axios');
const { createCanvas, loadImage } = require('canvas');
class ChainDashboard {
constructor(rpcUrl, apiKey) {
this.web3 = new Web3(new Web3.providers.HttpProvider(rpcUrl));
this.apiKey = apiKey;
this.alertThreshold = 100; // ETH
this.monitoredAddresses = new Map();
this.transactionHistory = [];
}
// 添加监控地址，就像电影中设置监控对象
addAddress(address, label, riskLevel = 'medium') {
this.monitoredAddresses.set(address.toLowerCase(), {
label,
riskLevel,
addedAt: Date.now(),
alerts: []
});
console.log(`[监控] 添加地址: ${label} (${address})`);
}
// 实时交易监听，就像电影中的实时监控
async startMonitoring() {
console.log('[链上追踪] 开始实时监控...');
const subscription = this.web3.eth.subscribe('pendingTransactions');
subscription.on('data', async (txHash) => {
try {
const tx = await this.web3.eth.getTransaction(txHash);
if (tx && tx.value) {
await this.analyzeTransaction(tx);
}
} catch (error) {
// 静默处理
}
});
return subscription;
}
// 分析交易，就像电影审查每一笔可疑资金
async analyzeTransaction(tx) {
const valueInEth = parseFloat(
this.web3.utils.fromWei(tx.value, 'ether')
);
const fromAddress = tx.from?.toLowerCase();
const toAddress = tx.to?.toLowerCase();
// 检查是否涉及监控地址
if (this.monitoredAddresses.has(fromAddress) ||
this.monitoredAddresses.has(toAddress)) {
const txRecord = {
hash: tx.hash,
from: tx.from,
to: tx.to,
value: valueInEth,
timestamp: Date.now(),
blockNumber: tx.blockNumber
};
this.transactionHistory.push(txRecord);
// 触发警报
if (valueInEth >= this.alertThreshold) {
this.triggerAlert(tx, valueInEth);
}
// 更新看板数据
this.updateDashboard(txRecord);
}
return tx;
}
// 触发警报，就像电影中的警报系统
triggerAlert(tx, value) {
const alert = {
type: 'HIGH_VALUE_TRANSACTION',
hash: tx.hash,
value: value,
timestamp: Date.now(),
severity: value >= 1000 ? 'critical' : 'warning'
};
console.log(`[警报] ${alert.severity.toUpperCase()}: ${value} ETH 转账`);
console.log(` 交易哈希: ${tx.hash}`);
console.log(` 从: ${tx.from}`);
console.log(` 到: ${tx.to}`);
// 更新监控地址的警报记录
const fromAddr = tx.from?.toLowerCase();
const toAddr = tx.to?.toLowerCase();
if (this.monitoredAddresses.has(fromAddr)) {
this.monitoredAddresses.get(fromAddr).alerts.push(alert);
}
if (this.monitoredAddresses.has(toAddr)) {
this.monitoredAddresses.get(toAddr).alerts.push(alert);
}
}
// 更新看板
updateDashboard(txRecord) {
// 在实际应用中，这里会更新WebSocket或SSE连接
// 让前端实时显示
console.log(`[看板] 新交易: ${txRecord.value} ETH`);
console.log(` 哈希: ${txRecord.hash.substring(0, 20)}...`);
}
// 生成追踪报告，就像电影中的案件报告
generateReport(targetAddress) {
const addr = targetAddress.toLowerCase();
const info = this.monitoredAddresses.get(addr);
if (!info) {
return { error: '地址未在监控中' };
}
const relatedTxs = this.transactionHistory.filter(
tx => tx.from?.toLowerCase() === addr ||
tx.to?.toLowerCase() === addr
);
// 构建资金流向
const flowMap = new Map();
relatedTxs.forEach(tx => {
const counterparty = tx.from?.toLowerCase() === addr
? tx.to : tx.from;
if (!flowMap.has(counterparty)) {
flowMap.set(counterparty, {
address: counterparty,
totalIn: 0,
totalOut: 0,
txCount: 0
});
}
const record = flowMap.get(counterparty);
if (tx.from?.toLowerCase() === addr) {
record.totalOut += tx.value;
} else {
record.totalIn += tx.value;
}
record.txCount++;
});
return {
targetAddress: targetAddress,
label: info.label,
riskLevel: info.riskLevel,
totalTransactions: relatedTxs.length,
uniqueCounterparties: flowMap.size,
alerts: info.alerts.length,
flowAnalysis: Array.from(flowMap.values())
.sort((a, b) => (b.totalIn + b.totalOut) - (a.totalIn + a.totalOut))
};
}
// 地址聚类分析
clusterAddresses(addresses) {
const clusters = new Map();
addresses.forEach(addr => {
const txs = this.transactionHistory.filter(
tx => tx.from?.toLowerCase() === addr.toLowerCase() ||
tx.to?.toLowerCase() === addr.toLowerCase()
);
// 找到共同对手方
const counterparties = new Set();
txs.forEach(tx => {
if (tx.from?.toLowerCase() === addr.toLowerCase()) {
counterparties.add(tx.to?.toLowerCase());
} else {
counterparties.add(tx.from?.toLowerCase());
}
});
// 寻找聚类
let assigned = false;
for (const [clusterId, members] of clusters) {
for (const member of members) {
if (counterparties.has(member.toLowerCase())) {
members.add(addr.toLowerCase());
assigned = true;
break;
}
}
if (assigned) break;
}
if (!assigned) {
const newClusterId = `cluster_${clusters.size + 1}`;
clusters.set(newClusterId, new Set([addr.toLowerCase()]));
}
});
return clusters;
}
// 链上取证
async forensicAnalysis(address) {
console.log(`[取证] 开始对地址 ${address} 进行链上取证...`);
const txs = this.transactionHistory.filter(
tx => tx.from?.toLowerCase() === address.toLowerCase() ||
tx.to?.toLowerCase() === address.toLowerCase()
);
// 时间线分析
const timeline = txs.sort((a, b) => a.timestamp - b.timestamp);
// 交易模式分析
const patterns = {
circularFlows: 0, // 循环交易
washTrading: 0, // 洗售交易
layering: 0 // 分层交易
};
// 检测循环交易
for (let i = 0; i < timeline.length - 2; i++) {
if (timeline[i].to === timeline[i + 1].from &&
timeline[i + 1].to === timeline[i + 2].from &&
timeline[i + 2].to === timeline[i].from) {
patterns.circularFlows++;
}
}
return {
address,
transactionCount: txs.length,
firstSeen: timeline[0]?.timestamp,
lastSeen: timeline[timeline.length - 1]?.timestamp,
patterns,
timeline: timeline.slice(0, 50) // 返回最近50条
};
}
}
// 使用示例
const dashboard = new ChainDashboard(
'https://mainnet.infura.io/v3/YOUR_PROJECT_ID',
'YOUR_API_KEY'
);
// 设置监控地址
dashboard.addAddress(
'0x1234567890abcdef1234567890abcdef12345678',
'可疑地址A',
'high'
);
dashboard.addAddress(
'0xabcdef1234567890abcdef1234567890abcdef12',
'可疑地址B',
'medium'
);
// 启动监控
dashboard.startMonitoring().then(subscription => {
console.log('监控已启动');
// 5分钟后生成报告
setTimeout(() => {
const report = dashboard.generateReport(
'0x1234567890abcdef1234567890abcdef12345678'
);
console.log('追踪报告:', JSON.stringify(report, null, 2));
}, 300000);
});
这个JavaScript看板就像电影中的"指挥中心"——实时监控、自动警报、数据可视化，每一个功能都是链上追踪的关键工具。在《万里追凶》中，如果FBI有这样的系统，主角们可能连雪地都走不出去就被抓了。
第五幕：链上追踪的伦理与局限
就像《万里追凶》中展现的"道德困境"——那笔钱本身就是赃款，追踪与被追踪之间存在着微妙的伦理边界。链上追踪同样面临着隐私与透明之间的张力。
从广播电视编导的视角来看，链上追踪就像一部纪录片——它记录的是真实发生的每一笔交易，但这种记录本身也可能被滥用。当"透明"成为区块链的核心价值，如何在追踪犯罪的同时保护普通用户的隐私，成为了一个需要平衡的命题。
画外音：镜头之外的思考
《万里追凶》的结局是悲剧性的——兄弟反目、家庭破碎，那笔从未被真正使用的钱成了所有人的诅咒。这种叙事告诉我们：追踪工具本身是中性的，关键在于使用它的人。
链上追踪技术本质上是一种"镜头语言"，它让不可见的经济活动变得可见。就像电影导演用镜头揭示人物的内心世界，链上调查员用交易数据揭示资金流动的真相。但正如任何强大的工具一样，它既可以被用来追查犯罪，也可能被用来侵犯隐私。
在这个万物皆可Token化的时代，技术的迭代往往比镜头切换更快。作为北京城市学院2021级广播电视编导的毕业生，我始终在影像与区块链的交汇处寻找共鸣。感谢阅读，我是王森涛，让我们在视听与去中心化的世界里，继续探索。
