TickDB SKILL 协议详解：如何让 AI 助手理解你的行情查询

当 AI 助手说“根据公开信息，NVDA 近期走势强劲”时，你有没有想过：它其实可以更聪明——直接调用 TickDB，拿到实时行情，告诉你买卖价差扩大了多少、depth 档位有什么变化？

这不是假设，而是 TickDB SKILL 协议正在实现的事。

本文从协议规范、解析机制到 Function Calling 全链路拆解，让你能为自己的 AI 助手接入 TickDB 的实时数据能力。

---

一、为什么需要 SKILL 协议

AI 助手在金融场景的困境，本质上是“语言理解”和“数据获取”的断裂。

传统接入模式是这样的：

1. 开发者写一个 Python 脚本
2. 脚本调用 TickDB REST API
3. 脚本返回数据给 AI
4. AI 把数据融入回答

这个链路有效，但有几个问题：

问题	影响
需要开发者介入	非技术人员无法直接使用
无法动态扩展	每新增一个查询场景，都需要改代码
多轮对话断链	上一轮问“NVDA”，下一轮问“走势”，AI 记不住上下文

SKILL 协议的本质，是给 AI 一份机器可读的能力清单。这份清单放在 http://skill.md/ 文件里，AI 读懂它之后，就能自主决定调用哪个接口、传入什么参数、把返回的数据变成自然语言回答。

---

二、协议文件的位置与格式

2.1 文件位置

TickDB SKILL 的规范文件托管在：http://skill.md/

这是一个标准的 Markdown 文件，但它的内容遵循严格的 JSON Schema 规范。

2.2 格式结构

完整的 SKILL 文件包含四个顶层字段：

name: tickdb-market-data          # SKILL 唯一标识符
version: "1.0"                    # 协议版本
description:                      # 人类可读的能力描述
  en: "Real-time market data API for stocks, crypto, futures..."
  zh: "股票、加密货币、期货、外汇的实时行情 API"
capabilities:                     # AI 可调用的能力列表
  - ...

其中，capabilities 是核心。每个 capability 对应一个可调用的功能。

---

三、capabilities 的结构：AI 怎么知道该调用什么

3.1 单个 capability 的结构

capabilities:
  - id: get_realtime_quote          # 能力 ID，全局唯一
    name:
      en: "Get Real-time Quote"
      zh: "获取实时行情"
    description:
      en: "Retrieve current price, volume, bid/ask for a symbol"
      zh: "获取指定标的的当前价格、成交量、买卖价"
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: "Symbol code, e.g., AAPL.US, BTC.USDT"
          required: true
        fields:
          type: array
          items:
            type: string
          description: "Fields to return: price, volume, bid, ask..."
      required: ["symbol"]
    returns:
      type: object
      properties:
        price:
          type: number
          description: "Last traded price"
        volume:
          type: integer
          description: "24h trading volume"
        timestamp:
          type: string
          description: "ISO 8601 timestamp"
    examples:
      - query: "What's the current price of AAPL?"
        params: { symbol: "AAPL.US", fields: ["price"] }

3.2 关键字段解析

字段	作用	AI 如何使用
id	能力唯一标识	作为 Function Calling 的函数名
name	多语言显示名	用于 UI 展示和能力列表
description	人类可读说明	AI 判断是否需要调用的依据
parameters	JSON Schema 定义	AI 构建调用请求的参数校验依据
required	必填参数列表	AI 必须填充，不能为空
examples	示例问答对	Few-shot 学习，提高意图识别准确率

3.3 参数类型支持

TickDB SKILL 规范支持以下参数类型：

# 基础类型
type: string, number, integer, boolean, array, object

# 枚举约束
symbol_type:
  type: string
  enum: ["stock", "crypto", "futures", "forex", "metal"]

# 格式约束
timestamp:
  type: string
  format: "date-time"          # ISO 8601

# 数组约束
intervals:
  type: array
  items:
    type: string
    enum: ["1m", "5m", "15m", "1h", "4h", "1d"]

AI 在构造请求时，会根据 parameters 的 schema 校验参数合法性，避免无效调用。

---

四、AI 解析 SKILL 的流程

当 AI 助手收到用户的查询时，它会经历以下解析流程：

用户输入 → 意图识别 → 能力匹配 → 参数提取 → 调用执行 → 响应整合 → 自然语言输出

4.1 意图识别（Intent Detection）

AI 首先判断用户想要什么类型的数据。

用户："NVDA 最近怎么样？"
AI 解析：
  - 关键词：NVDA、走势
  - 意图：查询股票行情 + 趋势分析
  - 置信度：0.87

这个阶段，AI 会参考 capabilities[].description 的多语言描述做匹配。

4.2 能力匹配（Capability Matching）

匹配到 get_realtime_quote 和 get_kline_history 两个能力。

AI 需要决定调用哪个，或者都调用：

决策：
  - "最近怎么样" → 调用 get_realtime_quote（当前行情）
  - "走势" → 调用 get_kline_history（历史 K 线）
  - 调用顺序：先实时行情，再历史 K 线（数据有依赖）

4.3 参数提取（Parameter Extraction）

从用户输入中提取 parameters 定义的参数：

用户："NVDA 最近怎么样？"
参数提取结果：
  {
    "symbol": "NVDA.US",      # 标准化格式（市场后缀）
    "fields": ["price", "volume", "change_percent"],
    "interval": "1d",
    "range": "5d"
  }

AI 还需要处理隐式参数：用户说“最近”，AI 需要翻译为具体的时间范围（如最近 5 个交易日）。

4.4 调用执行（Function Execution）

AI 调用 TickDB API：

# AI 生成的调用代码（概念示例）
response = tickdb.get_realtime_quote(symbol="NVDA.US", fields=["price", "volume"])

返回数据结构：

{
  "symbol": "NVDA.US",
  "price": 875.32,
  "volume": 45678900,
  "bid": 875.30,
  "ask": 875.34,
  "timestamp": "2026-04-15T16:00:00Z"
}

4.5 响应整合（Response Integration）

AI 将返回数据整合进自然语言：

“NVDA 当前价格 875.32 美元，较昨日收盘上涨 2.3%。过去 5 个交易日呈上升趋势，成交量较前期增加约 35%。目前买卖价差为 0.04 美元，流动性良好。”

---

五、Function Calling 的定义与实现

5.1 什么是 Function Calling

Function Calling 是大语言模型（LLM）与外部工具之间的桥接协议。它定义了三件事：

1. 工具定义：AI 知道有哪些工具可用
2. 调用生成：AI 根据用户意图生成调用参数
3. 响应处理：AI 将工具返回结果整合进回答

5.2 TickDB SKILL 的 Function Calling 规范

TickDB SKILL 协议基于 OpenAI 的 Function Calling 规范扩展，以下是完整的工具定义示例：

tools:
  - name: tickdb_get_realtime_quote
    description: "获取股票、加密货币等标的的实时行情数据"
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: "标的代码，格式：代码.市场，如 AAPL.US"
          example: "AAPL.US"
        fields:
          type: array
          description: "返回字段列表"
          items:
            type: string
            enum: ["price", "volume", "bid", "ask", "change", "change_percent"]
          example: ["price", "volume", "change_percent"]
      required: ["symbol"]

  - name: tickdb_get_kline_history
    description: "获取历史 K 线数据，用于技术分析或回测"
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: "标的代码"
          example: "BTC.USDT"
        interval:
          type: string
          description: "K 线周期"
          enum: ["1m", "5m", "15m", "30m", "1h", "4h", "1d", "1w"]
          example: "1h"
        start_time:
          type: string
          format: date-time
          description: "开始时间（ISO 8601）"
        end_time:
          type: string
          format: date-time
          description: "结束时间（ISO 8601）"
        limit:
          type: integer
          description: "最大返回条数（1-1000）"
          default: 100
      required: ["symbol", "interval"]

  - name: tickdb_subscribe_depth
    description: "通过 WebSocket 订阅订单簿深度数据"
    parameters:
      type: object
      properties:
        symbol:
          type: string
          description: "标的代码"
        depth:
          type: integer
          description: "档位数量（1-50）"
          default: 10
      required: ["symbol"]

5.3 调用流程的时序图

用户 → AI → TickDB API → AI → 用户

时序详解：
1. 用户输入："给我看看 NVDA 的订单簿"
2. AI 识别意图，匹配 tickdb_subscribe_depth
3. AI 生成调用参数：{symbol: "NVDA.US", depth: 10}
4. AI 调用 tickdb_subscribe_depth
5. TickDB WebSocket 返回 depth 数据（实时推送）
6. AI 将 depth 数据格式化为自然语言
7. AI 回答："NVDA 买一档 875.30 挂单 12300 股，卖一档 875.34 挂单 9800 股..."

5.4 多工具协调调用

复杂查询需要 AI 协调多个工具：

用户："NVDA 财报后走势怎样？买卖价差有什么变化？"

AI 调用序列：
1. tickdb_get_kline_history (symbol: "NVDA.US", interval: "5m", range: "1h")
   → 获取财报前后 1 小时 K 线
   
2. tickdb_get_realtime_quote (symbol: "NVDA.US", fields: ["bid", "ask"])
   → 获取当前买卖价
   
3. tickdb_subscribe_depth (symbol: "NVDA.US", depth: 10)
   → 订阅实时 depth，监控流动性变化

数据整合后输出综合分析。

---

六、TickDB SKILL 的独特设计

6.1 上下文感知（Context Awareness）

传统 Function Calling 每次调用都是独立的，AI 无法记住之前的上下文。

TickDB SKILL 引入上下文标记机制：

context:
  session_id: "usr_abc123_session_001"
  last_query:
    symbol: "NVDA.US"
    timestamp: "2026-04-15T16:00:00Z"
  implied_symbol: true     # 标记：上一轮已指定标的

这样，当用户说“再看看特斯拉”时，AI 知道 symbol 应该是 TSLA.US 而不是继续用 NVDA.US。

6.2 错误恢复与重试

AI 调用可能失败（网络、超限、标的不存在）。SKILL 规范定义了错误恢复策略：

error_handling:
  retry:
    max_attempts: 3
    backoff: "exponential"
    jitter: true
  fallback:
    - strategy: "use_cache"
      cache_ttl: 300        # 5 分钟缓存
    - strategy: "partial_response"
      min_fields: ["price"]

6.3 多市场统一接口

SKILL 规范定义了标的格式的统一规则：

{symbol}.{market}

股票：  AAPL.US, TSLA.US, 00700.HK
加密： BTC.USDT, ETH.USDT
期货： CL.FUT, GC.FUT
外汇： EUR/USD, GBP/USD
贵金属：XAU/USD

AI 无需为每个市场写独立逻辑，统一的 symbol 格式兼容所有资产类型。

---

七、生产级代码示例

以下代码展示如何基于 SKILL 规范实现一个调用 TickDB 的 Python 函数：

import os
import time
import random
import json
import requests
from typing import Optional, Dict, Any, List

# 环境变量存储 API Key
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
if not TICKDB_API_KEY:
    raise ValueError("请设置环境变量 TICKDB_API_KEY")

# ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio 实现异步并发

def call_tickdb_function(
    function_name: str,
    parameters: Dict[str, Any],
    max_retries: int = 3
) -> Dict[str, Any]:
    """
    基于 SKILL 规范的 Function Calling 实现
    
    Args:
        function_name: SKILL 定义的 capability ID
        parameters: AI 提取的参数
        max_retries: 最大重试次数
    
    Returns:
        API 响应的 data 字段
    """
    # 端点映射
    endpoint_map = {
        "tickdb_get_realtime_quote": "/v1/market/quote",
        "tickdb_get_kline_history": "/v1/market/kline",
        "tickdb_subscribe_depth": None,  # WebSocket，另行处理
    }
    
    endpoint = endpoint_map.get(function_name)
    if endpoint is None:
        raise ValueError(f"未知函数: {function_name}")
    
    # 指数退避重连
    base_delay = 1
    max_delay = 32
    
    for attempt in range(max_retries):
        try:
            headers = {"X-API-Key": TICKDB_API_KEY}
            response = requests.post(
                f"https://api.tickdb.ai/v1{endpoint}",
                headers=headers,
                json=parameters,
                timeout=(3.05, 10)  # 连接超时 3.05s，读取超时 10s
            )
            
            # 解析响应
            result = response.json()
            code = result.get("code", 0)
            
            if code == 0:
                return result.get("data", {})
            
            # 限频处理 (code: 3001)
            if code == 3001:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"限频触发，等待 {retry_after} 秒...")
                time.sleep(retry_after)
                continue
            
            # 其他错误码
            if code in (1001, 1002):
                raise ValueError(f"API Key 无效，请检查环境变量 TICKDB_API_KEY")
            if code == 2002:
                raise KeyError(f"交易品种不存在: {parameters.get('symbol')}")
            
            raise RuntimeError(f"API 错误 {code}: {result.get('message')}")
            
        except requests.exceptions.Timeout:
            # 超时重试
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)  # 抖动
            print(f"请求超时，第 {attempt + 1} 次重试，等待 {delay + jitter:.2f}s...")
            time.sleep(delay + jitter)
            
        except requests.exceptions.RequestException as e:
            # 网络错误重连
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            print(f"网络错误: {e}，第 {attempt + 1} 次重试...")
            time.sleep(delay + jitter)
    
    raise RuntimeError(f"重试 {max_retries} 次后仍失败")


def get_realtime_quote(symbol: str, fields: List[str]) -> Dict[str, Any]:
    """
    获取实时行情（对应 tickdb_get_realtime_quote）
    
    Example:
        >>> quote = get_realtime_quote("NVDA.US", ["price", "volume"])
        >>> print(quote)
        {'symbol': 'NVDA.US', 'price': 875.32, 'volume': 45678900}
    """
    return call_tickdb_function(
        "tickdb_get_realtime_quote",
        {"symbol": symbol, "fields": fields}
    )


# 使用示例
if __name__ == "__main__":
    try:
        # 调用示例
        result = get_realtime_quote("NVDA.US", ["price", "volume", "change_percent"])
        print(f"实时行情：{json.dumps(result, indent=2, ensure_ascii=False)}")
    except Exception as e:
        print(f"调用失败: {e}")

WebSocket 订阅 depth 的生产级实现

import os
import time
import json
import threading
import websocket

TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
WS_BASE_URL = "wss://stream.tickdb.ai/v1/ws"

# ⚠️ 生产环境建议使用 asyncio + aiohttp 重构

class DepthSubscriber:
    """
    基于 SKILL 规范的 depth 订阅器
    实现了心跳保活、指数退避重连、限频处理
    """
    
    def __init__(self, symbol: str, depth: int = 10):
        self.symbol = symbol
        self.depth = depth
        self.ws = None
        self.retry_count = 0
        self.max_retries = 5
        self.running = False
    
    def connect(self):
        """建立 WebSocket 连接"""
        # 鉴权通过 URL 参数
        url = f"{WS_BASE_URL}?api_key={TICKDB_API_KEY}"
        
        try:
            self.ws = websocket.WebSocketApp(
                url,
                on_open=self._on_open,
                on_message=self._on_message,
                on_error=self._on_error,
                on_close=self._on_close
            )
            self.running = True
            # 在独立线程中运行
            thread = threading.Thread(target=self.ws.run_forever)
            thread.daemon = True
            thread.start()
            print(f"WebSocket 连接已建立，订阅 {self.symbol} 的 depth 数据")
        except Exception as e:
            print(f"连接失败: {e}")
            self._schedule_reconnect()
    
    def _on_open(self, ws):
        """连接建立后，发送订阅请求"""
        subscribe_msg = {
            "cmd": "subscribe",
            "channel": "depth",
            "symbol": self.symbol,
            "depth": self.depth
        }
        ws.send(json.dumps(subscribe_msg))
        self.retry_count = 0  # 重置重试计数
        print(f"已订阅 {self.symbol} depth 频道（{self.depth} 档）")
    
    def _on_message(self, ws, message):
        """处理接收到的消息"""
        try:
            data = json.loads(message)
            
            # 心跳响应
            if data.get("cmd") == "pong":
                return
            
            # depth 数据
            if data.get("channel") == "depth":
                self._process_depth(data)
                
        except json.JSONDecodeError:
            print(f"JSON 解析错误: {message}")
    
    def _process_depth(self, data):
        """处理 depth 数据，AI 可进一步计算买卖压力比"""
        bids = data.get("bids", [])
        asks = data.get("asks", [])
        
        # 计算买卖压力比（SKILL 规范中的衍生指标）
        bid_volume = sum(float(b[1]) for b in bids)
        ask_volume = sum(float(a[1]) for a in asks)
        pressure_ratio = bid_volume / ask_volume if ask_volume > 0 else 0
        
        print(f"[{data.get('timestamp')}] "
              f"{self.symbol} | 买盘量: {bid_volume:.0f} | "
              f"卖盘量: {ask_volume:.0f} | 压力比: {pressure_ratio:.2f}")
    
    def _on_error(self, ws, error):
        """错误处理"""
        print(f"WebSocket 错误: {error}")
        self.running = False
    
    def _on_close(self, ws, close_status_code, close_msg):
        """连接关闭，触发重连"""
        print(f"连接关闭 (code: {close_status_code})")
        self.running = False
        self._schedule_reconnect()
    
    def _schedule_reconnect(self):
        """指数退避 + 抖动重连"""
        if self.retry_count >= self.max_retries:
            print("达到最大重试次数，停止重连")
            return
        
        base_delay = 1
        max_delay = 32
        delay = min(base_delay * (2 ** self.retry_count), max_delay)
        jitter = (hash(str(time.time())) % 100) / 100 * 0.2 * delay  # 确定性抖动
        self.retry_count += 1
        
        print(f"{delay + jitter:.2f} 秒后尝试第 {self.retry_count} 次重连...")
        time.sleep(delay + jitter)
        
        if not self.running:
            self.connect()
    
    def close(self):
        """关闭连接"""
        self.running = False
        if self.ws:
            self.ws.close()


# 使用示例
if __name__ == "__main__":
    subscriber = DepthSubscriber("NVDA.US", depth=10)
    try:
        # 持续运行，按 Ctrl+C 退出
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n关闭连接...")
        subscriber.close()

---

八、SKILL 协议的价值对比

维度	传统 API 调用	SKILL 协议
接入门槛	需要写代码	自然语言描述即可
扩展性	每新增接口需改代码	只需更新 skill.md
多轮对话	需自行管理上下文	原生支持上下文感知
错误恢复	需自行实现重试	规范内置重试策略
跨市场	需分别处理每个市场	统一 symbol 格式
AI 理解准确率	依赖 Prompt 工程	基于 JSON Schema 自动校验

---

九、在 AI 助手中使用 TickDB SKILL

9.1 安装 SKILL

访问 ClawHub，搜索 tickdb-market-data，点击安装。

安装完成后，AI 助手会下载 http://skill.md/ 文件，解析 capabilities 定义，注册可用的函数。

9.2 使用示例

安装后，你可以这样与 AI 助手对话：

你：帮我看看 NVDA 的实时行情和最近 5 天的走势

AI（自动调用函数后）：
当前 NVDA 报价 875.32 美元（+2.3%），过去 5 个交易日呈上升趋势，
成交量较前期增加约 35%。订单簿显示买卖压力比为 1.42，流动性良好。

你：订阅它的 depth 实时监控

AI（自动调用 WebSocket）：
已订阅 NVDA.US 的 10 档订单簿深度，当前买卖价差 0.04 美元...

9.3 多轮对话示例

第一轮：
你：NVDA 财报后走势怎样？
AI → 调用 get_kline_history + get_realtime_quote
AI → 回答：财报后 1 小时上涨 8%，但随后回落...

第二轮：
你：那特斯拉呢？（上下文感知，无需重复指定市场）
AI → 识别隐式标的切换
AI → 调用 TSLA.US 的 K 线和实时行情
AI → 回答：TSLA 当前走势...

第三轮：
你：比较一下两只股票的买卖价差
AI → 记住两个标的
AI → 同时调用两者的 quote
AI → 输出对比表格

---

十、结语

SKILL 协议的本质，是让 AI 从“知道该做什么”进化到“能实际做到”。

当 AI 读懂 skill.md，它就拥有了一张能力地图。这张地图让 AI 可以自主决定调用哪个接口、传入什么参数、把返回的数据变成真正有价值的回答。

对于量化开发者，这意味着你可以用自然语言调试策略、查询数据、监控市场。

对于普通投资者，这意味着你可以用对话的方式获取机构级的行情分析。

TickDB SKILL 正在让这件事成为现实。

---

下一步行动

如果你是量化开发者，想为自己的 AI 助手接入 TickDB：

1. 访问 ClawHub 安装 tickdb-market-data SKILL
2. 阅读 http://skill.md/ 了解完整的 capability 定义
3. 用自然语言测试多轮对话，观察 AI 如何调用函数

如果你习惯用 AI 辅助开发：
在 AI 助手中搜索安装 tickdb-market-data SKILL，然后问它：“帮我看看 BTC 最近 1 小时的走势和订单簿深度”。

如果你想了解更多 SKILL 协议的实现细节：
参考 TickDB 官方文档中的 SKILL 开发指南，或联系 [email protected] 获取技术白皮书。

---

本文不构成任何投资建议。市场有风险，投资需谨慎。