4、量化数据采集与接入：数据源类型、实时采集 vs 批量采集、API接入与文件导入、数据质量校验

量化交易这个行当，说白了就是跟数据打交道。你策略再牛，模型再花哨，数据源要是烂了，一切都是白搭。我见过太多团队，策略回测跑得飞起，一上实盘就崩，查到最后，往往是数据采集环节出了幺蛾子。

今天咱们就聊聊量化数据仓库的第一道关卡——数据采集与接入。这部分我踩过的坑不少，希望能帮你少走弯路。

4.1 数据源类型：行情、交易、财务、舆情

量化数据源，我习惯分成四大类。每一类都有它的脾气，采集方式也完全不同。

数据源类型	典型数据	更新频率	采集难点
行情数据	Tick、分钟线、日线、Level-2	毫秒级～秒级	高并发、低延迟、数据量大
交易数据	订单、成交、持仓、资金流水	实时（交易时段）	数据一致性、幂等性、回滚机制
财务数据	资产负债表、利润表、现金流量表	季度/年度	数据格式不统一、披露时间不确定
舆情数据	新闻、公告、社交媒体、研报	分钟级～小时级	非结构化、情感分析、噪声过滤

行情数据是实时性要求最高的。我记得刚入行时，为了拿到交易所的Level-2行情，折腾了好几个方案。最终选择了CTP（综合交易平台）的API，虽然接入门槛高，但数据质量最靠谱。

交易数据呢，说白了就是你的账户流水。这里有个坑——很多券商的交易接口，在极端行情下会丢包。我曾经在2015年股灾时，发现某券商API返回的成交记录少了十几笔，差点导致风控系统误判。所以，交易数据一定要做本地缓存和核对。

财务数据相对稳定，但坑在「时效性」和「修正值」。A股财报经常有更正公告，如果你只采集了初版数据，后面被修正了都不知道。我建议财务数据要保留版本号，每次更新都做增量覆盖。

舆情数据最头疼。你想想看，一条新闻出来，到底是利好还是利空？机器很难判断。我团队曾经用NLP模型分析某公司公告，结果把「计提减值准备」解读成利好，差点闹出笑话。所以舆情数据，我建议只做关键词提取和情感打分，最终决策还是得靠人。

4.2 实时采集 vs 批量采集

这个选择其实不难。核心就一句话：数据更新频率决定采集模式。

实时采集，我一般用在行情和交易数据上。技术栈通常是：

• 消息队列：Kafka 或 RabbitMQ，用来缓冲高并发写入
• 流处理框架：Flink 或 Spark Streaming，做实时清洗和聚合
• 时序数据库：InfluxDB 或 ClickHouse，存储高频行情

批量采集，适合财务和舆情数据。每天定时跑一次任务，把数据拉下来存到数据仓库里。常用的工具有：

• 调度框架：Airflow 或 DolphinScheduler
• ETL工具：DataX 或 Kettle
• 存储：Hive 或 StarRocks

这里有个经验之谈：不要为了实时而实时。我见过有人把财务数据也搞成实时采集，结果每天凌晨财报披露时，系统压力暴增，反而影响了行情数据的正常采集。财务数据一天拉一次完全够用，何必折腾呢？

4.3 API接入与文件导入

数据怎么进来？无非两种方式：API 和 文件。

API接入，我习惯用 RESTful 或 WebSocket。行情数据用 WebSocket，因为需要长连接推送；其他数据用 REST，简单稳定。

下面是一个简单的行情API接入示例（伪代码）：

# 伪代码：行情WebSocket接入
import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 写入Kafka
    kafka_producer.send('market_data', data)

def on_error(ws, error):
    # 重连逻辑
    logger.error(f"WebSocket error: {error}")
    time.sleep(5)
    ws.connect()

ws = websocket.WebSocketApp(
    "wss://api.xxx.com/market",
    on_message=on_message,
    on_error=on_error
)
ws.run_forever()

文件导入，常见的是 CSV、Excel、Parquet 格式。财务数据很多券商还是提供 CSV 文件，每天手动下载再上传到服务器。我建议自动化处理：写个脚本定时从FTP拉取文件，然后自动解析入库。

文件导入有个坑——编码问题。国内很多数据源还是 GBK 编码，而数据仓库一般用 UTF-8。我曾经因为没做编码转换，导致几百万条财务数据乱码，排查了一整天。所以，文件导入的第一步，永远是检查编码。

4.4 数据质量校验

数据采集进来了，不代表就能直接用。质量校验是最后一道防线。

我总结了一套「四步校验法」：

1. 完整性校验：检查数据是否缺失。比如行情数据，某只股票一天应该有240根分钟线，少了就是有问题。
2. 一致性校验：检查不同数据源是否对得上。比如交易数据，券商API返回的成交量和交易所的成交量应该一致。
3. 准确性校验：检查数值是否合理。比如某股票收盘价突然涨了100倍，那肯定是数据异常。
4. 时效性校验：检查数据是否及时。比如财务数据，财报披露后24小时内必须入库，超时就报警。

我曾经遇到过一个经典案例：某天策略突然大幅亏损，排查后发现是行情数据出了问题——某只股票的复权因子算错了，导致历史价格全部偏移。从那以后，我强制要求所有数据入库前必须做「回测校验」：用历史数据跑一遍策略，看结果是否和之前一致。

好了，数据采集与接入这块，核心就是这些。说白了，就是搞清楚数据从哪来、怎么来、来了怎么检查。下一层，咱们再聊数据清洗和标准化。

---

公众号:蓝海数据掘金营，微信deep3321