A 股实时行情 API 怎么接?一个 ticker 和 kline 不会各自为政的实时行情数据源

[复制链接]
发表于 2026-6-16 11:00:14 | 显示全部楼层 |阅读模式

马上注册,结交更多好友,享用更多功能,让你轻松玩转社区。

您需要 登录 才可以下载或查看,没有账号?立即注册

×
择要:接入行情 API 时,常见风险是差别端点的鉴权、时间字段、返回结构和错误分支必要分别确认——你以为同一个数据源的差别端点应该划一,但现实上大概 ticker 正常返回、kline 鉴权失败,大概两边都返回乐成码,timestamp 却是差别精度。本文以 TickDB 的 REST API 作为可复核示例,展示一个 ticker 和 kline 在鉴权和字段范例上保持划一性的数据源长什么样。假如你还没接过行情 API,这篇直接带你用最少代码跑通初次调用;假如你已经在用别的数据源,这篇给你一个可对照的左券划一性查抄清单。关键词:A股行情API、ticker、kline、REST API、TickDB、左券划一性。
一、需求界限

在开始写代码之前,先把本文要办理和不办理的标题说清楚。
本文要办理的:用最少代码完成一次 A 股 ticker 快照查询和一次 kline 汗青数据查询,并验证两个端点在鉴权、返回结构和关键字段范例上的划一性。跑通这两步,你就有了一个可复用的校验骨架。
本文不办理的:数据库写入、定时调治、断线回补、WebSocket 推送、多市场并行查询、生产级错误重试计谋。这些是另一个工程层级的标题,不在本文范围内。
实用读者:第一次接行情 API 的 Python 开辟者、必要验证 ticker 和 kline 端点左券划一性的后端工程师、想用最少代码跑通初次调用的量化初学者。
二、设置与环境变量

开始之前,确认三件事:

  • Python 3.8+ 已安装
  • requests 库已安装:pip install requests
  • 环境变量 TICKDB_API_KEY 已设置为你的 API Key
  1. export TICKDB_API_KEY="your_api_key_here"
复制代码
三、两个端点,可复用的校验骨架

接入行情 API 时,维护资源每每不在第一行代码,而在后续的扩展和排查。ticker 和 kline 假如鉴权方式差别、业务码语义差别、时间戳精度差别,你每多接一个端点就要多维护一套剖析逻辑。比及几十个品种的监控脚本跑起来、两边的数据必要合到一起做分析时,计划不划一的代价就会会集发作。
TickDB 的做法是把两个端点放在同一套规则下。你可以复用同一套鉴权、HTTP/JSON/业务码/空数据、Decimal 和时间字段校验骨架;但 ticker 与 kline 的字段路径和语义必须分别查对——ticker 返回 data[] 中的 last_price,kline 返回 data.klines[] 中的 OHLC,两者不能混用。业务码也需各自查抄,不可假设两个端点的错误码语义划一。
下面这张表,既是功能对照,也是左券划一性查抄的出发点。你可以拿它去对照你正在评估的数据源。
维度ticker汗青 kline端点/v1/market/ticker/v1/market/kline鉴权X-API-KeyX-API-Key(同一个 Header)焦点参数symbolssymbol + interval返回路径data[]data.klines[]代价字段last_priceopen, high, low, close时间字段timestamptime(均为毫秒整数)用途当前快照已竣事周期的汗青数据乐成码code=0code=0(需分别查抄)
ticker 的 last_price 不是 kline 的 close,两者不能混用。它们的鉴权方式和时间字段精度划一,但业务码必须各自验证,不假设跨端点同一。
四、TickerClient:一个短函数,查 A 股快照

[code]import osimport sysimport requestsfrom decimal import Decimal, InvalidOperationAPI_KEY = os.environ.get("TICKDB_API_KEY")BASE_URL = "https://api.tickdb.ai/v1"TIMEOUT = 10# 讲授示例,非生产级代码# 代价字段值为接口返回值占位阐明,不构成实时报价展示def fetch_ticker(symbols: str):    """获取 ticker 快照并校验关键字段。"""    if not API_KEY:        print("错误: 环境变量 TICKDB_API_KEY 未设置", file=sys.stderr)        sys.exit(1)    try:        resp = requests.get(            f"{BASE_URL}/market/ticker",            headers={"X-API-Key": API_KEY},            params={"symbols": symbols},            timeout=TIMEOUT,        )        resp.raise_for_status()    except requests.exceptions.ConnectionError as e:        sys.exit(f"毗连失败: {e}")    except requests.exceptions.Timeout as e:        sys.exit(f"哀求超时: {e}")    except requests.exceptions.HTTPError as e:        sys.exit(f"HTTP 错误: {e}")    except requests.exceptions.RequestException as e:        sys.exit(f"哀求非常: {e}")    try:        data = resp.json()    except ValueError as e:        sys.exit(f"JSON 剖析失败: {e}")    if data.get("code") != 0:        sys.exit(f"业务失败: code={data.get('code')}")    ticker_list = data.get("data", [])    if not isinstance(ticker_list, list) or not ticker_list:        sys.exit("ticker data 为空")    for item in ticker_list:        sym = item.get("symbol")        if sym != symbols:            sys.exit(f"symbol 不匹配: 渴望 {symbols}, 返回 {sym}")        raw_price = item.get("last_price")        if not isinstance(raw_price, str) or not raw_price:            sys.exit("last_price 缺失或为空")        try:            price = Decimal(raw_price)        except InvalidOperation:            sys.exit(f"last_price 无法剖析: {raw_price}")        if not price.is_finite():            sys.exit(f"last_price 非有限数: {raw_price}")        ts = item.get("timestamp")        if isinstance(ts, bool) or not isinstance(ts, int) or ts
回复

使用道具 举报

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表