第三章 数据清洗与预处理：把脏数据变成可分析的DataFrame
说实话，很多做量化交易的朋友，一上来就急着写策略、跑回测。但我得泼盆冷水——你拿到的原始资金流数据，十有八九是脏的。我见过太多人，模型跑得飞起，结果一查，数据里全是坑。嗯，这一章我们就来聊聊怎么把这些「脏数据」洗干净。
3.1 原始数据长什么样？
先看看我们拿到的原始数据。通常来自交易所或数据服务商，格式大概是这样的：
日期,股票代码,名称,主力净流入,主力净流入占比,超大单净流入,大单净流入,中单净流入,小单净流入
2024-01-02,600519,贵州茅台,352000000,0.12,210000000,142000000,-158000000,-194000000
2024-01-02,000858,五粮液,,0.08,95000000,52000000,-68000000,-79000000
2024-01-03,600519,贵州茅台,NaN,0.15,280000000,150000000,-190000000,-240000000
2024-01-04,000858,五粮液,9999999999,0.25,500000000,4999999999,-300000000,-200000000你看，这里就有几个典型问题：
- 缺失值：第二行主力净流入是空的，第三行直接写了NaN
- 异常值：第四行主力净流入是9999999999，明显是数据错误
- 格式不统一：有的字段是字符串，有的是数字，混在一起
核心原则：数据清洗不是「删掉所有问题数据」，而是「理解问题原因后做合理处理」。我个人的习惯是，先保留原始数据副本，再逐步清洗。
3.2 第一步：加载数据并初步探查
我用Pandas来加载数据。这里有个小技巧——别直接用默认参数，先看看数据的基本情况。
import pandas as pd
import numpy as np
# 加载原始数据
df = pd.read_csv('fund_flow_raw.csv', encoding='utf-8')
# 快速探查
print('数据形状:', df.shape)
print('列名:', df.columns.tolist())
print('数据类型:\n', df.dtypes)
print('前5行:\n', df.head())
print('缺失值统计:\n', df.isnull().sum())输出结果会告诉你：哪些列有缺失值，哪些列类型不对。比如「主力净流入」这一列，如果显示为object类型而不是float64，那就要小心了——里面可能混入了字符串。
我的经验：我曾经遇到过一个数据集，主力净流入列里混入了「--」这样的占位符，导致整列被识别为字符串。用pd.to_numeric()加errors='coerce'可以快速转换并标记异常值。
3.3 第二步：处理缺失值
缺失值处理没有银弹。我一般分三种情况：
| 缺失情况 | 处理方法 | 适用场景 |
|---|---|---|
| 少量缺失（<5%） | 直接删除该行 | 数据量大，缺失随机 |
| 中等缺失（5%-20%） | 填充均值/中位数 | 资金流数据波动不大 |
| 大量缺失（>20%） | 用前向填充或插值 | 时间序列数据 |
具体代码实现：
# 方法1：删除缺失行
df_clean = df.dropna(subset=['主力净流入', '超大单净流入'])
# 方法2：用中位数填充（我推荐中位数，因为资金流数据常有极端值）
median_value = df['主力净流入'].median()
df['主力净流入'].fillna(median_value, inplace=True)
# 方法3：前向填充（适合连续交易日数据）
df['主力净流入'].fillna(method='ffill', inplace=True)
# 方法4：线性插值（更平滑）
df['主力净流入'].interpolate(method='linear', inplace=True)避坑指南：我曾经在回测时用了均值填充，结果发现某只股票停牌期间的数据被填成了均值，导致策略在复牌后产生了虚假信号。后来我改用前向填充，效果好了很多。记住：时间序列数据，优先考虑时间维度的填充方法。
3.4 第三步：识别并处理异常值
异常值在资金流数据里很常见。比如某只股票一天内主力净流入突然变成100亿，这明显不合理。我通常用两种方法识别：
3.4.1 基于统计的方法
# 用IQR（四分位距）识别异常值
Q1 = df['主力净流入'].quantile(0.25)
Q3 = df['主力净流入'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
# 标记异常值
df['is_outlier'] = (df['主力净流入'] < lower_bound) | (df['主力净流入'] > upper_bound)
# 查看异常值
outliers = df[df['is_outlier']]
print(f'发现 {len(outliers)} 个异常值')3.4.2 基于业务逻辑的方法
统计方法有时会误判。比如茅台一天净流入5亿，对茅台来说是正常的，但对小盘股就是异常。所以我更常用业务逻辑：
# 设定合理的业务阈值
# 单只股票单日主力净流入不超过总市值的1%
df['市值'] = df['收盘价'] * df['总股本']
max_reasonable = df['市值'] * 0.01
# 标记异常
df['is_abnormal'] = abs(df['主力净流入']) > max_reasonable
# 处理：用前一日数据替换
df.loc[df['is_abnormal'], '主力净流入'] = df['主力净流入'].shift(1)核心思路：异常值处理的关键是「理解业务」。你想想看，一只股票一天内主力净流入超过市值的1%，这概率有多大？我一般会结合涨跌幅、成交量一起判断，多维度交叉验证。
3.5 第四步：格式统一与类型转换
数据清洗的最后一步，是把所有字段变成可分析的格式。我习惯这样做：
# 日期列转成datetime类型
df['日期'] = pd.to_datetime(df['日期'])
# 金额列转成数值类型（强制转换，错误值变NaN）
amount_cols = ['主力净流入', '超大单净流入', '大单净流入', '中单净流入', '小单净流入']
for col in amount_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
# 百分比列转成浮点数（去掉百分号）
df['主力净流入占比'] = df['主力净流入占比'].str.replace('%', '').astype(float) / 100
# 设置日期为索引（方便时间序列分析）
df.set_index('日期', inplace=True)
df.sort_index(inplace=True)3.6 最终输出：干净的DataFrame
清洗完成后，我会做一次最终检查：
# 最终质量检查
print('最终数据形状:', df.shape)
print('缺失值情况:\n', df.isnull().sum())
print('基本统计:\n', df[amount_cols].describe())
# 保存清洗后的数据
df.to_csv('fund_flow_clean.csv', encoding='utf-8')这时候的DataFrame，就可以放心拿去分析了。每一行代表一只股票在某个交易日的主力资金流向，没有缺失值，没有异常值，格式统一。
我的习惯：我会把清洗过程封装成一个函数，每次拿到新数据直接调用。这样既保证一致性，又节省时间。函数里我会加上详细的日志，记录每一步处理了多少数据，方便回溯。
3.7 本章知识体系
下面这张图，帮你理清数据清洗的完整流程：
数据清洗这件事，说白了就是「把脏活累活干在前面」。你花80%的时间把数据洗干净，后面20%的时间就能高效地做分析。反过来，如果你图省事跳过清洗，后面可能要花200%的时间来填坑。
我个人觉得，数据清洗是量化交易中最能体现「经验」的环节。同样的数据，新手可能直接删掉一半，老手却能通过合理的填充和转换，保留90%以上的有效信息。这个差距，就是长期实战积累出来的。