3. 数据采集与清洗：数据源、实时流处理与标准化
做风控系统，最怕什么？
不是模型不准，不是策略失效。而是——数据是脏的。
我见过太多团队，花三个月搭了一套漂亮的回测框架，结果上线第一天就崩了。为什么？交易所返回的字段名变了，或者某只股票突然多了一个除权除息字段没处理。嗯，数据采集和清洗，才是整个系统的地基。
3.1 数据源：从哪拿数据？
我个人习惯把数据源分成三类。每一类都有它的脾气，你得顺着来。
3.1.1 交易所API
这是最直接的方式。但也是最容易出幺蛾子的。
- REST API：适合拉历史数据。我一般用
requests库，加个重试机制。为什么？因为交易所经常限流，你请求太频繁，它直接给你返回 429。 - WebSocket：适合实时行情。我建议用
websockets库，心跳包一定要自己维护。我曾经遇到过，某交易所的 WebSocket 连接 30 分钟没发心跳，它也不断，就是不发数据了。你想想看，这多坑。
核心原则：永远假设 API 会挂。加超时、加重试、加熔断。
3.1.2 数据库
内部数据库，比如 PostgreSQL 或 ClickHouse，是存储清洗后数据的地方。
但要注意：
- 别直接在生产库上跑复杂查询。我建议用只读副本。
- 字段类型要严格。比如价格字段，用
DECIMAL(20,8)，别用FLOAT。为什么？浮点数精度问题，你算着算着，钱就对不上了。
3.1.3 CSV文件
嗯，这个最古老，但也最常用。特别是回测用的历史数据。
我踩过的坑：
- CSV 里混了 BOM 头，读进来第一列字段名多了个看不见的字符。
- 日期格式不统一。有的用
2024-01-01，有的用01/01/2024。 - 空值处理。有的用
NA，有的用null，有的直接空着。
我的习惯：不管什么数据源，进来第一件事——统一成 pandas DataFrame。后面所有清洗逻辑都基于 DataFrame 写，这样切换数据源时，清洗代码不用改。
3.2 实时数据流处理
做风控，实时性很重要。你总不能等爆仓了，数据才到吧？
我一般用 Kafka 做消息队列。结构大概是这样的：
你看，数据从不同源进来，统一进入 Kafka。清洗模块从 Kafka 消费，处理完再写回数据库。这样就算某个数据源挂了，也不影响整体流程。
注意：实时流处理一定要考虑「乱序」问题。比如，订单 A 先发，但网络延迟，后到。如果你按到达时间处理，逻辑就错了。我建议用事件时间，而不是处理时间。
3.3 数据清洗与标准化
这部分最枯燥，但也最重要。说白了，就是让数据变得「规规矩矩」。
3.3.1 缺失值处理
我遇到过最离谱的情况：某只股票停牌，API 返回的价格字段是 0。如果你没处理，算出来的收益率直接是负无穷。
常用的方法：
- 前向填充：用上一个有效值填充。适合价格、成交量。
- 插值：线性插值，适合中间缺失。
- 直接删除：如果缺失太多，这条数据就别要了。
# 我常用的清洗代码
import pandas as pd
def clean_price_data(df):
# 1. 去重
df = df.drop_duplicates(subset=['symbol', 'timestamp'])
# 2. 排序
df = df.sort_values(['symbol', 'timestamp'])
# 3. 缺失值前向填充
df['price'] = df.groupby('symbol')['price'].ffill()
# 4. 过滤异常值（价格不能为0或负数）
df = df[df['price'] > 0]
return df3.3.2 格式标准化
这个我深有体会。不同交易所，字段名都不一样。
| 字段含义 | 交易所A | 交易所B | 我的标准 |
|---|---|---|---|
| 交易对 | symbol | pair | symbol |
| 最新价 | last_price | close | price |
| 成交量 | volume | vol | volume |
| 时间戳 | timestamp | ts | timestamp |
我建议建一个映射表。所有数据进来，先过一遍映射，统一字段名。这样后面的代码就不用关心数据源了。
3.3.3 异常值过滤
嗯，这个要小心。不能一刀切。
比如，某只股票突然涨了 100 倍。可能是数据错误，也可能是真的发生了什么事（比如合并、拆分）。
我的做法：
- 用 Z-score 或 IQR 检测异常。
- 但只标记，不删除。人工复核后再决定。
避坑指南：我曾经因为没处理除权除息，回测结果看起来年化 50%，实盘一跑亏成狗。后来我加了一个字段 adjust_factor，所有价格都乘以这个因子，才把问题解决。
3.4 数据质量监控
数据清洗不是一次性的。你得持续监控。
我一般会监控这几个指标：
- 延迟：数据从产生到入库，花了多久？超过 5 秒就要报警。
- 缺失率：每个字段的缺失比例。超过 1% 就要查原因。
- 重复率：重复数据的比例。超过 0.1% 就要检查去重逻辑。
说白了，数据质量决定了风控系统的上限。模型再牛，数据是脏的，一切都是白搭。
我个人习惯，每天上班第一件事，先看数据质量看板。如果数据有问题，今天就不调模型了，先修数据。这个习惯救了我很多次。