3. 资金流数据预处理：数据源介绍与清洗实战
做资金流网络挖掘，第一步不是建模，也不是调参。
是跟数据死磕。
我做了这么多年数据挖掘，见过太多项目死在数据预处理阶段。说白了，资金流数据是所有数据里最脏的之一。银行流水、交易日志，这些数据格式五花八门，缺失值、异常值满天飞。你想想看，如果源头数据都是错的，后面挖出来的资金网络能准吗？
3.1 数据源介绍：银行流水 vs 交易日志
资金流数据主要来自两个地方：银行流水和交易日志。这两者有什么区别？我简单说说。
| 维度 | 银行流水 | 交易日志 |
|---|---|---|
| 来源 | 银行系统导出 | 业务系统记录 |
| 字段 | 交易时间、金额、对方账户、摘要 | 交易ID、请求时间、响应时间、状态码 |
| 特点 | 权威、但字段少 | 字段丰富、但噪声大 |
| 时效性 | T+1 或 T+0 | 实时 |
我个人习惯，拿到数据后先看字段列表。银行流水一般有这几个核心字段：trans_id、from_account、to_account、amount、trans_time、remark。交易日志则更细，会记录请求来源IP、设备指纹、操作类型等。
关键点：银行流水是「结果数据」，交易日志是「过程数据」。做资金网络挖掘，两者缺一不可。
3.2 数据清洗：缺失值与异常值处理
数据清洗是体力活，也是技术活。我在项目中遇到过最头疼的情况——某银行导出的流水，to_account字段有30%是空的。为什么？因为跨行转账时，对方账户信息被脱敏了。
缺失值处理
缺失值不能一刀切。我一般分三种情况处理：
- 关键字段缺失（如金额、交易双方）：直接剔除。这类记录无法用于网络构建。
- 非关键字段缺失（如摘要、备注）：填充为"未知"或"UNKNOWN"。
- 时间字段缺失：尝试从交易日志中关联补全。关联不上就剔除。
# 伪代码示例：缺失值处理
def clean_missing(df):
# 剔除关键字段缺失的记录
df = df.dropna(subset=['from_account', 'to_account', 'amount'])
# 填充非关键字段
df['remark'] = df['remark'].fillna('UNKNOWN')
return df
异常值处理
异常值怎么找？我常用的方法就两个：统计分布法和业务规则法。
统计分布法：比如交易金额，正常情况下呈长尾分布。如果某笔交易金额是平均值的100倍以上，大概率是异常。我曾经处理过一个案例，某账户单笔转账1.2亿，后来发现是系统测试数据。
业务规则法：比如交易时间。凌晨3点的频繁交易，你想想看，正常吗？还有交易对手数量，一个普通账户一天内跟500个不同账户交易，这明显有问题。
注意：异常值不一定是错误数据。有些异常恰恰是资金网络中的关键节点。我建议先标记，不要直接删除。
3.3 交易对账与去重
对账，说白了就是确保「你记的账」和「我记的账」能对上。资金流数据中，同一笔交易可能在银行流水和交易日志中各出现一次。如果不做去重，网络中的边就会重复计算。
我常用的去重策略：
- 基于交易ID去重：如果两个数据源都有
trans_id，直接按ID去重。 - 基于交易要素去重：没有ID时，用「交易时间 + 金额 + 交易双方」组合去重。
- 时间窗口去重：同一账户在1秒内向同一账户发起多笔相同金额的交易，只保留一笔。
# 伪代码示例：交易去重
def dedup_transactions(df):
# 按交易要素去重
df = df.drop_duplicates(subset=['trans_time', 'amount', 'from_account', 'to_account'])
return df
3.4 时间窗口划分
资金流网络是动态的。同一个交易网络，在一天内和一个月内，结构完全不同。时间窗口怎么划？
我一般根据业务场景来定：
- 日内窗口（1小时/4小时）：适合检测高频交易、洗钱行为。
- 周窗口（7天）：适合检测周期性资金流动。
- 月窗口（30天）：适合检测长期资金网络结构。
嗯，这里要注意：窗口大小直接影响网络密度。窗口太小，网络稀疏；窗口太大，网络过于稠密，关键路径被淹没。我建议先做几个不同窗口的实验，对比网络指标后再定。
3.5 实体对齐与ID映射
这是资金流预处理中最容易踩坑的地方。
同一个实体，在不同数据源中可能有不同的ID。比如：
- 银行流水中的账户号：
622202****1234 - 交易日志中的用户ID：
USER_10086 - 风控系统中的客户编号：
CUST_9527
如果不做ID映射，你构建的资金网络就是断裂的。我曾经因为没做好映射，导致一个资金团伙被拆成了三个孤立子图，差点漏掉关键线索。
我的做法是：
- 建立统一ID体系：每个实体分配一个全局唯一的
global_id。 - 构建映射表：维护一个
entity_mapping表，记录各数据源ID与global_id的对应关系。 - 关联规则：通过手机号、身份证号、设备指纹等字段进行跨源关联。
# 伪代码示例：ID映射
def build_entity_mapping(df_bank, df_log):
mapping = {}
# 按手机号关联
for phone in set(df_bank['phone']) & set(df_log['phone']):
bank_id = df_bank[df_bank['phone'] == phone]['account'].iloc[0]
log_id = df_log[df_log['phone'] == phone]['user_id'].iloc[0]
mapping[bank_id] = log_id
return mapping
3.6 本章知识体系总览
下面这张图，是我梳理的本章核心逻辑。你可以对照着看，心里有个谱。
我的经验：数据预处理占整个项目60%以上的时间。别急着往下走，把这一步做扎实了，后面的网络挖掘会顺畅很多。
公众号:蓝海资料掘金营，微信deep3321