3、数据接入层实战：数据源类型与质量校验
数据接入层，说白了就是量化系统的「水管」。水管要是堵了、漏了、流进来的是脏水，后面再牛的因子模型也白搭。我见过太多团队，因子研究做得花里胡哨，结果上线一跑，全是因为数据问题崩掉的。
今天咱们就把这层「水管」拆开看看。我会结合自己踩过的坑，把数据源类型、接入方式、质量校验这三块讲透。
3.1 数据源类型：行情、财务、另类
量化数据源，我习惯分成三大类。每一类的脾气秉性都不一样，接入策略也得区别对待。
3.1.1 行情数据
行情数据是量化最基础的「口粮」。包括日线、分钟线、Tick 级数据。这类数据的特点是：量大、频率高、对时效性要求极严。
我个人习惯把行情数据再细分成两类：
- Level-1 行情：普通买卖盘口，频率较低，适合做日频因子
- Level-2 行情：逐笔委托、逐笔成交，频率极高，适合做高频因子
嗯，这里要注意：Level-2 数据虽然信息量大，但存储成本也高得吓人。我见过有人把全市场 Level-2 数据存了三年，结果硬盘比策略还贵。
3.1.2 财务数据
财务数据是基本面因子的「命根子」。包括三大报表、业绩预告、审计意见等。这类数据的特点是：更新频率低（季报/年报）、但容易出错、且存在「预期差」。
我在项目中遇到过最坑的事：某公司财报里「净利润」字段，不同数据商给的数值差了 20%。后来一查，原来一家用的是「归母净利润」，另一家用的是「净利润总额」。你想想看，这种差异直接导致因子失效。
3.1.3 另类数据
另类数据是近年来的「新宠」。包括舆情数据、卫星图像、供应链数据、电商数据等。这类数据的特点是：非结构化、噪声大、但可能带来超额收益。
说实话，另类数据的水很深。我建议新手先从「结构化程度高」的另类数据入手，比如龙虎榜数据、大宗交易数据。别一上来就搞 NLP 舆情分析，那玩意儿坑太多。
3.2 数据接入方式：API、FTP、数据库
数据接入方式，说白了就是「怎么把数据从源头搬到你的系统里」。我常用的有三种方式，各有优劣。
3.2.1 API 接入
API 接入是最灵活的方式。适合实时行情、按需查询的场景。我一般用 Python 的 requests 库配合异步框架（如 aiohttp）来写。
import requests
import json
def fetch_tick_data(symbol, date):
url = f"https://api.dataprovider.com/v1/tick/{symbol}/{date}"
headers = {"Authorization": "Bearer YOUR_API_KEY"}
resp = requests.get(url, headers=headers)
if resp.status_code == 200:
return resp.json()
else:
raise Exception(f"API 请求失败: {resp.status_code}")
# 使用示例
data = fetch_tick_data("000001.SZ", "2024-01-15")
print(f"获取到 {len(data)} 条 Tick 数据")
3.2.2 FTP 接入
FTP 接入适合批量数据、定时同步的场景。比如每日收盘后的日线数据、财务数据。我习惯用 paramiko 库做 SFTP 接入，比普通 FTP 更安全。
import paramiko
import os
def download_from_sftp(host, port, username, password, remote_path, local_path):
transport = paramiko.Transport((host, port))
transport.connect(username=username, password=password)
sftp = paramiko.SFTPClient.from_transport(transport)
sftp.get(remote_path, local_path)
sftp.close()
transport.close()
print(f"文件已下载到: {local_path}")
# 使用示例
download_from_sftp(
host="ftp.dataprovider.com",
port=22,
username="quant_user",
password="secure_pass",
remote_path="/daily_data/2024/01/15.csv",
local_path="./data/20240115.csv"
)
FTP 接入有个坑：文件命名规则。不同数据商的命名方式千奇百怪。我建议在接入层统一做「文件名解析器」，把日期、品种、数据类型都解析出来。
3.2.3 数据库接入
数据库接入适合内部数据、历史数据回补的场景。比如从公司的 ClickHouse 或 InfluxDB 里拉数据。我一般用 SQLAlchemy 做 ORM 层，方便切换数据库类型。
from sqlalchemy import create_engine, text
import pandas as pd
def query_daily_data(engine, symbol, start_date, end_date):
query = text("""
SELECT trade_date, open, high, low, close, volume
FROM daily_data
WHERE symbol = :symbol
AND trade_date BETWEEN :start_date AND :end_date
ORDER BY trade_date
""")
df = pd.read_sql(query, engine, params={
"symbol": symbol,
"start_date": start_date,
"end_date": end_date
})
return df
# 使用示例
engine = create_engine("clickhouse://user:pass@localhost:8123/quant_db")
df = query_daily_data(engine, "000001.SZ", "2024-01-01", "2024-01-31")
print(f"查询到 {len(df)} 条日线数据")
3.3 数据质量校验
数据质量校验，是数据接入层最容易被忽视、但最重要的环节。我把它叫做「数据安检门」。数据进来之后，必须先过安检，才能进入后续的因子生产流水线。
3.3.1 校验维度
我一般从四个维度做数据质量校验：
| 校验维度 | 校验内容 | 常见问题 |
|---|---|---|
| 完整性 | 数据是否缺失、字段是否为空 | 某天某只股票没有行情数据 |
| 准确性 | 数值是否在合理范围内 | 股价出现负数、成交量异常大 |
| 一致性 | 不同数据源之间是否匹配 | 复权因子不一致、除权除息日对不上 |
| 时效性 | 数据是否在预期时间内到达 | 收盘后 2 小时数据还没到齐 |
3.3.2 校验流程
我设计了一套「三级校验」流程，大家可以参考：
- 一级校验（字段级）：检查字段类型、非空约束、数值范围。比如价格不能为负、日期不能为空。
- 二级校验（记录级）：检查记录之间的逻辑关系。比如开盘价、最高价、最低价、收盘价之间是否满足「最高价 ≥ 开盘价 ≥ 最低价」。
- 三级校验（跨源级）：对比不同数据源的数据。比如用 Wind 的数据校验聚宽的数据。
def validate_daily_data(df):
errors = []
# 一级校验：字段级
if df['close'].isnull().any():
errors.append("存在收盘价为空的数据")
if (df['close'] <= 0).any():
errors.append("存在收盘价小于等于0的数据")
# 二级校验：记录级
invalid_hl = df[df['high'] < df['low']]
if len(invalid_hl) > 0:
errors.append(f"存在 {len(invalid_hl)} 条最高价低于最低价的数据")
# 三级校验：跨源级（假设有另一个数据源 df_ref）
# 这里省略具体实现
if len(errors) > 0:
print("数据质量校验未通过：")
for err in errors:
print(f" - {err}")
return False
else:
print("数据质量校验通过 ✅")
return True
# 使用示例
df = pd.read_csv("./data/20240115.csv")
validate_daily_data(df)
3.4 知识体系总览
下面这张图，是我对数据接入层知识体系的总结。你可以把它当作「地图」，随时回来对照。
数据接入层，是整个因子生产流水线的「地基」。地基不牢，地动山摇。我建议你花时间把这块做扎实，后面会省去无数麻烦。
公众号:蓝海资料掘金营，微信deep3321