A股数据源怎么选？从历史数据下载到实时行情API，按用途做选择

A股数据源怎么选？从历史数据下载到实时行情API，按用途做选择

摘要： 寻找 A 股数据源时，历史回测与实时行情接入需要关注的条件并不相同。本文从历史数据下载、日线与分钟数据、实时行情 API 三类需求出发，说明如何判断数据范围、更新方式和接入成本，并给出试接与验证步骤，帮助你选择适合当前任务的数据来源。

选择 A股数据源，第一步是把任务写清楚。下载几年的日线做回测、分析某一天的分钟走势、给页面持续更新最新价格，虽然都叫“获取行情”，需要验证的条件却不同。

一个能返回最新价的接口，未必能下载指定年份的分钟历史；一份已经保存好的日线文件，也不能代替盘中持续更新的数据流。先确定用途，再比较候选来源，通常比先找一张供应商排行榜更容易做出决定。

先分清任务：历史下载，还是持续获取实时行情

可以先写一句需求：“我要哪些股票、哪个区间、什么粒度的数据，多久更新一次，最后用于什么功能。”例如，“下载十只股票过去三年的日线，每天收盘后补齐，用于价格策略回测”，就已经明确了大部分筛选条件。

当前任务	优先验证什么	最容易选错的地方
历史日线下载与回测	起止日期、复权方式、缺失和修订	把最新一段记录当成完整历史
历史分钟分析	指定日期能否返回、周期边界、交易时段	把近期分钟接口当成多年分钟档案
实时行情看板或提醒	快照或推送、报价时间、限制和恢复	把请求成功当成行情一直在更新

图 1：三类任务对应不同的验收重点，图示不是供应商排名。

“历史”和“实时”也不一定由同一个来源完成。研究库可以使用批量历史数据，在线应用使用另一条实时入口；但两者结合时，需要统一证券标识、时间、复权和单位，不能直接把两个返回表拼在一起。

历史数据下载：先验证区间，再看记录数

先查实际覆盖。 文档写“支持历史数据”，还不够回答你的问题。至少要确认最早可查日期、单次返回上限、分页或时间分段方式，以及当前权限是否允许指定起止时间。请求了三年，返回一千条，不代表三年已经下载完整；返回数量刚好达到上限时，更应该检查是否被截断。

日线与分钟线要分别验收。AKShare 官方文档中，东财 stock_zh_a_hist 提供指定周期和日期区间的历史行情；另一个 stock_zh_a_hist_min_em 接口的 1 分钟数据限定近 5 个交易日且不复权。这个限制属于具体接口，不能外推为整个工具库的统一能力，也不能因为设置了更早的开始日期，就认为一定能取得更早的数据。

再统一字段口径。 回测前至少记录证券代码、日期或时间、开高低收、成交量、成交额，以及复权方式。数量单位不能靠列名猜：相同的“成交量”可能有不同单位，成交额也不一定使用同一数量级。先按对应接口文档转换，再比较结果。

复权选择应服务于研究方法。不复权价格、前复权价格和后复权价格不能混用。发生公司行动或上游修订后，再次下载的历史价格可能变化，因此应保存查询日期、复权参数和原始版本，避免把数据版本变化误认为策略变化。

最后解释缺失。 缺少一条记录，可能来自休市、停牌、尚未上市、接口省略无成交周期，也可能是采集失败。先结合交易日历、证券状态和接口规则分类，再决定是否补数。不要把缺失价格填成 0，也不要默认每只股票每天都有相同数量的分钟记录。

一个可执行的历史试接办法是：先选三只股票和一个明确的日期区间，下载并保存原始结果；统计每只股票的最早、最晚时间、总行数和重复键；按日期列出缺口并解释原因。分钟数据再增加午间边界、开收盘附近和最后一根 K 线检查。仍在形成的 K 线不能直接当成已经结束的周期参与回测。

实时行情API：比较更新方式与异常行为

实时接口首先要区分两种消费方式：REST 快照是你发一次请求，得到一次当前状态；WebSocket 是建立连接并订阅后持续收消息。前者便于做少量自选股原型，后者适合按消息驱动更新。两种协议本身都不能证明低延迟，也不保证消息包含每一笔成交。

对于报价时间，至少保留两个时刻：上游记录的报价时间，以及客户端收到响应的时间。 两者的差包含上游数据年龄、传输和处理时间，还受本机时钟影响，不能直接叫“网络延迟”。连续请求成功但报价时间不变，需要结合交易时段和标的状态判断。

调用限制要换算到自己的规模。假设一次可查询 50 个标的，应用需要轮询 500 个标的，每 5 秒完成一轮，那么理想情况下每轮需要 10 次请求，即每分钟约 120 次；这还没计入重试、历史回补和其他使用者。这里是容量估算示例，不是任何套餐承诺。

异常行为也应纳入试接：无效密钥如何返回，超时是否可识别，部分标的缺失如何显示，触发限制后怎样退避，重连后是否需要重新订阅。应用可以保留最后有效值，但必须同时显示旧数据状态；空响应和失败都不应悄悄变成“最新价为 0”。

免费工具与商业接口：比较完成任务的总投入

免费工具适合用小样本验证研究思路，但需要把接口变化、补数、清洗和维护时间计入成本。商业接口提供哪些数据、权限和支持，要逐项核实；付费也不能替代数据验收。

AKShare 适合作为工具库入口来考察，选型时应落实到具体函数和上游来源。Tushare 则需要按接口核对权限：官方权限表把常规积分接口与独立权限接口分开，分钟数据不包含在常规积分范围内。拿到日线权限后，不能顺推为已经拥有分钟历史或实时数据权限。

任务与条件	可以先试哪类入口	继续采用的条件
学习采集、小范围日线研究	公开数据工具或满足权限的数据接口	覆盖目标区间，隔天可重跑，维护量可接受
多年分钟回测	明确提供目标分钟档案的数据服务	区间和粒度验收通过，权限与批量回补可行
长期运行的行情看板或提醒	有明确更新和调用规则的实时接口	时效、失败展示、恢复及调用容量满足要求

这张表给的是试接顺序，不是质量排名。比较总成本时，把数据权限、首次历史回补、日常调用、存储、异常修复和迁移工作一起列出；需要对外展示时，还应把对应使用许可作为采购核对项。低接口费用并不一定对应低总投入，套餐价格较高也不自动说明更适合你的用途。

TickDB 适合什么场景？

如果你的任务是把几只 A 股的最新价格接入网页看板、自选股页面，或为应用提供行情查询功能，可以将 TickDB 纳入实时行情 API 的试接范围。本文后面的示例使用其行情快照接口，检查最新价、报价时间和标的是否完整返回，适合用来验证这类需求的接入链路。

如果你的主要目标是下载多年日线做回测，或者构建连续的分钟历史数据库，则需要另行确认历史接口的覆盖区间、数据粒度、复权口径和账户权限。行情快照接口的返回结果，不能证明历史数据也满足这些要求。

对持续运行的行情应用，还应结合实际账户核对可访问的市场与标的、请求频率限制，以及断线或请求失败后的恢复方式。先完成少量标的试接，再决定是否扩展到完整业务规模。

用一个可执行示例检查字段与数据时间

下面用一个已核实字段的 REST 快照接口演示最小试接。它只验证两只股票的返回结构、价格与时间，不负责历史下载，也不能凭一次请求评定数据源质量。换用其他来源时，要先改请求和字段映射，不能假设返回结构相同。

代码仅使用 Python 3.10 及以上的标准库，无需安装第三方依赖。将其保存为 probe.py。在线模式需在本机配置环境变量 MARKET_API_KEY，并确保密钥具备目标行情权限；密钥不写入代码、输出报告或文章。

图 2：每次试接同时保留查询条件与原始数据。字段通过校验，只说明通过本次检查，不等于数据完整或持续实时。

import json
import os
import sys
import time
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

SYMBOLS = ("600519.SH", "000858.SZ")
URL = "https://api.tickdb.ai/v1/market/ticker"
TZ = timezone(timedelta(hours=8))


def inspect(payload, received_ms):
    if not isinstance(payload, dict) or str(payload.get("code")) != "0":
        raise ValueError("business response failed")
    data = payload.get("data")
    if not isinstance(data, list):
        raise ValueError("data must be a list")
    report = []
    for symbol in SYMBOLS:
        matches = [q for q in data if isinstance(q, dict)
                   and q.get("symbol") == symbol]
        if len(matches) != 1:
            report.append({"symbol": symbol, "state": "missing or duplicate"})
            continue
        q = matches[0]
        try:
            price = Decimal(str(q["last_price"]))
            stamp = q["timestamp"]
            if not price.is_finite() or price < 0:
                raise ValueError()
            if type(stamp) is not int or not 946684800000 <= stamp < 4102444800000:
                raise ValueError()
        except (KeyError, ValueError, InvalidOperation):
            report.append({"symbol": symbol, "state": "invalid fields"})
            continue
        quote_time = datetime.fromtimestamp(stamp / 1000, TZ)
        age = (received_ms - stamp) / 1000
        report.append({
            "symbol": symbol, "name": q.get("name") or symbol,
            "last_price": str(price), "quote_time": quote_time.isoformat(),
            "age_seconds": round(age, 3),
            "state": "future timestamp" if age < -5 else "fields valid",
        })
    return report


def main():
    offline = len(sys.argv) > 1
    if offline:
        payload = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
    else:
        key = os.environ.get("MARKET_API_KEY")
        if not key:
            raise ValueError("set MARKET_API_KEY first")
        query = "?symbols=" + ",".join(SYMBOLS) + "&type=stock"
        req = Request(URL + query, headers={"X-API-Key": key})
        with urlopen(req, timeout=10) as response:
            payload = json.load(response)
    received_ms = time.time_ns() // 1_000_000
    report = inspect(payload, received_ms)
    record = {"mode": "offline" if offline else "online",
              "checked_at_ms": received_ms, "requested": list(SYMBOLS),
              "payload": payload, "report": report}
    filename = f"probe-{time.time_ns()}.json"
    Path(filename).write_text(json.dumps(record, ensure_ascii=False, indent=2),
                              encoding="utf-8")
    print(json.dumps(report, ensure_ascii=False, indent=2))
    print("mode:", record["mode"], "saved:", filename)


if __name__ == "__main__":
    try:
        main()
    except HTTPError as exc:
        raise SystemExit(f"HTTP error: {exc.code}")
    except (URLError, TimeoutError, OSError, ValueError) as exc:
        raise SystemExit(f"Probe failed: {type(exc).__name__}")

配置好环境变量后，执行 python probe.py。每次成功解析业务响应后，程序都会保存一份带独立文件名的 JSON，其中包含请求代码、检查时刻、原始返回及检查结果。fields valid 表示本例的字段检查通过，不能翻译成“行情实时”。缺少某个标的或重复返回时，会单独标出；可选名称缺失则回退到代码。

本例按接口文档把 timestamp 解释为毫秒。代码中的 2000—2100 年范围是演示用的合理性检查，不能替代接口契约。明显超前的时间戳单独提示，数据年龄保留正负值，不通过截成 0 来隐藏时钟问题。

没有密钥时，可以先保存以下合成数据为 sample.json，执行 python probe.py sample.json：

{"code":0,"data":[
  {"symbol":"600519.SH","last_price":"10.20","timestamp":1792114200000},
  {"symbol":"000858.SZ","last_price":"20.30","timestamp":1792114200000}
]}

样本时间为 2026-10-16 09:30:00+08:00，价格是合成值。离线模式中的检查时刻是本次读取文件的时间，不能当成原始网络接收时间；报告会明确记录 offline。固定样本时间可能已经过去，也可能仍在未来；出现 future timestamp 是时间检查的预期提示，不代表字段格式错误。

建议依次做四项验证：保留原样，确认两行字段通过；删除一只股票，确认它被标为缺失；把 last_price 改为 null，确认不会被当成 0；把 code 改为非零，确认程序按业务失败退出。之后再用有效密钥试接，在有成交的时段按允许频率保存多次结果，比较报价时间是否推进，而不只比较价格是否变化。

本次已核对示例的请求地址、鉴权和字段，完成合成输入的离线验证；没有完成本机密钥直连、盘中连续采样或多家同条件性能测试。因此，这段代码是试接起点，不构成时效、覆盖或稳定性评级。

把选择写成一条可以验收的结论

最终结论应能对应回最初的任务：做历史日线研究，就确认指定区间、字段口径与缺失解释；做分钟回测，就确认目标日期和周期边界，而不只确认“有分钟接口”；做实时应用，就确认更新、容量和异常恢复。

先用小范围数据完成“取得、保存、检查、重跑”，再决定是否扩大范围或购买权限。选择 A股数据源的关键，是证明它能完成当前任务，并知道哪些条件仍未验证。