3. 数据采集与预处理：数据源介绍（CSV、数据库）、缺失值处理、重复值处理、数据类型转换
在构建异常交易检测模型之前，数据采集与预处理是决定模型质量的基础环节。原始数据通常来自异构数据源，包含大量噪声、缺失和冗余信息。本章将系统讲解如何从 CSV 文件和关系型数据库中采集交易样本，并完成缺失值、重复值处理及数据类型转换，确保输入模型的数据干净、一致、可用。
3.1 数据源介绍
异常交易样本的常见来源包括离线文件（CSV）和在线数据库。两种数据源在格式、访问方式和适用场景上存在显著差异。
| 特性 | CSV 文件 | 关系型数据库（如 MySQL / PostgreSQL） |
|---|---|---|
| 存储格式 | 纯文本，逗号分隔，无结构约束 | 结构化表，字段类型、约束、索引完整 |
| 访问方式 | 文件 I/O 读取，适合单机或小批量 | SQL 查询，支持并发、事务、权限控制 |
| 数据量级 | 通常 ≤ 1GB，适合原型验证 | 可扩展至 TB 级，适合生产环境 |
| 元数据 | 无或仅首行表头 | 完整的数据字典、字段注释、关系约束 |
| 典型场景 | 历史交易快照、离线分析、模型训练样本 | 实时交易流水、在线特征计算、增量更新 |
实践建议：在课程项目中，我们使用 CSV 作为初始样本导入工具，便于快速迭代；生产级风控系统则必须从数据库直接拉取，以保证数据一致性和时效性。
3.2 缺失值处理
缺失值是异常交易数据中的常见问题，可能由系统故障、字段可选、数据合并遗漏等原因造成。处理缺失值前，必须先识别缺失模式。
3.2.1 缺失模式识别
- 完全随机缺失（MCAR）：缺失与观测数据无关，如网络中断导致部分记录未写入。
- 随机缺失（MAR）：缺失依赖于其他观测变量，如高金额交易更可能缺失风控评分字段。
- 非随机缺失（MNAR）：缺失与缺失值本身有关，如欺诈交易故意隐藏用户 IP。
3.2.2 常用处理方法
| 方法 | 适用场景 | 注意事项 |
|---|---|---|
| 删除缺失行 | 缺失比例 < 5%，且为 MCAR | 可能丢失有效样本，导致样本偏差 |
| 均值/中位数填充 | 数值型特征，缺失比例 ≤ 20% | 降低方差，可能削弱异常检测能力 |
| 众数填充 | 分类型特征 | 可能引入模式偏差 |
| 前向/后向填充 | 时间序列交易数据 | 仅适用于有序数据 |
| 模型预测填充 | 缺失比例较高，特征间关联强 | 计算成本高，需避免数据泄露 |
# 示例：使用 pandas 进行缺失值处理
import pandas as pd
import numpy as np
# 读取交易样本
df = pd.read_csv('transaction_sample.csv')
# 查看缺失比例
missing_ratio = df.isnull().mean()
print("缺失比例:\n", missing_ratio[missing_ratio > 0])
# 删除缺失比例超过 50% 的列
df.drop(columns=df.columns[df.isnull().mean() > 0.5], inplace=True)
# 数值列用中位数填充
num_cols = df.select_dtypes(include=[np.number]).columns
df[num_cols] = df[num_cols].fillna(df[num_cols].median())
# 分类型用众数填充
cat_cols = df.select_dtypes(include=['object']).columns
for col in cat_cols:
df[col].fillna(df[col].mode()[0], inplace=True)
注意：在风控场景中，缺失值本身可能携带信息。例如，缺失“交易 IP”字段的记录，欺诈概率往往更高。建议在填充前先创建“是否缺失”的指示特征，保留缺失信号。
3.3 重复值处理
重复值会导致模型对相同样本过度学习，降低泛化能力，并扭曲统计分布。异常交易数据中的重复可能来自系统重试、数据合并、恶意刷单等。
3.3.1 重复类型
- 完全重复：所有字段完全相同，通常为系统冗余。
- 部分重复：关键字段（交易 ID、时间、金额）相同，但非关键字段有差异。
- 业务重复：同一笔交易被多次记录，如支付网关与银行流水重复。
3.3.2 处理策略
| 策略 | 操作 | 风控场景建议 |
|---|---|---|
| 完全去重 | 保留第一条，删除其余 | 适用于系统冗余导致的重复 |
| 基于关键字段去重 | 按交易 ID、时间、金额分组，保留最新或最完整记录 | 适用于业务重复 |
| 聚合去重 | 对重复组进行统计聚合（计数、均值） | 适用于需要保留重复频次信息的场景 |
# 示例：重复值检测与处理
# 检测完全重复行
duplicates = df.duplicated()
print(f"完全重复行数: {duplicates.sum()}")
# 删除完全重复
df.drop_duplicates(inplace=True)
# 基于关键字段去重：保留交易时间最新的记录
df.sort_values('transaction_time', ascending=False, inplace=True)
df.drop_duplicates(subset=['transaction_id', 'amount'], keep='first', inplace=True)
3.4 数据类型转换
数据类型不匹配是数据预处理中的常见陷阱。CSV 文件读取后，所有字段默认可能被识别为字符串；数据库导入后，时间戳、金额、分类编码等字段也需要显式转换。
3.4.1 常见转换场景
| 原始类型 | 目标类型 | 转换原因 | 示例 |
|---|---|---|---|
| 字符串 | datetime | 时间序列分析、窗口聚合 | '2025-01-15 14:30:00' → datetime |
| 字符串 | float / int | 数值计算、统计建模 | '1,234.56' → 1234.56 |
| 字符串 | category | 分类特征编码、减少内存 | 'high', 'medium', 'low' → category |
| int | bool | 二值标志位 | 0 / 1 → False / True |
# 示例：数据类型转换
# 时间戳转换
df['transaction_time'] = pd.to_datetime(df['transaction_time'])
# 金额字段：去除千分位逗号并转为浮点数
df['amount'] = df['amount'].str.replace(',', '').astype(float)
# 分类编码
df['risk_level'] = df['risk_level'].astype('category')
# 二值标志位
df['is_fraud'] = df['is_fraud'].astype(bool)
# 验证转换后的数据类型
print(df.dtypes)
3.5 数据预处理流程总览
以下 SVG 图展示了从原始数据源到干净训练样本的完整预处理流水线，包含本章涉及的三个核心步骤及其顺序关系。
流程要点：预处理顺序不可随意调换。先处理缺失值，再处理重复值，最后进行类型转换，可以避免因类型错误导致的去重失败或填充异常。建议在每一步后执行数据质量检查，确保转换结果符合预期。
3.6 数据质量检查清单
完成预处理后，建议使用以下清单验证数据质量，确保样本可用于模型训练。
- 完整性：所有关键字段无缺失值，缺失比例控制在 5% 以内。
- 唯一性：交易 ID 或业务主键无重复，重复率低于 0.1%。
- 一致性：金额、时间等数值字段类型正确，无异常格式。
- 时效性：时间戳字段在合理范围内，无未来时间或过旧数据。
- 平衡性：正负样本比例初步统计，为后续采样策略提供依据。
# 数据质量检查示例
def check_data_quality(df):
report = {}
report['total_rows'] = len(df)
report['missing_ratio'] = df.isnull().mean().max()
report['duplicate_ratio'] = df.duplicated().mean()
report['dtypes'] = df.dtypes.to_dict()
return report
quality_report = check_data_quality(df)
print(quality_report)