「錢被轉走了,鏈上是匿名的,肯定追不回來了吧?」——這是很多人的誤解。區塊鏈恰恰是公開、可追溯的。這篇先用 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,你就能搭出一個最小可用的鏈上資金追蹤器。它當然比不上專业的鏈上分析平台,但足以讓你理解「鏈上追蹤」到底是怎麼一回事:公開賬本 + 圖遍歷 + 地址標簽。真正的難點不在「能不能查」,而在跨鏈斷點、混幣、以及能否觸達可配合的落地平台。也正因如此,真正的追回往往不是一個人、一段腳本能完成的;德爾泰把鏈上取證、千萬級實體標簽庫與司法協同打通,才能把“看懂”真正變成“追得回”。
風險與合規提示:本文代碼僅用于學習、研究與合法的資產自保/取證場景,請勿用于侵犯他人隱私或任何非法用途。資產被盜被騙請第一時間報案,並通過合法途徑維權。