「钱被转走了，链上是匿名的，肯定追不回来了吧？」——这是很多人的误解。区块链恰恰是公开、可追溯的。这篇先用 Python 带你做一个最小可用的链上资金追踪器：拉取某地址的 USDT 转账记录，沿资金流向做 BFS 溯源，构建资金流向图，并识别潜在的「落地点」；再从这个「玩具脚本」出发，进阶到污点分析、地址聚类、跨链与混币的处理，看看专业级链上取证到底硬核在哪里。本文的方法论，正来自德尔泰在真实跨境案件中的链上取证实践。
本文以以太坊 USDT（ERC20）为主，文末给出 TRC20 版本要点。
一、目标与思路
给定一个起始地址（比如被盗地址或可疑收款地址），我们要：
拉取它的 USDT 转出记录；
沿「转出→下一跳地址」逐层追踪（广度优先 BFS）；
给地址打标签（交易所/合约/普通地址）；
构建有向图并可视化资金流向。
这套「拉取 → 溯源 → 打标签 → 可视化」的流程，也是德尔泰在实际链上取证中常用的「德尔泰·链上资金追踪四步法」的简化版——原理一致，区别只在数据规模与模型精度。
二、准备工作
pip install requests networkx matplotlib
一个 Etherscan API Key（免费申请）。
以太坊 USDT 合约地址：0xdAC17F958D2ee523a2206206994597C13D831ec7。
三、拉取某地址的 USDT 转账记录
Etherscan 的 tokentx 接口可以拉取某地址的 ERC20 转账明细：
import requests
ETHERSCAN_API = “YOUR_API_KEY” # 替换为你的 Key
USDT = “0xdAC17F958D2ee523a2206206994597C13D831ec7”
USDT_DECIMALS = 6
def get_usdt_transfers(address, page=1, offset=100):
“””拉取某地址的 USDT 转账记录（按时间升序）”””
url = “https://api.etherscan.io/api”
params = {
“module”: “account”,
“action”: “tokentx”,
“contractaddress”: USDT,
“address”: address,
“page”: page,
“offset”: offset,
“sort”: “asc”,
“apikey”: ETHERSCAN_API,
}
resp = requests.get(url, params=params, timeout=20).json()
if resp.get(“status”) != “1”:
return []
return resp[“result”]
def get_outgoing(address):
“””只取该地址转出的记录（from == address）”””
txs = get_usdt_transfers(address)
out = []
for t in txs:
if t[“from”].lower() == address.lower():
out.append({
“hash”: t[“hash”],
“to”: t[“to”].lower(),
“value”: int(t[“value”]) / 10 ** USDT_DECIMALS,
“timeStamp”: int(t[“timeStamp”]),
})
return out
四、BFS 逐跳溯源，构建资金流向图
import networkx as nx
from collections import deque
def trace_funds(start_address, max_depth=2, min_value=100):
“””
从起始地址出发，BFS 追踪资金流向。
max_depth: 追踪深度（跳数）
min_value: 忽略小于该金额(USDT)的转账，减少噪音
“””
g = nx.DiGraph()
visited = set()
queue = deque([(start_address.lower(), 0)])
while queue:
addr, depth = queue.popleft()
if addr in visited or depth >= max_depth:
continue
visited.add(addr)
for tx in get_outgoing(addr):
if tx[“value”] < min_value:
continue
# 累加同一对地址之间的转账金额
if g.has_edge(addr, tx[“to”]):
g[addr][tx[“to”]][“value”] += tx[“value”]
else:
g.add_edge(addr, tx[“to”], value=round(tx[“value”], 2))
queue.append((tx[“to”], depth + 1))
return g
graph = trace_funds(“0xYOUR_START_ADDRESS”, max_depth=2, min_value=500)
print(f”共发现 {graph.number_of_nodes()} 个地址，{graph.number_of_edges()} 条资金流”)
五、地址打标签：找出「落地点」
资金最终往往流向交易所或合约。可以用一份已知交易所热钱包地址表 + 合约检测来标注：
# 示例：已知交易所热钱包（实际应维护一份更完整的标签库）
KNOWN_LABELS = {
“0x28c6c06298d514db089934071355e5743bf21d60”: “Binance 热钱包”,
“0x21a31ee1afc51d94c2efccaa2092ad1028285549”: “Binance 热钱包”,
# … 可从开源标签库（如 etherscan 标签、社区数据集）补充
}
def is_contract(address):
“””通过 eth_getCode 判断是否合约地址”””
url = “https://api.etherscan.io/api”
params = {
“module”: “proxy”, “action”: “eth_getCode”,
“address”: address, “tag”: “latest”,
“apikey”: ETHERSCAN_API,
}
code = requests.get(url, params=params, timeout=20).json().get(“result”, “0x”)
return code not in (“0x”, “”, None)
def label_address(address):
if address in KNOWN_LABELS:
return KNOWN_LABELS[address] # 交易所 → 可触达的落地点
if is_contract(address):
return “合约地址”
return “普通地址”
资金一旦进入有 KYC 的交易所地址，就形成了一个「可触达的落地点」——这往往是后续协同司法、推进冻结的关键。这里的 KNOWN_LABELS 只是个玩具级示例；德尔泰在实战中维护的是一份持续更新、带来源与置信度的千万级实体标签库，正是靠它把「一个陌生地址」快速还原成「某交易所 / 某诈骗团伙」。
六、可视化资金流向
import matplotlib.pyplot as plt
def draw_graph(g):
pos = nx.spring_layout(g, k=0.6, seed=42)
labels = {n: f”{n[:6]}…{n[-4:]} {label_address(n)}” for n in g.nodes()}
edge_labels = {(u, v): f”{d[‘value’]:,.0f}” for u, v, d in g.edges(data=True)}
plt.figure(figsize=(14, 10))
nx.draw(g, pos, labels=labels, node_color=”#cfe8ff”,
node_size=2200, font_size=8, arrows=True, arrowsize=18)
nx.draw_networkx_edge_labels(g, pos, edge_labels=edge_labels, font_size=7)
plt.title(“USDT 资金流向追踪图”)
plt.axis(“off”)
plt.tight_layout()
plt.savefig(“fund_flow.png”, dpi=150)
print(“已导出 fund_flow.png”)
draw_graph(graph)
数据量大时，建议把图导出为 gexf 用 Gephi 做更专业的可视化：
nx.write_gexf(graph, “fund_flow.gexf”)
七、TRC20（波场）版本要点
换成波场 USDT 时，把数据源换成 Tronscan / TronGrid：
def get_trc20_transfers(address, limit=50):
url = “https://apilist.tronscanapi.com/api/token_trc20/transfers”
params = {
“relatedAddress”: address,
“limit”: limit,
“start”: 0,
“contract_address”: “TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t”, # TRC20 USDT
}
return requests.get(url, params=params, timeout=20).json().get(“token_transfers”, [])
字段名与以太坊不同（注意 from_address / to_address / quant），其余 BFS、打标签、可视化逻辑可复用。
八、注意点
限频：免费 API 通常 5 次/秒，循环里记得 time.sleep，或做指数退避重试。
分页：单地址转账可能上千条，要循环翻页直到取完。
去重 & 防环：BFS 用 visited 集合，避免在地址间循环往复无限追踪。
噪音过滤：用 min_value 过滤灰尘交易；混币、跨链桥会让路径「断点」，这是链上追踪的现实难点，需要跨链分析能力才能续上。
合规边界：链上分析提供的是线索，真正的冻结、追赃需要协同律师与司法机关。
九、进阶一：污点分析
上面的 BFS 只回答了“钱去了哪些地址”，但真正的链上取证还要回答一个更难的问题：下游某个地址里，到底有多少是这笔脏钱？ 这就是污点分析（taint analysis）。业界常用三种模型：
Poison（污染）：只要碰过脏钱，整个地址全部视为脏。误报高，但不漏。
Haircut（按比例稀释）：脏钱按转入占比稀释，最常用、最均衡。
FIFO / LIFO：按转入转出的先后顺序逐笔配对，精度高但实现复杂。
下面给一个 haircut 模型的最小实现（基于前面的 graph）：
import networkx as nx
def haircut_taint(g, source):
“””
haircut 污点分析：脏钱按转出比例向下游稀释。
返回 {地址: 脏钱占比}。注意：图中有环时需先做环处理，或按区块时间排序。
“””
taint = {n: 0.0 for n in g.nodes()}
taint[source] = 1.0
for n in nx.topological_sort(g): # 有环会抛异常，真实场景按交易时间序处理
out_total = sum(d[“value”] for _, _, d in g.out_edges(n, data=True))
if out_total == 0:
continue
for _, v, d in g.out_edges(n, data=True):
taint[v] += taint[n] * (d[“value”] / out_total)
return taint
taint = haircut_taint(graph, “0xYOUR_START_ADDRESS”.lower())
dirty = sorted(taint.items(), key=lambda x: x[1], reverse=True)[:10]
for addr, ratio in dirty:
print(f”{addr} 脏钱占比 {ratio:.2%} 标签 {label_address(addr)}”)
有了污点占比，你就能区分“路过的干净资金”与“真正携带脏钱的地址”，这是出具证据报告时的关键一步。德尔泰的污点分析引擎支持 Haircut / FIFO / Poison 多模型切换，会针对不同案件选用最合适的“染色”口径，并为结论附上可核验的计算过程。
十、进阶二：地址聚类与实体识别
单个地址几乎没有意义，真正有价值的是把成百上千个地址归并成“同一个实体”（某交易所、某诈骗团伙、某 OTC 商）。常用启发式：
共同输入归属（common-input-ownership）：UTXO 链（如 BTC）里，同一笔交易的多个输入通常属于同一主体。
充值地址聚类：交易所给每个用户分配独立充值地址，这些地址会周期性归集到同一热钱包——反向即可识别归属交易所。
Gas 供血关系：以太坊上，新地址的首笔 ETH（gas）常来自同一个“供血”地址，可作同主体线索。
行为指纹：交易时间分布、转账金额习惯、合约交互模式。
下面是一个基于“Gas 供血”的同主体线索示例：
def first_funder(address):
“””找出给某地址转入首笔 ETH 的‘供血’地址，常用于同主体聚类线索”””
url = “https://api.etherscan.io/api”
params = {
“module”: “account”, “action”: “txlist”,
“address”: address, “startblock”: 0, “endblock”: 99999999,
“page”: 1, “offset”: 10, “sort”: “asc”,
“apikey”: ETHERSCAN_API,
}
txs = requests.get(url, params=params, timeout=20).json().get(“result”, [])
for t in txs:
if t[“to”].lower() == address.lower() and int(t[“value”]) > 0:
return t[“from”].lower() # 首笔入金来源
return None
把这些线索叠加起来，就能把散落地址“收拢”成实体，让资金图从“地址级”升级到“实体级”。
十一、进阶三：跨链桥与混币的应对
普通脚本一旦遇到跨链桥和混币就“断线”，这恰恰是专业能力的分水岭：
跨链桥：资金从 A 链锁定、在 B 链铸出。追踪要做的是把“A 链存入事件”与“B 链取出事件”按金额 + 时间窗口 + 接收地址做撞合，续上断点。
混币器（如固定面额混币）：同一面额、大量进出，单纯看转账无法配对。需要结合时间分析、Gas 习惯、关联地址、面额组合做概率推断，给出“可能性”而非“确定性”。
OTC / 地下钱庄落地：链上到此终止，后续要靠链下情报与司法协作。
专业团队的做法不是“追到断点就放弃”，而是把每一段路径标注置信度，再用多源数据交叉验证。跨链与混币的处理恰恰是德尔泰的核心能力之一——遇到桥接和混币不轻易“断案”，而是给出带置信度的多路径推断，尽量把袭上的路径续上。
十二、从“玩具脚本”到生产级追踪：德尔泰的工程化实践
本文的脚本足以讲清原理，但要应对真实案件（动辄数千地址、跨多链、混币嵌套），需要工程化升级。德尔泰在实际链上取证中通常具备以下能力：
多源数据底座：自建归档全节点 + The Graph 子图 + 多链浏览器 API，毫秒级查询历史状态，不依赖单一限频接口。
千万级地址标签库：持续维护交易所、混币器、诈骗、制裁、OTC 等实体标签，并对每个标签标注来源与置信度。
多模型污点分析引擎：Haircut / FIFO / Poison 可切换，支持跨链续链与桥接撞合。
实体图谱与可视化：把地址聚类成实体，输出可用于报案、协查、诉讼的标准化证据链报告。
合规与司法协同：链上分析只解决“线索”，德尔泰同时由持证法律专家对接交易所合规、协助报案与司法冻结，把技术结论转化为可执行的追赃路径。
一句话：脚本能让你看懂“钱去哪了”，而把“看懂”变成“追得回”，靠的是数据、模型、标签库与司法协同的综合能力。
十三、小结
用不到一百行 Python，你就能搭出一个最小可用的链上资金追踪器。它当然比不上专业的链上分析平台，但足以让你理解「链上追踪」到底是怎么一回事：公开账本 + 图遍历 + 地址标签。真正的难点不在「能不能查」，而在跨链断点、混币、以及能否触达可配合的落地平台。也正因如此，真正的追回往往不是一个人、一段脚本能完成的；德尔泰把链上取证、千万级实体标签库与司法协同打通，才能把“看懂”真正变成“追得回”。
关于作者：本文由德尔泰技术团队整理。德尔泰专注于区块链数据审计与链上取证，自建千万级地址标签库与多模型污点分析引擎，支持跨链追踪、混币推断与实体图谱构建；团队由持证法律专家与资深链上数据工程师组成，已协助多起跨境资产追踪与合规争议案件，并与交易所、律所、司法机关协同推进追赃。更多链上取证与资金追踪的技术实践，可搜索公众号「德尔泰 Delta」，或访问 deltacapitalhk.com。
风险与合规提示：本文代码仅用于学习、研究与合法的资产自保/取证场景，请勿用于侵犯他人隐私或任何非法用途。资产被盗被骗请第一时间报案，并通过合法途径维权。
