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

"你不需要记住 API 文档，你只需要像问同事一样问 AI。"

凌晨 2:47，你被手机震醒。英伟达盘后发布财报，股价在盘后交易中剧烈波动。你想知道现在的盘口结构——买盘还是卖盘更厚？成交量有没有异常放大？

传统方案：打开终端 → 翻出 API 文档 → 写 Python 脚本 → 调试鉴权 → 等脚本跑完。数十分钟过去，市场可能已经变了。

现在，你可以直接问 AI：

"英伟达盘后盘口，买卖各五档，总量多少？"

AI 直接回答。这就是 TickDB SKILL 想要解决的事。

---

一、为什么 AI Agent 需要行情数据插件

1.1 量化研究员的三个尴尬时刻

时刻一：临时起意，想快速看一眼数据

"帮我查一下苹果过去一个月的日均成交量，看有没有异常。"

你有两个选择：打开券商 App（只能看 K 线图，不方便量化），或者打开 Python（至少十分钟）。大多数时候，你选择放弃。

时刻二：半夜被市场惊醒，想快速诊断

手机推送告诉你某只股票异动，你想确认"是不是假突破"。打开电脑太慢，看 App 不知道订单簿细节。

时刻三：跟同事讨论策略，需要临时跑个数据验证

"我觉得最近特斯拉的期权隐含波动率偏低，你觉得呢？"

如果能快速验证，数据支撑的讨论比拍脑袋有效得多。

1.2 AI Agent 的能力与缺口

大语言模型擅长理解自然语言、做推理、生成代码。但它不知道当前市场的实时状态。

传统解决方案是让 AI 调用外部 API。但这里有个门槛：

• 你得知道用哪个 API
• 你得写调用代码
• 你得处理鉴权、超时、限频
• 你得把返回的 JSON 解析成人类可读的回答

这个门槛拦住了大多数非工程师背景的量化爱好者，也让工程师在"快速验证一个想法"的场景下浪费大量时间。

SKILL 协议的本质：把 API 调用的复杂性封装成一个 AI 可以理解的工具，让用户只需要"说话"。

---

二、TickDB SKILL 是什么

2.1 概念拆解

SKILL 是 ClawHub 平台定义的一种 AI Agent 扩展协议。它本质上是一个自然语言到 API 调用的翻译层。

类比一下：

• 你去餐厅点菜，不需要知道后厨怎么运作，只需要说"要一份七分熟牛排"
• SKILL 就是那个把"七分熟牛排"翻译成后厨指令的服务员

TickDB SKILL 让你在 AI 对话中直接查询市场数据，而不需要写任何代码。

2.2 系统架构

┌─────────────────────────────────────────────────────────────┐
│                      AI Agent（ChatGPT/Claude/Gemini）       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   用户: "英伟达今天盘前成交了多少股？"                        │
│                    ↓                                        │
│          自然语言理解 + SKILL 调用                           │
│                    ↓                                        │
│   ┌─────────────────────────────────────────┐               │
│   │         TickDB SKILL                    │               │
│   │  · 自然语言 → 参数解析                   │               │
│   │  · 错误处理 + 重试逻辑                   │               │
│   │  · 结果格式化 + 上下文管理               │               │
│   └─────────────────────────────────────────┘               │
│                    ↓                                        │
│   ┌─────────────────────────────────────────┐               │
│   │         TickDB API                       │               │
│   │  · REST API / WebSocket                  │               │
│   │  · 历史 K 线 / 实时盘口 / 订单簿          │               │
│   └─────────────────────────────────────────┘               │
│                    ↓                                        │
│              TickDB 服务器集群                              │
└─────────────────────────────────────────────────────────────┘

用户与 AI 对话，AI 调用 TickDB SKILL，SKILL 与 TickDB 后端交互，最终返回结构化的市场数据。

2.3 功能覆盖矩阵

数据类型	支持情况	查询示例
实时行情（快照）	✅	"苹果现在什么价？"
历史 K 线	✅	"英伟达过去一周的日 K"
盘前/盘后数据	✅	"特斯拉盘后成交额多少？"
订单簿深度	✅	"苹果买卖各五档总量"
逐笔成交	✅（港股/数字货币）	"BTC 最近 10 笔成交"
板块/行业行情	✅	"科技股今天涨幅前五"
多标的对比	✅	"苹果 vs 微软谁的成交量大？"

---

三、安装配置：5 分钟上手

3.1 环境要求

要求	说明
AI Agent 平台	ChatGPT（Plus/Pro）、Claude（Pro/Enterprise）、Gemini（Advanced）
ClawHub 账号	用于安装和管理 SKILL
TickDB API Key	用于数据访问鉴权

3.2 安装步骤

第一步：在 ClawHub 安装 TickDB SKILL

1. 访问 clawhub.com
2. 注册/登录 ClawHub 账号
3. 搜索 "TickDB Market Data"
4. 点击安装，授权你使用的 AI Agent 平台

第二步：在 AI Agent 中激活

以 ChatGPT 为例：

1. 打开 ChatGPT（需要 Plus 或更高版本）
2. 点击 GPT Store
3. 找到 "My GPTs" 下的 TickDB Market Data
4. 启用

第三步：配置 API Key

在首次使用时，AI 会提示你输入 TickDB API Key。你可以：

• 直接在对话中输入一次（会话级有效）
• 在 GPT Settings 中存储（持久有效）

完整配置流程示意：

首次使用提示：
───────────────────────────────────
TickDB Market Data 需要你的 API Key。

1. 访问 tickdb.ai → 控制台 → 生成 API Key
2. 在下方输入（或粘贴）：
   
   X-API-Key: tkdb_xxxxxxxxxxxxxxxxxxxx
   
3. 开始查询。
───────────────────────────────────

3.3 快速验证

安装完成后，运行一个简单的验证查询：

"你好，我用苹果的代码 AAPL.US 查一下今天开盘价。"

如果返回类似下面的结果，说明安装成功：

查询结果：
───────────────────────────────────
标的：AAPL.US
数据类型：今日 K 线
开盘价：$182.63
当前时间：美东 09:47（盘中）
───────────────────────────────────

---

四、实战示例：你能问什么

4.1 实时行情查询

基础问法：

> 苹果现在什么价？成交量多少？

AI 返回：

AAPL.US 实时行情
───────────────────────────────────
当前价：$185.42
今日涨跌幅：+1.23%（+$2.25）
今日成交量：47.3M 股（均值 38.2M，超量 23.8%）
盘口快照：
  买一：$185.41 | 2,400 股
  卖一：$185.43 | 3,100 股
  买卖价差：$0.02（相对价差 0.01%）
───────────────────────────────────

4.2 历史 K 线回溯

问法：

> 英伟达过去一个月日 K，帮我算一下平均振幅

AI 返回：

NVDA.US 历史数据分析（近 30 个交易日）
───────────────────────────────────
平均收盘价：$118.45
平均日内振幅：4.32%（按收盘价计）
最大单日振幅：+11.2%（发生在 [具体日期]）
最近收盘价：$121.35
涨跌幅：+1.78%（近 30 日）
───────────────────────────────────

[注：AI 主动计算了波动率指标]

4.3 盘前盘后数据

问法：

> 特斯拉盘后成交了多少？价格怎么走的？

AI 返回：

TSLA.US 盘后行情（数据截至 21:30 ET）
───────────────────────────────────
盘后收盘价：$242.18
盘后涨跌幅：+2.34%（盘后时段）
盘后成交量：8.2M 股
价格走势：发布财报后快速拉升至 $245.50，随后回落至
          $242.18，呈现典型的"买预期卖事实"形态
盘口分析：
  买一：$242.10 | 5,200 股
  卖一：$242.25 | 8,800 股
  卖盘更厚，价格可能继续承压
───────────────────────────────────

4.4 订单簿深度分析

问法：

> 苹果盘口，买卖各五档，帮我算一下买卖压力比

AI 返回：

AAPL.US 订单簿深度（买卖各 5 档）
───────────────────────────────────
买盘（前 5 档累计）：89,300 股
卖盘（前 5 档累计）：112,400 股
买卖压力比：0.79（买盘/卖盘 < 1，卖压更大）
深度加权均价（VWAP）：
  买方 VWAP：$185.12
  卖方 VWAP：$185.48
  区间宽度：$0.36
───────────────────────────────────

[解读]：卖盘比买盘厚 25.8%，短期内价格可能承压。
       但若买盘在关键价位（$185.00）集中护盘，
       可能形成支撑。

4.5 条件告警配置

问法：

> 如果苹果跌幅超过 3%，或者成交量突然放大到均量 2 倍，通知我

AI 返回确认：

告警规则已配置
───────────────────────────────────
标的：AAPL.US
触发条件：
  ① 跌幅 > 3%（日内）
  ② 成交量 > 均量 2 倍（当日实时）
通知方式：对话内推送 + （可选）邮件/飞书
状态：监控中
───────────────────────────────────

当前条件未触发。继续监控中...

告警触发时的推送：

⚠️ 告警触发
───────────────────────────────────
标的：AAPL.US
触发条件：成交量放大
详情：
  当前成交量：78.2M 股（均量 38.2M 的 2.05 倍）
  当前价：$179.85（-3.10%）
  时间：美东 14:32
───────────────────────────────────

---

五、生产级代码：SKILL 背后的调用逻辑

虽然 SKILL 让你不需要写代码，但理解底层实现有助于你更好地使用它。以下是 TickDB REST API 的生产级调用示例，可以直接集成到你自己的 Agent 中。

5.1 历史 K 线查询

import os
import requests
from datetime import datetime, timedelta
import json


class TickDBMarketData:
    """TickDB 市场数据客户端 - 生产级实现"""
    
    BASE_URL = "https://api.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")
        
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        })
    
    def get_historical_kline(
        self,
        symbol: str,
        interval: str = "1d",
        limit: int = 30,
        end_time: int = None
    ) -> dict:
        """
        获取历史 K 线数据
        
        Args:
            symbol: 交易品种，如 "AAPL.US"
            interval: K 线周期，支持 1m/5m/15m/1h/4h/1d
            limit: 返回数量，默认 30
            end_time: 结束时间戳（毫秒），默认当前时间
        
        Returns:
            K 线数据字典
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit
        }
        if end_time:
            params["end_time"] = end_time
        
        response = self.session.get(
            f"{self.BASE_URL}/kline",
            params=params,
            timeout=(3.05, 10)  # 连接超时 3.05s，读取超时 10s
        )
        
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:
            # 限频处理
            retry_after = int(response.headers.get("Retry-After", 5))
            raise RateLimitError(f"触发限频，请在 {retry_after} 秒后重试")
        else:
            raise APIError(f"请求失败: {response.status_code}")
    
    def calculate_volatility(self, symbol: str, days: int = 30) -> dict:
        """
        计算历史波动率指标
        
        Args:
            symbol: 交易品种
            days: 计算周期
        
        Returns:
            波动率分析结果
        """
        data = self.get_historical_kline(symbol, interval="1d", limit=days)
        
        if data.get("code") != 0 or not data.get("data"):
            return {"error": "数据获取失败"}
        
        klines = data["data"]
        
        closes = [float(k["close"]) for k in klines]
        volumes = [float(k["volume"]) for k in klines]
        
        # 计算收益率序列
        returns = []
        for i in range(1, len(closes)):
            ret = (closes[i] - closes[i-1]) / closes[i-1]
            returns.append(ret)
        
        # 计算波动率指标
        avg_return = sum(returns) / len(returns) if returns else 0
        import statistics
        volatility = statistics.stdev(returns) if len(returns) > 1 else 0
        avg_volume = sum(volumes) / len(volumes)
        latest_volume = volumes[-1] if volumes else 0
        volume_ratio = latest_volume / avg_volume if avg_volume > 0 else 0
        
        latest_price = closes[-1] if closes else 0
        first_price = closes[0] if closes else 0
        total_return = (latest_price - first_price) / first_price if first_price else 0
        
        return {
            "symbol": symbol,
            "period_days": days,
            "avg_price": statistics.mean(closes),
            "latest_price": latest_price,
            "total_return": f"{total_return:.2%}",
            "daily_volatility": f"{volatility:.4f}",
            "annualized_volatility": f"{volatility * (252 ** 0.5):.4f}",
            "avg_volume": avg_volume,
            "volume_ratio": f"{volume_ratio:.2f}x",
            "latest_close": closes[-1] if closes else None,
            "klines": klines
        }


# 使用示例
if __name__ == "__main__":
    client = TickDBMarketData()
    
    # 查询苹果的波动率
    result = client.calculate_volatility("AAPL.US", days=30)
    
    print("=" * 50)
    print(f"AAPL.US 波动率分析（近 30 交易日）")
    print("=" * 50)
    print(f"最新收盘价: ${result['latest_price']:.2f}")
    print(f"近 30 日收益率: {result['total_return']}")
    print(f"日波动率: {result['daily_volatility']}")
    print(f"年化波动率: {result['annualized_volatility']}")
    print(f"成交量倍数: {result['volume_ratio']}")
    print("=" * 50)

5.2 实时盘口与买卖压力比

import os
import time
import requests
import json
import random
from typing import Optional


class TickDBWebSocketClient:
    """TickDB WebSocket 客户端 - 生产级实现"""
    
    WS_URL = "wss://api.tickdb.ai/v1/ws/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")
        
        self.ws = None
        self.reconnect_delay = 1
        self.max_reconnect_delay = 32
        self.base_delay = 1
    
    def connect(self, symbol: str, channels: list):
        """
        连接 WebSocket 并订阅频道
        
        Args:
            symbol: 交易品种，如 "AAPL.US"
            channels: 订阅频道，如 ["depth5", "ticker"]
        """
        url = f"{self.WS_URL}?api_key={self.api_key}"
        self.ws = self._create_connection(url)
        
        # 构建订阅消息
        subscribe_msg = {
            "cmd": "subscribe",
            "params": {
                "symbol": symbol,
                "channels": channels
            }
        }
        self.ws.send(json.dumps(subscribe_msg))
        print(f"已订阅 {symbol} 的 {channels} 频道")
    
    def _create_connection(self, url: str):
        """创建 WebSocket 连接，带心跳"""
        import websocket
        
        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.ws = ws
        return ws
    
    def _on_open(self, ws):
        print("WebSocket 连接已建立")
        # 启动心跳保活
        self._start_heartbeat()
    
    def _on_message(self, ws, message):
        """处理接收到的消息"""
        data = json.loads(message)
        
        if data.get("type") == "pong":
            return  # 心跳响应，忽略
        
        if data.get("type") == "snapshot":
            self._process_snapshot(data)
        elif data.get("type") == "update":
            self._process_update(data)
    
    def _process_snapshot(self, data: dict):
        """处理快照数据（订单簿）"""
        if "depth" in str(data):
            symbol = data.get("symbol", "unknown")
            depth_data = data.get("data", {})
            bids = depth_data.get("bids", [])
            asks = depth_data.get("asks", [])
            
            # 计算买卖压力比
            bid_total = sum(float(q) for _, q in bids)
            ask_total = sum(float(q) for _, q in asks)
            pressure_ratio = bid_total / ask_total if ask_total > 0 else 0
            
            print(f"\n{symbol} 订单簿快照")
            print(f"买盘总量: {bid_total:,.0f} 股")
            print(f"卖盘总量: {ask_total:,.0f} 股")
            print(f"买卖压力比: {pressure_ratio:.2f}")
            
            # 展示各档位详情
            print("\n买盘（前 5 档）:")
            for price, qty in bids[:5]:
                print(f"  ${price} | {int(float(qty)):,} 股")
            
            print("\n卖盘（前 5 档）:")
            for price, qty in asks[:5]:
                print(f"  ${price} | {int(float(qty)):,} 股")
    
    def _process_update(self, data: dict):
        """处理增量更新"""
        # 实时订单簿更新的处理逻辑
        pass
    
    def _start_heartbeat(self):
        """心跳保活"""
        import threading
        
        def send_ping():
            while self.ws and self.ws.sock and self.ws.sock.connected:
                try:
                    self.ws.send(json.dumps({"cmd": "ping"}))
                    time.sleep(30)  # 每 30 秒发送一次心跳
                except Exception as e:
                    print(f"心跳发送失败: {e}")
                    break
        
        thread = threading.Thread(target=send_ping, daemon=True)
        thread.start()
    
    def _on_error(self, ws, error):
        print(f"WebSocket 错误: {error}")
    
    def _on_close(self, ws, close_status_code, close_msg):
        print(f"WebSocket 连接关闭: {close_status_code}")
        self._schedule_reconnect()
    
    def _schedule_reconnect(self):
        """指数退避重连 + 抖动"""
        delay = min(self.reconnect_delay * (2 ** random.randint(0, 2)), 
                    self.max_reconnect_delay)
        jitter = random.uniform(0, delay * 0.1)  # 10% 抖动
        actual_delay = delay + jitter
        
        print(f"{actual_delay:.2f} 秒后尝试重连...")
        time.sleep(actual_delay)
        self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay)
    
    def close(self):
        """关闭连接"""
        if self.ws:
            self.ws.close()
            print("连接已关闭")


# 使用示例
if __name__ == "__main__":
    client = TickDBWebSocketClient()
    
    try:
        # 订阅苹果的 5 档订单簿
        client.connect("AAPL.US", ["depth5"])
        
        # 保持连接 60 秒
        time.sleep(60)
        
    except KeyboardInterrupt:
        print("\n用户中断")
    finally:
        client.close()

5.3 工程健壮性要点说明

上面两段代码包含了 TickDB 生产级接入的核心要素：

要素	实现	说明
心跳保活	_start_heartbeat()	WebSocket 每 30 秒发送 ping，防止连接被闲置断开
指数退避重连	_schedule_reconnect()	断开后递增等待时间，配合抖动避免惊群
超时设置	timeout=(3.05, 10)	HTTP 请求必须设置，防止无限等待
环境变量存储	os.environ.get()	API Key 不硬编码，支持密钥轮换
限频处理	RateLimitError	识别 429 响应，读取 Retry-After 头等待
连接超时	3.05s	略大于 3 的质数，避免 TCP 整数倍超时
工程预警注释	# ⚠️ 生产环境...	提示高频场景需用 aiohttp

---

六、产品能力对比

如果你的团队在评估是否接入 TickDB SKILL，下面的对比表可以帮助你快速了解 TickDB 相对于其他方案的差异。

能力维度	传统方式（自建爬虫）	通用行情 API	TickDB SKILL
数据覆盖	需要自己抓多个市场	通常只覆盖单一市场	覆盖股票、数字货币、外汇等 6 类资产
历史数据	数据碎片化，质量参差	通常不提供或仅 1-2 年	10 年级别、清洗对齐的美股历史 K 线
实时盘口	爬虫容易被封，且延迟高	通常不支持或仅快照	WebSocket 推送，深度 1-50 档（按市场）
订单簿数据	不支持	极少支持	支持 depth 频道
API 稳定性	依赖爬虫维护，SLA 无保障	有 SLA，但功能有限	企业级 SLA保障
AI 集成	需要自己写工具层	需要自己写工具层	原生 SKILL 协议，5 分钟接入
代码量	可能需要 500+ 行	可能需要 200+ 行	零代码（对话式）
维护成本	高（需要持续维护爬虫）	中（需要处理多数据源）	低（TickDB 官方维护）

核心差异：TickDB SKILL 的本质优势不是"数据更便宜"，而是"把数据查询的复杂度降到零"。当你的研究员问"苹果最近的成交量异动了吗"，他不需要写代码，不需要查文档，直接问 AI。

---

七、应用场景与选型建议

7.1 适合使用 TickDB SKILL 的场景

场景	传统方案痛点	SKILL 解决方案
快速市场诊断	需要写脚本、查文档、等数据	直接问 AI，秒级响应
盘中实时监控	自己写告警脚本，配置复杂	"如果 X 涨跌幅超过 Y，通知我"
策略讨论中的临时验证	打开 Python、等脚本跑完	直接问 AI，现场验证
给非技术合伙人展示数据	他看不懂终端，只能截图	自然语言解释，可视化呈现
高频事件后的快速复盘	手动翻历史，效率低	"财报前后各三天的订单簿变化"

7.2 不适合的场景

• 大规模回测：SKILL 适合探索性分析，不适合大批量数据导出。回测请直接用 TickDB REST API。
• 程序化交易：需要毫秒级延迟和确定性的执行保障的场景，请用原生 WebSocket API。
• 合规审计：需要完整操作日志的场景，请在 API 层接入，而不是通过 SKILL。

7.3 定价与接入门槛

层级	价格	数据权限	适合人群
免费版	免费	有限额度 REST API	个人爱好者，低频查询
专业版	$29/月起	实时数据、更多标的	认真做量化的个人开发者
机构版	定制报价	全量数据、企业 SLA	量化团队、合规机构

注册地址：tickdb.ai（无需信用卡即可开始）

---

八、结语

回到开篇的场景。

凌晨 2:47，你被英伟达的财报异动惊醒。你想知道现在的订单簿结构。

旧世界：你挣扎着打开电脑，翻出 API 文档，开始写脚本。15 分钟后，你终于拿到了数据——但市场已经变了。

新世界：你打开手机，AI 已经收到了英伟达的财报数据。你问它："买卖各五档，买卖压力比多少？"它回答："卖盘比买盘厚 41%，当前价格可能继续承压，建议关注 $240 一线的支撑。"

这不是魔法，这是 SKILL 协议把 API 调用封装成对话的必然结果。

TickDB SKILL 让市场数据的查询成本从"写代码 + 调试 + 运行"降到"说话"。这个降本幅度对于需要频繁查数据的量化研究员和爱好者来说，是质的飞跃。

对于已经在写代码的工程师，TickDB SKILL 也是一个很好的"快速验证工具"——你想测试一个想法，先用 SKILL 问一句，再决定要不要写完整代码。

---

下一步行动

如果你想体验对话式查行情：

1. 访问 clawhub.com 安装 TickDB Market Data SKILL
2. 访问 tickdb.ai 注册并获取 API Key
3. 在 AI Agent 中配置 API Key，开始查询

如果你需要生产级数据接入：

• 个人量化开发者：REST API + WebSocket 客户端（见本文第五节）
• 机构量化团队：联系 [email protected] 获取企业方案

如果你想了解更多 TickDB 产品能力：

• 历史 K 线数据：《如何获取 TickDB 10 年级历史 K 线数据》
• 实时盘口监控：《实时订单簿：WebSocket + depth 频道实战》
• 多市场数据覆盖：《TickDB 覆盖的 6 类资产与数据规格》

---

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

本文涉及的代码示例基于 TickDB API v1 版本实现，生产环境请参考官方最新文档。