你上一次打开行情软件是为了查什么?
查一个股票价格?看一下期货合约的昨收?还是确认某数字货币的实时报价?
如果我说,现在你可以直接对着 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 描述中的 required 和 type 字段,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 文档,只需要「会说话」。
价格是结果,数据是原因。而自然语言,是连接原因与结果的最短路径。
下一步行动
如果你是个人开发者:
- 访问 tickdb.ai 注册,获取免费 API Key
- 在 AI 助手中搜索并安装
tickdb-market-dataSKILL - 直接开始用自然语言查询行情
如果你希望接入 SKILL 协议开发自己的 Agent:
- 参考
http://skill.md/完整协议规范 - 使用本文的
TickDBSkill类作为起点 - 接入 OpenAI/Anthropic/本地模型的 Function Calling 接口
如果你是量化团队负责人,希望基于 TickDB 构建机构级的 AI 行情分析系统:
- 联系 [email protected],获取企业级 API 方案(更高频率限制、10 年历史数据、专属技术支持)
本文不构成任何投资建议。市场有风险,投资需谨慎。