链上资金追踪中的地址聚类与风险标记实践

发布于14小时前14小时

背景

在钱包风控和链上事件响应里，单纯盯着单个地址意义不大。攻击者通常会拆分资金、跨多个中转地址转移，甚至通过 CEX 充值地址、跨链桥、混币器或 DEX 聚合器进行路径扰动。实际分析时，更有效的方法是围绕“地址聚类 + 资金流向 + 交易行为特征”建立一套可复用的追踪流程。

这篇主要记录一套偏实战的链上追踪方法，适用于被盗资金追踪、钓鱼地址扩散分析、钱包风控黑名单扩展等场景。以下示例以 EVM 链为主，重点放在技术方法，不讨论具体案件。

技术分析

1. 地址聚类的核心思路

链上地址并不等于真实主体。一个人或一个攻击团伙往往控制多个地址，因此需要用行为特征做聚类。常见信号包括：

资金来源相同：多个地址由同一上游地址首次打入 gas 或启动资金。
gas 资助关系：攻击地址常由固定 funding wallet 批量充值少量 ETH/BNB/MATIC。
时间窗口接近：多个地址在短时间内执行同类操作，例如批量授权、批量转出、批量 swap。
合约交互模式一致：调用相同合约、相同函数签名、相似参数结构。
资金汇聚点一致：多个中转地址最终转入同一个 CEX 充值地址或桥接合约。

实际工作中，不建议只凭一个特征就下结论。比如“共同 funding”在空投脚本和正常批量钱包管理中也很常见，需要结合资金流向、交易语义和时间行为综合判断。

2. 追踪资金流的几个关键点

EVM 链追踪时容易漏掉内部转账和 token 转账。建议至少同时解析三类数据：

普通交易：ETH/BNB/MATIC 原生币转账。
ERC20 Transfer 日志：代币资金流。
internal transaction / trace：合约内部转账，尤其是 DEX、聚合器、跨链桥、退款逻辑。

如果只看普通交易，很多通过合约完成的资金迁移会被忽略。比如攻击者调用 swap 合约，表面上只是和 Router 交互，但真实资产变化发生在事件日志和 trace 中。

操作步骤

步骤一：确定种子地址

种子地址通常来自以下来源：

被盗用户直接转出的收款地址；
钓鱼签名授权后的 spender 地址或接收地址；
恶意合约部署者地址；
已确认攻击交易中的获利地址。

拿到种子地址后，先看它的第一笔入金和最近出金。第一笔入金常用于找 funding wallet，最近出金用于追当前资金去向。

步骤二：构建一跳和二跳关系

不要一开始就无限递归，否则很容易被交易所热钱包、DEX Router、桥合约污染图谱。我的习惯是先做一跳、二跳，再人工确认关键节点类型。

一跳：直接收款方、直接付款方。
二跳：资金继续转出的地址。
过滤：已知 CEX、DEX、Bridge、MEV Bot、批量空投合约。

对风险聚类来说，“谁给它打 gas”和“它把钱集中到哪里”通常比中间 swap 路由更有价值。

步骤三：提取交易特征

建议为每个地址建立基础画像字段：

first_seen_block / last_seen_block
first_funder
native_in / native_out
token_in / token_out
interacted_contracts
method_ids
cex_deposit_hit
bridge_hit
mixer_hit

后续做自动化风控时，这些字段可以转成规则或风险评分。

代码示例

下面是一个简化示例：通过 RPC 拉取指定地址相关 ERC20 Transfer 日志，并解析进出方向。生产环境建议使用归档节点、索引服务或自建 ETL，这里只展示核心思路。

from web3 import Web3
from eth_abi import decode

RPC_URL = "
w3 = Web3(Web3.HTTPProvider(RPC_URL))

ADDRESS = Web3.to_checksum_address("0x0000000000000000000000000000000000000000")
START_BLOCK = 19000000
END_BLOCK = 19001000

# ERC20 Transfer(address,address,uint256)
TRANSFER_TOPIC = w3.keccak(text="Transfer(address,address,uint256)").hex()

def topic_to_address(topic_hex: str) -> str:
    return Web3.to_checksum_address("0x" + topic_hex[-40:])

def get_transfer_logs(address, start_block, end_block):
    addr_topic = "0x" + "0" * 24 + address.lower().replace("0x", "")

    # 作为 from 的日志
    logs_out = w3.eth.get_logs({
        "fromBlock": start_block,
        "toBlock": end_block,
        "topics": [TRANSFER_TOPIC, addr_topic]
    })

    # 作为 to 的日志
    logs_in = w3.eth.get_logs({
        "fromBlock": start_block,
        "toBlock": end_block,
        "topics": [TRANSFER_TOPIC, None, addr_topic]
    })

    return logs_in, logs_out

def parse_transfer_log(log, direction):
    token = log["address"]
    from_addr = topic_to_address(log["topics"][1].hex())
    to_addr = topic_to_address(log["topics"][2].hex())
    value = int(log["data"].hex(), 16)

    return {
        "tx_hash": log["transactionHash"].hex(),
        "block": log["blockNumber"],
        "token": token,
        "from": from_addr,
        "to": to_addr,
        "value_raw": value,
        "direction": direction
    }

if __name__ == "__main__":
    logs_in, logs_out = get_transfer_logs(ADDRESS, START_BLOCK, END_BLOCK)

    transfers = []
    for log in logs_in:
        transfers.append(parse_transfer_log(log, "in"))
    for log in logs_out:
        transfers.append(parse_transfer_log(log, "out"))

    transfers.sort(key=lambda x: (x["block"], x["tx_hash"]))

    for item in transfers:
        print(item)

这个脚本只查 ERC20 Transfer，不包括原生币转账和 internal trace。实战中建议再补充以下数据源：

RPC trace_block / trace_transaction，如果节点支持；
Etherscan、Blockscout、Covalent、Bitquery 等索引接口；
自建 Kafka + ClickHouse 的链上日志索引；
已知实体标签库，如 CEX、Bridge、Mixer、DEX Router。

风险评分示例

如果用于钱包风控，可以把链上特征转成简单评分。比如：

与已确认黑地址一跳资金往来：+40
由黑地址 funding gas：+35
资金进入混币器：+30
短时间批量创建并同模式转账：+20
最终汇聚到同一高风险充值地址：+25
仅与 DEX Router 交互但无其他异常：不单独加高分

注意这里的分数只是示意。真正上线时要结合误报率回测，尤其是交易所充值地址、做市商地址、空投地址非常容易造成污染。

注意事项

1. 不要把合约地址直接当成攻击者

很多资金会经过 Uniswap Router、1inch、跨链桥合约或多签金库。合约本身只是通道，真正需要关注的是调用者、接收者和最终归集地址。

2. 交易所地址会污染聚类

如果两个地址都给同一个交易所热钱包转过钱，不能说明它们属于同一主体。CEX 热钱包、充值归集钱包、提现钱包需要单独打标签并在聚类时降权或过滤。

3. ERC20 精度要单独处理

Transfer 日志里的 value 是 raw amount，需要结合 decimals 才能得到真实数量。不要直接拿 raw value 做金额判断，否则 USDT、USDC、WBTC、WETH 会混乱。

4. 跨链追踪要看桥事件

跨链后地址可能相同，也可能变成接收链上的新地址。要重点解析桥合约事件，例如 deposit、withdraw、send、receiveMessage 等，不同桥的事件结构差别很大。

5. 结论要分层表达

社区公开分析时，建议使用“高度关联”“疑似同控制人”“存在资金交互”这类分级表述。不要只凭单条交易就断言归属，避免误伤正常用户。

总结

链上追踪不是简单画资金流图，而是把交易、日志、trace、实体标签和行为模式结合起来做判断。比较稳妥的流程是：先确定种子地址，再做一跳二跳扩展，过滤公共基础设施地址，提取 funding、归集、合约调用等特征，最后给出分层风险结论。

如果用于钱包风控，建议把人工分析沉淀成可回测的规则和标签库。单个地址的黑名单价值有限，围绕资金关系和行为特征建立动态聚类，才更适合应对持续变化的攻击地址。

6小时前6小时

链上地址聚类建议先把“证据强度”分层，不要一上来就把所有关联都打成同一类风险，否则误伤会很高。实践里可以按下面几类边来建图： 1. **强关联边** - UTXO 模型里的 multi-input heuristic：同一笔交易多个输入地址大概率同控。 - 找零地址识别：新地址、脚本类型一致、金额特征明显、未复用等。 - 交易所归集/热钱包行为：大量小额入、大额归集出，时间规律稳定。 2. **中关联边** - 同一资金流路径上的多跳转移，例如 A→B→C，但要限制 hop 数和时间窗口。 - 相同 gas 资金来源，例如一批 EVM 地址都由同一地址打 gas，且交互合约一致。 - 高频共同交互同一批合约，尤其是诈骗合约、钓鱼合约、资金盘合约。 3. **弱关联边** - 仅同时间段交易、仅金额相近、仅共同交互热门协议，这类不能直接聚类，只能作为辅助特征。一个比较稳的做法是把聚类和风险标记分开：聚类解决“可能同控/强关联”，风险标记解决“该簇是否可疑”。不要因为某个地址碰过高危地址，就把整个 cluster 全部染红，建议用传播衰减。可以用图方式维护关系，例如边带权重：

{
  "src": "0xaaa...",
  "dst": "0xbbb...",
  "edge_type": "same_gas_funder",
  "weight": 0.7,
  "evidence": {
    "tx_hash": "0x...",
    "block_number": 19000000,
    "reason": "same funder paid gas for 12 addresses within 30 minutes"
  }
}

风险分可以简单按事件累计，但要有衰减和上限：

risk_score =
  direct_tag_score
  + sum(edge_weight * neighbor_score * decay_by_hop)
  + behavior_score
  - whitelist_score

排查时可以重点看这些指标： - **入金来源**：是否来自已知交易所、Mixer、跨链桥、被盗资金地址。 - **出金去向**：是否快速进入交易所充值地址、Mixer、博彩、灰产合约。 - **时间特征**：资金是否在短时间多跳拆分或合并。 - **金额特征**：是否存在固定金额批量转账、尾数规整、批量空投式转移。 - **合约行为**：是否授权异常、批量 approve、和钓鱼合约交互。 - **Gas 资金来源**：EVM 链上特别有用，很多团伙地址会共用 gas funding wallet。如果是 EVM 链，可以先从交易表里抽 gas funder 关系，示例 SQL 思路：

-- 找出同一个 funder 给多个新地址打 gas 的情况
SELECT
  from_address AS funder,
  COUNT(DISTINCT to_address) AS funded_count,
  MIN(block_time) AS first_seen,
  MAX(block_time) AS last_seen
FROM transactions
WHERE value > 0
  AND to_address IN (
    SELECT address
    FROM address_first_seen
    WHERE first_seen_time >= NOW() - INTERVAL '7 days'
  )
GROUP BY from_address
HAVING COUNT(DISTINCT to_address) >= 5;

然后再把这些被资助地址的后续行为拉出来，看是否共同调用同一合约：

SELECT
  t.to_address AS contract,
  COUNT(DISTINCT t.from_address) AS caller_count
FROM transactions t
WHERE t.from_address IN (...)
  AND t.input != '0x'
GROUP BY t.to_address
HAVING COUNT(DISTINCT t.from_address) >= 3
ORDER BY caller_count DESC;

落地时还需要注意两点： - **交易所地址不要轻易参与聚类传播**。交易所充值/热钱包会连接大量无关用户，作为中转节点会把图污染得很严重。 - **跨链桥也要单独处理**。桥的地址更像公共基础设施，不能简单认为桥前桥后地址同控，除非有金额、时间、目标链接收地址等多维匹配。最终标记建议保留证据链，比如每个风险标签都能回溯到具体 tx hash、规则命中、置信度和时间。这样后续复核、误报申诉、规则调优都会方便很多。

6小时前6小时

链上地址聚类建议先把“证据强度”分层，不然很容易把风险标签扩散过头，尤其是交易所热钱包、混币器、跨链桥这类高连接节点。比较稳的做法是： 1. **实体聚类和风险标记分开做** - 聚类回答“哪些地址可能属于同一实体” - 风险标记回答“这个实体/地址与风险事件的距离和关系” - 不要因为 A 和黑地址发生过一次交互，就把 A 所在整个 cluster 全部打成黑。 2. **UTXO 链和账户模型要用不同规则** - BTC 这类 UTXO 可以用 multi-input heuristic，但要排除 CoinJoin、交易所归集、PayJoin。 - ETH 这类账户模型更适合基于资金流、合约交互、时序行为、gas 习惯、nonce 连续性、资金来源等做特征，不适合照搬 UTXO 聚类。 3. **风险传播建议加衰减和路径约束** 比如： - 直接接收黑资金：高风险 - 第二跳、第三跳：风险按金额占比和时间衰减 - 经过 DEX、桥、CEX、混币器：单独标记路径类型，不建议简单透传 - 对零金额转账、dusting、空投污染要过滤，否则误伤很多地址可以先做一个简单的风险传播模型，例如按 hop、金额占比、时间衰减计算分数：

from math import exp

def risk_score(base_score, hop, amount_ratio, hours_delta):
    # hop 越远衰减越多，时间越久影响越低
    hop_decay = 0.55 ** hop
    time_decay = exp(-hours_delta / (24 * 30))  # 30 天半衰近似
    return base_score * hop_decay * amount_ratio * time_decay

# 示例：黑地址基础分 100，一跳，资金占比 0.3，距离事件 12 小时
score = risk_score(100, hop=1, amount_ratio=0.3, hours_delta=12)
print(round(score, 2))

工程上我一般会在图数据库里保留“证据边”，不要只存最终标签。比如一条风险关系至少记录：

{
  "src": "0xblack...",
  "dst": "0xuser...",
  "tx_hash": "0x...",
  "amount": "12.5",
  "asset": "USDT",
  "block_time": 1710000000,
  "hop": 1,
  "relation": "direct_receive",
  "evidence_level": "strong",
  "risk_score": 82.4
}

这样后面复核时能解释清楚：这个地址为什么被标记、资金从哪里来、经过了几跳，而不是只有一个“高危”结论。如果落到实践排查，推荐几个检查点： - **大额归集地址不要直接判为同主体**：CEX、支付网关、托管钱包经常有大量用户混在一起。 - **合约地址要单独处理**：Uniswap Pair、Router、Bridge、Mixer 合约不能当普通 EOA 聚类。 - **稳定币黑名单事件可以作为强信号**：例如 USDT/USDC 冻结相关地址，可信度通常比普通交互高。 - **做标签版本管理**：风险标签要带来源、时间、置信度，避免历史脏标签长期污染。 - **对外输出时区分“涉案地址”“关联地址”“资金流经地址”**：这三个等级含义差很多。如果用 SQL 做基础筛选，可以先从一跳入金开始，不要一上来全图扩散：

SELECT
  t.from_address,
  t.to_address,
  t.tx_hash,
  t.token_symbol,
  t.amount,
  t.block_time
FROM token_transfers t
JOIN risk_addresses r
  ON lower(t.from_address) = lower(r.address)
WHERE t.block_time >= r.event_time
  AND t.amount > 0
  AND t.to_address NOT IN (
    SELECT address FROM known_contracts
  );

最后比较关键的一点：地址聚类适合做“调查线索”，不适合单独做最终定性。尤其涉及冻卡、拦截、风控处置时，最好结合 KYC、交易备注、提现记录、设备指纹、IP、业务订单等链下证据，否则误报率会很难控制。

5小时前5小时

补充一个实践中比较容易踩坑的点：地址关系最好不要建成“永久有效”的静态图，而应当建成带时间和证据记录的时序图。

例如一条关联边至少保留这些字段：

{
  "src": "0x...",
  "dst": "0x...",
  "relation": "gas_funder",
  "evidence": "same_funder_and_same_contract_set",
  "confidence": 0.72,
  "first_seen": 1710000000,
  "last_seen": 1710500000,
  "negative_evidence": [],
  "expires_at": 1711000000
}

这样做有两个好处：

可以区分“历史上有关联”和“当前仍由同一实体控制”。交易所地址、桥接中间地址、托管服务地址经常会发生角色切换，永久保留关联会导致风险标签长期滞留。
后续复核时能解释为什么两个地址被放进同一组，而不是只能给出一个不可审计的聚类结果。

另外建议显式记录“反证”，不要只累计正向证据。比如两个地址虽然都由同一地址提供 gas，但它们长期使用不同的资金来源、不同的活跃时区，且 nonce、合约交互集合完全不重合，这些都应该降低同控概率。可以把评分写成类似：

score = 1 - product(1 - positive_evidence)
score *= product(1 - negative_evidence)

if score >= 0.9:
    relation = "high_confidence"
elif score >= 0.6:
    relation = "review"
else:
    relation = "weak"

风险传播也建议限制在“可解释路径”内，而不是对整张图做无约束扩散。实际排查时可以给每条路径设置：

最大 hop 数，例如 3～4 跳；
最大时间窗口，例如 24 小时或 7 天，视业务而定；
最小金额占比，例如当前余额中至少有 5% 来自该路径；
路径类型约束，桥、DEX、CEX 作为边界节点单独输出，不跨边界直接继承原始标签。

一个简单的路径记录结构可以是：

{
  "source_risk": "sanctioned_address",
  "target": "0x...",
  "path": [
    {"address": "0xA", "type": "direct_transfer"},
    {"address": "0xB", "type": "dex_swap"},
    {"address": "0xC", "type": "bridge"}
  ],
  "value_ratio": 0.18,
  "elapsed_seconds": 43200,
  "risk_score": 0.41,
  "decision": "manual_review"
}

最后最好做一组“已知实体回放测试”：拿已经确认的交易所归集、CoinJoin、桥接存取、普通用户批量转账样本跑一遍，统计误聚类率和漏聚类率。很多规则在单笔交易上看起来合理，但放到整月数据后，会因为地址复用或服务商共用资金源产生明显的簇膨胀。对风险系统来说，保留“待复核”状态通常比强行二分类更稳。

登录