量化数据工程：Point-in-Time、数据管线与质量监控
本课目标：理解 Point-in-Time 原则为什么是数据工程的第一条军规，并学会用轻量工具搭一条”拉取、校验、入库”的可持续数据管线。
PIT 原则：回测只能用”当时知道的”
Point-in-Time（PIT）原则一句话就能说完：回测中任意时刻 T 能用的数据，必须是 T 时刻实际已经可以获得的版本。违反它就是前视偏差，而前视偏差是回测结果虚高最常见的来源。它的阴险之处在于不报错、不留痕：代码跑得通，曲线很漂亮，直到实盘表现与回测天差地别，你才被迫回头翻找数据里混进了哪些”当时不可能知道”的信息。
财务数据是重灾区。一份年报的”报告期”可能是去年 12 月 31 日，但”公告日”可能在今年 4 月。如果你的因子在 1 月就用上了这份年报的净利润，回测就用了未来数据。更隐蔽的是数据修正：公司可能对已披露财务数据进行更正，多数数据源只保留最新版本，你在回测里用的”干净数据”，在当时根本不存在。
行情数据同样有 PIT 问题：后复权价格用未来的分红送转信息调整了历史价格，直接拿后复权价算”历史上某日的涨跌停价”就会出错；指数成分股名单如果只用当前版本，会引入幸存者偏差。
落地原则很简单：所有随时间演化的数据，入库时都带两个时间戳——数据所属的时间（报告期/交易日）和数据可获得的时间（公告日/入库日）。回测查询永远按”可获得时间 ≤ T”过滤。
证券主数据：代码、退市与更名
A 股的证券标识远比想象中麻烦：
- 退市：股票退市后从多数行情接口里消失。如果数据库只有存续股票，任何横截面回测都自带幸存者偏差——你等于提前知道了谁能活下来。
- 更名与 ST：股票名称、ST 标记随经营状况变化。“剔除 ST”这类规则必须用当时的 ST 状态判断，而不是现在的。
- 代码复用与格式：不同数据源的代码格式各异（600000.SH、SH600000、600000），跨源合并前必须统一。
工程上的解法是维护一张证券主表（security master）：每只证券一个内部不变 ID，代码、名称、上市状态、ST 标记等属性全部带生效区间存储：
# 属性表结构示意：同一证券的属性变化各占一行# sec_id | attr | value | start_date | end_date# 000001 | name | XX银行 | 2018-01-01 | 2021-06-30# 000001 | name | YY银行 | 2021-07-01 | 9999-12-31def attr_asof(df, sec_id, attr, date): m = (df.sec_id == sec_id) & (df.attr == attr) & \ (df.start_date <= date) & (df.end_date >= date) return df.loc[m, 'value'].iloc[0]增量管线：拉取、校验、入库
个人量化数据库不需要复杂调度系统，但需要一条纪律严明的管线。所谓纪律，是指流程固定、可重复、出错时行为可预期——每天按同样的顺序跑三步：
- 拉取（Extract）：从数据源增量拉取新数据。记录每次拉取的时间、参数和行数，失败要能安全重试——这意味着写入必须幂等（重复运行不会产生重复数据，常用”先删后插”或按主键覆盖实现）。
- 校验（Validate）：新数据先进临时区，通过质量检查后才允许进正式库。检查不通过就告警并停止，绝不让脏数据污染正式表。
- 入库（Load）：按主键写入正式库，同时更新元数据表（各数据集的最后更新日期、行数）。
关键设计取舍是”原始层与加工层分离”：原始层只存数据源的原样数据，永不修改；加工层（复权价、因子值）全部可以从原始层重算。这样任何加工逻辑的 bug 都可以通过重跑修复，而不会丢失事实。
除日常增量外，管线还要预留两种运行模式：补数（backfill，指定日期区间重新拉取，用于修复历史缺口）和全量重建（从原始层重算全部加工层）。这两种模式在设计初期就要支持——等发现数据有问题再临时改代码，很容易在慌乱中引入新错误。调度方面，个人项目用系统自带的定时任务触发一个入口脚本即可，重点不是调度工具多先进，而是每次运行留下结构化日志：跑了哪一步、处理了多少行、耗时多久、是否告警。
def run_pipeline(trade_date): raw = fetch_daily_bars(trade_date) # 拉取 report = validate(raw) # 校验 if not report.passed: alert(report); return # 拦截脏数据 upsert(raw, table='daily_bars_raw') # 幂等入库 rebuild_derived(trade_date) # 重算加工层 log_meta(trade_date, rows=len(raw))数据质量检查清单
校验环节值得一张明确的清单，逐条写成断言：
| 检查项 | 示例规则 |
|---|---|
| 完整性 | 交易日历上的每个交易日都有数据；股票数量与预期偏差在阈值内 |
| 唯一性 | (证券, 日期) 主键无重复 |
| 值域 | 价格 > 0；最高价 ≥ 最低价；成交量 ≥ 0 |
| 逻辑一致 | 非停牌日涨跌幅在涨跌停限制内；开高低收关系成立 |
| 跨源对账 | 抽样若干股票与第二数据源比对收盘价，容忍度内一致 |
| 连续性 | 相邻交易日价格跳变超阈值时，检查是否有除权除息记录对应 |
其中”跨源对账”最容易被省略，但它是唯一能发现数据源本身系统性错误的手段。检查结果应该落库留痕，形成质量监控的时间序列——某个检查项的失败率突然上升，往往意味着数据源改了口径。
告警要分级：主键重复、价格为负这类硬错误直接阻断入库；股票数量偏离预期百分之几这类软异常只记录并提示人工复核。全部告警一视同仁的结果是”狼来了”——告警多到没人看，等于没有告警。每条检查规则最好附上一句”失败意味着什么”的注释，半年后的你会感谢现在的自己。
轻量存储：SQLite 与 Parquet 就够了
个人和小团队完全不需要一上来就上分布式数据库。一个务实的组合：
- SQLite 存证券主表、交易日历、元数据和质量检查结果——这些是小而关系复杂的数据，需要事务和灵活查询。
- Parquet 存日线行情、因子值这类大而规整的数据，按年或按月分区。列式存储配合 pandas 的谓词下推，读取速度远超逐行数据库。
import pandas as pd
df = pd.read_parquet( 'data/daily_bars/', filters=[('trade_date', '>=', '2024-01-01')], columns=['sec_id', 'trade_date', 'close', 'volume'],)这套组合的上限比想象中高：全 A 股十几年日线加常用因子，单机毫无压力。等真的碰到瓶颈，再考虑更重的方案也不迟。本文讨论数据工程方法，不构成投资建议。
本课小结
- PIT 是第一军规：所有随时间演化的数据都要带”所属时间”和”可获得时间”双时间戳。
- 证券主表用不变的内部 ID 加带生效区间的属性表，处理退市、更名与 ST 状态。
- 管线三段式：增量拉取（幂等）、临时区校验、正式入库；原始层永不修改，加工层随时可重算。
- 质量检查写成断言清单并落库留痕，跨源对账是发现数据源系统性错误的唯一手段。
- SQLite 管关系与元数据，Parquet 管大表，个人量化足够用很久。
动手练习：为日线行情数据写一个校验模块：实现完整性、唯一性、值域、开高低收逻辑四类检查，各返回通过/失败与问题明细。故意在测试数据里埋三个错误（重复行、负成交量、最高价低于最低价），验证你的模块能全部抓出来。