用自然语言查行情：TickDB SKILL + AI Agent 实战集成

"帮我看看英伟达最近三天的波动率变化，顺便对比一下特斯拉。"

这是凌晨两点，量化研究员老张在关掉回测框架后，对着 AI 助手说的一句话。几秒后，屏幕上出现了完整的波动率分析图表。

这件事在两年前还做不到——API Key、接口文档、数据格式转换、轮询逻辑，每一步都需要写代码。但今天，一句自然语言就够了。

实现这个能力的核心机制叫 SKILL。在 TickDB 的语境里，SKILL 是一套将行情数据接入 AI Agent 的标准化协议。你不需要写任何 HTTP 请求代码，不需要管理 WebSocket 连接，只需要像安装插件一样，把 SKILL 安装进你的 AI 助手，然后用自然语言提问。

本文拆解 SKILL 的工作原理、安装流程、支持的数据范围，以及在真实量化场景中的使用体验。

---

一、为什么 AI Agent 需要 SKILL

通用大模型有一个根本局限：知识有截止日期，能力有边界。它们能写代码、能推理，但不知道英伟达今天的收盘价是多少，不知道港股期权的隐含波动率怎么查，不知道以太坊的深度数据哪里拿。

想让 AI Agent 真正介入量化工作流，有三条路：

方案	实现难度	数据实时性	维护成本
直接调用 REST API	高，需要处理鉴权、限频、重试	高	高
用 Function Calling 封装 API	中，需要编写 Function Schema	高	中
SKILL 协议（标准化中间层）	低，一句指令完成安装	高	低，由平台维护

第三条路就是 SKILL 的价值所在。它在大模型和真实市场数据之间加了一层标准化适配层：API Key 管理、接口路由、响应格式化，全部由 SKILL 内部处理。对用户来说，看到的只有自然语言输入和结构化数据输出。

---

二、SKILL 协议的核心工作原理

2.1 三层架构

用户（自然语言）
    ↓
AI Agent（理解意图，调用 SKILL）
    ↓
SKILL（协议适配，数据路由）
    ↓
TickDB API（真实市场数据）

SKILL 在这个链路中扮演了两个关键角色：

意图路由：当用户说"英伟达的波动率"时，SKILL 需要判断用户想要的是历史波动率、隐含波动率还是已实现波动率，然后选择对应的 API 端点。这个判断过程可以通过 System Prompt 引导模型完成，也可以由 SKILL 内置的解析规则完成。

响应格式化：TickDB API 返回的是 JSON 格式的原始数据，SKILL 负责将其转换为对模型友好的结构化描述，比如"英伟达近3日收盘价分别为 881.25、902.10、887.45，对数收益率标准差为 0.017，对应年化波动率约 27.1%"。

2.2 支持的能力矩阵

根据 TickDB 当前的数据能力，SKILL 可调度的功能覆盖以下维度：

数据类型	支持的操作	示例问法
K 线数据（历史）	查询任意周期 K 线，支持指定时间范围	"苹果最近 20 根日 K 的成交量变化"
最新报价	获取实时价格、涨跌幅、成交量	"腾讯现在的价格是多少"
订单簿深度	多档深度数据（市场差异：美股1档/港股10档/币10档）	"以太坊买卖盘前10档深度"
可交易品种	查询支持的市场和标的列表	"港股科技股有哪些可查"
波动率分析	基于收益率计算历史波动率	"英伟达和特斯拉谁的波动率更高"
产业链关联	事件驱动场景下的关联标的查询	"阿尔忒弥斯2号概念股有哪些"

需要注意的是，SKILL 的数据边界与 TickDB 本身的能力完全对齐：美股历史 K 线最长 10 年（清洗对齐），逐笔成交数据（trades）暂不支持美股和 A 股，外汇、贵金属、指数等品种不支持 depth 订单簿。这些限制是产品层面的约束，SKILL 不会超出这个边界工作。

---

三、安装 SKILL：完整操作流程

3.1 前置准备

在开始之前，你需要准备两样东西：

• 一个已激活的 TickDB API Key（注册地址：tickdb.ai，免费层有额度限制）
• 一个支持 SKILL 扩展的 AI 客户端（OpenAI ChatGPT、Claude.ai、Gemini Advanced 等主流平台均已支持）

3.2 安装步骤

第一步：在 AI 助手中搜索 SKILL

打开你的 AI 客户端，导航到插件或扩展市场（通常在设置或侧边栏中），搜索 tickdb-market-data。

第二步：配置 API Key

安装完成后，系统会提示你配置 TickDB API Key。这一步是必要的——SKILL 需要通过你的 API Key 向 TickDB 发起数据请求。

API Key 配置方式：
1. 登录 tickdb.ai，进入控制台
2. 在「API Keys」页面生成一个新 Key
3. 将 Key 填入 SKILL 配置页面的「API Key」字段
4. 保存配置，完成绑定

第三步：验证连接

安装并配置完成后，建议先用一条简单指令验证连接是否正常：

请用 TickDB 查询苹果（AAPL.US）最新收盘价。

如果 SKILL 正常工作，你会收到包含具体价格数据的回复。如果返回错误，检查 API Key 是否正确、环境变量的读取权限是否正常。

---

四、生产级代码：SKILL 的底层调用逻辑

虽然 SKILL 的核心价值是"免代码"，但对于希望深入理解的量化开发者，这里给出 SKILL 底层调用 TickDB API 的生产级参考实现。

4.1 REST API：获取历史 K 线

import os
import requests
import time
import random


def get_historical_klines(symbol: str, interval: str = "1h", limit: int = 100):
    """
    从 TickDB 获取历史 K 线数据。
    
    参数：
        symbol: 交易品种，如 AAPL.US、TSLA.US
        interval: K 线周期，支持 1m/5m/15m/1h/4h/1d/1w
        limit: 返回数据条数，最大 1000
    返回：
        list[dict]: K 线数据列表，每条包含 timestamp/open/high/low/close/vol
    """
    api_key = os.environ.get("TICKDB_API_KEY")
    if not api_key:
        raise ValueError("未设置 TICKDB_API_KEY 环境变量")

    base_url = "https://api.tickdb.ai/v1/market/kline"
    headers = {"X-API-Key": api_key}
    params = {"symbol": symbol, "interval": interval, "limit": limit}

    # ⚠️ TickDB 限制单次请求最大 1000 条，超过需分页
    all_data = []
    current_page = 0
    max_page = (limit // 1000) + 1

    for page in range(max_page):
        if len(all_data) >= limit:
            break
        params["page"] = page
        params["page_size"] = 1000

        try:
            response = requests.get(
                base_url,
                headers=headers,
                params=params,
                timeout=(3.05, 10)  # (connect_timeout, read_timeout)
            )
            response.raise_for_status()
            data = response.json()

            code = data.get("code", 0)
            if code == 0:
                chunk = data.get("data", {}).get("klines", [])
                if not chunk:
                    break
                all_data.extend(chunk)
            elif code == 3001:
                # 请求频率超限，按服务端指定时间等待
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"限频触发，等待 {retry_after} 秒")
                time.sleep(retry_after)
                time.sleep(random.uniform(0.1, 0.5))  # 抖动，避免惊群
                continue
            else:
                print(f"API 返回错误码 {code}: {data.get('message')}")
                break

        except requests.exceptions.Timeout:
            print(f"请求超时 symbol={symbol}，跳过此页")
            break
        except requests.exceptions.RequestException as e:
            print(f"网络异常 symbol={symbol}: {e}")
            break

    return all_data[:limit]


def calculate_realized_volatility(klines: list[dict], annualize: bool = True) -> float:
    """
    基于 K 线数据计算已实现波动率。
    
    已实现波动率 = sqrt(sum(对数收益率^2)) * sqrt(年化系数)
    年化系数：日线用 sqrt(252)，小时线用 sqrt(252*6.5)
    """
    if len(klines) < 2:
        return 0.0

    log_returns = []
    for i in range(1, len(klines)):
        prev_close = float(klines[i - 1]["close"])
        curr_close = float(klines[i]["close"])
        if prev_close > 0:
            log_return = __import__("math").log(curr_close / prev_close)
            log_returns.append(log_return ** 2)

    if not log_returns:
        return 0.0

    realized_vol = sum(log_returns) ** 0.5

    if annualize:
        # 假设小时 K 线，每交易日约 6.5 小时
        periods_per_year = 252 * 6.5
        realized_vol *= (periods_per_year ** 0.5)

    return round(realized_vol, 4)


if __name__ == "__main__":
    # 示例：对比英伟达和特斯拉的已实现波动率
    symbols = ["NVDA.US", "TSLA.US"]

    for sym in symbols:
        klines = get_historical_klines(sym, interval="1h", limit=168)  # 约一周
        if klines:
            rv = calculate_realized_volatility(klines)
            print(f"{sym} 近7日已实现波动率（年化）: {rv:.2%}")

4.2 WebSocket：实时订单簿深度

对于需要实时监控买卖盘深度的场景，以下代码演示了如何通过 WebSocket 订阅 depth 频道：

import os
import json
import time
import random
import threading
import websocket


class TickDBDepthMonitor:
    """
    TickDB WebSocket 实时深度监控。
    
    ⚠️ 生产环境建议：
    - 使用 aiohttp/asyncio 实现异步并发订阅多个标的
    - 将数据写入本地缓存（Redis/内存数据库）而非直接打印
    - 添加心跳超时检测，防止连接假死
    """

    def __init__(self, api_key: str, symbol: str, on_update=None):
        self.api_key = api_key
        self.symbol = symbol
        self.on_update = on_update
        self.ws = None
        self.running = False
        self.retry_count = 0
        self.max_retries = 10
        self.base_delay = 2  # 初始重连等待（秒）

    def connect(self):
        """建立 WebSocket 连接"""
        ws_url = f"wss://stream.tickdb.ai/v1/ws/stream?api_key={self.api_key}&symbol={self.symbol}&channel=depth"
        self.ws = websocket.WebSocketApp(
            ws_url,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close,
            on_open=self._on_open
        )
        self.running = True
        # 在独立线程中运行，保持主线程响应
        thread = threading.Thread(target=self.ws.run_forever)
        thread.daemon = True
        thread.start()
        print(f"已连接到 TickDB WebSocket，订阅 {self.symbol} depth 频道")

    def _on_open(self, ws):
        """连接建立后发送心跳保活"""
        ws.send(json.dumps({"cmd": "ping"}))

    def _on_message(self, ws, message):
        """处理深度数据推送"""
        try:
            data = json.loads(message)
            # 心跳响应
            if data.get("type") == "pong":
                return

            bids = data.get("data", {}).get("bids", [])
            asks = data.get("data", {}).get("asks", [])

            if bids and asks:
                total_bid_vol = sum(float(b[1]) for b in bids[:10])
                total_ask_vol = sum(float(a[1]) for a in asks[:10])
                pressure_ratio = total_bid_vol / total_ask_vol if total_ask_vol > 0 else 0

                depth_snapshot = {
                    "symbol": self.symbol,
                    "bid_volume": round(total_bid_vol, 2),
                    "ask_volume": round(total_ask_vol, 2),
                    "pressure_ratio": round(pressure_ratio, 3)
                }

                if self.on_update:
                    self.on_update(depth_snapshot)
                else:
                    print(depth_snapshot)

                # 定期发送心跳保活
                ws.send(json.dumps({"cmd": "ping"}))

        except (json.JSONDecodeError, KeyError) as e:
            print(f"数据解析异常: {e}")

    def _on_error(self, ws, error):
        print(f"WebSocket 错误: {error}")

    def _on_close(self, ws, close_status_code, close_msg):
        """连接断开时自动重连"""
        self.running = False
        if self.retry_count < self.max_retries:
            delay = min(self.base_delay * (2 ** self.retry_count), 60)
            jitter = random.uniform(0, delay * 0.1)  # 添加抖动
            self.retry_count += 1
            print(f"连接断开，{delay + jitter:.1f} 秒后第 {self.retry_count} 次重连...")
            time.sleep(delay + jitter)
            self.connect()
        else:
            print("达到最大重连次数，停止监控")

    def stop(self):
        self.running = False
        if self.ws:
            self.ws.close()


def example_callback(snapshot: dict):
    """当买卖压力比骤变时触发告警"""
    if snapshot["pressure_ratio"] > 2.5 or snapshot["pressure_ratio"] < 0.4:
        print(f"⚠️ 告警：{snapshot['symbol']} 买卖压力比异常 {snapshot['pressure_ratio']}")


if __name__ == "__main__":
    api_key = os.environ.get("TICKDB_API_KEY")
    if not api_key:
        print("请先设置 TICKDB_API_KEY 环境变量")
        exit(1)

    monitor = TickDBDepthMonitor(
        api_key=api_key,
        symbol="NVDA.US",
        on_update=example_callback
    )
    monitor.connect()

    # 持续运行 60 秒后停止
    try:
        time.sleep(60)
    finally:
        monitor.stop()

4.3 限频与错误处理要点

在上述两段代码中，有几个生产级工程细节值得特别说明：

限频响应码 3001：TickDB 对 API 请求有频率限制。当触发 3001 错误时，服务端会在响应头中返回 Retry-After 字段，告知客户端需要等待的秒数。直接 sleep 一个固定值（常见错误）是不可靠的——应该读取服务端返回的具体数值。多个客户端同时以固定值等待，会在服务端限频窗口结束后产生惊群效应，因此建议添加随机抖动。

接口区分：/v1/market/kline 获取的是历史已结束周期的 K 线数据，适合回测和事后分析；/v1/market/kline/latest 获取的是当前正在形成的 K 线，适合实时展示。混用这两个接口是常见错误来源。

---

五、SKILL 的实际使用体验

5.1 能做到的事

基于实测，以下场景 SKILL 可以流畅完成：

场景一：跨品种波动率对比

用户：对比英伟达、特斯拉、AMD 最近 30 个交易日的已实现波动率。

SKILL 响应：
- 英伟达（NVDA.US）：年化波动率 41.2%
- 特斯拉（TSLA.US）：年化波动率 38.7%
- AMD（AMD.US）：年化波动率 35.1%
结论：英伟达波动率最高，三者差距在 6 个百分点以内。

场景二：实时盘口异常监控

用户：帮我监控苹果 AAPL.US 的盘口深度，如果买卖压力比超过 2.5 或低于 0.4，告诉我。

SKILL 响应（实时推送）：
- 监测已开启
- [告警] 当前买卖压力比 2.73，买盘深度远超卖盘，注意流动性不对称风险。

场景三：产业链关联查询

用户：SpaceX 星舰第九次试飞相关的美股和港股标的有哪些？

SKILL 响应：
- 波音（BA.US）：NASA 主力承包商，星舰项目相关
- 洛克希德马丁（LMT.US）：联合发射联盟大股东
- AST SpaceMobile（ASTS.US）：低轨卫星通信，直接受益

5.2 当前限制

SKILL 的能力边界是清晰的，以下场景暂不支持：

• 逐笔成交（trades）：tick 级成交数据暂不支持美股和 A 股，无法用于高频订单流分析
• 盘前盘后价格：SKILL 返回的数据以 TickDB 实际接入的市场为准，部分品种盘前盘后数据可能缺失
• 预测性分析：SKILL 只提供数据查询和基于历史数据的统计分析，不生成买卖建议

---

六、SKILL 与传统方案的价值对比

维度	传统 API 接入	SKILL 方案
安装配置时间	30 分钟 - 数小时	<5 分钟
需要的代码量	50-200 行（鉴权、解析、错误处理）	0 行
数据种类覆盖	取决于开发者接入多少接口	自动对齐 TickDB 全量数据能力
自然语言交互	不支持	完全支持
实时监控配置	需编写监控逻辑	口头描述即可
适合人群	有 API 经验的开发者	分析师、研究员、量化新手
深度定制能力	高	有限（受 SKILL 接口限制）

SKILL 和传统 API 接入不是替代关系，而是互补关系：SKILL 适合快速探索和日常查询，当需要构建高度定制化的交易系统或进行大规模回测时，直接调用 REST/WebSocket API 仍然是更灵活的选择。

---

七、适用场景与部署建议

7.1 适合使用 SKILL 的场景

场景	说明
盘前快速扫盘	"帮我看看今天哪些科技股开盘跳空超过 2%"
策略效果验证	"我的趋势策略在过去一年表现如何"
异常监控	"特斯拉盘口深度异常时通知我"
产业链研究	"哪些公司直接受益于某个政策/事件"
数据探索	"港股上市的新能源汽车股有哪些，它们最近的成交量变化"

7.2 分场景部署建议

场景	推荐方案
个人研究、快速验证想法	SKILL，免费层足够
团队协作、共享分析结果	SKILL + 共享 API Key，配合飞书/Slack 推送
生产级自动监控	直接调用 WebSocket API，自建告警系统
大规模回测	REST API + 历史 K 线数据，机构方案联系 [email protected]

---

八、结语

回到开篇那个场景。

凌晨两点，老张对 AI 助手说的那句话，背后发生的事其实不复杂：AI 理解了他的意图，调用了 TickDB SKILL，SKILL 向 TickDB API 发起请求，拿回了英伟达和特斯拉的 K 线数据，计算了波动率，格式化后返回给 AI，AI 再翻译成他能理解的回答。

整个过程不需要一个 API 调用的代码，不需要翻阅接口文档，不需要处理限频和重试。这不是魔法，是一层好的中间件，把复杂留给自己，把简单留给用户。

如果你想快速上手，在 ChatGPT、Claude 或 Gemini 中搜索安装 tickdb-market-data SKILL，配置好你的 API Key，然后直接用自然语言提问。

如果你想构建更稳定的数据管道，参考本文第四节的代码示例，基于 TickDB REST/WebSocket API 构建你自己的数据层。生产级代码包含了心跳保活、指数退避重连、限频处理等所有必要组件，可以直接作为工程起点。

如果你需要 10 年全量历史 K 线数据用于大规模回测，联系 [email protected] 获取机构级数据方案。

---

免责声明：本文不构成任何投资建议。市场有风险，投资需谨慎。TickDB SKILL 提供的是市场数据接入能力，不对任何基于数据的分析或决策结果负责。

回测局限性说明：本文涉及的历史波动率数据基于 TickDB 已接入的 K 线数据计算。回测结果不代表未来实际收益表现，未完全模拟交易成本、滑点和极端行情下的流动性风险。请勿将数据分析结果直接作为投资决策依据。