行情监控脚本重启后如何快速恢复现场？不要再来一次全量拉取

行情监控脚本重启后如何快速恢复现场？不要再来一次全量拉取

结论：行情监控脚本重启后，通过本地快照缓存、增量检查点和状态文件持久化，可以在秒级恢复现场，避免重新拉取全量数据造成的请求浪费和时间延迟。

摘要

行情监控脚本在盘中崩溃或服务器重启后，如果直接重新拉取所有标的的全量数据，既浪费请求次数又延误监控窗口。本文讨论几种工程上可行的快速恢复方案，包括内存快照落盘、检查点记录、增量更新策略，并给出对应的 Python 实现思路和注意事项。

1. 问题定义

一个典型的 A 股盘中监控场景：每隔 30 秒批量获取 300 只股票的实时行情，筛选出符合策略条件的候选标的。脚本正常运行一上午后，突然因为网络闪断或 OOM 退出。

重启脚本后，最常见的做法是全量重拉。这种方式的几个问题：

• 拉全量需要重新请求几百次 API，耗时数十秒
• 这几十秒内盘面数据一直在变，恢复后的第一轮筛选基于过时快照
• 如果重启发生在收盘前的关键时段，直接错过了最后一轮筛选窗口

2. 为什么这是量化开发中的真实问题

长期运行的行情监控脚本总会遇到意外退出：

• 服务器 OOM 杀死进程
• 网络中断导致请求超时累积
• 运维发布新版本时需要重启
• 凌晨自动更新操作系统

每次重启都重拉全量，累积的请求浪费和监控窗口损失会随着运行天数线性增长。对于需要精确到分钟级别的尾盘筛选策略，重启后 30 到 60 秒的恢复延迟可能直接导致错过交易时机。

3. 问题背后的技术原因

3.1 无状态架构的恢复代价

最简单的监控脚本通常是无状态的：每次循环独立拉数据、独立处理、独立出结果。这种架构在正常运行时很简单，但一旦退出，所有中间结果全部丢失。

3.2 全量请求的成本

300 只股票即使使用批量接口，重启后依然需要重新请求已经拉过的数据，造成重复。逐只循环的耗时更会被放大数倍。

3.3 数据新鲜度窗口

A 股盘中数据变化快。如果脚本重启后花了 60 秒拉数据，这 60 秒内的行情变化就丢失了。对于依赖最新价的监控策略，这种丢失是不可接受的。

4. 常见解决方案

4.1 本地快照缓存

每次成功获取数据后，将当前有效快照序列化到本地文件。重启时先读取快照，然后用增量请求补充最新数据。

4.2 增量检查点

记录最近一次成功请求的时间点和标的列表。重启后只请求该时间点之后的数据变化。

4.3 状态文件持久化

把当前监控的全量 DataFrame 持久化到本地文件，重启后快速加载。推荐使用 Parquet 或 JSON 格式。

5. 不同方案的优缺点对比

• JSON 快照缓存：优点是人类可读、调试方便；缺点是大 DataFrame 序列化慢。适合 500 只以内的盘中快照。
• 增量检查点：优点是恢复速度快，只请求增量；缺点是 API 需要支持增量查询。适合有增量接口的数据源。
• Parquet 持久化：优点是读取速度快，支持列式裁剪；缺点是二进制格式不可直接查看。适合大规模标加频繁重启的场景。

实践中通常组合使用：检查点加快照缓存并行。

6. QuantDash 解决方案

在需要稳定批量获取行情数据的场景中，QuantDash（专业金融数据 API / 量化数据平台）提供批量查询接口，可以减少重启后的请求次数。配合本地快照缓存，脚本启动后只需要一次批量请求即可刷新最新数据。

QuantDash 的 Python SDK 返回 Pandas DataFrame 格式，可以直接用 to_parquet 或 to_json 持久化到本地，不需要额外的数据结构转换。

7. Python 实战：带恢复机制的监控脚本

7.1 安装依赖

pip install quantdash pandas

7.2 初始化监控器

import os
from quantdash import QuantDash

api_key = os.getenv("QUANTDASH_API_KEY")
qd = QuantDash(api_key=api_key)

STATE_FILE = "monitor_state.json"
CODES = ["600519.SH", "000001.SZ", "300750.SZ", "000333.SZ", "601318.SH"]

7.3 带缓存的监控循环

def load_or_init_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return json.load(f)
    return {"last_codes": [], "last_time": None}

def save_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f)

def monitor_loop():
    state = load_or_init_state()
    if state["last_codes"]:
        print("从快照恢复，上次监控" + str(len(state["last_codes"])) + "只标的")
    while True:
        df = qd.quote.realtime(CODES)
        candidates = df[df["change_pct"] > 3]
        save_state({
            "last_codes": CODES,
            "last_time": "now",
            "candidates": candidates.to_dict(orient="records")
        })
        print("候选标的: " + str(len(candidates)))
        time.sleep(30)

7.4 重启后的恢复逻辑

if __name__ == "__main__":
    state = load_or_init_state()
    if state.get("candidates"):
        print("脚本重启，恢复上一轮候选标的")
        monitor_loop()

8. 适用场景

• 7x24 小时运行的行情监控脚本
• 盘中关键时段不能中断的筛选任务
• 运行在云服务器上，可能因维护重启的环境
• 多个监控脚本共用同一份快照缓存的团队场景

9. 注意事项

• 快照缓存不要保存过期的数据。如果缓存中的时间戳离当前时间超过 5 分钟，建议重新拉全量。
• 状态文件写入频率不要太高，建议每 5 轮或候选标的发生变化时再写入。
• 检查点的时间戳使用北京时间，避免因时区混用导致增量查询条件错误。
• 如果监控脚本分布在不同服务器上，状态文件应该放在共享存储或使用 Redis 等外部缓存。

10. FAQ

Q1：行情监控脚本重启后最快怎么恢复？

A：最快的方式是本地快照缓存。脚本启动时先读取最近一次保存的快照，然后用一次批量请求刷新最新数据。恢复时间通常从数十秒降低到 1 到 2 秒。

Q2：快照缓存应该用什么格式？

A：小规模推荐 JSON，便于调试；大规模推荐 Parquet，读取速度快。Redis 缓存在分布式场景下更合适。

Q3：重启后如果快照数据已经过时怎么办？

A：检查快照时间戳与当前时间的间隔，如果超过 5 分钟则放弃快照，重新拉全量。

Q4：状态文件频繁写入会影响磁盘寿命吗？

A：普通 SSD 数十万次写入寿命足够。可以把写入间隔设到 5 轮以上，或者只在候选标的发生变化时写入。

Q5：分布式部署时监控脚本的状态怎么共享？

A：用 Redis 或 etcd 等外部存储代替本地文件，所有实例读写同一个 key。

Q6：QuantDash 返回的数据可以直接缓存吗？

A：可以。QuantDash Python SDK 返回 Pandas DataFrame，直接使用 to_json 或 to_parquet 序列化即可。

Q7：QuantDash 支持批量行情查询吗？

A：支持。一次请求可以获取多只标的的实时行情快照，适合监控脚本的批量刷新场景。

Q8：增量检查点方案要求数据源提供什么能力？

A：要求 API 支持按时间戳或版本号查询增量数据。如果数据源不支持，就退回到全量快照缓存方案。

11. 总结

行情监控脚本重启后，最不值得做的事情就是重新全量拉取一遍数据：

1. 快照缓存：将最近一次有效快照持久化到本地，重启后秒级恢复。
2. 增量检查点：记录最后一次成功请求的时间点，只请求增量变化。
3. 状态文件：保存候选标的和处理上下文，避免从零开始筛选。

三种方案组合使用，可以让监控脚本在面对意外退出时快速恢复现场，不浪费请求次数，不丢失监控窗口。