4. 数据获取与预处理：从CSV/数据库获取历史交易数据、数据清洗、缺失值处理、时间序列对齐
数据是量化回测的基石。在坏账率哨兵监控场景中，原始数据通常来自交易系统导出的CSV文件或关系型数据库。本章将系统讲解如何从不同数据源获取历史交易数据，并完成清洗、缺失值处理与时间序列对齐，为后续回测与实时监控提供干净、连续、可计算的数据集。
4.1 数据源接入：CSV与数据库
根据数据规模与更新频率，选择不同的接入方式：
| 数据源类型 | 适用场景 | 读取方式 | 注意事项 |
|---|---|---|---|
| CSV文件 | 历史批量数据、离线回测 | pandas.read_csv() |
编码、分隔符、日期格式 |
| MySQL/PostgreSQL | 在线交易系统、增量更新 | pandas.read_sql() + SQLAlchemy |
连接池、SQL注入、时区 |
| ClickHouse | 高吞吐、列式存储、实时查询 | clickhouse-driver + DataFrame |
批量拉取、分区裁剪 |
实践建议：在回测阶段优先使用CSV快照，保证数据可复现；在生产监控阶段切换为数据库实时查询，并加入数据版本校验。
4.1.1 CSV读取与字段标准化
import pandas as pd
# 读取CSV，指定日期列与解析格式
df = pd.read_csv(
'transactions_2024.csv',
parse_dates=['trade_date'],
dtype={
'order_id': str,
'user_id': str,
'amount': float,
'status': str
},
encoding='utf-8'
)
# 统一列名风格
df.columns = [col.strip().lower().replace(' ', '_') for col in df.columns]
print(df.info())4.1.2 数据库批量拉取
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://user:pass@host:3306/trade_db')
query = """
SELECT order_id, user_id, trade_date, amount, status
FROM transactions
WHERE trade_date >= '2024-01-01' AND trade_date < '2025-01-01'
ORDER BY trade_date
"""
df_db = pd.read_sql(query, engine, parse_dates=['trade_date'])
print(f"拉取记录数: {len(df_db)}")4.2 数据清洗：脏数据识别与处理
原始交易数据常包含以下脏数据，需逐项清洗：
- 重复记录：同一订单ID出现多次，需去重保留最新状态。
- 异常值：交易金额为负、金额超过阈值（如单笔>1亿）、状态字段拼写错误。
- 格式不一致：日期格式混用（2024/01/01 vs 2024-01-01）、用户ID大小写不统一。
- 逻辑矛盾：交易时间早于用户注册时间、坏账标记与还款记录冲突。
清洗流水线示例：
# 1. 去重：保留最后一条记录
df = df.drop_duplicates(subset='order_id', keep='last')
# 2. 过滤异常金额
df = df[(df['amount'] > 0) & (df['amount'] < 1e8)]
# 3. 标准化状态字段
valid_status = {'normal', 'overdue', 'bad_debt', 'settled'}
df = df[df['status'].isin(valid_status)]
# 4. 日期格式统一
df['trade_date'] = pd.to_datetime(df['trade_date'], errors='coerce')
# 5. 删除逻辑矛盾记录
df = df[df['trade_date'] >= df['user_reg_date']]4.3 缺失值处理策略
交易数据中的缺失值通常出现在以下字段：
| 字段 | 缺失原因 | 处理策略 |
|---|---|---|
amount |
系统异常未记录 | 用该用户历史均值填充，或标记后剔除 |
status |
流程中断 | 根据前后状态推断，或设为'unknown' |
user_reg_date |
用户信息缺失 | 从用户维度表关联补齐 |
repay_date |
未还款 | 保留NaN，在坏账计算中特殊处理 |
注意：在坏账率计算中，缺失的还款日期不应随意填充，应视为“未还款”状态，直接参与逾期天数计算。随意填充会扭曲坏账率指标。
4.4 时间序列对齐
回测要求所有数据在统一的时间频率上对齐。交易数据通常是事件驱动（非等间隔），需要重采样为固定频率（如日频、周频）。
4.4.1 重采样与聚合
# 设置日期索引
df.set_index('trade_date', inplace=True)
# 按日重采样，聚合坏账金额与笔数
daily_agg = df.resample('D').agg({
'amount': 'sum',
'order_id': 'count',
'bad_debt_flag': 'sum' # 1表示坏账
}).rename(columns={
'amount': 'total_amount',
'order_id': 'total_orders',
'bad_debt_flag': 'bad_debt_orders'
})
# 计算日坏账率
daily_agg['bad_rate'] = daily_agg['bad_debt_orders'] / daily_agg['total_orders']
# 填充缺失日期（无交易日期补0）
full_index = pd.date_range(start='2024-01-01', end='2024-12-31', freq='D')
daily_agg = daily_agg.reindex(full_index, fill_value=0)
print(daily_agg.head())4.4.2 多资产时间对齐
当回测涉及多个资产或用户群组时，需对齐各自的时间序列：
# 假设有多个用户组的时间序列
grouped = df.groupby('user_group').resample('D').agg({
'amount': 'sum',
'bad_debt_flag': 'sum'
}).reset_index()
# 透视成宽表：行为日期，列为用户组
pivot_table = grouped.pivot_table(
index='trade_date',
columns='user_group',
values='bad_debt_flag',
aggfunc='sum'
).fillna(0)
print(pivot_table.head())4.5 数据质量校验清单
在数据进入回测引擎前，建议执行以下校验：
- 完整性：检查日期范围是否覆盖回测区间，无连续缺失超过3天。
- 一致性：坏账标记与还款记录逻辑一致，无“已还款”仍标记为坏账。
- 准确性：抽样核对原始交易金额与聚合后总额。
- 时效性：数据时间戳在合理范围内，无未来日期。
自动化校验脚本：建议将上述校验封装为
validate_dataset()函数，在每次数据加载后自动执行，输出校验报告。若校验失败，则阻止回测启动并告警。
完成数据获取与预处理后，下一阶段将进入特征工程与坏账率指标计算。干净、对齐的数据集是后续所有量化分析的基础，请务必在项目初期建立标准化的数据管道。