摘要

你用 Python 接入了 A 股实时行情接口,第一次请求返回 HTTP 200 和 code=0,价格看起来正常。但接口返回成功不等于数据可用——data 可能是空数组、symbol 可能跟请求不一致、last_price 可能是无法解析的字符串、timestamp 语义可能跟你想的不一样。本文给出一套首次接入后的字段校验流程:先看懂返回结构,再逐字段检查 symbol、价格和 timestamp,最后用 raw_snapshot 留痕。附带 Python 伪代码、业务侧留痕字段表和 7 个失败分支。读完你会拿到一张可直接集成到取数脚本里的检查清单。

一、这篇文章解决什么,不解决什么

适合谁看

你会用 Python 调一个 A 股实时行情接口,拿到了 HTTP 200 和 code=0,但你不确定:这个价格能直接展示吗?symbol 跟请求时一致吗?timestamp 是哪个时刻?出错时有没有留下足够信息供事后排查?

解决什么

告诉你接口返回成功后,应该按什么顺序检查哪些字段。只解决一件事:怎样判断一次行情查询的结果是否真的“可用”。

读完你能拿走什么

  • 一张字段检查顺序:先看返回结构,再查 symbol,再验价格,再读 timestamp,最后留 raw_snapshot。
  • 一张业务侧留痕字段表:8 个字段,供你直接加到自己的取数脚本里。
  • 一段 Python 伪代码:演示从 data 数组中匹配 symbol、解析 last_price、检查 timestamp 的完整流程。
  • 7 个失败分支:每个失败场景的处理方式。

不解决什么

不讨论投资、选股、策略有效性。不比较数据源优劣。不承诺数据永远正确。

二、为什么 HTTP 200 和 code=0 仍然不能直接入库

HTTP 200 只代表“服务端收到了请求并给出了响应”。code=0 只代表服务端处理了你的请求。它们不承诺以下任何一件事:

  • data 数组一定非空。无效 symbol 可能返回 code=0message=successdata=[]——如果你只看 code,会以为一切正常。
  • 返回的 symbol 跟你请求的 symbol 逐字符一致。
  • last_price 是可解析的数值,而不是空字符串或无法解析的占位符。
  • timestamp 的精度和语义跟你以为的一致。
  • 请求失败时返回的信息能被你的代码正确识别。

接口返回成功是传输层和业务层的初步确认。数据是否可用,是业务侧必须独立完成的判断。

下面按顺序逐层检查。

三、先看返回结构:code、message、data 数组

第一步不是查具体字段,而是确认整个响应体的结构是否符合预期。

一个典型的 ticker 返回结构是顶层 codemessagedata,其中 data 是数组。你的校验逻辑第一步就是确认这三件事:code 是否为 0、data 是否为 list、data 是否非空。

空数组是独立失败分支,不能只看 code 就跳过。

如果 code 非 0,或者 data 不是 list,或者 data 为空——直接进入失败分支,不要继续往下查 symbol 和价格。错误 Key 会返回 HTTP 401 和结构化错误信息,这属于鉴权失败,不应重试。

四、查 symbol:返回的代码跟你请求的一样吗

为什么先查它

你请求了 600519,返回的 symbol 可能是 600519.SH,也可能完全不同。symbol 不一致,后面所有检查都白做——你在检查一个不是你想要的品种的数据。

怎么查

data 数组中按 symbol 找到匹配的 item。如果找不到,属于 symbol 不匹配失败分支。找到后,拿出请求参数里的 symbol 和响应 item 里的 symbol,逐字符比对。如果文档没明确约定自动补后缀的行为,任何静默修正都应视为需要确认的异常。

检查项 通过标准 失败处理
data 非空且为数组 isinstance(data, list)len(data) > 0 阻断,记录空数据
symbol 存在于 data 中 能从 data 数组中匹配到请求的 symbol 阻断,记录 symbol 不匹配
symbol 一致性 request_symbol == response_symbol 阻断,记录差异

五、验 last_price:用 Decimal,别用 float

为什么它比 timestamp 更优先

价格是行情的核心。如果 last_price 不可解析,timestamp 再精确也没用。last_price 等价格/数量类字段应按字符串处理,客户端使用 Decimal(str(value)) 解析。直接拿 float 处理,精度损失会在后续计算中累积。

怎么查

Decimal 解析,检查 is_finite()。解析失败或非有限值时,阻断入库,不默认为 0。把 NaN 或 Infinity 写进数据库,比空值更难排查。

检查项 通过标准 失败处理
价格可解析性 非空字符串,Decimal 解析成功且 is_finite() 阻断,记录原始值,不补默认值

六、读 timestamp:13 位毫秒不代表低延迟

为什么它不是“能转成数字就行”

ticker 的 timestamp 是 13 位毫秒时间戳。但这个精度只代表字段格式——不要把时间戳精度写成低延迟或数据新鲜度承诺。 高精度时间戳只能说明源头记录得有多细,不能说明数据多快到达你的脚本。

更关键的是语义:这个 timestamp 是交易所生成行情的时刻,还是服务端处理完数据的时刻?文档是否说明了生成位置?如果没有说明,至少标注“语义待确认”,不直接参与跨市场排序。

怎么查

先确认类型为 int 且非 bool,再确认单位(13 位为毫秒)。如果文档没有说明时间戳的生成位置,标注“语义待确认”。

检查项 通过标准 失败处理
timestamp 可解释性 类型为 int 非 bool,13 位毫秒 标注“语义待确认”,保留原始值

七、分支处理:空 data、无效 symbol、非交易时段、权限错误

前三个字段都通过了。但空 data 怎么处理?非交易时段的数据能用吗?权限错误要不要重试?

空 data 分支:无效 symbol 可能返回 code=0message=successdata=[]。这是合法返回,不是异常——但它意味着你请求的品种没有数据。你的代码必须把空数组作为独立失败分支处理,不能只看 code。

非交易时段分支:如果数据源提供 market_state 或 status 字段,则纳入校验——非交易时段的价格应标注状态,不触发误报。如果数据源不提供这类字段,你的业务侧需要结合交易日历、交易时段和空数据分支自行判断:当前时间是否在交易时段内?空 data 是因为非交易时段还是 symbol 无效?

权限错误分支:错误 Key 返回 HTTP 401 和结构化错误信息。鉴权失败不应重试。

检查项 通过标准 失败处理
空 data 处理 区分“非交易时段”和“无效 symbol” 按错误类型分支处理
权限错误 HTTP 401,结构化错误信息 阻断,不重试

八、raw_snapshot 怎么留痕

前面全部通过,数据入库。三个月后你复查某个回测结果,怀疑某天价格有问题。如果你只存了处理后的价格,没有保存当时的原始返回体——你永远无法确认是数据源当时就返回了这个值,还是你的处理逻辑在某个环节改了它。

raw_snapshot 不是日志,是证据。 它不帮你判断对错,但它让每一次判断都可被复查。

怎么留

把完整的响应 JSON 序列化后保存,同时生成一个哈希摘要用于快速校验。注意脱敏:Key 和 Token 不进入快照。

字段 含义 用途
raw_snapshot 完整响应 JSON 字符串 事后回放、复现现场
raw_snapshot_ref 哈希摘要 快速校验原始数据是否被修改
checked_at 客户端校验通过时间 区分行情时间和校验时间

九、业务侧留痕字段表

以下字段为业务侧建议记录的留痕字段,用于接入校验和事后排查。它们都不是任何行情数据源的原生字段。

字段名 来源 含义
source 客户端定义 数据源标识
symbol 请求参数 请求的标的代码
request_params 客户端记录 请求参数快照
checked_at 客户端生成 校验通过并入库的时间
payload_timestamp 响应字段 行情数据自带的时间戳
raw_snapshot_ref 客户端生成 原始响应体哈希摘要
validation_status 客户端判断 校验状态:passed / partial / failed
failure_reason 客户端判断 失败原因

十、Python 伪代码:ticker 快照校验骨架

下面的代码演示校验流程。这是教学骨架,不是完整生产代码。 所有字段路径以数据源文档和实测为准。

from decimal import Decimal, InvalidOperation
from datetime import datetime, timezone
import hashlib, json
from typing import Dict, Optional


def validate_ticker_snapshot(response: Dict,
                              request_symbol: str,
                              request_params: Dict) -> Dict:
    """
    校验一次 A 股实时行情快照的返回结果。
    教学骨架,字段路径以数据源文档和实测为准。
    """
    report = {"checked_at": datetime.now(timezone.utc).isoformat(),
              "source": None,
              "symbol": request_symbol,
              "request_params": request_params,
              "checks": {},
              "validation_status": "passed",
              "failure_reason": None,
              "raw_snapshot_ref": None}

    # 保存原始快照引用(不保存 Key/Token)
    snapshot = json.dumps(response, ensure_ascii=False, default=str)
    report["raw_snapshot_ref"] = hashlib.sha256(snapshot.encode()).hexdigest()[:16]

    # ---- 1. 检查顶层结构 ----
    code = response.get("code")
    message = response.get("message", "")
    data = response.get("data")

    # code 非 0:业务错误
    if code != 0:
        report["validation_status"] = "failed"
        report["failure_reason"] = f"business error: code={code}, message={message}"
        return report

    # data 必须是 list
    if not isinstance(data, list):
        report["validation_status"] = "failed"
        report["failure_reason"] = f"data is not a list: {type(data).__name__}"
        return report

    # data 为空数组(无效 symbol 可能返回此状态)
    if len(data) == 0:
        report["validation_status"] = "failed"
        report["failure_reason"] = "data is empty array"
        return report

    # ---- 2. 从 data 数组中按 symbol 匹配 item ----
    item: Optional[dict] = None
    for d in data:
        if isinstance(d, dict) and d.get("symbol") == request_symbol:
            item = d
            break

    if item is None:
        report["validation_status"] = "failed"
        returned_symbols = [d.get("symbol") for d in data if isinstance(d, dict)]
        report["failure_reason"] = f"symbol not found in data, returned: {returned_symbols}"
        return report

    # ---- 3. last_price 校验 ----
    last_price = item.get("last_price")
    price_ok = False
    if isinstance(last_price, str) and last_price.strip():
        try:
            d = Decimal(last_price)
            price_ok = d.is_finite()
        except (InvalidOperation, ValueError):
            pass
    report["checks"]["price"] = {"passed": price_ok, "raw": last_price}
    if not price_ok:
        report["validation_status"] = "failed"
        report["failure_reason"] = f"last_price unparseable: {last_price}"
        return report

    # ---- 4. timestamp 校验 ----
    timestamp = item.get("timestamp")
    ts_ok = False
    if isinstance(timestamp, int) and not isinstance(timestamp, bool):
        # 13 位毫秒时间戳
        if len(str(timestamp)) == 13:
            ts_ok = True
    report["checks"]["timestamp"] = {"passed": ts_ok, "raw": timestamp}
    if not ts_ok:
        report["validation_status"] = "failed"
        report["failure_reason"] = f"timestamp cannot be interpreted: {timestamp}"
        return report

    return report

代码要点

  • 先查 code,再查 data 类型和是否为空数组。
  • data 数组中按 symbol 遍历匹配 item,而不是用 data[0] 假设第一个就是。
  • last_priceDecimal(str(value)) 解析,检查 is_finite()
  • timestamp 必须是 int 且非 bool,13 位毫秒。
  • 每个失败分支都填写 validation_statusfailure_reason
  • 原始快照的哈希引用在函数开头就保存,确保即使后续校验失败也不会丢失。

十一、7 个失败分支

# 失败场景 触发条件 处理方式
code 非 0 业务错误 阻断,记录 code 和 message
data 不是 list 返回结构异常 阻断,记录实际类型
data 为空数组 无效 symbol 等 阻断,区分非交易时段和无效 symbol
symbol 不匹配 data 中找不到请求的 symbol 阻断,列出实际返回的 symbol 列表
last_price 不可解析 空字符串、非有限值等 阻断,不补默认值
timestamp 不可解释 类型错误或位数不对 阻断,标注语义待确认
raw_snapshot 未保存 快照保存失败 标记为不可复查

十二、TickDB 在这套流程里的位置

上面这套字段校验流程不依赖任何特定数据源。如果你用 TickDB 作为统一实时行情 API 的候选入口,它在这套流程中的工程角色是:

  • REST ticker:端点 /v1/market/ticker,参数 symbols,鉴权 X-API-Key。返回顶层 code/message/data,其中 data 是数组,每个 item 包含 symbollast_pricetimestamp 等字段。适合定时快照查询和字段核对。
  • WebSocket:适合持续行情推送。首次接入后的字段校验逻辑同样适用于推送的每条数据。
  • MCP:适合 AI 工具按需查询真实行情,不替代 REST 的定时探测和字段校验。

TickDB 不内置接入校验流程——本文的字段检查、留痕逻辑和失败分支处理需要你自己实现。 所有业务侧留痕字段(validation_statusfailure_reasonraw_snapshot_refchecked_atrequest_params 等)均为自定义设计,不是 TickDB 原生字段。具体端点、字段路径、错误码枚举和返回结构以官方文档和实测为准。

十三、常见问题

Q1:A 股实时行情 API 返回 200 和 code=0,是不是就能直接用了?

不能。无效 symbol 可能返回 code=0message=successdata=[]——空数组必须作为独立失败分支处理。必须逐字段检查后再入库或展示。

Q2:Python 获取股票实时行情,优先查哪个字段?

先看 data 是不是 list 且非空,再从数组中按 symbol 找匹配 item。结构不对,后面所有字段检查都不可靠。

Q3:last_price 为什么要用 Decimal 解析?

价格/数量类字段按字符串处理,客户端应使用 Decimal(str(value))。直接用 float 会引入浮点精度损失,后续计算会累积误差。

Q4:timestamp 是 13 位毫秒,说明数据延迟很低吗?

不。13 位毫秒只代表字段格式——高精度时间戳只能说明源头记录得有多细,不能说明数据多快到达你的脚本。不要把时间戳精度写成低延迟或数据新鲜度承诺。

Q5:TickDB 的 ticker 返回可以直接入库吗?

TickDB 返回结构化 JSON,但入库前仍建议按本文流程逐项校验,并保存 raw_snapshot。接口返回成功不等于数据在你的业务场景里完全可用。

十四、边界声明

  • 一次 symbol 查询成功只证明此时此刻该品种可返回有效数据,不推导所有 A 股品种在所有时段均可正常返回。
  • 本文所有业务侧留痕字段均为自定义设计,不是任何行情数据源的原生字段。
  • TickDB 不内置接入校验流程——本文的字段检查、留痕逻辑和失败分支处理需要你自己实现。
  • 本文不涉及投资建议、收益判断、策略有效性。
  • 不写延迟、SLA、覆盖数量、价格、排名。
  • 所有端点、字段路径和 timestamp 语义以官方文档和实测为准。

你们接入 A 股实时行情 API 后,第一件事是校验 symbol 和 last_price,还是直接展示?有没有把 raw_snapshot 留痕写进取数脚本?

📡 本文以 TickDB 作为 A 股实时行情 API 示例。文中代码为 Python 伪代码骨架,不依赖任何特定数据源的端点或字段。本文仅讨论接入校验的工程方法,不构成投资建议。

Logo

智能硬件社区聚焦AI智能硬件技术生态,汇聚嵌入式AI、物联网硬件开发者,打造交流分享平台,同步全国赛事资讯、开展 OPC 核心人才招募,助力技术落地与开发者成长。

更多推荐