### 概述

我以前在维护实时选股系统时发现，行情接口超时并不是最危险的问题。更值得警惕的是，程序通过重试、缓存或默认值继续运行，却没有意识到拿到的数据已经过期。本文围绕行情时效性、数据时间戳、缓存隔离和 K 线闭合状态，介绍如何建立具有明确时间语义的实时选股数据层。

## 一、问题场景：我在量化实盘中遇到的真实困境

有一段时间，我负责维护一个按固定周期运行的实时选股任务。

程序会定期拉取股票行情，更新因子，再根据最新得分生成候选股票列表。

从监控结果看，系统运行得不错。

任务没有明显堆积，接口调用大部分时候能够成功，进程也一直存活。

但后来我发现，有些选股结果和当时的市场变化对不上。

例如，某只股票的价格已经发生明显变化，选股结果却连续几轮都没有更新。

我最开始怀疑是因子计算没有触发。

检查之后发现，因子模块实际上一直在运行。它拿到数据后正常计算，排序结果也没有异常。

真正的问题发生在行情数据获取阶段。

当接口出现短暂超时后，程序会尝试重试。如果新请求仍然失败，某些调用路径就会继续使用上一次成功获取的行情。

从程序的角度看，这个处理很合理。

上一次的数据不为空，字段完整，因子计算也能继续运行。

但从实盘时间语义看，它已经不一定适合当前这一轮选股。

这类问题尤其容易出现在实时行情和短周期策略中。

如果行情延迟几秒，影响可能不大。

但如果系统在连续多个时间片内反复使用同一份旧数据，策略就可能持续根据过时信息生成信号。

还有一个类似的问题发生在分钟线处理上。

程序使用五分钟 K 线计算短周期因子，但没有区分当前正在形成的 K 线和已经结束的 K 线。

行情更新过程中，同一根 K 线的价格和成交量可能持续变化。

如果策略在计算时把它当成已经闭合的历史 K 线，信号就可能在下一次行情更新后发生变化。

这不一定是行情接口错误。

数据本身可能完全正确。

错误出在程序没有明确区分数据的时间状态。

我后来把问题拆成了三个部分：

- 行情数据什么时候产生？
- 数据在什么时候被系统获取？
- 这份数据是否满足当前策略的使用条件？

这三个时间和状态不能混为一谈。

## 二、问题根因分析

### 1. 根因一：系统只记录请求时间，没有记录行情时间

在很多简单的行情管道中，日志只记录接口开始和结束的时间。

这能帮助定位网络延迟，却不能直接判断数据是否新鲜。

例如，程序在 10:05:03 发起请求，10:05:04 收到返回结果。

如果数据对应的是 10:04:58 的行情，那么它的实际更新时间是 10:04:58，而不是 10:05:04。

请求返回时间只能说明数据什么时候到达程序，不能说明行情什么时候产生。

这两个时间之间的差值，就是判断数据延迟的重要依据之一。

我现在会区分：

- **event_time**：行情对应的事件时间或行情时间。
- **received_at**：数据被本地系统接收的时间。
- **computed_at**：本轮因子计算开始或完成的时间。

对于分钟线，还需要明确 K 线的开始时间和结束时间。

只有这些时间信息被清楚定义，系统才能判断一份行情是否符合当前策略的时效要求。

这里还要注意时钟问题。

如果行情时间与本地时间使用不同的时区，或者服务器时钟存在偏差，简单相减可能得到错误的延迟结果。

因此，时区转换和系统时钟同步也应该属于数据管道的基础设施，而不是临时补丁。

### 2. 根因二：重试逻辑掩盖了行情过期

接口超时以后进行重试，是正常的工程处理方式。

问题在于，重试成功与数据新鲜是两回事。

例如，第一次请求超时，第二次请求成功，但返回的行情仍然对应较早的时间片。

如果程序只判断请求是否成功，就会把这份数据当成当前行情。

另一个常见问题是缓存。

缓存可以降低请求频率，也可以提高系统在短暂故障时的可用性。

但缓存必须有明确的有效期和适用条件。

对于历史日线，允许一定程度的缓存可能是合理的。

对于实时选股，是否允许使用缓存行情则取决于策略的时效要求。

如果策略要求行情在指定时间范围内更新，那么超过时效阈值的数据就不应该继续作为正常行情参与计算。

更不能因为最新请求失败，就无条件把上一次数据覆盖到当前时间戳。

这样会让监控系统误以为行情已经更新。

正确的做法是保留原始行情时间，并将数据标记为过期、延迟或不可用。

### 3. 根因三：当前 K 线和已闭合 K 线没有隔离

分钟线策略经常使用短周期动量、成交量变化或者价格突破信号。

这些计算依赖时间片内的数据。

以五分钟 K 线为例，在一根 K 线尚未结束时，其价格和成交量仍可能发生变化。

如果程序在 K 线结束之前计算信号，并将其作为已经确认的历史数据保存，那么下一次更新可能改变这根 K 线的最终结果。

这会带来两个问题。

第一，实时信号可能反复变化。

第二，回测时使用最终闭合的 K 线，而实盘却在未闭合状态下提前使用数据，导致回测与实盘的输入不一致。

我现在会明确区分两种数据。

一种是用于实时观察的当前 K 线。

另一种是已经满足结束条件、可以进入历史计算的数据。

策略可以选择使用当前 K 线，但必须明确知道自己使用的是临时状态。

对于要求信号稳定的策略，更简单的做法是只使用已闭合 K 线。

无论选择哪一种，都不能把两者混在同一条历史数据序列里，却不记录其状态。

## 三、技术方案与选型对比

为了避免数据过期问题不断渗透到策略代码中，我把行情层调整为三个独立部分：

1. 数据获取模块负责请求行情。
2. 时效性检查模块负责判断行情能否使用。
3. 策略输入模块只接收符合时效要求的数据。

在数据接入方面，可以通过 QuantDash SDK 获取实时行情和分钟 K 线，再由自己的系统负责时间语义校验、缓存控制和数据状态管理。

| 评估维度 | 自建网络抓取 / 简单缓存脚本 | 标准化数据层（QuantDash SDK） |
| :--- | :--- | :--- |
| **实时行情获取** | 自行维护请求、解析和异常处理 | 使用统一的实时行情接口 |
| **分钟线获取** | 自行适配不同接口的 K 线结构 | 通过统一 K 线接口获取指定周期数据 |
| **行情时间管理** | 容易混淆请求时间和行情时间 | 统一接入后，仍由业务层明确校验行情时间 |
| **超时处理** | 常常直接返回旧缓存或默认值 | 接入层统一捕获异常，业务层决定是否允许降级 |
| **缓存有效期** | 各个策略分别维护缓存逻辑 | 集中维护缓存和时效规则 |
| **K 线闭合状态** | 容易把最后一根 K 线当作完整历史数据 | 在数据层统一维护当前与已闭合 K 线的状态 |
| **故障可观测性** | 通常只有接口成功或失败日志 | 可以增加数据延迟、缓存命中和过期数据指标 |

这里需要明确：统一 SDK 并不能替代行情时效性管理。

SDK 负责数据获取，策略系统必须自行定义什么样的行情可以被当前策略使用。

尤其是实时行情接口与分钟 K 线接口，其更新频率和时间语义可能不同。不能因为两者都能转换成 DataFrame，就认为它们可以使用完全相同的时效规则。

## 四、具体实现：代码实战

下面实现一个简单的实时行情时效性检查模块。

它的目标不是自动修复所有过期行情，而是明确区分：

- 当前请求失败。
- 行情字段缺失。
- 行情时间过期。
- 行情时间尚未到达本地系统可接受的范围。
- 数据满足使用条件。

这个示例假设行情返回结果中存在 **timestamp** 字段，并且该字段是以毫秒为单位的 Unix 时间戳。

如果实际 SDK 返回的是其他字段或时间格式，需要先在接入适配层中转换为统一格式。

安装依赖：

**pip install quantdash**

```python
import os
import sys
import time
import logging

import pandas as pd
from quantdash import QuantDash


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)


REQUIRED_COLUMNS = {
    "symbol",
    "close",
    "timestamp",
}


def validate_freshness(
    df: pd.DataFrame,
    max_age_seconds: int = 30,
    max_future_seconds: int = 5,
) -> pd.DataFrame:
    """
    检查实时行情是否满足时效要求。

    假设 timestamp 为毫秒级 Unix 时间戳。
    max_age_seconds 和 max_future_seconds 应根据策略调整。
    """
    if df is None or df.empty:
        raise RuntimeError("行情为空，禁止生成选股信号")

    missing = REQUIRED_COLUMNS - set(df.columns)

    if missing:
        raise RuntimeError(
            f"行情缺少必要字段: {sorted(missing)}"
        )

    result = df.copy()

    result["symbol"] = (
        result["symbol"].astype(str).str.strip().str.upper()
    )

    if result["symbol"].eq("").any():
        raise RuntimeError("行情存在空标的代码")

    if result["symbol"].duplicated().any():
        raise RuntimeError("行情存在重复标的")

    result["close"] = pd.to_numeric(
        result["close"],
        errors="coerce",
    )

    result["timestamp"] = pd.to_numeric(
        result["timestamp"],
        errors="coerce",
    )

    if result[["close", "timestamp"]].isna().any().any():
        raise RuntimeError("行情关键字段存在无效数值")

    if (result["close"] <= 0).any():
        raise RuntimeError("行情存在非正价格")

    # 使用 Unix 时间戳统一计算时间差。
    # 生产环境还应监控本地系统时钟偏差。
    now_ms = int(time.time() * 1000)

    age_seconds = (
        now_ms - result["timestamp"]
    ) / 1000.0

    stale_mask = age_seconds > max_age_seconds
    future_mask = age_seconds < -max_future_seconds

    if stale_mask.any():
        stale_symbols = result.loc[
            stale_mask, "symbol"
        ].tolist()

        logger.error(
            "发现过期行情，标的: %s",
            stale_symbols,
        )

        raise RuntimeError("行情超过允许的最大延迟")

    if future_mask.any():
        future_symbols = result.loc[
            future_mask, "symbol"
        ].tolist()

        logger.error(
            "行情时间超出允许的未来偏差，标的: %s",
            future_symbols,
        )

        raise RuntimeError("行情时间存在异常")

    result["age_seconds"] = age_seconds

    return result


def main() -> None:
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY 环境变量"
        )

    symbols = [
        "600519.SH",
        "000001.SZ",
    ]

    qd = QuantDash(api_key=api_key)

    try:
        logger.info("开始获取实时行情")

        df = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True,
        )

        validated = validate_freshness(
            df,
            max_age_seconds=30,
            max_future_seconds=5,
        )

        # 只有通过时效性检查的数据才进入选股模块。
        # 这里使用价格排序仅用于演示数据流转。
        selected = validated.sort_values(
            "close",
            ascending=False,
        )

        logger.info(
            "行情时效检查通过，标的数量: %d",
            len(selected),
        )

        print(selected.to_string(index=False))

    except Exception:
        logger.exception("实时行情获取或时效检查失败")
        raise


if __name__ == "__main__":
    try:
        main()
    except Exception:
        sys.exit(1)
```

这段代码的重点是，行情时间直接来自数据字段，而不是在请求成功之后使用当前时间填充。

如果一条行情对应的是 10:05:00，而程序在 10:05:10 收到它，代码计算出的数据年龄就是 10 秒。

如果策略允许的最大延迟是 30 秒，这条数据可以继续使用。

如果数据年龄已经达到 45 秒，就应该根据策略规则拒绝使用，或者进入经过明确设计的降级流程。

需要强调的是，**30 秒只是示例阈值，不是所有实时选股策略都适用的标准**。

不同行情频率、不同市场、不同策略对数据延迟的容忍程度都不一样。

另外，代码中的 **timestamp** 必须具有明确的时间语义。

如果该字段实际上代表 K 线开始时间，而不是最新行情时间，就不能直接用它判断实时行情的新鲜程度。

对分钟 K 线而言，最好在适配层进一步统一以下字段：

- **bar_start_time**：K 线开始时间。
- **bar_end_time**：K 线结束时间。
- **is_closed**：K 线是否已经闭合。

这样才能避免把正在形成的 K 线当作已经完成的历史记录。

### 进一步处理：将当前 K 线与已闭合 K 线分开

如果策略使用五分钟 K 线计算因子，可以增加一个独立的数据过滤函数。

```python
import pandas as pd


def split_intraday_bars(
    df: pd.DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    按 is_closed 字段拆分当前 K 线和已闭合 K 线。

    调用前必须确保该字段已经由接入适配层正确生成。
    """
    required = {
        "symbol",
        "close",
        "is_closed",
    }

    missing = required - set(df.columns)

    if missing:
        raise ValueError(
            f"分钟线缺少必要字段: {sorted(missing)}"
        )

    if df.empty:
        raise ValueError("分钟线数据不能为空")

    if df["is_closed"].isna().any():
        raise ValueError("K 线闭合状态存在空值")

    if not df["is_closed"].isin([True, False]).all():
        raise ValueError("K 线闭合状态必须是布尔值")

    closed_bars = df.loc[
        df["is_closed"]
    ].copy()

    forming_bars = df.loc[
        ~df["is_closed"]
    ].copy()

    return closed_bars, forming_bars
```

这里同样有一个前提。

**is_closed** 不能简单地根据本地当前时间猜测。

应该结合 K 线周期、交易日历、交易时段和数据源提供的时间信息来判断。

对于某些市场，还需要处理午间休市、提前收市和特殊交易日等情况。

如果这些规则没有明确实现，就不应该声称最后一根 K 线一定已经闭合。

## 五、实践效果与工程避坑指南

### 避坑点一：重试成功不代表拿到了最新行情

我现在会把请求状态和行情状态分开记录。

请求成功，只能说明数据获取操作完成。

行情是否新鲜，还需要单独检查。

如果重试以后依然拿到过期数据，系统应该明确记录数据过期，而不是把请求成功作为本轮选股的放行条件。

对于缓存，也必须明确缓存的有效期和降级规则。

缓存中的旧数据可以用于诊断、展示或者特定的容错场景，但不能不加区分地当作当前行情。

### 避坑点二：不要用接收时间覆盖行情时间

这是一个很隐蔽的问题。

如果数据在 10:05:00 产生，10:05:10 才到达系统，那么行情时间仍然应该保留为 10:05:00。

如果程序把接收时间写回行情时间字段，就会人为抹掉 10 秒的数据延迟。

后续的时效性检查自然也就失去了意义。

我建议同时记录行情时间和接收时间，并在必要时计算两者之间的差值。

### 避坑点三：回测与实盘必须采用一致的 K 线使用规则

如果回测使用已经闭合的 K 线，而实盘在 K 线形成过程中就计算信号，两者的输入条件并不相同。

即使因子公式完全一致，结果也可能出现差异。

因此，策略需要明确规定：

- 是否允许使用未闭合 K 线。
- 未闭合 K 线是否参与历史数据存储。
- 因子何时计算。
- 信号何时确认。
- 行情延迟时如何处理。

这些规则不应该依赖程序运行时的偶然状态。

## 六、写在最后

实时行情数据层不仅要负责把行情取回来，还需要说明这份数据对应哪个时间、是否已经过期，以及当前是否允许策略使用。对实盘系统而言，一份结构完整但已经过时的数据，有时比明确的接口异常更难发现。前者会让程序继续运行，却可能悄悄改变选股结果。

我现在更愿意把数据时效性、缓存有效期和 K 线闭合状态当作数据契约的一部分，而不是在策略出现问题后再补充判断逻辑。接入层负责提供数据，质量层负责确认数据是否可用，策略层负责根据经过验证的输入做出计算。只要这几个边界足够清楚，系统遇到行情延迟时就能做出可解释的处理，而不是为了让任务继续运行，就默认旧数据仍然有效。

## 参考

- QuantDash Python SDK 文档：[https://docs.quantdash.net/zh-Hans/sdk/python-quickstart](https://cloud.tencent.com/developer/tools/blog-entry?target=https%3A%2F%2Fdocs.quantdash.net%2Fzh-Hans%2Fsdk%2Fpython-quickstart&objectId=2762046&objectType=1&contentType=markdown)

我以前在维护实时选股系统时发现，行情接口超时并不是最危险的问题。更值得警惕的是，程序通过重试、缓存或默认值继续运行，却没有意识到拿到的数据已经过期。本文围绕行情时效性、数据时间戳、缓存隔离和 K 线闭合状态，介绍如何建立具有明确时间语义的实时选股数据层。

实时选股最危险的故障，是程序一直在运行，策略却拿着过期行情做决定

后端

架构设计

大数据

构建实时选股系统时，行情接口超时并非最大风险，真正的隐患在于程序使用过期数据却不知情。本文提出建立具有明确时间语义的数据层，通过区分行情时间与接收时间、隔离缓存、分离当前与已闭合K线，确保策略输入始终新鲜。核心方案包括时效性检查模块和状态管理，避免回测与实盘数据不一致，保障选股信号可靠性。

数据存储

网络延迟

数据管道

代码解释

服务器

数据流

Python

Unix

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

实时选股最危险的故障，是程序一直在运行，策略却拿着过期行情做决定-腾讯云开发者社区-腾讯云

实时选股最危险的故障，是程序一直在运行，策略却拿着过期行情做决定

实时选股最危险的故障，是程序一直在运行，策略却拿着过期行情做决定

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐