从数据消费到智能协作：企业级 SKILL 扩展实战指南

“你们的风控系统能不能在流动性枯竭时自动告警？同时考虑不同市场的正常波动范围。”

这是一个量化风控总监在凌晨 3 点向 AI 助手发出的指令。如果 AI 能理解这句话并直接执行，这意味着什么？

传统方案需要：写定时任务轮询 API → 解析数据 → 实现告警逻辑 → 维护告警阈值配置 → 处理各种边界情况。整个流程需要数周开发时间，且每次修改告警规则都需要工程师介入。

通过 SKILL 协议，同一句话的响应变成：AI 理解意图 → 调用 monitor_liquidity_event Function → 返回结构化告警。修改规则只需用自然语言描述新需求。

这不是假设。这是 SKILL 协议正在解决的核心问题：让 AI 从“辅助查询”升级为“自动执行”。

---

一、标准 API 的边界：为什么企业需要 SKILL

在讨论 SKILL 之前，先明确一个常见误解：TickDB 已有完整的 REST API 和 WebSocket 接口，为什么还需要 SKILL？

答案在于抽象层次的差异。

TickDB API 提供的是底层数据获取能力：K 线查询、深度数据、实时成交推送。这些接口设计精密、性能优秀，但返回的是原始数据。

能力维度	TickDB API	自定义 SKILL
抽象层次	底层数据（K线、深度、成交）	高层业务（事件检测、信号生成、告警触发）
调用方式	代码调用（需要处理分页、限频、错误码）	自然语言描述需求，AI 自动执行
业务逻辑	需要在调用方代码中自行实现	在 Function 定义中封装，团队共享复用
组合能力	单数据源获取	多数据源组合、跨市场分析
AI 集成	不支持	原生支持 AI 意图理解与调用

举一个具体场景：交易风控系统需要在特定条件下触发告警。

用 API 实现：需要写一个定时任务轮询 TickDB API → 解析返回的订单簿数据 → 计算买卖压力比 → 判断是否触发告警阈值 → 发送飞书/企微通知。每次修改告警逻辑（如调整阈值、改变计算方式）都需要工程师介入代码。

用 SKILL 实现：定义一个 monitor_liquidity_event Function，输入股票代码和告警阈值，返回告警状态。AI 理解自然语言指令后自动调用。修改规则只需用自然语言描述。

这就是 SKILL 的核心价值：把业务逻辑封装成 AI 可调用的 Function，让开发者从“写代码调用 API”升级为“描述需求让 AI 执行”。

---

二、SKILL 协议架构解析

2.1 什么是 SKILL

SKILL 是一种 AI Agent 的技能定义协议。你可以把它理解为 AI 的“工具箱”：每个 SKILL 是一组相关 Function 的集合，定义了 AI 可以执行的操作。

类比人类的工作场景：一个量化研究员掌握的工具包里有“获取财报数据”、“计算因子暴露度”、“生成回测报告”等技能。SKILL 协议就是为 AI 定义这个技能包的规范。

TickDB-market-data SKILL
├── get_kline          # 获取 K 线数据
├── get_depth          # 获取订单簿深度
├── subscribe_quote    # 订阅实时行情
├── monitor_event      # 监控市场事件 ← 企业可扩展
└── analyze_signal     # 分析交易信号 ← 企业可扩展

2.2 Function 的核心要素

Function 是 SKILL 的基本单元。一个 Function 由以下要素构成：

名称与描述：清晰描述 Function 的功能，让 AI 理解何时应该调用。

输入输出规范：用 JSON Schema 定义输入参数和输出结构，确保 AI 能正确构造请求和解析响应。

执行代码：实际执行业务逻辑的代码片段。

Function: monitor_liquidity_event
输入: symbol (string), market (string), threshold (object)
输出: { event_type, pressure_ratio, alert_level, timestamp }
执行: 检查订单簿深度，计算买卖压力比，判断是否触发告警

2.3 SKILL 调用流程

当用户向 AI 发送“监控苹果公司流动性枯竭”时，SKILL 协议的完整执行流程如下：

用户请求
    ↓
AI 意图理解 → 识别需要调用 monitor_liquidity_event
    ↓
参数构造 → { symbol: "AAPL.US", market: "NASDAQ", threshold: {...} }
    ↓
Function 执行 → 调用 monitor_liquidity_event，执行业务逻辑
    ↓
返回结果 → { event_type: "LIQUIDITY_DRY", pressure_ratio: 5.8, ... }
    ↓
AI 汇总输出 → 向用户呈现结构化的告警信息

这个流程的关键在于：业务逻辑被封装在 Function 中，AI 只需要理解用户意图并调用相应 Function，无需理解底层的 API 调用细节、数据解析逻辑、错误处理机制。

---

三、生产级 Function 开发实战

3.1 Function 定义规范

基于 SKILL 协议规范，一个完整的 Function 定义包含以下结构：

# skill.yaml - SKILL 定义文件
name: tickdb-market-data
version: 1.0.0
functions:
  - name: monitor_liquidity_event
    description: |
      监控指定股票的流动性事件。当买卖压力比超出正常范围时，
      触发相应级别的告警。适用于财报发布、重大事件前后的市场异常监控。
    input_schema:
      type: object
      properties:
        symbol:
          type: string
          description: 股票代码，如 AAPL.US、TSLA.US
        market:
          type: string
          description: 市场代码，如 NASDAQ、NYSE、HKEX
          enum: [NASDAQ, NYSE, HKEX, SSE, SZSE]
        threshold:
          type: object
          properties:
            critical:
              type: number
              description: 严重告警阈值（买卖压力比），默认 5.0
            warning:
              type: number
              description: 警告告警阈值（买卖压力比），默认 3.0
          required: [critical, warning]
      required: [symbol, market]
    output_schema:
      type: object
      properties:
        symbol:
          type: string
        event_type:
          type: string
          enum: [NORMAL, WARNING, CRITICAL]
        pressure_ratio:
          type: number
        bid_volume:
          type: integer
        ask_volume:
          type: integer
        spread:
          type: number
        timestamp:
          type: string
          format: date-time
        recommendation:
          type: string
          description: 针对当前状态的建议

3.2 生产级 Function 实现

Function 的执行代码需要考虑生产环境的各种边界情况。以下是一个完整的生产级实现示例：

"""
TickDB SKILL Function: monitor_liquidity_event
监控指定股票的流动性事件，返回买卖压力比和告警级别
"""
import os
import time
import random
import requests
from typing import Dict, Any, Optional
from datetime import datetime

# 常量配置
BASE_URL = "https://api.tickdb.ai/v1"
DEFAULT_TIMEOUT = (3.05, 10)  # HTTP 超时设置
MAX_RETRIES = 3
BASE_RETRY_DELAY = 1.0

class TickDBClient:
    """TickDB API 客户端，含心跳、重连、限频处理"""
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("未设置 TICKDB_API_KEY 环境变量")
        self.headers = {"X-API-Key": self.api_key}
        self._rate_limit_until = 0
    
    def _wait_if_rate_limited(self):
        """限频处理：检查并等待限速窗口"""
        now = time.time()
        if now < self._rate_limit_until:
            wait_time = self._rate_limit_until - now
            print(f"⏳ 限频等待 {wait_time:.2f}s")
            time.sleep(wait_time)
    
    def _request_with_retry(self, method: str, endpoint: str, **kwargs) -> Dict:
        """带指数退避和抖动的重试机制"""
        kwargs.setdefault("headers", self.headers)
        kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
        
        for attempt in range(MAX_RETRIES):
            try:
                self._wait_if_rate_limited()
                
                response = requests.request(method, f"{BASE_URL}{endpoint}", **kwargs)
                
                # 限频处理 (code: 3001)
                if response.status_code == 429 or (
                    response.headers.get("Content-Type", "").startswith("application/json") and
                    (data := response.json()).get("code") == 3001
                ):
                    retry_after = int(response.headers.get("Retry-After", 5))
                    self._rate_limit_until = time.time() + retry_after
                    print(f"⚠️ 限频触发，等待 {retry_after}s")
                    time.sleep(retry_after)
                    continue
                
                # 其他错误处理
                if response.status_code == 401:
                    raise ValueError("API Key 无效，请检查 TICKDB_API_KEY 环境变量")
                if response.status_code == 404:
                    raise ValueError(f"接口 {endpoint} 不存在")
                if not response.ok:
                    raise RuntimeError(f"请求失败: {response.status_code} {response.text}")
                
                result = response.json()
                if result.get("code") == 1001:
                    raise ValueError("API Key 无效")
                if result.get("code") == 2002:
                    raise KeyError(f"交易品种不存在")
                
                return result
                
            except requests.exceptions.Timeout:
                print(f"⏰ 请求超时（尝试 {attempt + 1}/{MAX_RETRIES}）")
                if attempt == MAX_RETRIES - 1:
                    raise
                delay = BASE_RETRY_DELAY * (2 ** attempt) + random.uniform(0, 0.5)
                time.sleep(delay)
                
            except requests.exceptions.ConnectionError as e:
                print(f"🔌 连接错误（尝试 {attempt + 1}/{MAX_RETRIES}）: {e}")
                if attempt == MAX_RETRIES - 1:
                    raise
                delay = BASE_RETRY_DELAY * (2 ** attempt) + random.uniform(0, 0.5)
                time.sleep(delay)
        
        raise RuntimeError("达到最大重试次数")

    def get_depth(self, symbol: str, depth: int = 10) -> Dict[str, Any]:
        """
        获取订单簿深度数据
        ⚠️ 美股 depth 仅支持 1 档，港股/数字货币支持多档
        """
        return self._request_with_retry(
            "GET", 
            f"/market/depth", 
            params={"symbol": symbol, "depth": depth}
        )

def calculate_pressure_ratio(bids: list, asks: list) -> float:
    """计算买卖压力比"""
    if not bids or not asks:
        return 1.0  # 无数据时返回中性值
    
    bid_volume = sum(level.get("volume", 0) for level in bids)
    ask_volume = sum(level.get("volume", 0) for level in asks)
    
    if ask_volume == 0:
        return float('inf') if bid_volume > 0 else 1.0
    
    return bid_volume / ask_volume

def determine_event_type(pressure_ratio: float, threshold: Dict) -> str:
    """根据买卖压力比判断事件类型"""
    critical = threshold.get("critical", 5.0)
    warning = threshold.get("warning", 3.0)
    
    # 极端值处理
    if pressure_ratio > 10 or pressure_ratio < 0.1:
        return "CRITICAL"
    
    if pressure_ratio >= critical or pressure_ratio <= (1.0 / critical):
        return "CRITICAL"
    if pressure_ratio >= warning or pressure_ratio <= (1.0 / warning):
        return "WARNING"
    return "NORMAL"

def generate_recommendation(event_type: str, pressure_ratio: float) -> str:
    """根据事件类型生成建议"""
    if event_type == "CRITICAL":
        if pressure_ratio > 1:
            return "流动性极度偏向买方，建议关注卖方Depth快速恢复的信号，可能存在潜在卖压"
        else:
            return "流动性极度偏向卖方，建议关注买方Depth快速积累的信号，可能存在潜在买压"
    elif event_type == "WARNING":
        if pressure_ratio > 1:
            return "买方压力增加，关注价差扩大信号"
        else:
            return "卖方压力增加，关注价差收窄信号"
    else:
        return "流动性处于正常范围"

def monitor_liquidity_event(
    symbol: str,
    market: str,
    threshold: Optional[Dict] = None
) -> Dict[str, Any]:
    """
    SKILL Function: monitor_liquidity_event
    
    监控指定股票的流动性事件，基于订单簿深度计算买卖压力比，
    判断是否触发告警阈值。
    
    参数:
        symbol: 股票代码，如 AAPL.US
        market: 市场代码，如 NASDAQ
        threshold: 告警阈值配置
    
    返回:
        包含事件类型、压力比、建议的结构化数据
    """
    # 参数校验
    if threshold is None:
        threshold = {"critical": 5.0, "warning": 3.0}
    
    if not symbol or not isinstance(symbol, str):
        raise ValueError("symbol 参数无效")
    
    # 初始化客户端
    try:
        client = TickDBClient()
    except ValueError as e:
        raise ValueError(f"初始化 TickDB 客户端失败: {e}")
    
    # ⚠️ 美股 depth 仅支持 1 档
    max_depth = 1 if market in ("NASDAQ", "NYSE") else 10
    
    # 获取深度数据
    try:
        depth_data = client.get_depth(symbol, depth=max_depth)
    except KeyError as e:
        raise KeyError(f"交易品种 {symbol} 不存在，请检查代码是否正确")
    except ValueError as e:
        raise ValueError(str(e))
    
    # 解析订单簿数据
    data = depth_data.get("data", {})
    bids = data.get("bids", [])
    asks = data.get("asks", [])
    
    if not bids or not asks:
        raise ValueError(f"获取 {symbol} 深度数据失败，订单簿为空")
    
    # 计算指标
    pressure_ratio = calculate_pressure_ratio(bids, asks)
    event_type = determine_event_type(pressure_ratio, threshold)
    recommendation = generate_recommendation(event_type, pressure_ratio)
    
    # 计算价差
    best_bid = float(bids[0].get("price", 0))
    best_ask = float(asks[0].get("price", 0))
    spread = best_ask - best_bid
    spread_pct = (spread / best_bid * 100) if best_bid > 0 else 0
    
    return {
        "symbol": symbol,
        "market": market,
        "event_type": event_type,
        "pressure_ratio": round(pressure_ratio, 4),
        "bid_volume": sum(level.get("volume", 0) for level in bids),
        "ask_volume": sum(level.get("volume", 0) for level in asks),
        "spread": round(spread, 4),
        "spread_pct": round(spread_pct, 4),
        "timestamp": datetime.now().isoformat(),
        "recommendation": recommendation
    }

# 测试入口
if __name__ == "__main__":
    result = monitor_liquidity_event(
        symbol="AAPL.US",
        market="NASDAQ",
        threshold={"critical": 5.0, "warning": 3.0}
    )
    print(f"监控结果: {result}")

代码说明：

1. 心跳与重连：通过 _request_with_retry 实现指数退避重连，指数因子为 2，最大延迟封顶。
2. 抖动：在退避延迟中加入随机抖动 [0, delay * 0.1]，避免惊群效应。
3. 限频处理：识别 code:3001 错误码和 429 状态码，读取 Retry-After 头等待。
4. 超时设置：HTTP 请求设置 (3.05, 10) 超时元组（连接超时 3.05s，读取超时 10s）。
5. 环境变量：API Key 从 TICKDB_API_KEY 环境变量读取，不硬编码在代码中。
6. 工程预警：注释中标注了美股 depth 仅支持 1 档的限制。

3.3 多 Function 组合场景

企业级应用中，单个 Function 往往不够用。考虑一个更复杂的场景：财报发布前后的流动性监控与事件记录。

# 组合多个 Function 实现复杂业务逻辑
class EarningsMonitorSkill:
    """
    财报事件监控 SKILL
    组合 monitor_liquidity_event 和 analyze_volatility 实现联动分析
    """
    
    def __init__(self, api_key: str = None):
        self.client = TickDBClient(api_key)
        self.threshold = {"critical": 5.0, "warning": 3.0}
    
    def monitor_pre_earnings(self, symbol: str, minutes: int = 30) -> Dict:
        """
        财报发布前的流动性预热监控
        在事件窗口期持续监控，记录异常时点
        """
        market = self._get_market(symbol)
        
        # 初始化监控记录
        records = []
        start_time = time.time()
        
        print(f"🔍 开始监控 {symbol}，预计持续 {minutes} 分钟")
        
        while time.time() - start_time < minutes * 60:
            try:
                result = monitor_liquidity_event(symbol, market, self.threshold)
                records.append(result)
                
                # 异常时打印告警
                if result["event_type"] != "NORMAL":
                    print(f"⚠️ [{result['timestamp']}] {symbol} 触发 {result['event_type']}，压力比 {result['pressure_ratio']}")
                
                time.sleep(5)  # 每 5 秒检查一次
                
            except Exception as e:
                print(f"⚠️ 监控异常: {e}")
                time.sleep(10)  # 异常后等待更长时间
        
        return self._aggregate_results(records)
    
    def _get_market(self, symbol: str) -> str:
        """根据代码推断市场"""
        if symbol.endswith(".US"):
            return "NASDAQ"
        elif symbol.endswith(".HK"):
            return "HKEX"
        return "NASDAQ"
    
    def _aggregate_results(self, records: list) -> Dict:
        """聚合监控结果"""
        if not records:
            return {"status": "NO_DATA", "message": "未获取到监控数据"}
        
        event_counts = {"NORMAL": 0, "WARNING": 0, "CRITICAL": 0}
        for r in records:
            event_counts[r["event_type"]] += 1
        
        avg_pressure = sum(r["pressure_ratio"] for r in records) / len(records)
        
        return {
            "total_records": len(records),
            "event_distribution": event_counts,
            "avg_pressure_ratio": round(avg_pressure, 4),
            "max_pressure_ratio": round(max(r["pressure_ratio"] for r in records), 4),
            "min_pressure_ratio": round(min(r["pressure_ratio"] for r in records), 4),
            "status": "COMPLETED"
        }

这个组合示例展示了 SKILL 的真正威力：通过组合多个 Function，企业可以构建复杂的业务逻辑，AI 只需要理解高层意图并编排调用。

---

四、私有部署方案

4.1 企业级部署的三种模式

当企业决定引入 SKILL 协议时，部署模式是首要决策点。以下是三种主流方案的对比：

维度	云端托管	混合部署	完全私有
SKILL 管理	ClawHub 云端	混合：云端 SKILL + 私有 Function	完全私有 ClawHub
行情数据	TickDB 云端 API	TickDB 云端 + 企业内网数据源	完全内网数据源
数据安全	⚠️ 数据出内网	⚠️ 部分数据出内网	✅ 全数据内网
部署复杂度	低	中	高
适用场景	概念验证、非敏感场景	核心逻辑需保密，但接受云端行情	金融合规、数据敏感、高延迟敏感
典型客户	初创公司、量化个人用户	中型量化团队	机构风控、券商自营

4.2 完全私有部署架构

对于数据安全要求极高的机构，推荐完全私有部署。架构如下：

┌─────────────────────────────────────────────────────────────┐
│                    企业内网环境                               │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐  │
│  │  TickDB     │    │  ClawHub     │    │  企业私有   │  │
│  │  私有集群   │◄───│  Mirror      │◄───│  数据源      │  │
│  └──────────────┘    └──────────────┘    └──────────────┘  │
│         │                   │                                 │
│         └─────────┬─────────┘                                 │
│                   ▼                                          │
│  ┌────────────────────────────────────────────────────────┐ │
│  │              企业内部 SKILL 开发环境                    │ │
│  │  • SKILL SDK (私有 ClawHub 版本)                        │ │
│  │  • 本地调试工具                                          │ │
│  │  • CI/CD 流水线                                         │ │
│  └────────────────────────────────────────────────────────┘ │
│                          │                                   │
│                          ▼                                   │
│  ┌────────────────────────────────────────────────────────┐ │
│  │              AI 助手集成层                              │ │
│  │  • 配置企业 ClawHub 地址                                │ │
│  │  • 自动发现内网 SKILL                                   │ │
│  │  • 统一鉴权与审计                                       │ │
│  └────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

完全私有部署的关键组件：

1. TickDB 私有集群：部署在企业内网的 TickDB 实例，支持私有数据源接入。
2. ClawHub Mirror：企业内网的 ClawHub 镜像，同步公开 SKILL 的同时管理私有 SKILL。
3. 私有 SKILL 开发环境：提供 SDK 和调试工具，让企业开发者可以本地开发和测试 SKILL。
4. 统一鉴权层：企业 SSO 集成，所有 AI 调用都经过审计。

4.3 混合部署：务实的折中方案

对于大多数企业，混合部署是务实的起点——既能享受云端基础设施的便利，又能保护核心业务逻辑。

┌──────────────────────────────┐     ┌──────────────────────────────┐
│         云端                  │     │        企业内网              │
│                              │     │                              │
│  ┌────────────────────────┐  │     │  ┌────────────────────────┐  │
│  │   TickDB Cloud API     │  │     │  │   企业私有 Function     │  │
│  │   (行情数据)           │  │     │  │   (核心业务逻辑)        │  │
│  └───────────┬────────────┘  │     │  └───────────┬────────────┘  │
│              │               │     │              │               │
│              ▼               │     │              ▼               │
│  ┌────────────────────────┐  │     │  ┌────────────────────────┐  │
│  │   ClawHub (SKILL管理)  │  │◄───►│  │   ClawHub Mirror       │  │
│  └────────────────────────┘  │     │  │   (内网同步)           │  │
│                               │     │  └────────────────────────┘  │
└───────────────────────────────┘     └──────────────────────────────┘
                    │                                    │
                    └──────────────┬─────────────────────┘
                                   ▼
                    ┌──────────────────────────────┐
                    │     AI 助手 (双 ClawHub)     │
                    │  • 云端公开 SKILL             │
                    │  • 内网私有 SKILL             │
                    └──────────────────────────────┘

在混合部署模式下：

• 云端 TickDB API 负责获取公开市场行情
• 私有 ClawHub Mirror 托管企业的核心业务 SKILL（如专有的风控规则、因子计算逻辑）
• AI 助手 同时连接云端 ClawHub 和企业内网 ClawHub Mirror，根据场景选择调用

---

五、TickDB SKILL 能力对比

能力维度	标准 API 调用	TickDB 公开 SKILL	企业定制 SKILL
调用方式	代码手动调用	自然语言 + AI 自动执行	自然语言 + AI 执行私有逻辑
抽象层次	底层数据获取	业务封装（事件监控、信号生成）	完全自定义业务逻辑
业务逻辑	需要在调用方代码中实现	已封装，开箱即用	完全在企业控制下
团队协作	代码级共享	团队共享 SKILL 定义	企业内统一管理
数据源	TickDB	TickDB	TickDB + 企业私有数据源
AI 集成	不支持	原生支持	原生支持
部署模式	纯云端	纯云端	云端/混合/完全私有
适用场景	个人量化开发	通用行情监控与分析	机构级风控、合规、投研

选择建议：

• 个人开发者：标准 API 调用更灵活，适合快速验证想法
• 团队量化项目：TickDB 公开 SKILL 可以快速构建通用监控能力
• 机构级应用：企业定制 SKILL 保护核心逻辑，满足合规要求

---

结语

SKILL 协议的核心价值，在于重新定义 AI 与数据的关系。

在传统模式中，数据是静态资源——开发者写代码获取数据、处理数据、消费数据。在 SKILL 模式下，数据是动态能力——开发者定义 Function，AI 理解意图并执行。从“数据消费”到“智能协作”，这是本质的区别。

对于企业级场景，SKILL 协议带来的改变更为深远：

• 风控系统：从“定期报表”到“实时告警”，一句话配置告警规则
• 投研平台：从“数据导出再分析”到“AI 自动执行因子计算”，效率提升 10 倍
• 合规监控：从“人工巡检”到“AI 自动巡检”，7×24 小时无死角

这不是技术的炫技，而是实实在在的效率革命。

---

下一步行动

如果你是个人开发者，想快速体验 SKILL 驱动的行情查询，访问 ClawHub 安装 tickdb-market-data SKILL，用自然语言探索行情数据。

如果你是企业技术负责人，正在评估 SKILL 协议在风控或投研场景的落地路径，联系 [email protected] 获取私有部署架构方案和报价。

如果你是 SKILL 开发者，想基于 TickDB 扩展自己的业务 SKILL，参考本文档的 Function 定义规范开始开发，或联系 [email protected] 获取 SKILL 开发工具包和文档。

---

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