TickDB REST API 从入门到生产：一个 Key 调通 4 大市场的正确姿势

TickDB Research ·

你拿到 TickDB 的 API Key，打开文档，对着 GET /v1/market/ticker 发了一个请求。返回 code: 0，数据拿到了。但第二天你加了一个参数 type=stocks，同一个端点却返回了 0 条结果——因为正确的值是 stock，单数。文档没写错，只是你没注意到。这种"文档不会提前告诉你"的细节，才是从入门到生产之间真正的路。

一、你在接入行情 REST API 时，真正卡住的地方

网上从来不缺"5 分钟快速上手"的教程，但你照着跑通第一个请求之后，真正开始写业务代码时，卡住你的往往不是 HTTP 请求本身，而是一些更基础的问题：

你接 REST 行情 API 时遇到的问题通用 API 文档为什么没覆盖根因
type= 参数传了 stocks，品种查询返回 0 条，不报错文档可能用描述性语言"股票类型"，而非精确参数值正确值是 stock（单数），不是 stocks。MCP 实测：只有 stock/crypto/futures/forex/indices 五个值有效
symbols 和 symbol 用错了端点，返回空但没报错不同端点的参数名是单数还是复数有严格区分/ticker 用 symbols，/kline 用 symbol
想查估值指标，搜 fundamental 找不到端点估值端点叫 calc-index，MCP 工具名叫 get_market_metricsTickDB 的端点命名体系与其他 API 不同
限流后按自己的节奏重试，连续被封文档写了错误码 3001，但没展开退避策略HTTP 响应头里的 Retry-After 才是服务端给的退避窗口
A 股后缀是 .SH 还是 .SSE？港股带不带前导零？期货带不带后缀？文档有格式说明，但分散在不同页面需要一个统一的 symbol 格式速查表

这些问题，如果你从一个多数据源的环境迁移过来，会被进一步放大——每个源的参数命名、错误码、symbol 格式都不一样，每换一次数据源，你都要重写半套解析和错误处理逻辑。

这正是 TickDB 作为一个统一实时行情数据层要解决的核心问题：一套接口、一套字段、一套错误码，覆盖 4 大市场、6 大资产类别、近 4 万个品种。 接下来的指南，不会把 API 文档重抄一遍，而是沿着"从第一个请求到可运行的完整实现"这条主线，把每个你需要做对的设计决策讲清楚。

二、TickDB REST API 的设计：不是背端点，是理解规则

TickDB 的 REST API 不是一组随机的 URL，而是遵循几条明确的设计约束。一旦你理解了这些约束，就不需要死记硬背每个端点的参数名。

1. 快照 vs 时序：复数 symbols 与单数 symbol

这个命名规则贯穿了整个 API：

  • 查询快照类数据（当前行情、最新 K 线、估值指标）时，参数用 symbols（复数），可以一次传多个品种代码，最多 50 个。
  • 查询时序类数据（历史 K 线、盘口深度、成交明细）时，参数用 symbol（单数），一次只查一个品种。
你想查什么端点参数
某些品种的当前行情快照GET /v1/market/tickersymbols=
某只股票最近 n 根历史 K 线GET /v1/market/klinesymbol=
某些品种最近一根 K 线GET /v1/market/kline/latestsymbols=
某只股票的盘口深度（10 档）GET /v1/market/depthsymbol=
某只股票的成交明细GET /v1/market/tradessymbol=
某些品种的估值指标（PE/PB）GET /v1/market/calc-indexsymbols=

为什么这么设计？ 快照查询的场景通常是"看一眼当前价"，用户大概率想同时看好几个品种，批量查询可以减少请求次数。而 K 线或深度，每个品种的数据量本身就不小，单品种查询更利于做分页、缓存和错误隔离。

2. 品种筛选：type= 的精确值

用 GET /v1/symbols/available 查询"有哪些品种可以交易"时，type 参数的精确值经 MCP 实测确认，只有以下 5 个有效——这是本文写作时通过直接调用 MCP 工具验证的完整列表：

type 值资产类别可查询的市场
stock股票（含 ETF）CN / HK / US
crypto加密货币Global
futures期货CN（中金所 + 五大商品交易所）
forex外汇Global
indices指数Global

注意：stocks（复数）和 cryptocurrencies 虽然在直觉上合理，但实测返回 0 条结果且不报错。ETF 不是独立的 type 值，而是包含在 stock 类型下。一个典型的查询：GET /v1/symbols/available?market=CN&type=futures 会一次性返回中金所、上期所、大商所、郑商所、广期所、上海能源交易所的全部期货品种。

3. 估值端点不叫 fundamental

如果你从其他金融数据 API 迁移过来，很可能习惯性地搜索 "fundamental" 来查估值指标。TickDB 里对应的端点叫 /v1/market/calc-index——覆盖了 PE（市盈率）、PB（市净率）等估值数据。在 MCP 工具体系里，它的工具名叫 get_market_metrics。实测注意：get_market_metrics 对港股返回的 symbol 格式为 stock:HK:700，而非标准长格式 700.HK，处理返回结果时需要兼容这种内部格式。

4. 错误码不是扁平的：什么该重试，什么该立刻停

TickDB 的错误码大致分为两类，处理策略完全不同：

错误码含义处理策略
3001请求频率超限（限流）等待 Retry-After 头指定的秒数，然后重试。如果没有这个头，再用指数退避自己算。最多重试 3 次。
429（HTTP 状态码）HTTP 层面的限流同上
1001API Key 无效立即停止，不要重试。检查 Key 是否过期或被撤销。
1002请求中没有提供 API Key立即停止。检查 HTTP Header 是否包含了 X-API-Key。
1004API Key 权限不足立即停止。确认你的账户是否有该端点的访问权限。
网络超时TCP 连接超时重试 1–2 次，超过则告警。
2002symbol 不存在或未被识别停止，检查 symbol 格式，或通过 /v1/symbols/available 确认该品种是否在覆盖范围内。

为什么不能所有错误都统一重试？ 因为 1001、1002、1004 这类问题不会因为"等一会儿再试"就自己修复，无限重试只会浪费你的配额，甚至导致 IP 被暂时封禁。区分可恢复错误和不可恢复错误，是编写可靠代码的基本要求。

5. Symbol 格式速查

跨市场查询时，symbol 格式最容易出错。以下是经 MCP 实测确认的格式规则：

市场正确格式示例错误格式
上交所 A 股代码.SH600519.SH（贵州茅台）600519.SSE
深交所 A 股代码.SZ300750.SZ（宁德时代）300750.SZSE
北交所 A 股代码.BJ831445.BJ—
港股代码.HK（无前导零）700.HK（腾讯控股）0700.HK
美股代码.USAAPL.US（苹果）AAPL.NASDAQ
中金所期货代码本身，无后缀IF2606（沪深300股指期货）IF2606.CFE
加密货币交易对BTCUSDT—

在代码里，最好用正则把这些格式校验做在请求发出之前——本文第三部分的代码就包含了一个完整的格式校验器。

三、可运行的三层封装实现

以下代码实现了上述所有设计原则：参数校验层在请求前拦截格式错误，请求执行层统一处理鉴权、限流退避和所有错误码，字段映射层把不同端点的原始字段翻译成消费方可直接使用的结构化数据，价格字段采用 Decimal 保护精度。

环境准备

pip install requests python-dotenv

在项目根目录创建 .env 文件（不要提交到版本控制）：

TICKDB_API_KEY=你的TickDB API Key
TICKDB_REST_URL=https://api.tickdb.ai

完整代码：tickdb_rest_client.py

# tickdb_rest_client.py
# TickDB REST API 客户端实现
# 三层封装：参数校验 → 请求执行与错误处理 → 字段映射
# Python 3.10+ 可直接运行

import os
import re
import time
import json
import requests
from decimal import Decimal, InvalidOperation
from typing import Optional, Dict, List
from dotenv import load_dotenv

load_dotenv()

# ================== 配置 ==================
API_KEY = os.getenv("TICKDB_API_KEY")
BASE_URL = os.getenv("TICKDB_REST_URL", "https://api.tickdb.ai")
MAX_RETRIES = 3

# ================== Symbol 格式校验器 ==================
SYMBOL_PATTERN = re.compile(
    r'^(\d{6}\.(SH|SZ|BJ)|[1-9]\d{0,4}\.HK|[A-Z]{1,5}\.US|[A-Z]{2,4}[A-Z0-9]{2,8}|[A-Z]{2}\d{4})$'
)

# MCP 实测确认的 type 参数完整可选值（5 个）
VALID_TYPES = {"stock", "crypto", "futures", "forex", "indices"}

def validate_symbols(symbols: str) -> List[str]:
    """校验 symbol 格式，返回清洗后的列表。格式错误直接抛出异常。"""
    if not symbols or not symbols.strip():
        raise ValueError("symbols 不能为空")
    sym_list = [s.strip() for s in symbols.split(",") if s.strip()]
    if not sym_list:
        raise ValueError("symbols 解析后为空")
    invalid = [s for s in sym_list if not SYMBOL_PATTERN.match(s)]
    if invalid:
        raise ValueError(
            f"以下 symbol 格式可能有误: {invalid}。"
            f"A 股格式: 600519.SH，港股: 700.HK，美股: AAPL.US，期货: IF2606"
        )
    return sym_list

def validate_type(type_str: str) -> str:
    if type_str and type_str not in VALID_TYPES:
        raise ValueError(
            f"type={type_str} 不在可选范围。"
            f"MCP 实测确认的完整可选值: {sorted(VALID_TYPES)}"
        )
    return type_str


# ================== 请求执行层 ==================
class TickDBClient:
    """统一请求客户端，处理鉴权、限流退避和所有业务错误码。"""

    def __init__(self, api_key: str = None, base_url: str = None):
        self.api_key = api_key or API_KEY
        self.base_url = base_url or BASE_URL
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": self.api_key})
        self.session.timeout = 10

    def _request(self, method: str, path: str, params: dict = None,
                 retry_count: int = 0) -> dict:
        if retry_count > MAX_RETRIES:
            raise Exception(f"重试 {MAX_RETRIES} 次后仍失败")

        url = f"{self.base_url}{path}"
        try:
            if method == "GET":
                resp = self.session.get(url, params=params)
            elif method == "POST":
                resp = self.session.post(url, json=params)
            else:
                raise ValueError(f"不支持的 HTTP 方法: {method}")
        except requests.exceptions.Timeout:
            raise Exception(f"请求超时: {url}")
        except requests.exceptions.ConnectionError:
            raise Exception(f"无法连接到 {self.base_url}，请检查网络")

        # HTTP 429 限流
        if resp.status_code == 429:
            retry_after = resp.headers.get("Retry-After", "5")
            try:
                wait = float(retry_after)
            except (ValueError, TypeError):
                wait = 5
            print(f"[限流] HTTP 429，等待 {wait}s...")
            time.sleep(wait)
            return self._request(method, path, params, retry_count + 1)

        data = resp.json()
        code = data.get("code")

        if code == 0:
            return data

        # 3001 限流（业务层）
        if code == 3001:
            retry_after = resp.headers.get("Retry-After", "5")
            try:
                wait = float(retry_after)
            except (ValueError, TypeError):
                wait = 5
            print(f"[限流] code=3001，等待 {wait}s...")
            time.sleep(wait)
            return self._request(method, path, params, retry_count + 1)

        # 不可恢复错误
        if code == 1001:
            raise Exception("API Key 无效 (1001)，请检查 .env 中的 TICKDB_API_KEY")
        if code == 1002:
            raise Exception("未提供 API Key (1002)，请检查请求头 X-API-Key")
        if code == 1004:
            raise Exception("API Key 权限不足 (1004)，请确认账户权限范围")
        if code == 2002:
            raise Exception(f"Symbol 不存在或未被识别 (2002): {params}")

        raise Exception(f"未知错误 code={code}: {data.get('message', '无错误信息')}")

    # ---------- 行情端点 ----------
    def get_ticker(self, symbols: str) -> dict:
        """实时行情快照。symbols 为逗号分隔，最多 50 个。"""
        validate_symbols(symbols)
        return self._request("GET", "/v1/market/ticker", {"symbols": symbols})

    def get_kline(self, symbol: str, interval: str = "1d", limit: int = 100) -> dict:
        """历史 K 线。interval 如 1m/5m/1h/1d/1w。"""
        validate_symbols(symbol)
        return self._request("GET", "/v1/market/kline",
                             {"symbol": symbol, "interval": interval, "limit": limit})

    def get_kline_latest(self, symbols: str, interval: str = "1d") -> dict:
        """最近一根 K 线。symbols 可传多个。"""
        validate_symbols(symbols)
        return self._request("GET", "/v1/market/kline/latest",
                             {"symbols": symbols, "interval": interval})

    def get_depth(self, symbol: str) -> dict:
        """盘口深度（10 档）。支持美股、港股、A 股、加密货币。"""
        validate_symbols(symbol)
        return self._request("GET", "/v1/market/depth", {"symbol": symbol})

    def get_trades(self, symbol: str) -> dict:
        """
        成交明细。
        MCP 实测：美股（AAPL.US）和港股（700.HK）返回真实成交数据；
        A 股（600519.SH）返回空数组。期货待测。
        """
        validate_symbols(symbol)
        return self._request("GET", "/v1/market/trades", {"symbol": symbol})

    def get_calc_index(self, symbols: str) -> dict:
        """
        估值指标（PE、PB 等）。
        端点名为 calc-index，MCP 工具名为 get_market_metrics。
        注意：MCP 工具返回的港股 symbol 格式为 'stock:HK:700'，非标准长格式。
        """
        validate_symbols(symbols)
        return self._request("GET", "/v1/market/calc-index", {"symbols": symbols})

    def get_available_symbols(self, market: str = None, type_: str = None) -> dict:
        """
        可用品种列表。market: CN/HK/US/Global。
        type_ 可选值（MCP 实测确认）: stock/crypto/futures/forex/indices
        """
        params = {}
        if market:
            params["market"] = market
        if type_:
            params["type"] = validate_type(type_)
        return self._request("GET", "/v1/symbols/available", params)

    def get_kline_intervals(self) -> dict:
        """支持的 K 线周期列表。"""
        return self._request("GET", "/v1/market/intervals/kline")


# ================== 字段映射层 ==================
def map_ticker_fields(raw_data: list) -> list:
    """将 ticker 原始字段转为统一消费格式，价格用 Decimal 保护精度。"""
    results = []
    for d in raw_data:
        try:
            price = Decimal(str(d.get("last_price", "0")))
            vol = Decimal(str(d.get("volume_24h", "0")))
        except (InvalidOperation, ValueError):
            continue
        results.append({
            "symbol": d.get("symbol"),
            "last_price": price,
            "volume_24h": vol,
            "timestamp_ms": d.get("timestamp"),
            "high_24h": d.get("high_24h"),
            "low_24h": d.get("low_24h"),
        })
    return results

def map_kline_fields(raw_klines: list) -> list:
    """将 K 线原始字段转为统一消费格式。"""
    results = []
    for k in raw_klines:
        results.append({
            "time_ms": k.get("time"),
            "open": k.get("open"),
            "high": k.get("high"),
            "low": k.get("low"),
            "close": k.get("close"),
            "volume": k.get("volume"),
        })
    return results


# ================== 快速验证 ==================
if __name__ == "__main__":
    client = TickDBClient()

    # 1. 查看中金所及商品期货品种
    print("=== 期货品种查询 ===")
    try:
        data = client.get_available_symbols(market="CN", type_="futures")
        products = data.get("data", {}).get("products", [])
        print(f"共找到 {len(products)} 个期货品种（含中金所/上期所/大商所/郑商所/广期所/上海能源）")
    except Exception as e:
        print(f"查询失败: {e}")

    # 2. 跨市场行情快照
    print("\n=== A股+港股+美股+加密货币行情 ===")
    try:
        data = client.get_ticker("600519.SH,700.HK,AAPL.US,BTCUSDT")
        mapped = map_ticker_fields(data.get("data", []))
        for item in mapped:
            print(f"{item['symbol']}: 最新价={item['last_price']}, 24h成交量={item['volume_24h']}")
    except Exception as e:
        print(f"查询失败: {e}")

    # 3. K 线
    print("\n=== 贵州茅台日线（最近 5 根） ===")
    try:
        data = client.get_kline("600519.SH", interval="1d", limit=5)
        klines = map_kline_fields(data.get("data", {}).get("klines", []))
        for k in klines:
            print(f"时间={k['time_ms']}, 开={k['open']}, 收={k['close']}, 量={k['volume']}")
    except Exception as e:
        print(f"查询失败: {e}")

核心解读：三层封装的好处在于，当你的业务从"查一个品种"扩展到"批量查询 + 定时任务"时，你不需要在每个调用点重复写限流判断和 Decimal 转换。参数校验器会确保格式错误在请求之前就被拦截，而不是等到 API 返回 2002 才发现 symbol 写错了。

四、REST 与 MCP：同一种数据能力，两种接入方式

TickDB 除了 REST API，还提供了等价的 MCP 工具。同一份行情数据，在 AI 编码环境中（如 Claude Code、Cursor、Codex）可以直接通过 MCP 调用，而不需要手写 HTTP 请求。下表列出了常用能力的对应关系，方便你在两种模式间切换：

数据能力REST 端点MCP 工具名
实时行情GET /v1/market/tickerget_ticker
历史 K 线GET /v1/market/klineget_kline
最近 K 线GET /v1/market/kline/latestget_kline_latest
盘口深度GET /v1/market/depthget_order_book
成交明细GET /v1/market/tradesget_recent_trades
估值指标GET /v1/market/calc-indexget_market_metrics
品种列表GET /v1/symbols/availableget_available_symbols
K 线周期GET /v1/market/intervals/klineget_kline_intervals

如果你日常工作中有大量时间在 AI 编码助手里完成，可以直接在 MCP 配置中添加 mcp.tickdb.ai，省去手写和维护 REST 客户端的工程投入。本文的 Python 客户端路径，则更适合独立脚本、生产服务、自动化流水线等对运行环境有完整控制力的场景。详细文档见 docs.tickdb.ai。

五、结尾

TickDB REST API 的入门非常简单——拿到 Key，一个 curl 就能看到数据。但从"能调通"到"可靠运行"，中间隔着一整套需要自己补完的工程逻辑：参数为什么是单数、端点为什么叫这个名字、限流后究竟该等多久、type= 到底有哪几个有效值。

一个好的 API 设计，不是让你少写代码，而是让你在写代码时少做决策——当你知道 symbols 一定用于快照、symbol 一定用于时序，当你知道错误码 1001 和 3001 有截然不同的处理路径，当你知道 type=stock 而非 type=stocks，你就不再是"调接口"，而是在"用接口"。

现在，打开你的编辑器，把这篇文章里的 TickDBClient 拷进项目里，然后把 .env 中的 Key 填上。第一个请求返回 code: 0 的时候，你已经站在了可靠运行的起点上，而不是 Demo 的终点。

📡 数据示例由 TickDB.ai 提供

标签：TickDB / REST API / 实时行情 / 金融数据 / 接口设计 / Python

相关文章

系列文章第 3 篇：TickDB REST 快速开始：完成第一次可核验的行情查询

> 这是 TickDB 实时行情接入系列的第 3 篇。前一篇解决“接入路径怎么选”，本文解决“REST 第一次查询”。 很多开发者第一次接行情数据，卡住的不是“有没有 API”，而是另一个更实际的问题： 先说结论：第一次验证，只跑通一条 REST 快照请求 第一次使用 TickDB REST API，不需要先搭复杂工程。你可以先完成一个可核验闭环：准备 API Key，向 传入真实 ，再确认响应

TickDB Research ·

DeepSeek 和 ChatGPT 在金融数据接入上的真实差距：别让“API 兼容”替你回答选型问题

> 你把 从 改成 ，代码跑通了。但当你给 Agent 配了两个金融数据工具，同一个 query 在两个模型上的行为开始分化——这才是需要关注的地方。 一、当“全面兼容”遇上金融数据 DeepSeek API 全面兼容 OpenAI SDK。改一行 ，换个 参数，你的 Function Calling 代码就能在两个模型上跑通。 但“跑通”和“跑对”之间，隔着一整套金融数据接入场景下才会暴露

TickDB Research ·

想用扣子 / Coze 工作流查行情，为什么第一步不是搭流程，而是确认四类证据？

想让扣子 / Coze 工作流查询外部行情数据，最先要证明的不是流程画布能否搭出来，而是四类证据是否成立：能否请求外部服务、能否安全传递密钥、能否解析可核对的返回、失败时能否停止生成结论。本文仅写验证清单，不把尚未取得官方配置证据的能力写成教程。 很多人想做的事情很自然：让一个工作流接入外部行情数据，再让 AI 根据查询结果回答问题。

TickDB Research ·

老虎、富途、长桥相关主体被查处后，港股新股行情还能怎么看？

2026 年 5 月 22 日，证监会发布消息称，八部门已联合印发《综合整治非法跨境证券期货基金经营活动实施方案》，该正式通知落款为 2026 年 5 月 9 日；同日发布的案件公告披露，近日已对老虎、富途、长桥境内外相关主体作出行政处罚事先告知。 对关注港股新股或上市首日波动的内地投资者而言，更值得先回答的问题不是某条新闻对应的股价变化，而是： 看到一只港股新股的信息或行情，是否就意味着可以参与

TickDB Research ·

为什么 DeepSeek 不能自己知道实时行情？正确做法是把查询接成 Tool Calls

DeepSeek 接实时行情，上一篇代码需要补上的 3 个可信度约束 5 月 23 日，我发布过一篇《DeepSeek Function Calling 接实时行情：从工具定义到多轮查询的完整示例》。 那篇文章解决的是：如何把行情接口包装成模型可调用的工具。 但复盘之后，我发现还有一个更重要的问题需要单独说明： > 工具接通了，不等于当前行情答案就可信。 对于“苹果现在多少钱”“腾讯今天涨跌如何”

TickDB Research ·