当 AI 问你"想买点英伟达"时，你知道它有多狼狈吗
一个真实场景：你在 Cursor 里随口说"帮我看看英伟达最近的走势"，AI 助手大概率会回复一段关于如何申请 API Key 的废话，或者直接告诉你"我无法获取实时数据"。
这背后不是一个能力问题，而是一个协议问题——AI 不知道如何将"英伟达的走势"这个模糊意图，转化为结构化的行情查询请求。它没有统一的方式理解：查询哪个标的？用什么时间周期？返回什么字段？
这就是 SKILL 协议要解决的核心问题。
一、为什么 AI 需要行情数据的"翻译层"
通用大模型在对话上有惊人表现，但在金融市场数据面前，它面临三重困境：
语义歧义。用户说"英伟达最近怎么样"，AI 需要推断是日线还是小时线？"最近"是指近一周、近一月、还是年初至今？这些对人类交易员是常识，对 AI 是零样本推理。
数据源异构。不同数据源有完全不同的命名体系和接口设计。Yahoo Finance 用 NVDA，TickDB 用 NVDA.US，Polygon 用 Nvidia，Binance 用 BTCUSDT。AI 如果没有显式指引，极可能构造出错误的标的代码。
上下文断裂。一次对话中的多轮查询之间存在隐式关联。用户问完"AAPL 当前价格"，再问"它比昨天涨了多少"——AI 必须记住前序查询结果，或者具备回溯上下文的能力。通用模型对此没有标准机制。
SKILL 协议的本质是一份合约：它告诉 AI，当用户提到某个市场数据意图时，应该调用哪个函数、传入什么参数、从哪里读取答案。
二、skill.md：一份给 AI 看的接口说明书
2.1 文件定位与基本结构
http://skill.md/ 是一个遵循固定 Schema 的 Markdown 文件，托管在 TickDB 官网或指定 CDN 路径上。AI 助手在初始化阶段（或按需拉取）读取这份文件，将其内化为"对 TickDB 行情数据的理解能力"。
文件结构分为四个顶级区块：
# SKILL Specification
## identity # 技能身份
## instructions # 行为指令（最重要）
## functions # 函数定义（AI 可调用的能力）
## examples # 对话样例
每一区块的内容量和精度，直接决定 AI 调用 TickDB 数据时的准确率。
2.2 identity：声明身份与能力边界
## identity
name: tickdb-market-data
version: "1.0"
description: >
TickDB Market Data SKILL provides real-time and historical
market data for global equities, digital currencies, futures,
forex, and commodities. Designed for quantitative trading and
investment research.
这一区块的作用是让 AI 在多 SKILL 协作场景下做选择：当用户安装了多个数据 SKILL 时，AI 需要根据 description 判断哪个 SKILL 更适合处理本次请求。
2.3 instructions：AI 的行为宪法
instructions 是整个协议中最容易被低估的部分。它不是简单告诉 AI"你能查什么"，而是告诉 AI"在什么场景下用什么方式查、查到后怎么返回"。
## instructions
## context_window
You maintain full conversation context. When the user asks
follow-up questions like "对比一下" or "再看其他标的", you
must reference previous query parameters (symbol, interval,
etc.) rather than requiring the user to repeat them.
## parameter_convention
- All symbol parameters MUST use the exchange-suffixed format:
`{TICKER}.{EXCHANGE}` — e.g., `AAPL.US`, `BTC.USDT`
- Supported exchanges: US (us equity), HK (hk equity),
Binance (digital currency), etc.
- Interval format: `1m`, `5m`, `15m`, `30m`, `1h`, `4h`,
`1d`, `1w`
- All timestamps use UTC
## response_format
When returning data, you MUST:
1. Format numbers with appropriate precision (price → 2dp,
volume → integer)
2. Include a brief market interpretation (e.g., "currently
in overbought territory")
3. Attach a one-line risk note for leveraged instruments
## error_handling
When encountering API errors:
- code 1001/1002: Prompt user to check API Key configuration
- code 2002: Inform user the symbol is not supported, suggest
alternatives
- code 3001: Wait and retry silently (do not surface to user)
注意 context_window 这一条——它解决了上文提到的"多轮对话上下文断裂"问题。AI 被显式要求记住前序参数，而非每次重新解析。
2.4 functions：Function Calling 的定义规范
functions 区块是整个 SKILL 协议的核心，直接决定了 AI 能够执行哪些操作。每个函数遵循 OpenAI Function Calling 的 Schema，同时扩展了 TickDB 特有的参数语义。
## functions
namespace functions {
// 获取单个标的实时行情快照
function get_realtime_quote: (_: {
// 标的代码（必填）
symbol: string,
// 可选返回字段列表，不填则返回全部
fields?: string[],
}) => any;
// 获取 K 线数据（历史回测）
function get_kline: (_: {
// 标的代码（必填）
symbol: string,
// K 线周期（必填）：1m | 5m | 15m | 30m | 1h | 4h | 1d | 1w
interval: string,
// 起始时间（必填）：ISO 8601 UTC
start_time: string,
// 结束时间（必填）：ISO 8601 UTC
end_time: string,
// 每页数据量上限，最大 1000
limit?: number,
}) => any;
// 实时 WebSocket 订阅
function subscribe_realtime: (_: {
// 标的代码
symbol: string,
// 订阅频道：quote | kline | depth | trades
channel: string,
// K 线周期（仅 kline 频道需填）
interval?: string,
}) => any;
}
函数定义中的每个字段都附带类型和约束说明。interval 参数限定为枚举值而非自由文本，这是避免 AI 构造非法参数的关键——模型只能从 1m|5m|...|1w 中选择，而非随意生成。
三、AI 如何解析 skill.md
3.1 解析链路
当 AI 助手（如 Claude Desktop、Cursor AI、GPTs）安装了 TickDB SKILL 后，解析链路如下：
用户输入 → AI 意图识别 → 匹配函数签名 → 构造参数 → 调用函数 → 格式化结果 → 返回用户
第一步：意图映射。AI 将用户自然语言映射到 functions 中定义的某个函数。映射依据包括：函数 description、用户查询的动词（"查"→quote，"看历史"→kline，"跟踪"→subscribe）和名词（"价格"→quote，"走势"→kline）。
第二步：参数提取。AI 从用户查询中提取各字段值。例如用户说"看看苹果最近一周的日线"，AI 识别出：
symbol= "AAPL.US"（通过标的规范化规则映射）interval= "1d"（通过"日线"关键词匹配）start_time= 当前日期减 7 天end_time= 当前日期
第三步：约束校验。AI 在构造参数后做自检：symbol 是否包含交易所后缀？interval 是否在枚举范围内？时间范围是否超过单次查询上限？这一步的准确性高度依赖 instructions 中的约束说明是否充分。
第四步：结果格式化。API 返回原始 JSON，AI 根据 instructions.response_format 将其转化为人类可读的描述文本，同时附上市场解读。
3.2 解析质量的决定因素
解析准确率取决于三个变量：
| 变量 | 影响 | 优化方向 |
|---|---|---|
instructions 清晰度 |
AI 对模糊意图的推断准确率 | 补充边界场景说明（如"用户说'走势'默认 1d，除非另有说明"） |
functions 参数约束 |
非法参数率 | 使用枚举而非自由文本；必填/可选严格区分 |
examples 样例数量 |
零样本推理置信度 | 覆盖高频场景和易混淆场景（如"AAPL"和"AAPL.US"的歧义处理） |
四、生产级 SKILL 集成代码
以下示例展示如何在 Python 环境中实现 SKILL 协议的核心调用逻辑。代码包含完整的错误处理、限频处理和重试机制。
import os
import json
import time
import random
import requests
from datetime import datetime, timedelta, timezone
class TickDBSKILL:
"""
TickDB Market Data SKILL 协议实现
支持：实时行情、历史 K 线、WebSocket 订阅
"""
BASE_URL = "https://api.tickdb.ai/v1/market"
WS_URL = "wss://stream.tickdb.ai/v1/market"
def __init__(self, api_key: str = None):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError("API Key 未配置，请设置 TICKDB_API_KEY 环境变量")
def _headers(self) -> dict:
return {"X-API-Key": self.api_key}
def _validate_symbol(self, symbol: str) -> str:
"""规范化标的代码，确保包含交易所后缀"""
if "." not in symbol:
raise ValueError(
f"标的代码 '{symbol}' 缺少交易所后缀，"
"请使用格式如 AAPL.US, BTC.USDT"
)
return symbol.upper()
def get_realtime_quote(self, symbol: str, fields: list = None) -> dict:
"""
获取实时行情快照
对应 SKILL 函数：get_realtime_quote
"""
symbol = self._validate_symbol(symbol)
params = {"symbol": symbol}
if fields:
params["fields"] = ",".join(fields)
response = requests.get(
f"{self.BASE_URL}/quote",
headers=self._headers(),
params=params,
timeout=(3.05, 10) # (connect_timeout, read_timeout)
)
return self._handle_response(response)
def get_kline(
self,
symbol: str,
interval: str,
start_time: str,
end_time: str,
limit: int = 1000,
) -> dict:
"""
获取 K 线数据（用于历史回测）
对应 SKILL 函数：get_kline
约束：时间范围不超过 500 天（1m 周期）或 10 年（1d 周期）
"""
valid_intervals = {"1m", "5m", "15m", "30m", "1h", "4h", "1d", "1w"}
if interval not in valid_intervals:
raise ValueError(
f"非法周期 '{interval}'，有效值：{valid_intervals}"
)
symbol = self._validate_symbol(symbol)
params = {
"symbol": symbol,
"interval": interval,
"start_time": start_time,
"end_time": end_time,
"limit": min(limit, 1000),
}
response = requests.get(
f"{self.BASE_URL}/kline",
headers=self._headers(),
params=params,
timeout=(3.05, 10)
)
return self._handle_response(response)
def _handle_response(self, response: requests.Response) -> dict:
"""
统一错误处理
对应 skill.md instructions.error_handling
"""
try:
data = response.json()
except json.JSONDecodeError:
raise RuntimeError(f"API 返回了无效的 JSON：{response.text[:200]}")
code = data.get("code", 0)
if code == 0:
return data.get("data")
error_messages = {
1001: "API Key 无效，请检查 TICKDB_API_KEY 环境变量",
1002: "API Key 缺失，请先在 tickdb.ai 注册并生成 Key",
2002: "标的代码不存在，请确认使用正确格式（如 AAPL.US）",
}
if code == 3001:
# ⚠️ 限频触发，静默等待后重试（不向用户暴露）
retry_after = int(response.headers.get("Retry-After", 5))
time.sleep(retry_after)
return None
message = error_messages.get(code, data.get("message", "未知错误"))
raise RuntimeError(f"[code:{code}] {message}")
# 使用示例
if __name__ == "__main__":
skill = TickDBSKILL()
# 实时行情查询
quote = skill.get_realtime_quote("AAPL.US")
print(f"AAPL 当前价格: {quote['close']}")
# 历史 K 线（近 30 个交易日日线）
end = datetime.now(timezone.utc)
start = end - timedelta(days=30)
klines = skill.get_kline(
symbol="AAPL.US",
interval="1d",
start_time=start.isoformat(),
end_time=end.isoformat(),
)
print(f"获取到 {len(klines)} 根 K 线")
⚠️ 工程提醒：上述实现适用于低频查询场景。若需毫秒级实时监控，应将 WebSocket 订阅部分改为
asyncio异步架构，并添加心跳保活和指数退避重连逻辑。
五、WebSocket 实时订阅的 SKILL 扩展
skill.md 中的 subscribe_realtime 函数定义了一种比轮询更高效的模式——AI 引导用户建立 WebSocket 订阅，让 TickDB 主动推送行情更新，而非反复轮询。
import json
import time
import websocket
import threading
import random
class TickDBWebSocket:
"""
TickDB WebSocket 实时订阅（SKILL subscribe_realtime）
包含心跳保活、指数退避重连、限频自适应处理
"""
MAX_RETRIES = 10
BASE_DELAY = 1
MAX_DELAY = 60
def __init__(self, api_key: str = None, on_message=None):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self.on_message = on_message or (lambda msg: print(msg))
self.ws = None
self.running = False
self._retry_count = 0
def subscribe(self, symbol: str, channel: str, interval: str = None):
"""
建立 WebSocket 订阅
channel 选项：quote | kline | depth | trades
"""
symbol = symbol.upper()
channel = channel.lower()
if "." not in symbol:
raise ValueError("标的代码必须包含交易所后缀，如 AAPL.US")
self.running = True
self._connect_and_subscribe(symbol, channel, interval)
def _connect_and_subscribe(self, symbol, channel, interval):
"""
带指数退避重连的连接逻辑
"""
while self.running and self._retry_count < self.MAX_RETRIES:
try:
url = (
f"wss://stream.tickdb.ai/v1/market?"
f"api_key={self.api_key}&symbol={symbol}&channel={channel}"
)
if interval:
url += f"&interval={interval}"
self.ws = websocket.WebSocketApp(
url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
)
# 运行 WebSocket 连接（含心跳）
self.ws.run_forever(ping_interval=20, ping_timeout=10)
self._retry_count = 0
except Exception as e:
if not self.running:
break
self._retry_count += 1
delay = min(
self.BASE_DELAY * (2 ** self._retry_count) +
random.uniform(0, 1), # 抖动，避免惊群
self.MAX_DELAY
)
print(f"[重连] 第 {self._retry_count} 次，{delay:.1f} 秒后重试")
time.sleep(delay)
def _on_message(self, ws, message):
"""消息处理：支持限频自适应（code:3001）"""
try:
data = json.loads(message)
code = data.get("code", 0)
if code == 3001:
retry_after = int(data.get("retry_after", 5))
time.sleep(retry_after)
return
if code == 0:
self.on_message(data.get("data"))
else:
print(f"[异常] code:{code} {data.get('message')}")
except json.JSONDecodeError:
print(f"[解析错误] {message[:100]}")
def _on_error(self, ws, error):
print(f"[WebSocket 错误] {error}")
def _on_close(self, ws, code, reason):
if self.running:
print(f"[连接断开] code:{code} reason:{reason}")
def unsubscribe(self):
"""主动关闭订阅"""
self.running = False
if self.ws:
self.ws.close()
# 使用示例
def handle_price_alert(data):
symbol = data.get("symbol")
price = data.get("close")
change_pct = data.get("change_pct", 0)
print(f"⚡ {symbol} 当前价 {price}，涨跌 {change_pct:+.2f}%")
ws = TickDBWebSocket(on_message=handle_price_alert)
ws.subscribe(symbol="NVDA.US", channel="quote")
⚠️ 高频场景注意：单个 WebSocket 连接可同时订阅多个标的（逗号分隔），但需遵守 TickDB 的连接数限制。建议机构用户使用连接池管理，而非每个标的独立开一个连接。
六、SKILL 协议能力对比
| 能力维度 | 通用 AI（无 SKILL） | SKILL 增强 AI |
|---|---|---|
| 标的识别 | 依赖训练数据，容易出错 | 协议显式规定 .US/.HK 等格式规范 |
| 周期映射 | 模糊，AI 自由推断 | instructions 规定默认行为，减少歧义 |
| 多轮对话上下文 | 需要在 prompt 中手动管理 | context_window 条款强制要求 AI 记忆 |
| 错误处理 | 通用的网络错误回复 | 针对 1001/2002/3001 的定向处理方案 |
| 实时数据 | 完全不支持 | WebSocket subscribe_realtime 频道 |
| 数据字段解释 | 数字罗列，无市场解读 | response_format 规定附加市场含义 |
| 格式规范化 | 小数位数不统一 | 协议规定 price → 2dp，volume → integer |
| 风险提示 | 无 | 杠杆类品种自动附加风险说明 |
七、SKILL 协议的扩展路径
当前 skill.md v1.0 聚焦于行情查询的核心场景。协议设计上预留了三个扩展方向：
多 SKILL 协作。当用户同时安装 TickDB（行情）和另一个新闻 SKILL（事件）时，identity.description 决定了 AI 如何分配任务——"用户问 NVDA 的价格和业绩"由 TickDB 处理价格部分，新闻 SKILL 处理业绩部分。
策略回测联动。在 functions 中增加 run_backtest 函数，AI 可以直接根据用户描述生成回测请求，实现"查询—分析—回测"的单轮对话闭环。
告警规则定义。扩展 subscribe_realtime 支持结构化告警条件的声明式描述，如 AI 接收"苹果跌破 150 块时提醒我"，自动翻译为 depth 频道的阈值监控。
结语
SKILL 协议的本质，是将 AI 与金融市场数据之间的"最后一公里"问题，从隐式推理变成显式合约。
skill.md 不是一个营销文档，而是一份给 AI 看的技术规范。它的质量——instructions 的完整性、functions 的约束精度、examples 的覆盖广度——直接决定了 AI 能否可靠地理解用户的行情意图，并构造出正确的查询请求。
对于量化开发者，这意味着 AI 辅助编程的边界从"写代码"扩展到"理解数据"；对于 TickDB，这意味着产品能力可以通过 AI 生态自动触达用户，而无需每个用户都手动阅读 API 文档。
下一步行动
如果你想直接体验 TickDB SKILL：
在 AI 助手中搜索并安装 tickdb-market-data SKILL，然后尝试用自然语言查询行情。
如果你需要在自己的产品中集成 SKILL 协议：
参考 http://skill.md/ 规范文件，按四个顶级区块（identity / instructions / functions / examples）构建你专属的 SKILL 文档，并确保 functions 中的每个参数都附有明确的类型约束和枚举值。
如果你想用历史数据验证策略：
访问 tickdb.ai，注册获取免费 API Key（无需信用卡），通过 REST API 获取 10 年级别的美股历史 K 线数据。
如果你在构建量化系统：TickDBSKILL 和 TickDBWebSocket 两个类的完整源码可在 TickDB 开发者文档中获取，支持 Python 和 JavaScript 双语言。
本文不构成任何投资建议。市场有风险，投资需谨慎。TickDB Market Data SKILL 是数据访问工具，不提供交易执行或投资决策功能。