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

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

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

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

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

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

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

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

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

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

• 查询快照类数据（当前行情、最新 K 线、估值指标）时，参数用 symbols（复数），可以一次传多个品种代码，最多 50 个。
• 查询时序类数据（历史 K 线、盘口深度、成交明细）时，参数用 symbol（单数），一次只查一个品种。

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

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

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

注意：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 的错误码大致分为两类，处理策略完全不同：

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

5. Symbol 格式速查

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

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

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

以下代码实现了上述所有设计原则：参数校验层在请求前拦截格式错误，请求执行层统一处理鉴权、限流退避和所有错误码，字段映射层把不同端点的原始字段翻译成消费方可直接使用的结构化数据，价格字段采用 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 请求。下表列出了常用能力的对应关系，方便你在两种模式间切换：

如果你日常工作中有大量时间在 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

通过 TickDB API 获取实时行情数据

一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。