从一次 API 超时到完整数据容灾：我如何设计可恢复的量化数据管道
### 📌 摘要 > 我曾经把量化数据 API 当成一个普通函数调用：调用成功就继续，调用失败就报错。真正进入实盘后，我才发现这种设计远远不够。一个偶发超时可能造成任务堆积、数据重复请求、旧数据进入策略，甚至让整个盘中任务失去时序。本文从我的工程实践出发，介绍我如何围绕 Python 量化 API 建立“请求、重试、校验、缓存、降级”的数据管道。 ## 一、问题场景：我在量化实盘中遇到的真实困境 我刚开始搭建量化数据服务的时候，代码其实非常简单。 类似这样： ```javascript df = API.get(...) ``` 拿到数据以后直接交给策略。 研究阶段这么写完全没有问题。 因为研究员手动运行一次脚本，失败了重新运行就可以。 但当我把程序放到自动任务里以后，问题开始暴露出来。 有一次夜间数据任务需要处理一批股票。 某个时间段 API 出现短暂延迟。 第一批请求没有按预期返回。 调度程序认为任务没有完成，于是启动下一轮。 于是服务器上同时出现了： ```javascript 任务 A：等待 API 任务 B：等待 API 任务 C：开始请求 ``` 后来 API 恢复了。 但是我的服务器并没有马上恢复正常。 因为之前积累的任务已经开始同时处理数据。 这次故障让我重新设计了整个数据访问层。 我后来给自己的原则是： > **数据接口应该允许失败，但数据管道不能因为一次失败就失去控制。** ## 二、问题根因分析 ### 1. 单次 API 调用不是数据管道 这是我现在最强调的一点。 下面这种代码： ```javascript df = qd.klines.get(...) ``` 只是一个 API 调用。 真正的数据管道至少应该考虑： ```javascript 请求 ↓ 异常处理 ↓ 数据校验 ↓ 数据落地 ↓ 任务状态 ↓ 失败恢复 ``` 如果把这些逻辑全部放在策略函数里，系统很快会变得难以维护。 ### 2. “重试”必须和“幂等”一起设计 假设一个任务要获取某一天的数据。 第一次请求失败。 第二次成功。 如果我的程序不知道第一次到底有没有完成，就可能产生： ```javascript 重复数据 重复文件 重复任务 ``` 所以我现在处理历史数据时，会尽量让任务以： ```javascript symbol + period + start_time + end_time + adjust ``` 这样的组合确定数据范围。 同一个查询重复执行，应该能够被识别为同一份数据任务。 ### 3. 缓存不是简单地保存一个 DataFrame 很多人说： > 超时了就用缓存。 但我认为缓存最重要的问题是： **这份缓存有多新？** 例如： ```javascript 缓存更新时间：14:15 当前时间：14:35 ``` 那么对于一个盘中策略，这份数据可能已经没有意义。 因此我现在设计缓存时，会同时保存： ```javascript 数据 更新时间 查询参数 数据来源 数据版本 ``` 即使暂时不把所有元数据都写入数据库，也应该至少在文件名或任务记录中保留查询上下文。 ## 三、解决方案探索与选型 ### 方案一：失败以后直接退出 这是最简单的。 优点是不会继续使用错误数据。 缺点也明显： 一次偶发网络抖动就会导致整个任务失败。 对于批量历史数据任务，这个策略太脆弱。 ### 方案二：无限重试 这个方案看起来可靠，实际上我非常反对。 因为： ```javascript API 出问题 ↓ 客户端不断请求 ↓ 请求数量增加 ↓ 服务端压力增加 ↓ API 更难恢复 ``` 这实际上可能形成反馈循环。 ### 方案三：有限重试 + 指数退避 这是我现在更常用的方案。 例如： ```javascript 第一次失败 ↓ 1 秒 第二次失败 ↓ 2 秒 第三次失败 ↓ 任务失败 ``` 如果服务端存在短暂抖动，通常有机会恢复。 如果服务端真的不可用，也不会让客户端无限占用资源。 ### 方案四：批量数据 + 本地落地 对于历史数据，我现在更喜欢： ```javascript API ↓ 批量查询 ↓ DataFrame ↓ 校验 ↓ 本地文件 ``` 而不是策略每次运行都重新请求全部历史数据。 这样可以把： ```javascript 研究阶段 ``` 和： ```javascript 生产数据同步阶段 ``` 分开。 ## 四、具体实现：代码实战 下面这份代码实现一个比较轻量的数据同步器。 它没有依赖 QuantDash SDK 没有公开说明的内部参数，而是只使用 SDK 已公开的 K 线查询接口。 ```python import os import time from pathlib import Path from quantdash import QuantDash MAX_RETRIES = 3 CACHE_DIR = Path("quant_cache") def validate_dataframe(df, symbol: str): if df is None or df.empty: return False, "DataFrame 为空" required_columns = { "symbol", "trade_date", "open", "high", "low", "close", "volume", } missing = required_columns.difference(df.columns) if missing: return False, f"缺少字段: {sorted(missing)}" if symbol not in set(df["symbol"].astype(str)): return False, "返回数据中没有目标标的" if df[ ["open", "high", "low", "close"] ].isna().any().any(): return False, "价格字段存在空值" if (df["high"] {path}" ) if __name__ == "__main__": main() ``` 这里我有一个很重要的工程习惯： **只有数据校验通过以后，我才覆盖缓存。** 因为如果 API 返回了空数据或者异常数据，我不希望程序把原本正确的缓存直接覆盖掉。 这其实是一个非常简单的“最后一份正确数据”保护机制。 ### 1. 使用批量接口减少请求数量 当数据规模进一步扩大以后，我会优先使用批量接口。 ```python import os from pathlib import Path from quantdash import QuantDash def main(): api_key = os.getenv("QUANTDASH_API_KEY") if not api_key: raise RuntimeError( "未配置 QUANTDASH_API_KEY 环境变量" ) qd = QuantDash(api_key=api_key) symbols = [ "600519.SH", "000001.SZ", "000858.SZ", "600000.SH", ] output_dir = Path("batch_cache") output_dir.mkdir( parents=True, exist_ok=True, ) try: result = qd.klines.batch( symbols, period="1d", count=30, adjust="forward", to_dataframe=True, show_progress=True, ) for symbol, df in result.items(): if df.empty: print( f"{symbol}: 返回为空" ) continue missing = { "symbol", "trade_date", "open", "high", "low", "close", "volume", }.difference(df.columns) if missing: print( f"{symbol}: " f"缺少字段 {sorted(missing)}" ) continue path = output_dir / ( symbol.replace(".", "_") + ".csv" ) df.to_csv( path, index=False, encoding="utf-8-sig", ) print( f"{symbol}: " f"{len(df)} 条数据已保存" ) except Exception as exc: print( f"批量数据同步失败: {exc}" ) if __name__ == "__main__": main() ``` ### 2. 历史数据最好使用明确的时间范围 如果我是做增量数据同步，我不会每次都重新下载全部历史数据。 例如： ```python import os import datetime from quantdash import QuantDash def main(): api_key = os.getenv("QUANTDASH_API_KEY") if not api_key: raise RuntimeError( "未配置 QUANTDASH_API_KEY 环境变量" ) qd = QuantDash(api_key=api_key) start = int( datetime.datetime( 2026, 5, 26, ).timestamp() * 1000 ) end = int( datetime.datetime( 2026, 6, 1, ).timestamp() * 1000 ) try: df = qd.klines.get( "600519.SH", period="1d", start_time=start, end_time=end, to_dataframe=True, ) if df.empty: print( "指定时间范围没有数据" ) return print( f"本次获取 {len(df)} 条记录" ) print( df[ [ "trade_date", "open", "close", "volume", ] ].to_string(index=False) ) except Exception as exc: print( f"增量同步失败: {exc}" ) if __name__ == "__main__": main() ``` 这样我可以在自己的数据任务里建立非常清晰的状态： ```javascript 2026-05-26 ↓ 2026-06-01 ``` 下一次任务继续从上一次成功位置开始。 ## 五、实践效果与避坑指南 ### 避坑一：不要在失败时覆盖好数据 这是我认为最值得优先做的一件事。 错误逻辑： ```javascript API 返回异常 ↓ 保存 ↓ 覆盖旧数据 ``` 正确逻辑应该是： ```javascript API 返回 ↓ 校验 ├─ 失败 → 保留旧数据 └─ 成功 → 更新数据 ``` ### 避坑二：API Key 不要进入代码仓库 我现在统一使用： ```javascript QUANTDASH_API_KEY ``` 环境变量。 Python 代码只读取： ```javascript api_key = os.getenv("QUANTDASH_API_KEY") ``` 而不是： ```javascript api_key = "真实密钥" ``` 这一点看起来很基础，但我见过不少量化项目在 Git 提交记录里泄露密钥。 ### 避坑三：429 不要当成普通网络异常处理 如果服务端提示请求频率过高，我不会继续快速重试。 因为： ```javascript 429 ↓ 立即重试 ↓ 继续 429 ``` 没有任何意义。 我的原则是降低请求频率，并遵循服务端返回的等待信息。 ### 避坑四：缓存必须有“新鲜度”概念 缓存解决的是： > API 暂时不可用怎么办？ 而不是： > 永远使用旧数据。 所以如果是实时策略，我不会因为“缓存存在”就无条件使用它。 我会先判断： ```javascript 当前时间 - 缓存时间 ``` 是否还在策略允许的范围内。 ## 六、总结与思考 我现在看量化 API，已经很少把它理解成一个简单的“数据查询工具”。 在生产环境里，它实际上是整个数据管道的一部分。 真正成熟的结构应该是： ```javascript 数据源 ↓ 请求控制 ↓ 异常处理 ↓ 重试 ↓ 数据校验 ↓ 缓存 ↓ 策略 ``` 这样即使外部 API 偶尔发生超时，故障也会被限制在数据层，而不会直接扩散到策略层。 我目前使用 QuantDash Python SDK 时，也更倾向于把它放在这样的数据访问层里。SDK 可以负责把行情数据以 Pandas 形式交给我，而超时、重试、缓存和数据质量控制，则应该由我的应用工程负责。 我越来越相信一个原则： > **量化系统不是要求每一次请求都成功，而是要求请求失败以后系统仍然可控。** 这才是我理解的量化数据工程。 ## Q&A ### Q1：是不是所有数据都应该做缓存？ 不是。 历史数据非常适合缓存。 实时行情则要根据策略对数据新鲜度的要求决定。 如果策略需要最新盘口，那么几分钟以前的缓存通常没有意义。 ### Q2：缓存应该保存 CSV 还是 Parquet？ 小型研究项目使用 CSV 很方便，容易查看，也容易调试。 如果数据量很大，我通常会考虑列式存储格式。 但无论使用什么格式，最重要的还是保证： ```javascript 数据 查询范围 数据时间 版本 ``` 这些信息能够被追踪。 ### Q3：批量 API 是不是一定比单个 API 请求好？ 不是绝对的。 批量请求可以减少请求次数，但单次任务的失败影响范围也可能更大。 我的经验是： ```javascript 小规模实时请求 → 单个请求 历史批量同步 → 批量请求 ``` 然后再结合限流和失败率做调整。 ### Q4：如果 API 长时间不可用怎么办？ 我不会让策略无限等待。 对于历史数据任务，可以暂停并等待下一轮同步。 对于实时策略，则需要根据策略的时间敏感程度决定： ```javascript 等待 跳过本轮 使用符合新鲜度要求的缓存 切换备用数据源 ``` 关键是这个决策应该提前设计，而不是故障发生以后临时决定。
