你上一次打开行情软件是为了查什么?

查一个股票价格?看一下期货合约的昨收?还是确认某数字货币的实时报价?

如果我说,现在你可以直接对着 AI 说「帮我看看英伟达今晚盘后的报价,以及台积电最近一根小时K的高低点」,然后得到结构化的数据回复——整个过程不需要你记住任何 API endpoint,不需要拼装 HTTP 参数,也不需要写一行 Python。

这不是在描述未来。这是 TickDB SKILL 协议今天就能做到的事。


一、为什么自然语言查行情是个工程问题

自然语言查询听起来很直觉,但它的实现链路远比想象中复杂。

一个 AI 助手要「听懂」你的查询并返回准确数据,需要跨越三重鸿沟:

语义鸿沟:用户的「刚才」「最近一根」「今晚盘后」在时间维度上表达模糊,AI 需要将这些口语化表述映射到精确的时间区间和交易时段。

知识鸿沟:AI 模型本身并不知道 TickDB 有哪些接口、每个接口的参数是什么,它需要一个标准化的描述体系来「学会」调用正确的工具。

数据鸿沟:即便 AI 生成了正确的调用请求,数据源是否支持低延迟的实时查询?历史数据的时间范围够不够?不同资产类别的数据格式是否统一?

前两重鸿沟的交汇点,就是 SKILL 协议要解决的问题。


二、SKILL 协议:让 AI 学会调用你的 API

2.1 协议的核心设计思想

http://skill.md/ 定义了一套标准协议,用于将外部 API 的能力描述为 AI Agent 可理解的「技能清单」。它的核心逻辑非常简洁:

API 不是给 AI 用的。SKILL 描述才是。

开发者为每个 API 端点编写一份 SKILL 描述文档(通常是一个结构化的 Markdown 或 JSON Schema),这份文档精确描述了:

  • 这个接口叫什么(自然语言名称)
  • 它能做什么(功能描述)
  • 需要哪些参数(参数名、类型、约束、示例值)
  • 返回什么数据(字段说明、单位、格式)
  • 有什么使用限制(频率限制、错误码含义)

AI 助手在加载 SKILL 描述后,就拥有了「调用这个 API 的知识」,而不需要事先在模型权重中记忆这些细节。

2.2 SKILL 描述的结构示例

以下是一个简化后的 TickDB 行情查询 SKILL 描述片段,用于展示协议的核心结构:

# SKILL: TickDB Market Data Query

## identity
name: tickdb_market_data
version: 1.0.0
description: 查询全球多个市场的实时行情、历史K线和订单簿深度数据

## tools

### get_realtime_quote
description: 获取指定交易品种的实时行情快照,包括最新价、涨跌幅、成交量等
parameters:
  symbol:
    type: string
    required: true
    description: 交易品种代码,如 AAPL.US、BTC.USDT、GC.CMD
    examples:
      - AAPL.US
      - BTC.USDT
      - GC.CMD
  fields:
    type: string[]
    required: false
    description: 需要返回的行情字段,默认返回全部字段
    examples:
      - [last, change, change_pct, volume]

这份描述的作用等价于为 AI 提供了一份「API 使用说明书」,但用 AI 能理解的自然语言和结构化字段编写。

2.3 SKILL 协议与传统 API 文档的本质区别

维度 传统 API 文档 SKILL 描述
目标读者 开发者(人类) AI Agent(机器可读+人可读)
参数描述 偏技术(类型、格式) 偏语义(这个参数做什么、怎么用)
示例值 偏代码片段 偏自然语言(AAPL.US 而不是 "AAPL.US"
错误处理 代码级异常 业务级错误含义(用户该怎么响应)
上下文感知 支持关联描述(接口之间的关系)

三、Function Calling:AI 调用工具的标准协议

3.1 Function Calling 是什么

Function Calling(函数调用)是主流大语言模型在 2023 年下半年普遍支持的一种能力。它的本质是:

当 AI 判断用户的请求需要执行某个动作时,它生成一个结构化的调用请求,而不是直接输出文字。

这个请求包含函数名、参数名和参数值,外部系统接收后执行真实的计算或查询,然后将结果返回给 AI,AI 再将结果转译为人类可读的回答。

整个过程形成了经典的「工具调用循环」:

用户自然语言查询
    ↓
AI 理解意图,选择工具(SKILL 描述提供知识)
    ↓
AI 生成 Function Calling 请求 { tool: "get_realtime_quote", args: {...} }
    ↓
外部系统执行 API 调用,获取真实数据
    ↓
数据返回给 AI
    ↓
AI 将结构化数据转译为自然语言回复

3.2 Function Calling 的关键设计

一个典型的 Function Calling 请求结构如下:

{
  "tool_calls": [
    {
      "id": "call_abc123",
      "type": "function",
      "function": {
        "name": "get_realtime_quote",
        "arguments": "{\"symbol\": \"NVDA.US\"}"
      }
    }
  ]
}

关键设计点:

参数约束:通过 SKILL 描述中的 requiredtype 字段,AI 生成严格类型正确的参数。模型不会凭空捏造一个不存在的股票代码——如果用户输入了无效代码,AI 会基于 SKILL 描述中的 examples 提示用户修正。

意图判断:AI 并不是每次都调用工具。当用户的问题可以直接基于模型知识回答时(如「什么是K线」),AI 不会触发 Function Calling。工具调用是「有需要才触发」的。

批量调用:在复杂场景下,单次回复可能需要调用多个工具(如「帮我比较英伟达和苹果的今日行情」会触发两次 get_realtime_quote),AI 可以并行生成多个 tool_calls


四、实战:构建一个 TickDB 自然语言行情助手

接下来进入工程环节。我将展示一个完整的实现:从 SKILL 描述的定义,到 AI Agent 的接入配置,再到自然语言查询的端到端运行流程。

4.1 定义 TickDB 的 SKILL 描述

首先,我们需要为 TickDB 的核心能力编写 SKILL 描述。以下是一个针对行情查询和K线获取的完整描述:

# SKILL: TickDB Market Intelligence

## identity
provider: TickDB
version: 1.0.0
description: >
  TickDB 提供全球多个市场的实时行情和历史K线数据。
  支持美股、港股、数字货币、期货、外汇、贵金属六大类资产。
  通过自然语言接口,可以查询实时价格、历史走势和订单簿深度。

## capabilities
- 实时行情快照(价格、涨跌、成交量、买卖盘口)
- 历史K线查询(1m/5m/15m/1h/4h/1d/1w 多周期)
- 多品种批量查询
- 支持中文品种名称和代码混用查询

## tools

### get_realtime_quote
description: 获取一个或多个交易品种的实时行情数据
parameters:
  type: object
  properties:
    symbols:
      type: array
      items: type: string
      description: 交易品种代码列表
      examples:
        - AAPL.US
        - [NVDA.US, TSLA.US]
        - BTC.USDT
      required: true
    fields:
      type: array
      items: type: string
      description: 返回字段,缺省返回全部字段
      enum: [symbol, last, open, high, low, close, volume, amount, change, change_pct, bid, ask, timestamp]
      required: false
  required: [symbols]

### get_historical_kline
description: 获取指定交易品种的历史K线数据,用于策略回测和技术分析
parameters:
  type: object
  properties:
    symbol:
      type: string
      description: 交易品种代码
      examples: [AAPL.US, BTC.USDT, GC.CMD]
      required: true
    interval:
      type: string
      description: K线周期
      enum: [1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w]
      required: true
    start_time:
      type: string
      format: unix_timestamp
      description: 起始时间戳(Unix 秒)
      required: true
    end_time:
      type: string
      format: unix_timestamp
      description: 结束时间戳(Unix 秒),缺省为当前时间
      required: false
    limit:
      type: integer
      description: 数据条数上限,最大 1000
      default: 100
      required: false

### get_orderbook_depth
description: 获取订单簿深度数据,分析买卖盘结构和流动性
parameters:
  type: object
  properties:
    symbol:
      type: string
      description: 交易品种代码
      examples: [AAPL.US, BTC.USDT]
      required: true
    depth:
      type: integer
      description: 深度档位数,1-50,默认10
      default: 10
      required: false

4.2 接入 AI Agent:完整 Python 实现

以下是接入主流 AI 平台(以 OpenAI 兼容接口为例)的完整生产级代码:

import os
import json
import time
import random
from typing import Optional
from dataclasses import dataclass, field
from datetime import datetime

import requests


@dataclass
class TickDBConfig:
    """TickDB 配置"""
    api_key: str = field(default_factory=lambda: os.environ.get("TICKDB_API_KEY", ""))
    base_url: str = "https://api.tickdb.ai/v1"
    max_retries: int = 3
    timeout: tuple = (3.05, 10)  # (connect_timeout, read_timeout)

    def __post_init__(self):
        if not self.api_key:
            raise ValueError(
                "未设置 TICKDB_API_KEY 环境变量。"
                "请运行: export TICKDB_API_KEY='your_key_here'"
            )


class TickDBSkill:
    """
    TickDB SKILL 协议实现。
    提供 SKILL 描述注册、Function Calling 路由和执行能力。
    """

    # SKILL 描述(来自 http://skill.md/ 规范)
    SKILL_MANIFEST = {
        "name": "tickdb_market_data",
        "version": "1.0.0",
        "description": "查询全球多个市场的实时行情、历史K线和订单簿深度数据",
        "tools": [
            {
                "type": "function",
                "function": {
                    "name": "get_realtime_quote",
                    "description": "获取一个或多个交易品种的实时行情数据",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "symbols": {
                                "type": "array",
                                "items": {"type": "string"},
                                "description": "交易品种代码列表,如 AAPL.US、BTC.USDT"
                            },
                            "fields": {
                                "type": "array",
                                "items": {"type": "string"},
                                "description": "返回字段,缺省返回全部字段",
                                "enum": ["symbol", "last", "change", "change_pct", "volume", "timestamp"]
                            }
                        },
                        "required": ["symbols"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_historical_kline",
                    "description": "获取指定交易品种的历史K线数据,用于策略回测和技术分析",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "symbol": {"type": "string", "description": "交易品种代码"},
                            "interval": {
                                "type": "string",
                                "description": "K线周期",
                                "enum": ["1m", "5m", "15m", "30m", "1h", "4h", "1d", "1w"]
                            },
                            "start_time": {"type": "string", "description": "起始时间戳(Unix秒)"},
                            "end_time": {"type": "string", "description": "结束时间戳(Unix秒)"},
                            "limit": {"type": "integer", "description": "数据条数上限", "default": 100}
                        },
                        "required": ["symbol", "interval", "start_time"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_orderbook_depth",
                    "description": "获取订单簿深度数据,分析买卖盘结构和流动性",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "symbol": {"type": "string", "description": "交易品种代码"},
                            "depth": {"type": "integer", "description": "深度档位数,默认10", "default": 10}
                        },
                        "required": ["symbol"]
                    }
                }
            }
        ]
    }

    def __init__(self, config: Optional[TickDBConfig] = None):
        self.config = config or TickDBConfig()
        self._tool_map = {
            "get_realtime_quote": self._fetch_realtime_quote,
            "get_historical_kline": self._fetch_kline,
            "get_orderbook_depth": self._fetch_orderbook,
        }

    # ─── Tool Handlers ───────────────────────────────────────────────────────

    def execute_tool(self, tool_name: str, arguments: dict) -> dict:
        """AI Agent 的 Function Calling 入口"""
        handler = self._tool_map.get(tool_name)
        if not handler:
            return {"error": f"Unknown tool: {tool_name}"}
        return handler(arguments)

    def _fetch_realtime_quote(self, args: dict) -> dict:
        """获取实时行情"""
        symbols = args.get("symbols", [])
        fields = args.get("fields")

        params = {"symbol": ",".join(symbols)}
        if fields:
            params["fields"] = ",".join(fields)

        data = self._request("GET", "/market/ticker", params=params)
        return {"content": [{"type": "text", "text": json.dumps(data, ensure_ascii=False)}]}

    def _fetch_kline(self, args: dict) -> dict:
        """获取历史K线"""
        params = {
            "symbol": args["symbol"],
            "interval": args["interval"],
            "start_time": args["start_time"],
            "limit": args.get("limit", 100),
        }
        if "end_time" in args:
            params["end_time"] = args["end_time"]

        data = self._request("GET", "/market/kline", params=params)
        return {"content": [{"type": "text", "text": json.dumps(data, ensure_ascii=False)}]}

    def _fetch_orderbook(self, args: dict) -> dict:
        """获取订单簿深度"""
        params = {
            "symbol": args["symbol"],
            "depth": args.get("depth", 10),
        }
        data = self._request("GET", "/market/depth", params=params)
        return {"content": [{"type": "text", "text": json.dumps(data, ensure_ascii=False)}]}

    # ─── HTTP Layer ─────────────────────────────────────────────────────────

    def _request(self, method: str, path: str, params: Optional[dict] = None) -> dict:
        """带重试和限频处理的 HTTP 请求"""
        headers = {"X-API-Key": self.config.api_key}
        url = f"{self.config.base_url}{path}"

        last_error = None
        for attempt in range(self.config.max_retries):
            try:
                response = requests.request(
                    method,
                    url,
                    headers=headers,
                    params=params,
                    timeout=self.config.timeout
                )

                # 限频处理(code: 3001)
                if response.status_code == 429 or (
                    response.headers.get("Content-Type", "").startswith("application/json")
                    and response.json().get("code") == 3001
                ):
                    retry_after = int(response.headers.get("Retry-After",
                                 response.headers.get("X-RateLimit-Reset", 5)))
                    print(f"[限频] 等待 {retry_after} 秒后重试...")
                    time.sleep(retry_after)
                    continue

                response.raise_for_status()
                result = response.json()

                # 业务层错误处理
                if result.get("code") != 0:
                    raise RuntimeError(f"TickDB API 错误: {result}")

                return result.get("data", result)

            except requests.exceptions.Timeout:
                last_error = f"请求超时(尝试 {attempt + 1}/{self.config.max_retries})"
                delay = min(1.0 * (2 ** attempt), 8.0) + random.uniform(0, 0.5)
                time.sleep(delay)
            except requests.exceptions.RequestException as e:
                last_error = f"网络错误: {e}"
                delay = min(1.0 * (2 ** attempt), 8.0) + random.uniform(0, 0.5)
                time.sleep(delay)

        raise RuntimeError(f"请求失败,已重试 {self.config.max_retries} 次: {last_error}")

4.3 AI Agent 核心:自然语言理解层

SKILL 描述和 API 调用层就绪后,现在需要接入大语言模型来实现自然语言理解。以下是一个完整的 Agent 实现:

import openai


class MarketDataAgent:
    """
    基于 SKILL 协议的自然语言行情查询 Agent。
    支持 OpenAI 兼容接口(GPT-4o、Claude 等)。
    """

    SYSTEM_PROMPT = """你是一个专业的金融市场数据助手。

你拥有以下工具(SKILL)可供调用。当你需要查询实时数据时,
必须使用工具调用来获取最新信息,而不是凭记忆回答。

关于 TickDB:
- 支持美股(如 AAPL.US、NVDA.US)、港股、数字货币、外汇等
- 所有代码格式为「代码.市场」如 BTC.USDT、GC.CMD
- 行情时间均为 UTC 时间
- 品种查询支持中文名称(如「英伟达」→ NVDA.US)

重要原则:
1. 只调用工具,不捏造数据
2. 用户问「刚才」「最近」等模糊时间,你需要转换为具体时间戳
3. 返回数据后,用简洁的自然语言总结关键数据
4. 不提供任何投资建议"""

    def __init__(self, skill: TickDBSkill, model: str = "gpt-4o"):
        self.skill = skill
        self.model = model
        self.client = openai.OpenAI()
        self.messages = [{"role": "system", "content": self.SYSTEM_PROMPT}]

        # 注册 SKILL 清单(用于 AI 理解可用工具)
        self.available_tools = skill.SKILL_MANIFEST["tools"]

    def chat(self, user_query: str, max_turns: int = 5) -> str:
        """
        处理自然语言查询的完整流程。
        支持多轮工具调用(AI 可能在一次回答中多次调用工具)。
        """
        self.messages.append({"role": "user", "content": user_query})

        turns = 0
        while turns < max_turns:
            turns += 1

            # Step 1: AI 决定是否调用工具
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.available_tools,
                tool_choice="auto",
                temperature=0.1,  # 低温度确保工具选择稳定性
            )

            assistant_message = response.choices[0].message
            self.messages.append({
                "role": "assistant",
                "content": assistant_message.content,
                "tool_calls": assistant_message.tool_calls
            })

            # Step 2: 如果没有工具调用,说明 AI 已有最终回答
            if not assistant_message.tool_calls:
                final_text = assistant_message.content or ""
                self.messages.append({"role": "user", "content": "[对话结束标记]"})
                return final_text

            # Step 3: 执行所有工具调用
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                arguments = json.loads(tool_call.function.arguments)

                print(f"[工具调用] {tool_name} → {json.dumps(arguments, ensure_ascii=False)}")

                try:
                    result = self.skill.execute_tool(tool_name, arguments)
                    tool_result = json.dumps(result, ensure_ascii=False, indent=2)
                except Exception as e:
                    tool_result = json.dumps({"error": str(e)}, ensure_ascii=False)

                # 将工具结果反馈给 AI
                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                })

        return "[对话达到最大轮次限制]"

4.4 运行示例

# 初始化
agent = MarketDataAgent(skill=TickDBSkill())

# 查询一:实时行情
print(agent.chat("英伟达和苹果现在的股价是多少?涨了多少?"))

# 输出:
# [工具调用] get_realtime_quote → {"symbols": ["NVDA.US", "AAPL.US"], "fields": ["last", "change", "change_pct"]}
# 英伟达(NVDA.US)当前报价 875.32 美元,日涨幅 +3.45%,上涨 29.22 美元。
# 苹果(AAPL.US)当前报价 189.84 美元,日涨幅 +1.12%,上涨 2.10 美元。

# 查询二:历史K线
print(agent.chat(
    "看一下比特币最近24小时的1小时K线,收盘价最高的是哪根?"
))

# 输出:
# [工具调用] get_historical_kline → {"symbol": "BTC.USDT", "interval": "1h", 
#         "start_time": "1745347200", "limit": 24}
# 最近24小时(UTC)比特币1小时K线中,收盘价最高的是 2026-04-22 18:00 那根,
# 收盘价为 97,234.50 USDT,当日最高触及 97,890.00。

# 查询三:订单簿深度
print(agent.chat("给我看看苹果的订单簿,买卖盘压力大不大?"))

# 输出:
# [工具调用] get_orderbook_depth → {"symbol": "AAPL.US", "depth": 10}
# 苹果当前买一~买十档合计 23,450 股,卖一~卖十档合计 18,920 股,
# 买卖压力比约为 1.24,买盘略占优势。买卖价差 0.02 美元(0.01%),
# 市场流动性正常,未见明显失衡。

五、从 SKILL 到 AI Native:协议设计的深层价值

5.1 为什么 SKILL 协议比普通 Plugin 更适合 API 提供商

目前主流 AI 平台提供的 Plugin/Tool 扩展机制主要分为两类:OpenAI 的 Tool Use 和 Anthropic 的 Tool Use。二者在底层都依赖 Function Calling,但 SKILL 协议在 API 提供商侧提供了额外的结构化保障:

语义层与传输层分离。API 的技术实现(HTTP 请求格式、鉴权方式、错误码体系)不需要暴露给 AI。SKILL 描述只描述「做什么」和「怎么用」,AI 基于这个描述生成调用请求,由 SKILL 协议的运行时负责将请求翻译为正确的 HTTP 调用。

渐进式能力扩展。当 TickDB 新增接口时,只需在 SKILL 描述中注册新的工具定义,AI Agent 无需重新配置即可理解并调用新接口。这比传统的 Plugin 热更新更灵活。

5.2 时间语义理解:自然语言查询的技术难点

在所有自然语言查询中,时间是最复杂的语义维度。「最近」「刚才」「今天」「开盘以来」这些表述在不同资产类别、不同交易所规则下含义完全不同:

自然语言时间 美股(UTC-5/-4) 数字货币(24h) 期货(有结算日)
「今天」 最近一个交易日 当日00:00 UTC至今 合约所在交易日
「最近」 通常指最近1根K线或5分钟 最近1根K线 最近1根K线
「开盘以来」 盘前+盘中所有成交 00:00 UTC起 合约开盘至今

SKILL 描述中通过 examples 字段提供的时间示例,帮助 AI 建立「时间表述 → 时间戳」的映射能力。但更准确的时间解析,通常需要在 Agent 层增加一个「时间预处理」环节:

from datetime import datetime, timezone

def normalize_time_reference(text: str) -> tuple[str, str]:
    """
    将自然语言时间引用转换为 Unix 时间戳。
    实际生产中建议接入专门的 timeNL 解析库。
    """
    now = datetime.now(timezone.utc)
    text_lower = text.lower()

    if "最近1小时" in text_lower or "最近一根" in text_lower:
        start = int((now.timestamp()) - 3600)
        end = int(now.timestamp())
    elif "最近24小时" in text_lower:
        start = int((now.timestamp()) - 86400)
        end = int(now.timestamp())
    elif "今天" in text_lower:
        start = int(datetime.combine(now.date(), datetime.min.time()).timestamp())
        end = int(now.timestamp())
    else:
        # 默认返回最近100条K线
        start = 0
        end = int(now.timestamp())

    return str(start), str(end)

5.3 多工具协作:复杂查询的路由策略

在更复杂的场景中,单次用户查询可能需要调用多个不同来源的工具。例如:「帮我看看英伟达的股价,再查一下期权市场的隐含波动率」——这需要同时调用 TickDB(股价)和另一个期权数据源(IV)。

SKILL 协议支持在同一 SKILL 清单中注册多个不同提供商的工具:

# 扩展 SKILL 清单,支持多数据源
agent = MarketDataAgent(skill=TickDBSkill())
agent.available_tools.extend(OptionsDataSkill().SKILL_MANIFEST["tools"])
agent.available_tools.extend(EconomicCalendarSkill().SKILL_MANIFEST["tools"])

AI 在收到查询后,会自动分析用户意图,将请求拆分为多个工具调用,并按依赖顺序执行(如先获取股价,再查相关期权链)。


六、部署方案

场景 推荐配置 说明
个人研究 单 Agent 实例 + GPT-4o mini 成本最低,适合学习和小规模验证
量化研究 Agent 集群 + 本地模型(Llama3) + TickDB 历史数据 数据不出本地,适合策略保密需求
机构应用 多 Agent 协作 + 专业模型 + 企业级 TickDB 方案 支持高并发,有 SLA 保障

结语

SKILL 协议的核心价值,不是「让 AI 能查行情」,而是将数据能力从代码层抽象到语义层

一旦你的 API 有了 SKILL 描述,它就不再是一个需要程序员编写的工具,而是一个 AI Agent 可以理解、选择和调用的技能。这种转变的意义远超过「省几行代码」——它意味着数据的消费方式正在从「程序员调用 API」进化到「AI 理解意图后调用数据」。

对于量化开发者,这意味着你的策略回测框架、市场监控系统甚至风控仪表盘,都可以用自然语言来驱动。对于数据平台而言,这意味着每一个 SKILL 都是一个 AI Native 的分发渠道——用户不需要记住你的 API 文档,只需要「会说话」。

价格是结果,数据是原因。而自然语言,是连接原因与结果的最短路径。


下一步行动

如果你是个人开发者

  1. 访问 tickdb.ai 注册,获取免费 API Key
  2. 在 AI 助手中搜索并安装 tickdb-market-data SKILL
  3. 直接开始用自然语言查询行情

如果你希望接入 SKILL 协议开发自己的 Agent

  • 参考 http://skill.md/ 完整协议规范
  • 使用本文的 TickDBSkill 类作为起点
  • 接入 OpenAI/Anthropic/本地模型的 Function Calling 接口

如果你是量化团队负责人,希望基于 TickDB 构建机构级的 AI 行情分析系统:

  • 联系 [email protected],获取企业级 API 方案(更高频率限制、10 年历史数据、专属技术支持)

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