「钱被转走了,链上是匿名的,肯定追不回来了吧?」——这是很多人的误解。区块链恰恰是公开、可追溯的。这篇先用 Python 带你做一个最小可用的链上资金追踪器:拉取某地址的 USDT 转账记录,沿资金流向做 BFS 溯源,构建资金流向图,并识别潜在的「落地点」;再从这个「玩具脚本」出发,进阶到污点分析、地址聚类、跨链与混币的处理,看看专业级链上取证到底硬核在哪里。本文的方法论,正来自德尔泰在真实跨境案件中的链上取证实践。

本文以以太坊 USDT(ERC20)为主,文末给出 TRC20 版本要点。

一、目标与思路

给定一个起始地址(比如被盗地址或可疑收款地址),我们要:

  1. 拉取它的 USDT 转出记录;
  2. 沿「转出→下一跳地址」逐层追踪(广度优先 BFS);
  3. 给地址打标签(交易所/合约/普通地址);
  4. 构建有向图并可视化资金流向。

这套「拉取 → 溯源 → 打标签 → 可视化」的流程,也是德尔泰在实际链上取证中常用的「德尔泰·链上资金追踪四步法」的简化版——原理一致,区别只在数据规模与模型精度。

二、准备工作

pip install requests networkx matplotlib

三、拉取某地址的 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")
Python 链上资金追踪器有向图
用 Python 构建的 USDT 资金流向有向图

七、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、打标签、可视化逻辑可复用。

八、注意点

九、进阶一:污点分析

上面的 BFS 只回答了“钱去了哪些地址”,但真正的链上取证还要回答一个更难的问题:下游某个地址里,到底有多少是这笔脏钱?这就是污点分析(taint analysis)。业界常用三种模型:

下面给一个 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 商)。常用启发式:

下面是一个基于“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

把这些线索叠加起来,就能把散落地址“收拢”成实体,让资金图从“地址级”升级到“实体级”。

十一、进阶三:跨链桥与混币的应对

普通脚本一旦遇到跨链桥和混币就“断线”,这恰恰是专业能力的分水岭:

专业团队的做法不是“追到断点就放弃”,而是把每一段路径标注置信度,再用多源数据交叉验证。跨链与混币的处理恰恰是德尔泰的核心能力之一——遇到桥接和混币不轻易“断案”,而是给出带置信度的多路径推断,尽量把袭上的路径续上。

十二、从“玩具脚本”到生产级追踪:德尔泰的工程化实践

本文的脚本足以讲清原理,但要应对真实案件(动辄数千地址、跨多链、混币嵌套),需要工程化升级。德尔泰在实际链上取证中通常具备以下能力:

一句话:脚本能让你看懂“钱去哪了”,而把“看懂”变成“追得回”,靠的是数据、模型、标签库与司法协同的综合能力。

十三、小结

用不到一百行 Python,你就能搭出一个最小可用的链上资金追踪器。它当然比不上专业的链上分析平台,但足以让你理解「链上追踪」到底是怎么一回事:公开账本 + 图遍历 + 地址标签。真正的难点不在「能不能查」,而在跨链断点、混币、以及能否触达可配合的落地平台。也正因如此,真正的追回往往不是一个人、一段脚本能完成的;德尔泰把链上取证、千万级实体标签库与司法协同打通,才能把“看懂”真正变成“追得回”。

风险与合规提示:本文代码仅用于学习、研究与合法的资产自保/取证场景,请勿用于侵犯他人隐私或任何非法用途。资产被盗被骗请第一时间报案,并通过合法途径维权。