准实时行情采集器的稳定性设计：客户端限速、自适应节拍与断档补数

原创

用户9138916

修改于 2026-11-01 13:44:47

850

准实时行情采集要稳定，关键在于把“限速”从被动挨打变成主动控制：在客户端用令牌桶约束请求速率，用自适应节拍让轮询频率贴合数据实际刷新节奏，再用基于时间戳的幂等写入和日内分时补数保证数据完整。本文给出一个单进程采集器的分层设计、组件职责和关键代码，示例代码基于 Python 行情 SDK 编写，读者可替换为自己使用的任意行情数据源。

一、问题拆解

采集链路的“不稳定”通常由四类原因叠加而成，处理手段各不相同：

现象	根因	对应设计
频繁收到 429	请求数被放大（逐只请求、多进程重复采集）	请求合并 + 客户端令牌桶
拿到大量重复数据	轮询频率高于数据刷新频率	自适应节拍
偶发超时、连接中断	网络抖动	有上限的指数退避
重启后序列有缺口	进程中断期间无采集	日内分时补数

前两类占多数，且完全可以在客户端解决，这也是本文的重点。

二、总体架构

            ┌──────────────┐
            │  Scheduler   │  自适应节拍（3s ~ 15s）
            └──────┬───────┘
                   │ tick
            ┌──────▼───────┐       ┌──────────────┐
            │ RequestGate  │──────▶│ 行情数据接口 │
            │ 令牌桶+退避   │◀──────│ 快照/分时    │
            └──────┬───────┘       └──────────────┘
                   │ DataFrame
            ┌──────▼───────┐
            │  Deduper     │  symbol → last timestamp
            └──────┬───────┘
                   │ changed rows
     ┌─────────────▼─────────────┐
     │ Sink（CSV/Parquet/队列）    │──▶ 策略 / 监控
     └───────────────────────────┘
            ▲
            │ 重启时触发
     ┌──────┴───────┐
     │ Backfiller   │  日内分时补齐当日分钟序列
     └──────────────┘

设计原则只有一条：全系统只有 RequestGate 一个出口访问数据接口，策略进程只读 Sink，不直接请求。

三、组件职责

组件	职责	关键参数
Scheduler	决定下一轮何时执行	基础间隔 3s、上限 15s
RequestGate	速率控制、错误分级、退避	桶容量、补充速率、最大重试次数
Deduper	按标的时间戳过滤未变化行	内存字典，重启后由 Sink 恢复
Sink	追加写入，按交易日分文件	写入格式
Backfiller	启动时补齐当日分钟序列	标的列表

四、RequestGate：令牌桶与错误分级

令牌桶在客户端提前削峰：即使上层逻辑出错连续触发请求，实际发出的调用也不会超过设定速率。错误按 HTTP 语义分级：429、超时、连接失败可重试；401、403 属于配置问题，直接抛出。

import random
import threading
import time

from quantdash import (AuthenticationError, ConnectionError, PermissionError,
                       QuantDash, RateLimitError, TimeoutError)


class RequestGate:
    def __init__(self, rate=1.0, capacity=3, max_tries=5):
        self.rate, self.capacity, self.max_tries = rate, capacity, max_tries
        self.tokens, self.updated = float(capacity), time.monotonic()
        self.lock = threading.Lock()
        self.retries = 0

    def _acquire(self):
        while True:
            with self.lock:
                now = time.monotonic()
                self.tokens = min(self.capacity,
                                  self.tokens + (now - self.updated) * self.rate)
                self.updated = now
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
            time.sleep(1 / self.rate)

    def call(self, fn):
        for attempt in range(self.max_tries):
            self._acquire()
            try:
                return fn()
            except (AuthenticationError, PermissionError):
                raise
            except (RateLimitError, ConnectionError, TimeoutError):
                self.retries += 1
                if attempt == self.max_tries - 1:
                    raise
                time.sleep(min(30, 2 ** attempt) * random.uniform(0.5, 1))


qd = QuantDash()  # 从环境变量 QUANTDASH_API_KEY 读取凭证
gate = RequestGate(rate=1.0, capacity=3)

速率参数建议从保守值起步，结合监控中的重试次数逐步调整。

五、Scheduler：自适应节拍

以示例所用的数据源为例，实时快照约 3 秒刷新一次。节拍策略是：有数据变化时保持 3 秒；连续多轮无变化（午休、收盘后）则逐步放大到 15 秒；一旦发生重试，同样放大间隔，主动给接口“降压”。

def run_collector(universe="CN_Stock", base=3.0, ceiling=15.0, rounds=6):
    interval, last_ts, log = base, {}, []
    for i in range(rounds):
        t0, retries_before = time.monotonic(), gate.retries
        df = gate.call(lambda: qd.quotes.get(universes=universe, to_dataframe=True))
        changed = df[df["timestamp"] != df["symbol"].map(last_ts)]
        last_ts.update(zip(changed["symbol"], changed["timestamp"]))
        if gate.retries > retries_before or changed.empty:
            interval = min(ceiling, interval * 1.5)
        else:
            interval = base
        cost = time.monotonic() - t0
        log.append((i + 1, len(df), len(changed), round(cost, 2), round(interval, 1)))
        time.sleep(max(0.0, interval - cost))
    return log


for row in run_collector(rounds=6):
    print("轮次=%d 行数=%d 变化=%d 耗时=%.2fs 下一间隔=%.1fs" % row)

收盘后运行时可以看到间隔从 3 秒逐级放大到 15 秒，盘中则稳定在 3 秒附近。

六、Deduper 与 Sink：幂等写入

去重键为“标的 + 快照时间戳”。写入端按交易日分文件、只追加，重启时读取当日文件最后一批记录恢复 last_ts 字典，即可保证同一条快照不会被重复写入。下游按同样的键做 upsert，即使上游偶尔重复投递，结果依然一致。

七、容量估算

以 A 股全市场为例，单次快照返回 5500 余行、18 列，每轮请求 1 次。按 3 秒节拍、每交易日 4 小时连续竞价计算，全天约 4800 次调用；若改为逐只请求，同样的覆盖需要两千多万次调用，二者相差三个数量级以上。

写入侧的压力取决于变化行数而非返回行数。盘中多数标的每轮都有更新，单轮写入量接近全量；午休与收盘后变化行数趋近于零，自适应节拍会同时把调用频率降下来。因此存储容量按“交易时段全量 × 轮数”预估即可，非交易时段可以忽略不计。

八、Backfiller：重启后的断档补数

快照描述的是“此刻状态”，进程中断期间的变化无法从快照恢复。对需要连续分钟序列的标的，在启动时调用日内分时接口批量补齐当日数据，与快照数据按时间对齐后再进入下游。

def backfill_today(symbols):
    data = gate.call(lambda: qd.klines.intraday_batch(symbols))
    return {s: len(bars["timestamp"]) for s, bars in data.items()}


print(backfill_today(["600519.SH", "000001.SZ", "510300.SH"]))

补数列表应限于策略真正关心的标的；全市场分钟级历史更适合收盘后通过分钟 K 线接口按区间同步（分钟 K 线为独立权限）。

九、监控指标

指标	采集方式	告警建议
单轮耗时	Scheduler 日志	连续 5 轮超过基础间隔
返回行数	每轮 len(df)	较前一交易日均值偏离超过 5%
变化行数	Deduper 输出	交易时段连续 10 轮为 0
重试次数	RequestGate 计数	每分钟超过 3 次
认证/权限错误	异常捕获	出现即告警