凌晨 3:17，你被手机震醒。

某个你关注了 72 小时的标的突然放量——不是普通的那种，是“可能影响接下来一周走势”的那种。你本能地想摸出笔记本电脑，打开终端，调出订单簿，再切到 5 分钟 K 线。

但你忍住了，因为你知道：打开终端需要 30 秒，输入密码需要 5 秒，启动脚本需要 10 秒，等数据返回需要 5 秒。45 秒后，你可能已经错过了最佳观测窗口。

如果你可以直接问 AI 助手：

“帮我看看 BTC/USDT 过去 1 小时的订单簿变化，重点关注卖方深度变化。”

然后在 3 秒内拿到一张结构化的数据表格，你会选择哪个？

这不是“AI 取代专业工具”的叙事，而是工作流的重新分层：日常监控交给对话，高频决策交给终端，策略回测交给历史数据库。对于需要持续跟踪市场但又不想被终端绑架的开发者，这种分层正在成为新的工程常态。

本文拆解 TickDB SKILL 的接入方式、查询能力边界，以及一个完整的对话式监控场景。

一、SKILL 协议是什么

在直接进入安装流程之前，有必要解释一下 SKILL 的定位。

SKILL 是一种 AI Agent 扩展协议，它允许第三方服务以结构化的方式接入 AI 助手的推理链路。一个 SKILL 通常包含：

组件 作用
Manifest（清单文件） 描述 SKILL 的能力边界、调用方式、参数规范
Action（动作定义） 定义 AI 可以调用的具体操作，如“查询实时行情”“获取历史 K 线”
Schema（参数模式） 规定每个动作的输入输出格式，确保 AI 能正确构造请求和解析响应

类比一下：如果 AI Agent 是操作系统，SKILL 就是应用程序。当你在 AI 助手中安装 TickDB SKILL 后，AI 就获得了“查询 TickDB 数据”的能力——不是通过模糊的提示词猜测，而是通过结构化的 API 调用。

这意味着什么？

在没有 SKILL 的情况下，你尝试让 AI 查询股价，它可能：

  1. 调用一个过时的第三方 API
  2. 构造了一个不存在的端点
  3. 返回了一堆“基于公开数据训练”的模糊数字

在有 SKILL 的情况下，AI 查询股价：

  1. 读取 SKILL 的 Action 定义
  2. 构造符合 TickDB 规范的请求参数
  3. 调用 TickDB 官方 API
  4. 解析结构化响应，返回给你

这是一个有契约保证的数据管道，不是“尽力而为”的模糊推理。

二、安装 TickDB SKILL

2.1 环境准备

在开始之前，确保你满足以下条件：

要求 说明
AI 助手客户端 支持 SKILL 协议的消费级或企业级 AI 助手
TickDB 账号 需要在 tickdb.ai 注册并获取 API Key
网络环境 能访问 tickdb.ai API

目前主流支持 SKILL 协议的 AI 助手包括 ClawHub 及部分企业定制客户端。如果你使用的是 ChatGPT、Claude 或 Gemini 的原生版本，可能需要通过特定渠道（如 GPT Store、企业版）获取。

2.2 安装步骤

以支持 SKILL 协议的 AI 助手为例：

步骤 1：在 AI 助手中搜索 SKILL

在 AI 助手的插件/技能商店中，搜索 tickdb-market-data。

步骤 2：点击安装并绑定 API Key

安装过程中，AI 会提示你输入 TickDB API Key。如果你还没有：

  1. 访问 tickdb.ai 并注册
  2. 登录后进入控制台 → API Keys → 生成新 Key
  3. 复制 Key 并粘贴到 SKILL 安装界面

步骤 3：验证安装

安装完成后，测试一下：

用户：/tickdb status

如果返回类似以下内容，说明安装成功：

TickDB SKILL 已激活
- API Key: ****3f7a（已绑定）
- 可用端点: /kline, /depth, /trades
- 今日调用量: 0 / 1000（免费层）

三、能查什么：SKILL 能力边界

安装完成后，你的 AI 助手就具备了以下能力模块。

3.1 实时行情查询

用户：查一下苹果最新的股价和日内波动

[tickdb-market-data] 正在查询 AAPL.US 实时数据...

✅ 实时行情 (AAPL.US)
- 当前价格: 187.42 USD
- 今日开盘: 185.60 USD
- 日内最高: 188.15 USD
- 日内最低: 185.23 USD
- 日内涨跌幅: +1.01%
- 今日成交量: 23.5M 股
- 查询时间: 2026-04-15 14:30:05 EST

这个能力来自 TickDB 的 /kline/latest 接口，返回当前交易日已闭合的最新 K 线数据（注意：不是未闭合的实时 K 线）。

3.2 历史 K 线回溯

用户：帮我看看以太坊最近 7 天的日线走势，重点关注成交量变化

[tickdb-market-data] 正在查询 ETH/USDT 7日日线数据...

✅ ETH/USDT 日线 (近7日)
| 日期       | 开盘    | 最高    | 最低    | 收盘    | 成交量    | 涨跌  |
|-----------|--------|--------|--------|--------|----------|------|
| 04-09    | 3,245  | 3,312  | 3,198  | 3,287  | 12.3B USD | +1.3% |
| 04-10    | 3,287  | 3,341  | 3,265  | 3,298  | 11.8B USD | +0.3% |
| 04-11    | 3,298  | 3,456  | 3,278  | 3,412  | 18.7B USD | +3.5% |
| 04-12    | 3,412  | 3,489  | 3,356  | 3,371  | 15.2B USD | -1.2% |
| 04-13    | 3,371  | 3,398  | 3,301  | 3,334  | 14.1B USD | -1.1% |
| 04-14    | 3,334  | 3,367  | 3,298  | 3,345  | 13.5B USD | +0.3% |
| 04-15    | 3,345  | 3,378  | 3,318  | 3,352  | 11.9B USD | +0.2% |

📊 成交量观察:
- 4月11日成交量环比+58.5%，当日涨幅+3.5%，量价齐升
- 之后4天成交量持续萎缩，价格进入震荡
- 当前成交量处于近期低位（低于7日均值12.8%）

这个能力来自 /kline 接口，支持指定时间范围和周期（1m/5m/1h/1d）。

3.3 订单簿深度查询（数字货币）

用户：看一下 BTC/USDT 现在的买卖盘深度，重点关注前5档

[tickdb-market-data] 正在查询 BTC/USDT depth 快照...

✅ BTC/USDT 订单簿快照
- 查询时间: 2026-04-15 14:32:18 UTC
- 买卖价差: 1.5 USD (0.0017%)

【卖方深度 Top 5】
| 档位 | 价格 (USDT) | 挂单量 (BTC) | 累计量    |
|-----|------------|-------------|----------|
| 卖1  | 67,234.50  | 12.84       | 12.84    |
| 卖2  | 67,235.20  | 8.21        | 21.05    |
| 卖3  | 67,236.80  | 15.37       | 36.42    |
| 卖4  | 67,239.50  | 6.92        | 43.34    |
| 卖5  | 67,243.10  | 22.15       | 65.49    |

【买方深度 Top 5】
| 档位 | 价格 (USDT) | 挂单量 (BTC) | 累计量    |
|-----|------------|-------------|----------|
| 买1  | 67,233.00  | 18.63       | 18.63    |
| 买2  | 67,231.50  | 9.44        | 28.07    |
| 买3  | 67,229.80  | 25.31       | 53.38    |
| 买4  | 67,227.20  | 11.76       | 65.14    |
| 买5  | 67,224.50  | 14.28       | 79.42    |

📊 深度分析:
- 买卖压力比: 1.21 (买方略强)
- 前5档总卖量: 65.49 BTC vs 前5档总买量: 79.42 BTC
- 卖方流动性集中在 67,236-67,243 区间（阻力位）
- 买方在 67,229-67,224 有较强承接（支撑位）

这是 TickDB depth 频道的核心能力之一。值得注意的是，美股只支持 1 档深度，港股和数字货币支持 10 档甚至更多。

3.4 逐笔成交流（数字货币/港股）

用户：过去1分钟BTC/USDT的大单成交有哪些？

[tickdb-market-data] 正在查询 BTC/USDT trades (最近1分钟)...

✅ BTC/USDT 逐笔成交 (最近1分钟)
| 时间       | 方向 | 价格      | 成交量    | 成交额       |
|-----------|------|----------|----------|-------------|
| 14:32:15  | 买   | 67,234.50 | 2.85 BTC | 191,618 USD |
| 14:32:08  | 卖   | 67,233.80 | 1.23 BTC | 82,618 USD  |
| 14:32:03  | 买   | 67,234.20 | 5.41 BTC | 363,697 USD ⚡大单 |
| 14:31:58  | 卖   | 67,234.00 | 0.87 BTC | 58,474 USD  |
| 14:31:45  | 买   | 67,233.50 | 3.72 BTC | 250,148 USD ⚡大单 |
| ...       | ...  | ...       | ...      | ...         |

📊 大单统计 (成交额 > 100,000 USD):
- 大单数量: 4 笔
- 大单总额: 1,025,162 USD
- 买方主导: 2笔 / 卖方主导: 2笔
- 大单买卖比: 1:1 (均衡)

这个能力来自 /trades 接口。注意：美股和 A 股不支持 trades 接口，数字货币和港股可以。

3.5 组合查询：自然语言的真正威力

SKILL 的价值不只是“让 AI 调 API”，而是组合多个能力实现复杂查询：

用户：我关注的NVDA今晚要发财报了，帮我查一下：
1. 过去1年NVDA的季度财报后走势（K线）
2. 财报前一天的期权隐含波动率（如果有数据的话）
3. 最近3天的盘前成交量有没有异常

[tickdb-market-data] 正在执行组合查询...

📋 NVDA 财报前监控面板

【1】历史财报后走势 (近4次季报)
| 财报日期   | 当日开盘  | 当日收盘  | 当日振幅  | 收盘涨跌   |
|-----------|----------|----------|----------|----------|
| 2025-10-23 | 135.20   | 142.85   | +7.65%  | +5.66%   |
| 2025-07-24 | 118.40   | 121.30   | +4.85%  | +2.45%   |
| 2025-04-24 | 98.50    | 92.30    | -8.35%  | -6.29%   |
| 2025-01-22 | 85.20    | 88.75    | +5.20%  | +4.17%   |

💡 历史规律:
- 4次财报中3次上涨，1次下跌
- 振幅范围: 4.85% ~ 8.35%
- 平均收盘涨幅: +1.50%

【2】盘前成交量监控 (近3日)
| 日期       | 盘前成交量 | 环比变化   | 正常吗？          |
|-----------|-----------|-----------|-----------------|
| 04-14     | 28.5M     | +18.2% ⚠️ | 偏高（放量）     |
| 04-13     | 24.1M     | +3.4%    | 正常范围         |
| 04-12     | 23.3M     | —        | 基准             |

⚠️ 告警:
- 4月14日盘前成交量较基准日+18.2%，高于近30日均值
- 可能反映市场对今晚财报预期偏乐观

【3】数据说明
- 期权隐含波动率数据暂未接入 SKILL，可通过其他渠道补充
- 历史走势仅供参考，不构成未来预测

这种组合能力才是 SKILL 真正有价值的地方：AI 帮你做了数据聚合和初步分析，你只需要关注决策本身。

四、底层实现：SKILL 的技术架构

4.1 工作原理

用户输入 (自然语言)
       ↓
AI Agent 理解意图
       ↓
匹配到 tickdb-market-data SKILL
       ↓
读取 Action 定义 (如 "query_kline")
       ↓
构造 API 请求 (GET /v1/market/kline)
       ↓
发送请求到 TickDB API
       ↓
接收 JSON 响应
       ↓
AI 解析并格式化输出
       ↓
返回给用户

4.2 如何在代码中调用 TickDB API（绕过 SKILL）

虽然 SKILL 提供了对话式体验，但在某些场景下你可能需要直接调用 API——比如在你的交易系统中集成 TickDB 数据。以下是生产级的 Python 实现：

import os
import time
import random
import logging
from typing import Optional, Dict, Any, List
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class TickDBClient:
    """TickDB API 客户端 - 生产级实现"""
    
    BASE_URL = "https://api.tickdb.ai/v1"
    
    def __init__(self, api_key: Optional[str] = None):
        """
        初始化客户端
        
        Args:
            api_key: API Key，优先读取环境变量 TICKDB_API_KEY
        """
        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})
        
        # 重连状态
        self._retry_count = 0
        self._max_retries = 5
        self._base_delay = 1.0  # 基础重连延迟（秒）
        self._max_delay = 32.0  # 最大重连延迟（秒）
    
    def _calculate_delay(self, retry: int) -> float:
        """
        计算带抖动的指数退避延迟
        
        公式: delay = min(base * (2 ** retry), max_delay) + random(0, delay * 0.1)
        """
        delay = min(self._base_delay * (2 ** retry), self._max_delay)
        jitter = random.uniform(0, delay * 0.1)
        return delay + jitter
    
    def _handle_rate_limit(self, response: requests.Response) -> float:
        """
        处理限频响应
        
        返回需要等待的秒数
        """
        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", 5))
            logger.warning(f"触发限频，等待 {retry_after} 秒")
            return float(retry_after)
        return 0
    
    def _request_with_retry(
        self, 
        method: str, 
        endpoint: str, 
        params: Optional[Dict] = None,
        timeout: tuple = (3.05, 10)
    ) -> Dict[str, Any]:
        """
        带重连和限频处理的请求方法
        
        Args:
            method: HTTP 方法 (GET/POST)
            endpoint: API 端点
            params: 查询参数
            timeout: (连接超时, 读取超时)
        
        Returns:
            API 响应的 JSON 数据
        
        Raises:
            ValueError: API Key 无效或品种不存在
            RuntimeError: 未知错误或重连次数超限
        """
        url = f"{self.BASE_URL}{endpoint}"
        
        for retry in range(self._max_retries + 1):
            try:
                response = self.session.request(
                    method=method,
                    url=url,
                    params=params,
                    timeout=timeout
                )
                
                # 检查限频
                wait_time = self._handle_rate_limit(response)
                if wait_time > 0:
                    time.sleep(wait_time)
                    continue
                
                # 解析响应
                if response.headers.get("Content-Type", "").startswith("application/json"):
                    data = response.json()
                    code = data.get("code", 0)
                    
                    if code == 0:
                        return data
                    
                    # 错误处理
                    error_messages = {
                        1001: "API Key 无效",
                        1002: "API Key 缺失",
                        2002: f"交易品种不存在: {params.get('symbol')}",
                        3001: "请求频率超限"
                    }
                    raise ValueError(
                        f"API 错误 {code}: {error_messages.get(code, data.get('message', '未知错误'))}"
                    )
                
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.Timeout:
                logger.warning(f"请求超时 (尝试 {retry + 1}/{self._max_retries + 1})")
                if retry < self._max_retries:
                    time.sleep(self._calculate_delay(retry))
                continue
                
            except requests.exceptions.ConnectionError as e:
                logger.warning(f"连接错误 (尝试 {retry + 1}/{self._max_retries + 1}): {e}")
                if retry < self._max_retries:
                    time.sleep(self._calculate_delay(retry))
                continue
        
        raise RuntimeError(f"重试 {self._max_retries + 1} 次后仍失败")
    
    def get_kline(
        self,
        symbol: str,
        interval: str = "1h",
        limit: int = 100,
        start_time: Optional[int] = None,
        end_time: Optional[int] = None
    ) -> List[Dict[str, Any]]:
        """
        获取 K 线数据
        
        Args:
            symbol: 交易品种，如 "AAPL.US", "BTC.USDT"
            interval: K 线周期，如 "1m", "5m", "1h", "1d"
            limit: 返回数量 (1-1000)
            start_time: 开始时间戳（秒）
            end_time: 结束时间戳（秒）
        
        Returns:
            K 线数据列表
        """
        params = {"symbol": symbol, "interval": interval, "limit": limit}
        if start_time:
            params["start_time"] = start_time
        if end_time:
            params["end_time"] = end_time
        
        # ⚠️ 此接口用于获取历史 K 线，不适合实时监控的高频场景
        # 实时监控建议使用 WebSocket depth 频道（见下方）
        result = self._request_with_retry("GET", "/market/kline", params=params)
        return result.get("data", [])
    
    def get_kline_latest(self, symbol: str, interval: str = "1h") -> Optional[Dict[str, Any]]:
        """
        获取最新的 K 线数据（已闭合周期）
        
        适用于: 当前价格查询、收盘后的日线数据
        不适用于: 盘中实时价格（应使用 WebSocket）
        """
        params = {"symbol": symbol, "interval": interval}
        result = self._request_with_retry("GET", "/market/kline/latest", params=params)
        data = result.get("data")
        return data[0] if data else None
    
    def get_depth(self, symbol: str, limit: int = 10) -> Optional[Dict[str, Any]]:
        """
        获取订单簿深度
        
        ⚠️ 注意:
        - 美股: 仅支持 1 档
        - 港股/数字货币: 支持 10 档
        - 不支持: 外汇、贵金属、指数
        
        Returns:
            包含 bids 和 asks 的字典
        """
        params = {"symbol": symbol, "limit": limit}
        result = self._request_with_retry("GET", "/market/depth", params=params)
        return result.get("data")
    
    def calculate_pressure_ratio(self, depth_data: Dict[str, Any], levels: int = 5) -> float:
        """
        计算买卖压力比
        
        Args:
            depth_data: get_depth() 返回的数据
            levels: 计算深度（前 N 档）
        
        Returns:
            压力比 = 买方深度 / 卖方深度
            > 1: 买方占优
            < 1: 卖方占优
            ≈ 1: 均衡
        """
        bids = depth_data.get("bids", [])
        asks = depth_data.get("asks", [])
        
        buy_volume = sum(float(bid[1]) for bid in bids[:levels])
        sell_volume = sum(float(ask[1]) for ask in asks[:levels])
        
        if sell_volume == 0:
            return float('inf')
        
        return buy_volume / sell_volume


# ============================================================
# 使用示例
# ============================================================

if __name__ == "__main__":
    # 初始化客户端
    client = TickDBClient()
    
    # 示例 1: 查询 BTC/USDT 订单簿
    print("=" * 50)
    print("示例 1: 订单簿深度查询")
    print("=" * 50)
    
    depth = client.get_depth("BTC.USDT", limit=10)
    if depth:
        pressure_ratio = client.calculate_pressure_ratio(depth, levels=5)
        print(f"买卖压力比: {pressure_ratio:.2f}")
        print(f"买方前5档总量: {sum(float(b[1]) for b in depth['bids'][:5]):.2f} BTC")
        print(f"卖方前5档总量: {sum(float(a[1]) for a in depth['asks'][:5]):.2f} BTC")
    
    # 示例 2: 查询 NVDA 日线
    print("\n" + "=" * 50)
    print("示例 2: 历史 K 线查询")
    print("=" * 50)
    
    klines = client.get_kline("NVDA.US", interval="1d", limit=7)
    print(f"获取到 {len(klines)} 条 K 线数据")
    if klines:
        print(f"最新收盘价: ${klines[0]['close']}")
    
    # 示例 3: 查询最新价格
    print("\n" + "=" * 50)
    print("示例 3: 实时价格查询")
    print("=" * 50)
    
    latest = client.get_kline_latest("AAPL.US", interval="1h")
    if latest:
        print(f"AAPL 当前价格: ${latest['close']}")
        print(f"查询时间: {latest.get('close_time', 'N/A')}")

⚠️ 生产环境高频场景建议：上述代码使用 requests 库，适合低频调用（每分钟几次）。如果你需要实时监控（每秒多次），请使用 aiohttp + asyncio 实现异步架构，或直接使用 TickDB 的 WebSocket 频道（详见深度监控章节）。

4.3 WebSocket 实时监控（深度集成场景）

对于需要毫秒级响应的场景，SKILL 的轮询模式可能不够。以下是 WebSocket 的生产级实现：

import json
import time
import asyncio
import random
import logging
from typing import Callable, Optional

try:
    import websockets
    from websockets.exceptions import ConnectionClosed
except ImportError:
    raise ImportError("请先安装: pip install websockets")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class TickDBWebSocket:
    """TickDB WebSocket 客户端 - 生产级实现"""
    
    WS_URL = "wss://ws.tickdb.ai/v1/ws"
    
    def __init__(self, api_key: Optional[str] = None):
        """
        初始化 WebSocket 客户端
        
        Args:
            api_key: API Key，优先读取环境变量 TICKDB_API_KEY
        """
        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: Optional[websockets.WebSocketClientProtocol] = None
        self._running = False
        self._retry_count = 0
        self._max_retries = 10
        self._base_delay = 1.0
        self._max_delay = 60.0
    
    async def connect(self):
        """建立 WebSocket 连接"""
        # ⚠️ WebSocket 使用 URL 参数传递 API Key
        url = f"{self.WS_URL}?api_key={self.api_key}"
        
        try:
            self.ws = await websockets.connect(url, ping_interval=20)
            self._retry_count = 0
            logger.info("WebSocket 连接已建立")
        except Exception as e:
            logger.error(f"连接失败: {e}")
            raise
    
    async def subscribe(self, channel: str, symbol: str):
        """
        订阅频道
        
        Args:
            channel: 频道名，如 "depth", "kline", "trades"
            symbol: 交易品种
        """
        if not self.ws:
            raise RuntimeError("请先调用 connect()")
        
        subscribe_msg = {
            "cmd": "subscribe",
            "channel": channel,
            "symbol": symbol
        }
        await self.ws.send(json.dumps(subscribe_msg))
        logger.info(f"已订阅: {channel} @ {symbol}")
    
    async def unsubscribe(self, channel: str, symbol: str):
        """取消订阅"""
        if not self.ws:
            return
        
        unsubscribe_msg = {
            "cmd": "unsubscribe",
            "channel": channel,
            "symbol": symbol
        }
        await self.ws.send(json.dumps(unsubscribe_msg))
        logger.info(f"已取消订阅: {channel} @ {symbol}")
    
    async def heartbeat(self):
        """
        心跳保活
        
        WebSocket 连接需要在无数据时发送 ping，
        服务器响应 pong 以保持连接活跃
        """
        if self.ws:
            try:
                # TickDB 使用 JSON 格式的 ping/pong
                await self.ws.send(json.dumps({"cmd": "ping"}))
                logger.debug("心跳已发送")
            except Exception as e:
                logger.warning(f"心跳发送失败: {e}")
    
    async def listen(
        self, 
        callback: Callable[[dict], None],
        heartbeat_interval: int = 20
    ):
        """
        监听消息
        
        Args:
            callback: 消息处理回调函数
            heartbeat_interval: 心跳间隔（秒）
        """
        self._running = True
        last_heartbeat = time.time()
        
        while self._running:
            try:
                if not self.ws:
                    await self.connect()
                
                # 非阻塞接收，设置短超时以便检查心跳
                try:
                    message = await asyncio.wait_for(
                        self.ws.recv(), 
                        timeout=1.0
                    )
                    data = json.loads(message)
                    
                    # 处理 pong 响应
                    if data.get("cmd") == "pong":
                        logger.debug("收到 pong 响应")
                        continue
                    
                    # 处理限频响应
                    if data.get("code") == 3001:
                        retry_after = int(data.get("retry_after", 5))
                        logger.warning(f"触发限频，等待 {retry_after} 秒")
                        await asyncio.sleep(retry_after)
                        continue
                    
                    # 调用回调处理数据
                    callback(data)
                    
                except asyncio.TimeoutError:
                    # 超时，检查是否需要发送心跳
                    pass
                
                # 定期发送心跳
                if time.time() - last_heartbeat > heartbeat_interval:
                    await self.heartbeat()
                    last_heartbeat = time.time()
                
            except ConnectionClosed as e:
                logger.warning(f"连接断开: {e}")
                await self._reconnect(callback, heartbeat_interval)
            except Exception as e:
                logger.error(f"监听异常: {e}")
                await self._reconnect(callback, heartbeat_interval)
    
    async def _reconnect(
        self, 
        callback: Callable[[dict], None],
        heartbeat_interval: int = 20
    ):
        """指数退避重连"""
        if self._retry_count >= self._max_retries:
            logger.error("重连次数超限，停止重试")
            self._running = False
            return
        
        delay = min(self._base_delay * (2 ** self._retry_count), self._max_delay)
        jitter = random.uniform(0, delay * 0.1)
        wait_time = delay + jitter
        
        logger.info(f"{wait_time:.1f} 秒后尝试重连 (第 {self._retry_count + 1} 次)")
        await asyncio.sleep(wait_time)
        
        self._retry_count += 1
        self._running = True
    
    async def close(self):
        """关闭连接"""
        self._running = False
        if self.ws:
            await self.ws.close()
            logger.info("WebSocket 连接已关闭")


# ============================================================
# 使用示例
# ============================================================

async def on_depth_update(data: dict):
    """订单簿更新回调"""
    if data.get("type") == "depth_snapshot":
        symbol = data.get("symbol")
        bids = data.get("bids", [])
        asks = data.get("asks", [])
        
        # 计算压力比
        buy_vol = sum(float(b[1]) for b in bids[:5])
        sell_vol = sum(float(a[1]) for a in asks[:5])
        ratio = buy_vol / sell_vol if sell_vol > 0 else float('inf')
        
        print(f"[{symbol}] 压力比: {ratio:.2f} | 买方: {buy_vol:.2f} | 卖方: {sell_vol:.2f}")
        
        # 告警逻辑（示例）
        if ratio > 2.0:
            print(f"⚠️ 极端信号: 买方压力是卖方的 {ratio:.1f} 倍")
        elif ratio < 0.5:
            print(f"⚠️ 极端信号: 卖方压力是买方的 {1/ratio:.1f} 倍")


async def main():
    """主函数"""
    ws = TickDBWebSocket()
    
    try:
        await ws.connect()
        # ⚠️ 仅示例订阅一个品种，生产环境按需订阅
        await ws.subscribe("depth", "BTC.USDT")
        await ws.listen(callback=on_depth_update)
    except KeyboardInterrupt:
        print("\n收到中断信号，正在关闭...")
    finally:
        await ws.close()


if __name__ == "__main__":
    # ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio
    asyncio.run(main())

五、场景对比：SKILL vs 直接 API vs 专业终端

维度 TickDB SKILL 直接 API 调用 专业终端
使用门槛 对话式，无需编程 需要写代码 需要学习界面操作
响应速度 3-10 秒（网络 + AI 生成） <1 秒（纯数据） <1 秒
适合场景 日常监控、快速查询、策略思路验证 系统集成、高频监控 精细交易操作
数据格式 自然语言描述 + 表格 结构化 JSON 图形化
自动化能力 低（需要人工触发） 高（程序自动调用） 中
成本 消耗 SKILL 调用额度 消耗 API 调用额度 软件订阅费

选择建议：

  • SKILL：日常盯盘、快速验证想法、写策略文档时查数据
  • API：交易系统内嵌、自动化策略、批量数据获取
  • 终端：手动下单、需要图形化看盘、复杂订单操作

三者的关系不是替代，而是互补。

六、部署方案：选型指南

6.1 按使用场景

场景 推荐方案 理由
个人投资者，日常监控 TickDB SKILL 零学习成本，手机也能用
个人量化开发者 SKILL + API 双轨 SKILL 查想法，API 做回测
量化团队（3人以下） SKILL + API + 基础监控面板 分工：研究员用 SKILL，工程师用 API
机构级量化团队 完整 API + 定制化监控 + SKILL SKILL 用于晨会和快速决策

6.2 按数据需求

数据需求 SKILL 支持 API 支持
实时价格 ✅ ✅
历史 K 线 ✅ ✅
订单簿深度（数字货币） ✅ ✅
订单簿深度（美股） ⚠️ 仅1档 ⚠️ 仅1档
逐笔成交（数字货币/港股） ✅ ✅
逐笔成交（美股/A股） ❌ ❌

七、常见问题

Q1: SKILL 调用是否消耗 API 配额？

是的。SKILL 底层调用的仍然是 TickDB API，因此会消耗你的 API 调用额度。免费层用户每日 1000 次，专业版更高。

Q2: SKILL 支持多少个品种？

与 API 一致。TickDB 支持数字货币、美股、港股、外汇、贵金属、指数六大类资产。但注意：trades 接口不支持美股和 A 股，depth 美股仅 1 档。

Q3: SKILL 能替代专业交易终端吗？

不能。SKILL 的定位是辅助决策而非执行交易。它擅长数据聚合、快速查询、初筛分析，但不适合需要毫秒级响应的精细交易操作。

Q4: API Key 安全性如何保障？

  • SKILL 安装时只授权读取权限，无法执行交易
  • API 调用走 HTTPS/WSS 加密通道
  • 建议定期轮换 API Key

结语

回到开头那个场景。

凌晨 3:17，你被手机震醒。如果你已经安装了 TickDB SKILL，正确的打开方式不是摸出笔记本电脑，而是：

“帮我查一下 BTC/USDT 过去 1 小时的订单簿变化，重点关注卖方深度变化。”

3 秒后，你有了答案。5 秒后，你判断这只是正常波动，关掉手机继续睡。

工具的价值不是让你更忙，而是帮你在对的时刻做对的判断，然后把时间还给你。

下一步行动

如果你想快速体验 SKILL：

  1. 访问 tickdb.ai 注册（免费，无需信用卡）
  2. 在控制台生成 API Key
  3. 在支持的 AI 助手中搜索并安装 tickdb-market-data SKILL

如果你需要系统级集成：

  1. 参考本文的 Python 客户端代码
  2. 免费层适合个人开发者验证想法
  3. 专业版/企业版提供更高调用配额和 SLA 保障

如果你想在 AI 助手中直接使用：
在 AI 助手的技能商店中搜索 tickdb-market-data 并安装。

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