# A 股批量行情的完整性与时效校验

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-12794355/77f4e0d25b7c2fdfd50cd10271795067.png)


批量接口返回了五千条行情，是否就说明五千只股票的数据都已收到？如果其中有重复记录、范围外证券，或者沿用了旧价格，单看条数就会得出错误结论。

检查这类数据，可以分成两个问题：目标证券是否都有记录，以及记录的价格和时间是否适合当前使用。下面用一组模拟数据实现这两类检查。示例只使用 Python 标准库，不需要访问行情服务；代码按顺序放进同一个文件即可运行。

## 清单决定检查范围

先保存一份带有效日期的证券清单。若目标是沪深京上市人民币普通股，就把 ETF、指数、债券等放在其他集合中；停牌股票仍属于上市集合，不能因为当天没有成交就从清单删除。

清单可以通过上交所、深交所和北交所公布的股票列表核对。实际取得清单时要检查分页、证券类型和有效日期，找到列表入口并不代表已经取得完整数据。保存原始文件和下载时间，便于以后解释清单差异。

代码使用字符串保存，保留前导零，并把交易所纳入标识。接入不同数据源时，先把它们的代码映射到同一套标识，再进行集合比较。清单中的新增、删除和代码变更应有生效日期，不能把一次下载缺页当作证券退出市场。

在实际系统中，可以保留三个集合：U 是目标清单，S 是数据源目录，R 是本轮返回代码。U 与 S 的差异用于检查覆盖范围和代码映射；目标中已经能查询、这一轮却没有返回的证券，再检查采集失败和初始化状态。无论使用哪个数据源，返回记录数都不能代替这些集合关系。

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-12794355/d6483a108aa475331c8d7328d8531baf.png)


*图 1：先用清单确定范围，再分别检查目录缺口、本轮缺失和范围外记录。去重前必须保留重复明细。*

## 构造一组会暴露问题的输入

以下代码中的证券标识、价格和时间都是模拟值。`A.SH` 等符号只用于演示，不应发送给真实接口。目标集合有四个成员，返回数据却有五条：A 重复，C 缺失，另有一条范围外的 ETF；B 的源时间较旧，D 的价格为零且时间超前。

```python
import json
from collections import Counter
from decimal import Decimal, InvalidOperation

NOW = 1_790_733_600_000  # 固定的模拟参照时间，毫秒
expected = {"A.SH", "B.SZ", "C.BJ", "D.SH"}
rows = [
    {"symbol": "A.SH", "price": "10.20", "ts": NOW - 1000},
    {"symbol": "A.SH", "price": "10.20", "ts": NOW - 1000},
    {"symbol": "B.SZ", "price": "8.50", "ts": NOW - 120000},
    {"symbol": "D.SH", "price": "0", "ts": NOW + 10000},
    {"symbol": "ETF.SH", "price": "1.20", "ts": NOW},
]
```

这里采用一个很小的内部数据结构：`symbol` 是映射后的证券标识，`price` 是最新成交价，`ts` 是源数据时间。真实响应的字段名可能不同，应在适配层转换；它不是任何供应商的原始响应格式。

## 先核对集合 再校验字段

先统计每个代码出现的次数，再构造返回集合。不能一开始就把记录存进以代码为键的字典，否则重复数据会被覆盖，后续检查便看不到它。

下面的函数同时返回缺失、额外、重复和结构异常明细。价格要求能解析为有限正数；这只是“可作为本例最新成交价使用”的条件，零值也可能是数据源表达暂无成交的约定，需要单列处理，不能直接归咎于数据源错误。

```python
def inspect_quotes(expected, rows, now_ms,
                   max_age_ms=30000, future_ms=2000):
    if max_age_ms < 0 or future_ms < 0:
        raise ValueError("时间阈值不能为负数")
    if type(now_ms) is not int or now_ms < 946684800000:
        raise ValueError("参照时间应为2000年后的毫秒整数")
    counts, invalid_rows, issues = Counter(), [], []

    for index, row in enumerate(rows):
        symbol = row.get("symbol") if isinstance(row, dict) else None
        if not isinstance(symbol, str) or not symbol.strip():
            invalid_rows.append(index)
            continue
        counts[symbol] += 1
        if symbol not in expected:
            continue
        flags = []

        try:
            value = row.get("price")
            if value is None or isinstance(value, bool):
                raise ValueError()
            price = Decimal(str(value))
            if not price.is_finite() or price <= 0:
                raise ValueError()
        except (InvalidOperation, ValueError):
            flags.append("价格不可用")

        ts = row.get("ts")
        if type(ts) is not int or ts < 946684800000:
            flags.append("时间格式或单位待核")
        elif ts > now_ms + future_ms:
            flags.append("时间超前待核")
        elif now_ms - ts > max_age_ms:
            flags.append("源数据年龄待核")
        if flags:
            issues.append({"row": index, "symbol": symbol,
                           "flags": flags})

    returned = set(counts)
    return {
        "missing": sorted(expected - returned),
        "extra": sorted(returned - expected),
        "duplicate": sorted(s for s, n in counts.items() if n > 1),
        "invalid_rows": invalid_rows,
        "coverage": len(expected & returned) / len(expected)
                    if expected else None,
        "issues": issues,
    }
```

示例用 2000 年之后的毫秒整数作为时间格式检查下界，适用于这里的当前行情任务，并非所有历史数据的通用规则。明显的秒级时间会落入待核，微秒级或远未来数值会触发超前检查；程序不会猜测单位后自动修正。覆盖率只表示“目标代码有记录”的比例，不能代表记录内容可用的比例。

## 运行并解释检查结果

把下面两行接在前面的代码之后，保存为 `quote_check.py`，运行 `python3 quote_check.py`。固定模拟时间使结果可重复，不会因为运行日期改变而把所有样本都判为过旧。

```python
report = inspect_quotes(expected, rows, NOW)
print(json.dumps(report, ensure_ascii=False, indent=2))
```

输出如下，`row` 为从零开始的输入行号：

```json
{
  "missing": ["C.BJ"],
  "extra": ["ETF.SH"],
  "duplicate": ["A.SH"],
  "invalid_rows": [],
  "coverage": 0.75,
  "issues": [
    {"row": 2, "symbol": "B.SZ", "flags": ["源数据年龄待核"]},
    {"row": 3, "symbol": "D.SH",
     "flags": ["价格不可用", "时间超前待核"]}
  ]
}
```

五条返回记录，只覆盖四个目标中的三个，覆盖率为 75%。D 虽然有记录，但内容不可直接使用；这说明完整性与字段质量必须分开报告。对重复的 A，先检查两条内容是否相同，再决定去重。同一代码、同一时间却出现不同价格时，应记录冲突并复核，不能默认最后一条就是正确值。

`B.SZ` 的源时间比参照时间早 120 秒，因此超过示例的 30 秒观察阈值。结果写成“待核”，因为年龄超限还不足以证明传输故障。30 秒与 2 秒未来容差都是本例参数，应根据业务容忍时间、数据源时间语义及机器时钟条件配置。

## 时间检查必须带上交易状态

源时间、本地接收时间、业务读取时间应分开保存。源数据年龄由读取时间减去源时间得到；缓存年龄由读取时间减去接收时间得到。两个差值回答的问题不同：刚收到的响应也可能包含旧行情。

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-12794355/a2c1902c0c6ce6eace127e1e05ab4397.png)


*图 2：三种时间口径的示意。源时间到接收时间的差值还可能包含上游处理、缓存与时钟偏差，不能直接当作网络延迟。*

检查保存的快照时，可将当时接收时间传给 `now_ms`，分析收到数据那一刻的年龄；检查当前缓存时传入当前时间。应用内部统一用毫秒运算，显示时再转换时区。不要用本地接收时间替换源时间，否则每次重新获取同一份旧数据都会显得“刚刚更新”。

最新成交价没有变化，可能是没有新成交；增量通道未推送某只股票，也可能只是没有字段变化。因此，连接心跳正常、记录存在、时间较新、价格可用应分别记录，不能用其中一个条件替代其他条件。

还要用交易日历、市场阶段和独立的停复牌信息解释结果。午休、休市或已确认停牌时，不直接沿用连续交易阶段的报警条件；状态未知就保留待核，不根据零成交量或价格没变反推停牌。这段示例没有自动接入状态服务，所以它输出的是诊断线索，而不是最终可交易判断。

## 接入真实数据时保留相同检查入口

采集与校验可以分开：采集层负责请求、权限、分页、压缩和原始响应保存，适配层转换证券标识、价格字段和时间单位，最后把标准化记录交给同一个检查函数。检查结果保留原始行号，便于回查响应。

不同数据源的响应先经过适配层，再进入同一套检查逻辑。适配时应保留原始值，明确字段语义与时间单位；标识转换失败或字段缺失应作为异常保留，不能悄悄丢弃整条记录。下图给出一种字段映射示例。

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-12794355/ec9f211df61a2c0c31b1fda67f314190.png)


*图 3：快照字段映射示例。适配负责统一结构，校验负责判断记录是否符合使用条件。*

按批获取数据时，给每轮采集分配轮次标识，把每批接收时间和失败状态单独保存。某批失败后，整轮应标为部分完成，不能把旧缓存混进本轮后宣称覆盖齐全。批量响应中的证券也未必具有同一个源时间，不能拿整轮完成时间代表每只证券的更新时间。

使用增量更新时，检查对象应是“当前维护的证券状态表”，而不是某一个增量消息。重连后建立新的恢复轮次，对照清单重新确认缺口；如果协议没有明确的快照结束标记，不能因收到第一批数据就判为初始化完成。对较旧记录、重复记录和同时间冲突分别处理，必要时用快照复核。

## 将检查结果用于排查

实际落地时，每轮先核对任务是否全部完成，再分别统计缺失记录、重复记录、字段异常和时间待核。缺失优先回查清单版本、映射结果与批次失败；时间待核先确认单位、时钟和市场状态。保存原始响应与本轮检查结果，修复后用同一轮样本复算，才能判断问题是否消除。

这套检查可以发现明确的集合缺口、无效字段和需要复核的时间异常。它不能单独证明成交价正确，也不能证明更新过程从未丢消息。进一步验证时，应保留同一证券、同一价格类型和相近时间窗口下的独立参考样本，先排除口径差异，再判断数据是否有误。

批量接口返回了五千条行情，是否就说明五千只股票的数据都已收到？如果其中有重复记录、范围外证券，或者沿用了旧价格，单看条数就会得出错误结论。

A 股批量行情的完整性与时效校验

编程语言

开发工具

数据库

返回五千条行情并不能证明覆盖五千只股票。检查数据需从两个维度入手：目标证券是否都有记录，以及记录的价格和时间是否可用。先核对集合关系，统计缺失、额外及重复记录，再逐一校验价格正数性和时间合理性。条数不能代替集合覆盖率和字段质量检查，时间与价格问题需分开报告并保留原始行号追踪。

网络延迟

数据结构

代码解释

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

A 股批量行情的完整性与时效校验-腾讯云开发者社区-腾讯云

A 股批量行情的完整性与时效校验

A 股批量行情的完整性与时效校验

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐