Python 获取 A 股实时行情:HTTP 200 到手后,先验 symbol、last_price 和 timestamp 再入库
摘要
你用 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=0、message=success、data=[]——如果你只看 code,会以为一切正常。- 返回的 symbol 跟你请求的 symbol 逐字符一致。
last_price是可解析的数值,而不是空字符串或无法解析的占位符。- timestamp 的精度和语义跟你以为的一致。
- 请求失败时返回的信息能被你的代码正确识别。
接口返回成功是传输层和业务层的初步确认。数据是否可用,是业务侧必须独立完成的判断。
下面按顺序逐层检查。
三、先看返回结构:code、message、data 数组
第一步不是查具体字段,而是确认整个响应体的结构是否符合预期。
一个典型的 ticker 返回结构是顶层 code、message、data,其中 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=0、message=success、data=[]。这是合法返回,不是异常——但它意味着你请求的品种没有数据。你的代码必须把空数组作为独立失败分支处理,不能只看 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_price用Decimal(str(value))解析,检查is_finite()。- timestamp 必须是 int 且非 bool,13 位毫秒。
- 每个失败分支都填写
validation_status和failure_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 包含symbol、last_price、timestamp等字段。适合定时快照查询和字段核对。 - WebSocket:适合持续行情推送。首次接入后的字段校验逻辑同样适用于推送的每条数据。
- MCP:适合 AI 工具按需查询真实行情,不替代 REST 的定时探测和字段校验。
TickDB 不内置接入校验流程——本文的字段检查、留痕逻辑和失败分支处理需要你自己实现。 所有业务侧留痕字段(
validation_status、failure_reason、raw_snapshot_ref、checked_at、request_params等)均为自定义设计,不是 TickDB 原生字段。具体端点、字段路径、错误码枚举和返回结构以官方文档和实测为准。
十三、常见问题
Q1:A 股实时行情 API 返回 200 和 code=0,是不是就能直接用了?
不能。无效 symbol 可能返回 code=0、message=success、data=[]——空数组必须作为独立失败分支处理。必须逐字段检查后再入库或展示。
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 伪代码骨架,不依赖任何特定数据源的端点或字段。本文仅讨论接入校验的工程方法,不构成投资建议。
更多推荐


所有评论(0)