4、数据源接入与数据管理：实时行情、历史数据、数据清洗、数据存储（SQLite/Parquet/时序数据库）与数据更新机制
量化策略从研究走向实盘，第一个真正的工程化挑战就是数据。回测时数据不完整、延迟几秒或许只是影响研究结论的精度，但实盘中数据断流、字段错位、时间戳漂移或存储写入失败，轻则产生错误信号，重则导致仓位错乱甚至程序崩溃。本章围绕数据源接入、历史与实时数据管理、清洗规则、存储选型以及更新机制展开，目标是建立一套可靠、可审计、可回放的数据基础设施。
4.1 数据源接入：实时行情与历史数据的获取路径
核心原理：量化实盘的数据源分为实时行情流和历史数据批量获取两大类。实时行情通常通过交易所或券商的行情网关以 TCP/WebSocket 长连接推送，常见协议包括 CTP（国内期货）、Binary Protocol（美股）、WebSocket JSON（加密货币）等。历史数据则通过 REST API、FTP 文件下载或数据库直连获取。接入层需要解决三个核心问题：连接稳定性、协议兼容性和数据完整性。
接入架构设计：建议采用独立的数据接入服务（Data Feeder），与策略引擎解耦。该服务负责维护连接、自动重连、心跳检测、消息解码和初步校验。接入层应具备以下能力：
- 多路冗余：关键行情源至少配置主备两路，切换时间控制在 1 秒以内。
- 协议适配：将不同数据源的字段统一映射为内部标准格式（如统一时间戳为纳秒级 Unix Time，统一价格为浮点型）。
- 增量快照：对于 Level-2 行情，需同时处理逐笔成交和十档快照，两者需通过序列号对齐。
- 历史数据分片：大范围历史数据请求应分片拉取，每片 1000-5000 根 K 线，避免超时。
local_ts 和 exchange_ts 两个字段。
4.2 数据清洗：从原始数据到可用数据的转换规则
核心原理：原始行情数据往往存在重复推送、乱序到达、字段缺失、价格异常（如零价格、涨跌停价格错误）等问题。数据清洗的目标是生成一份连续、一致、可复现的数据集，供策略计算和回测使用。清洗规则必须可配置、可审计，且不能破坏原始数据。
清洗流程与规则：
- 去重：同一时间戳、同一合约、同一价格、同一成交量的记录只保留一条。对于 Tick 数据，还需根据交易所序列号去重。
- 排序：按时间戳升序排列，对于乱序数据，允许 500ms 内的延迟重排，超过阈值则丢弃并告警。
- 缺失值处理：Bar 数据缺失时，根据前收盘价和涨跌停价判断是否停牌；若停牌则填充前收盘价并标记
is_suspended=1，否则视为异常并告警。 - 异常值过滤：价格小于等于 0、成交量小于 0、买卖价差为负等明显错误直接剔除。对于涨跌停价格，需结合合约涨跌停板幅度校验。
- 复权因子：历史数据需标注复权方式（前复权/后复权/不复权），并在存储时保留原始价格和复权因子两个字段，避免后续重复计算。
| 清洗步骤 | 输入 | 输出 | 异常处理策略 |
|---|---|---|---|
| 去重 | 原始 Tick/Bar | 唯一记录集 | 保留首条，记录重复次数 |
| 排序 | 乱序记录 | 时间升序 | 延迟 >500ms 丢弃并告警 |
| 缺失填充 | 缺失 Bar | 填充后 Bar | 停牌填充前收盘，否则告警 |
| 异常过滤 | 含异常值记录 | 干净记录 | 剔除并记录原因 |
| 复权标注 | 原始价格 | 价格+复权因子 | 因子缺失时拒绝入库 |
4.3 数据存储选型：SQLite、Parquet 与时序数据库的协同
核心原理：没有一种存储引擎能同时满足实时读写、历史分析和事务性要求。实盘系统通常采用分层存储策略：SQLite 用于存储交易记录、配置参数、账户信息等关系型数据；Parquet 列式文件用于历史研究和批量回测；时序数据库（如 InfluxDB、TDengine、ClickHouse）用于存储高频 Tick 和 Bar 数据，支持高效的时间范围查询和聚合计算。
选型对比与适用场景：
| 存储类型 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|
| SQLite | 零配置、事务支持、单文件易备份 | 并发写入弱、不适合海量时序 | 交易记录、策略参数、账户信息 |
| Parquet | 列式压缩、扫描高效、与 Pandas/Polars 无缝集成 | 不可变、不适合频繁更新 | 历史日线/分钟线研究、离线回测 |
| 时序数据库 | 高吞吐写入、自动分区、降采样、保留策略 | 运维成本高、SQL 方言差异 | Tick 级行情、实时监控、盘中策略 |
存储架构建议：实时行情先写入时序数据库（热数据），同时异步导出为 Parquet 文件（温数据）用于研究。每日收盘后，将当日数据从时序库归档到 Parquet，并清理时序库中超过 30 天的原始 Tick 数据（仅保留降采样后的 1 分钟 Bar）。SQLite 独立存储交易和配置，定期备份到本地磁盘和云存储。
-- SQLite 交易记录表示例
CREATE TABLE trades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
strategy_id TEXT NOT NULL,
symbol TEXT NOT NULL,
side TEXT CHECK(side IN ('BUY','SELL')),
price REAL NOT NULL,
volume INTEGER NOT NULL,
trade_time INTEGER NOT NULL, -- 纳秒时间戳
status TEXT DEFAULT 'FILLED',
created_at INTEGER DEFAULT (strftime('%s','now'))
);4.4 数据更新机制：增量同步、校验与生命周期管理
核心原理：数据更新机制决定了数据的时效性和一致性。实盘系统需要同时处理实时流式更新和定时批量更新。实时更新依赖订阅推送，批量更新则通过定时任务（如每日 15:30 拉取当日日线）完成。更新机制的核心是增量同步——只拉取上次同步点之后的新数据，避免全量重复传输。
更新流程设计：
- 交易日历驱动：维护一张交易日历表，标记每个交易日的开收盘时间、午休时间、夜盘时段。更新任务仅在有效交易时段内运行。
- 增量游标：为每个数据源维护一个游标（如最后同步时间戳或最后一条记录的序号）。每次拉取时携带游标，服务端返回增量数据。
- 断点续传：若拉取过程中断，记录当前游标，下次从断点继续。对于文件型数据（如 Parquet），采用“先写临时文件，校验后原子重命名”的策略。
- 数据校验：每次更新完成后，对比源端和本地的最新时间戳、记录条数、数据指纹（如 CRC32）。不一致则触发补录任务。
- 版本快照：每日收盘后生成一份数据版本快照（包含当日所有数据的哈希值），便于回溯和审计。
背景：策略依赖 5 分钟 Bar 数据，数据源为 CTP 行情网关和第三方历史数据商。某日开盘后，策略连续 15 分钟未产生信号，但行情界面显示价格正常波动。
排查过程：① 检查策略日志，发现数据队列为空；② 检查数据接入服务，发现 WebSocket 连接正常，但心跳响应超时；③ 检查数据存储，发现时序数据库写入延迟高达 8 秒；④ 进一步定位，发现第三方数据商的增量接口返回了重复数据，导致清洗层去重逻辑阻塞。
根因：数据商在开盘瞬间推送了 3 次相同的快照，清洗层去重模块使用 Redis 分布式锁，锁竞争导致处理线程阻塞。
解决方案：① 将去重逻辑改为基于时间窗口的内存去重（如 500ms 窗口），不再依赖分布式锁；② 增加数据接入服务的独立线程池，隔离不同数据源的清洗任务；③ 添加监控告警，当数据队列积压超过 1000 条时自动报警。
结果：故障恢复时间从 15 分钟缩短至 30 秒，后续未再出现同类问题。
复盘经验：数据清洗逻辑必须保持轻量且无阻塞；分布式锁在高频数据场景下慎用；每个数据源应有独立的故障隔离和监控指标。
4.5 数据质量监控与运维复盘
核心原理：数据质量监控是数据管理的最后一道防线。监控指标包括：数据延迟（本地时间与交易所时间差）、数据完整性（预期条数 vs 实际条数）、数据一致性（相邻 Bar 的价格连续性）、数据源健康状态（连接数、心跳丢失率）。监控系统应具备多级告警：延迟超过 1 秒为警告，超过 5 秒为严重，超过 30 秒触发策略暂停。
监控与复盘清单：
- 延迟监控：每 5 秒计算一次最新 Tick 的本地接收时间与交易所时间戳差值，绘制时间序列图。
- 完整性校验：每日收盘后统计每个合约的 Bar 数量，与交易所公布的合约明细对比。
- 异常波动检测：当某合约价格变动超过前收盘价 ±10% 且无对应公告时，触发人工复核。
- 存储容量规划：Tick 数据每月约占用 50-200GB（视合约数量），需提前规划磁盘扩容和归档策略。
- 定期复盘：每周进行一次数据质量复盘，分析告警记录、补录任务执行情况、存储增长趋势，并更新清洗规则库。
数据管理是一个持续迭代的过程。建议团队建立“数据变更日志”，记录每一次清洗规则调整、存储结构变更、数据源切换操作。实盘系统的稳定性，很大程度上取决于数据基础设施的健壮性——这往往比策略本身更值得投入精力。