低代码量化：用 AI Agent + TickDB SKILL 实现自然语言行情查询
凌晨两点，你从床上坐起来。
不是失眠，是突然想到一个交易想法：最近港股科技股的财报季，买入那些营收增长超过 50% 但股价还没动的标的，胜率会不会更高？
这个想法在脑子里盘旋了五分钟，然后你打开电脑——
然后你发现，你不知道怎么查“港股科技股里，营收增长超过 50% 但近一个月涨幅小于 5% 的股票”。你去同花顺和东方财富的网页上找了半天，没找到这个筛选功能。你想写个脚本，但不会 Python。你想找量化工具，但发现都要收费而且学习曲线陡峭。
你的交易直觉很好，但你的工具链跟不上。
这不是你一个人的问题。量化交易行业存在一个巨大的“能力折叠”：策略idea的产生是毫秒级的，但把它变成可执行的查询，需要几天的代码工作。中间隔着的，是 API 文档、数据格式转换、错误处理这些工程细节。
AI Agent 的出现，正在打破这个折叠。
本文拆解一套低代码量化方案：用自然语言描述你的数据需求，让 AI Agent 自动完成查询、筛选、信号生成的全流程。你不需要写代码，但你能做的事情，比大多数写代码的人还多。
一、为什么自然语言是量化查询的下一个接口
1.1 传统 API 查询的断裂感
在讨论自然语言查询之前，我们需要理解为什么传统的 API 方式让大多数人止步。
假设你想查询腾讯控股（0700.HK）最近 30 天的日线数据，用于计算均线策略。按照传统方式，你需要：
- 找到 TickDB 的 API 文档
- 理解
symbol参数的格式（0700.HK还是Tencent） - 理解
interval参数（1d还是day） - 理解
limit参数（返回多少条） - 写一段 Python 代码，配置 API Key，处理错误
- 运行代码，解析 JSON 结果
这不是高深的技术，但它打断了你的思路。对于一个“想验证交易想法”的用户来说，他关心的是“我想知道 30 日均线是否上穿 60 日均线”，而不是“limit 参数应该填多少”。
人脑处理的是语义，API 处理的是语法。中间需要一个翻译层。
1.2 Function Calling：让 AI 调用工具的桥梁
Function Calling（函数调用）是 2023 年下半年大模型最重要的能力升级之一。它的核心思想是：不再让 AI 只输出文字，而是让它在合适的时机调用预定义的工具函数。
一个典型的 Function Calling 流程：
用户：帮我查一下英伟达最近的股价走势
AI理解意图 → 判断需要调用行情查询工具 → 生成函数调用请求
→ 函数返回数据 → AI 将数据格式化为用户能理解的回答
在这个流程里，AI 不需要知道 TickDB 的 API Key 怎么配，不需要知道 symbol 参数是什么格式——它只需要知道：当用户问股价问题时，调用 get_stock_price 这个函数，传入股票代码和时间范围。
具体的技术细节，被封装在函数的定义（Function Schema）里。
1.3 SKILL 协议：Function Calling 的领域特定扩展
SKILL 协议是 ClawHub 平台上的 AI 工具扩展标准。它在 Function Calling 的基础上，增加了几个关键能力：
| 能力 | 传统 Function Calling | SKILL 协议 |
|---|---|---|
| 参数校验 | 由 Agent 代码处理 | 在 SKILL 层完成，支持类型检查和约束 |
| 认证信息 | 需要在 Agent 代码中硬编码 | SKILL 定义中声明，运行时注入 |
| 调用日志 | 分散在 Agent 代码中 | 统一记录在 SKILL 层 |
| 能力发现 | 需要 Agent 开发者手动配置 | Agent 可动态查询已安装 SKILL 的能力列表 |
对于量化场景，这意味着：TickDB SKILL 只需要定义一次，所有支持 SKILL 协议的 AI Agent 都可以直接调用它。用户不需要关心数据从哪里来，只需要关心他想问什么。
二、TickDB SKILL 的能力边界
在开始写代码之前，我们需要明确 TickDB SKILL 能做什么、不能做什么。这决定了它的使用场景。
2.1 核心能力
TickDB SKILL 当前暴露以下四大类能力：
行情数据查询
- 实时价格和 K 线快照
- 历史 K 线数据（支持多种时间周期）
- 支持的资产类型：数字货币（BTC、ETH 等主流币种）、港股、美股（不含逐笔成交数据）
深度数据查询
- 订单簿深度（depth）：港股 10 档、数字货币 10 档
- 买卖盘挂单量计算
符号查询
- 查询某市场/某类型下支持的全部交易品种
- 符号映射（代码与名称互转）
时间序列计算
- 基于历史 K 线计算均线、波动率等常见指标
2.2 能力边界
理解限制同样重要：
| 限制 | 说明 |
|---|---|
| 不支持美股逐笔成交 | trades 接口不支持美股和 A 股，但历史 K 线数据可用 |
| 深度数据不支持外汇、贵金属、指数 | 仅支持港股和数字货币 |
| 不包含财务数据 | TickDB 不提供财报基本面数据 |
| 计算能力有限 | SKILL 层做简单指标计算，复杂策略建议导出后在本地处理 |
这些边界划定了一个清晰的定位：TickDB SKILL 是“数据查询层”，不是“策略执行层”。它的职责是把自然语言问题翻译成数据请求，然后把数据返回给用户。策略判断和交易执行，仍在用户的掌控之中。
三、从零构建：AI Agent + TickDB SKILL 实战
3.1 系统架构
在写代码之前，先看清楚整体架构：
用户（自然语言）
↓
AI Agent（理解意图，选择工具）
↓
SKILL 协议层（参数校验，认证管理）
↓
TickDB API（数据源）
整个链路中，用户的唯一工作是写好问题。AI Agent 负责理解问题、决定调用哪个 SKILL、传递什么参数。SKILL 层负责把自然语言参数转换为 API 所需的精确格式，并处理认证。TickDB API 负责返回干净的数据。
3.2 环境准备
# Python 环境
# 需要安装 OpenAI SDK 和支持 SKILL 协议的 Agent 框架
pip install openai clawhub-sdk
# 环境变量配置
export OPENAI_API_KEY="sk-..."
export TICKDB_API_KEY="your_tickdb_api_key"
3.3 第一步：定义 TickDB SKILL
SKILL 本质上是一个 JSON Schema，描述了工具的能力、参数和返回值。下面的定义涵盖 TickDB 四大核心能力：
# tickdb_skill.py
TICKDB_SKILL = {
"name": "tickdb_market_data",
"version": "1.0.0",
"description": "获取全球主流市场的实时行情和历史 K 线数据。支持数字货币、港股、美股。",
"capabilities": [
{
"name": "get_realtime_quote",
"description": "获取单个交易品种的实时行情快照（价格、涨跌幅、成交量等）",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "交易品种代码，格式为 CODE.MARKET，如 BTC.USDT、BTC.US、0700.HK"
}
},
"required": ["symbol"]
},
"returns": "实时行情对象（价格、成交量、波动等）"
},
{
"name": "get_historical_klines",
"description": "获取历史 K 线数据，支持日线、小时线、分钟线等多种周期",
"parameters": {
"type": "object",
"properties": {
"symbol": {"type": "string", "description": "交易品种代码"},
"interval": {
"type": "string",
"description": "K 线周期：1m/5m/15m/30m/1h/4h/1d/1w",
"enum": ["1m", "5m", "15m", "30m", "1h", "4h", "1d", "1w"]
},
"limit": {
"type": "integer",
"description": "返回条数，最大 1000",
"default": 100
}
},
"required": ["symbol", "interval"]
},
"returns": "K 线数组（时间戳、开盘、收盘、最高、最低、成交量）"
},
{
"name": "get_order_book_depth",
"description": "获取订单簿深度数据（买卖盘挂单量和价差）",
"parameters": {
"type": "object",
"properties": {
"symbol": {"type": "string", "description": "交易品种代码"}
},
"required": ["symbol"]
},
"returns": "订单簿深度（买一至买十、卖一至卖十的挂单量和价格）"
},
{
"name": "list_available_symbols",
"description": "查询支持的全部交易品种",
"parameters": {
"type": "object",
"properties": {
"market": {
"type": "string",
"description": "市场类型：crypto（数字货币）/ stock_us（美股）/ stock_hk（港股）",
"enum": ["crypto", "stock_us", "stock_hk"]
}
}
},
"returns": "该市场下的全部交易品种列表"
}
],
"auth": {
"type": "api_key",
"env_var": "TICKDB_API_KEY",
"description": "API Key 需要在 TickDB 控制台生成"
}
}
这段 JSON Schema 是 SKILL 的核心。它告诉 AI Agent：
- 这个 SKILL 能做什么（四大能力）
- 每个能力需要什么参数、参数是什么格式
- 认证信息存在哪个环境变量里
一个重要工程细节：参数定义中的 enum 和 description 不是给人看的，是给 AI 看的。AI 会根据这些信息，决定用户输入应该映射到哪个枚举值。所以 interval 的 enum 写的是 1m/5m/15m... 而不是 minute_1/hour_1...——因为用户在自然语言中会说的词，更接近前者。
3.4 第二步：实现 SKILL 执行层
SKILL 定义只是告诉 AI“能做什么”，真正的数据获取发生在执行层：
# skill_executor.py
import os
import time
import random
import requests
class TickDBSkillExecutor:
"""TickDB SKILL 的执行器，实现与 TickDB API 的交互"""
BASE_URL = "https://api.tickdb.ai/v1"
def __init__(self):
self.api_key = os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError("环境变量 TICKDB_API_KEY 未设置")
self.session = requests.Session()
self.session.headers.update({"X-API-Key": self.api_key})
def _request_with_retry(self, method, endpoint, **kwargs):
"""
带指数退避的重试机制
遇到限频错误（3001）时，自动读取 Retry-After 并等待
"""
max_retries = 3
base_delay = 1.0
for attempt in range(max_retries):
try:
# 强制设置超时，避免请求挂起
kwargs.setdefault("timeout", (3.05, 10))
response = self.session.request(method, endpoint, **kwargs)
data = response.json() if response.headers.get("content-type", "").startswith("application/json") else {}
# 处理 TickDB 错误码
if data.get("code") == 0:
return data.get("data")
code = data.get("code", 0)
if code == 3001:
# 请求频率超限，等待后重试
retry_after = int(response.headers.get("Retry-After", 5))
wait_time = retry_after + random.uniform(0, 0.5)
print(f"[TickDB] 限频触发，等待 {wait_time:.1f} 秒后重试...")
time.sleep(wait_time)
continue
elif code in (1001, 1002):
raise PermissionError(f"TickDB API Key 无效或缺失: {data.get('message')}")
elif code == 2002:
raise ValueError(f"交易品种不存在: {kwargs.get('params', {}).get('symbol')}")
else:
raise RuntimeError(f"TickDB 错误 {code}: {data.get('message')}")
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
print(f"[TickDB] 请求超时，{delay:.1f} 秒后重试...")
time.sleep(delay)
else:
raise
raise RuntimeError("达到最大重试次数")
def get_realtime_quote(self, symbol: str) -> dict:
"""获取实时行情快照"""
return self._request_with_retry(
"GET",
f"{self.BASE_URL}/market/kline/latest",
params={"symbol": symbol, "interval": "1d"}
)
def get_historical_klines(self, symbol: str, interval: str, limit: int = 100) -> list:
"""获取历史 K 线数据"""
return self._request_with_retry(
"GET",
f"{self.BASE_URL}/market/kline",
params={"symbol": symbol, "interval": interval, "limit": limit}
)
def get_order_book_depth(self, symbol: str) -> dict:
"""获取订单簿深度"""
return self._request_with_retry(
"GET",
f"{self.BASE_URL}/market/depth",
params={"symbol": symbol}
)
def list_available_symbols(self, market: str = None) -> list:
"""获取可用交易品种"""
params = {}
if market:
params["market"] = market
return self._request_with_retry(
"GET",
f"{self.BASE_URL}/symbols/available",
params=params
)
这段代码的工程要点：
- 指数退避 + 抖动：每次重试等待时间是
base * 2^attempt，加上随机抖动（0~50%），避免惊群效应。 - 限频自适应：遇到 3001 错误时，读取
Retry-After头（TickDB 返回的精确等待秒数），而不是硬编码等待时间。 - 超时设置：
(3.05, 10)表示连接超时 3.05 秒、读取超时 10 秒。3.05 不是笔误，是为了让请求在网关超时前主动断开，避免占用连接。 - Session 复用：使用
requests.Session()复用 TCP 连接，减少高频调用时的连接建立开销。
3.5 第三步：构建 AI Agent
现在把 SKILL 定义和执行器组装成完整的 Agent：
# agent.py
import json
from openai import OpenAI
from skill_executor import TickDBSkillExecutor
class QuantAgent:
"""量化查询 Agent，支持自然语言交互"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.executor = TickDBSkillExecutor()
# SKILL 定义（来自 tickdb_skill.py）
self.skills = [TICKDB_SKILL]
# 对话历史
self.messages = [
{
"role": "system",
"content": """你是一个量化数据助手。用户会用自然语言提问，你需要：
1. 理解用户的查询意图
2. 从 SKILL 列表中选择合适的工具
3. 如果需要多个工具，先完成第一个再继续
4. 把工具返回的数据转换为用户能理解的分析
注意：
- 不要编造数据，所有数据必须来自工具调用
- 如果用户的问题超出工具能力范围，明确告知
- 保持回答简洁，聚焦于回答用户的问题"""
}
]
def chat(self, user_input: str) -> str:
"""处理用户输入，返回分析结果"""
self.messages.append({"role": "user", "content": user_input})
while True:
# 调用大模型，让它决定是否需要调用工具
response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=[
{
"type": "function",
"function": {
"name": skill["name"] + "_" + str(i)
if self._count_skills_with_name(skill["name"]) > 1
else skill["name"],
"description": skill["description"],
"parameters": skill["capabilities"][0] if skill["capabilities"] else {}
}
}
for i, skill in enumerate(self.skills)
for cap in skill["capabilities"]
] if self.skills else None,
tool_choice="auto"
)
assistant_message = response.choices[0].message
self.messages.append(assistant_message)
# 检查是否需要调用工具
if assistant_message.tool_calls:
for call in assistant_message.tool_calls:
tool_name = call.function.name
arguments = json.loads(call.function.arguments)
# 映射到执行器的具体方法
result = self._execute_tool(tool_name, arguments)
# 把结果返回给 Agent
self.messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False)
})
else:
# 没有工具调用，直接返回回复
return assistant_message.content
def _execute_tool(self, tool_name: str, arguments: dict) -> dict:
"""将工具调用映射到执行器方法"""
method_map = {
"get_realtime_quote": self.executor.get_realtime_quote,
"get_historical_klines": self.executor.get_historical_klines,
"get_order_book_depth": self.executor.get_order_book_depth,
"list_available_symbols": self.executor.list_available_symbols,
}
method = method_map.get(tool_name)
if method:
return method(**arguments)
raise ValueError(f"未知工具: {tool_name}")
def _count_skills_with_name(self, name: str) -> int:
return sum(1 for s in self.skills if s.get("name") == name)
# 使用示例
if __name__ == "__main__":
agent = QuantAgent()
# 自然语言查询
queries = [
"BTC 现在多少钱？",
"帮我看一下以太坊最近 30 天的走势，要日线数据",
"港股里腾讯和阿里的当前买卖盘深度怎么样？"
]
for q in queries:
print(f"\n{'='*50}")
print(f"用户: {q}")
print(f"Agent: {agent.chat(q)}")
3.6 第四步：实际运行测试
将上述代码保存为三个文件：
project/
├── tickdb_skill.py # SKILL 定义
├── skill_executor.py # 执行器
├── agent.py # Agent 主程序
└── .env # 环境变量（不在 Git 中）
运行结果示例：
$ python agent.py
==================================================
用户: BTC 现在多少钱？最近 7 天走势怎么样？
Agent: 截至目前，比特币（BTC/USDT）最新价格约为 67,450 美元，24 小时涨幅 2.3%。
最近 7 天走势（每日收盘价）：
- 11月18日: 64,200 美元
- 11月19日: 65,100 美元 (+1.4%)
- 11月20日: 66,800 美元 (+2.6%)
- 11月21日: 67,450 美元 (+1.0%)
- 7日均价: 65,930 美元
- 当前价格高于均价 2.3%，短期偏强。
波动率方面，这 7 天的日均振幅为 3.2%，属于正常范围。
Agent 自动完成了：
- 调用
get_realtime_quote获取当前价格 - 调用
get_historical_klines获取 7 日日线数据 - 计算涨跌幅、均线、波动率
- 格式化输出
全程用户只说了一句话。
四、场景化示例：从自然语言到量化信号
4.1 场景一：均线金叉扫描
用户输入：帮我找一下哪些主流币种目前处于 5 日均线上穿 20 日均线的状态？
Agent 处理流程：
# Agent 内部自动完成以下步骤：
# 1. 列出所有数字货币
symbols = list_available_symbols("crypto")
# 2. 获取每个币种的 20 日 K 线（需要足够数据计算均线）
for symbol in ["BTC.USDT", "ETH.USDT", "BNB.USDT", ...]:
klines = get_historical_klines(symbol, interval="1d", limit=25)
# 计算 MA5 和 MA20
ma5 = avg(klines[-5:]["close"])
ma20 = avg(klines[-20:]["close"])
ma5_prev = avg(klines[-6:-1]["close"])
ma20_prev = avg(klines[-21:-1]["close"])
# 判断金叉
if ma5_prev <= ma20_prev and ma5 > ma20:
print(f"{symbol}: MA5={ma5:.2f}, MA20={ma20:.2f} ✅ 金叉")
输出示例：
目前满足 5/20 日均线金叉的主流币种：
- ETH/USDT: MA5=3,245, MA20=3,180（刚形成 1 天）
- SOL/USDT: MA5=158.3, MA20=156.8（形成 2 天）
- AVAX/USDT: MA5=35.6, MA20=35.1（形成 3 天）
注意：金叉本身不是买入信号，需要结合成交量、趋势强度综合判断。
4.2 场景二：财报前流动性监控
用户输入：英伟达下周三发布财报，我需要监控盘后盘口的流动性变化，帮我设计一个监控方案
Agent 输出：
好的，以下是一个基于 TickDB depth 频道的流动性监控方案：
# 监控脚本伪代码
import websocket, time, json
def on_message(ws, message):
data = json.loads(message)
depth = data.get("depth", {})
bids = depth.get("bids", []) # 前 10 档买盘
asks = depth.get("asks", []) # 前 10 档卖盘
# 计算买卖压力比
bid_volume = sum(float(b[1]) for b in bids)
ask_volume = sum(float(a[1]) for a in asks)
pressure_ratio = bid_volume / ask_volume if ask_volume > 0 else 0
# 计算加权价差
spread = float(asks[0][0]) - float(bids[0][0])
spread_pct = spread / float(bids[0][0]) * 100
print(f"[{time.strftime('%H:%M:%S')}] 压力比={pressure_ratio:.2f}, 价差={spread_pct:.3f}%")
# 告警逻辑
if spread_pct > 0.5: # 价差超过 0.5% 触发告警
send_alert(f"流动性预警: {pressure_ratio:.2f} / {spread_pct:.3f}%")
# ⚠️ 实际生产环境建议使用 asyncio/aiohttp 处理高频数据
ws = websocket.WebSocketApp(
"wss://api.tickdb.ai/v1/ws/depth?symbol=NVDA.US&api_key=YOUR_KEY",
on_message=on_message
)
ws.run_forever(ping_interval=30) # 心跳保活
关键监控指标解读：
- 买卖压力比 > 2：买盘明显强于卖盘，可能上涨
- 买卖压力比 < 0.5：卖盘强于买盘，可能下跌
- 价差扩大至 0.5% 以上：市场对方向产生分歧，波动加大
4.3 场景三：跨市场联动分析
用户输入：最近港股科技股和数字货币市场走势有没有相关性？
Agent 处理流程：
- 获取港股科技股指数代表（腾讯、阿里、美团、京东、小米）近 30 日数据
- 获取 BTC、ETH 近 30 日数据
- 计算相关系数
输出示例：
近 30 日相关性分析：
标的 BTC 相关系数 ETH 相关系数 腾讯 (0700.HK) 0.42 0.38 阿里 (9988.HK) 0.35 0.31 美团 (3690.HK) 0.28 0.25 结论：港股科技股与加密市场存在中等正相关，但相关性不足以构成跨市场套利机会。建议关注宏观流动性事件的同步冲击。
五、技术细节：为什么 SKILL 层值得单独构建
看完上面的代码，你可能会问：为什么不直接在 Agent 代码里硬编码 API 调用？
有三个原因。
第一，可复用性。 一个 SKILL 定义，可以被任何支持 SKILL 协议的平台加载。ClawHub 上的 AI 助手、Cursor 的 Copilot、任何自建 Agent 平台——只要它们理解 SKILL 协议，就能调用 TickDB。这比你每次在 Agent 代码里手动配置要高效得多。
第二，安全隔离。 SKILL 层处理 API Key 的注入和校验，用户在自然语言中不可能泄露凭据。相比之下，如果把 API 调用逻辑写在提示词或代码里，prompt injection 攻击可能让恶意输入窃取密钥。
第三，能力演进。 TickDB 未来可能新增接口（比如美股逐笔成交数据），SKILL 定义只需要更新一次，所有 Agent 立即获得新能力，而不需要逐个修改 Agent 代码。
六、部署建议：选对工具组合
不是所有场景都需要从零构建 Agent。以下是三种部署路径的选择建议：
| 场景 | 推荐方案 | 成本 | 复杂度 |
|---|---|---|---|
| 快速尝试验证想法 | 直接在 ClawHub 安装 tickdb-market-data SKILL，用官方 AI 助手 |
免费 | 零 |
| 个人量化研究 | 用 Python 构建轻量 Agent（如本文示例） | API 费用 | 低 |
| 团队协作/生产环境 | 自建 Agent 平台，私有化部署 SKILL | 服务器成本 | 中高 |
ClawHub 方案（最快上手）：
- 访问 ClawHub，搜索并安装
tickdb-market-dataSKILL - 在支持的 AI 助手中加载该 SKILL
- 直接用自然语言开始查询
自建方案（本文代码）：
适用于需要深度定制、或者需要在现有系统中集成的场景。代码已在第三章完整给出，可以直接 fork 改编。
结语
回到开头的场景。
凌晨两点那个突然想到的交易想法——港股科技股里，营收增长超过 50% 但股价还没动的标的——现在你可以直接问了：
“帮我找一下港股科技股中，近一期财报营收增长超过 50% 的有哪些？再看其中近一个月涨幅小于 5% 的”
如果 TickDB 本身不包含财报数据（确实不包含），Agent 会如实告诉你：“我没有财报数据，但我可以帮你监控那些你手动筛选出的标的的盘口变化。”
这不是能力的边界，而是边界的诚实。
低代码的目标不是让所有人都不写代码，而是让有想法的人不被工具门槛拦在门外。当你的交易直觉和数据之间只隔着一个自然语言问题，量化研究的天花板会比你想象的更高。
下一步行动
如果你想快速尝试验证：
- 访问 ClawHub，搜索安装
tickdb-market-dataSKILL - 在支持的 AI 助手中加载，开始自然语言查询
如果你想深入定制：
- 访问 tickdb.ai 注册，获取免费 API Key
- 基于本文的代码框架，构建你自己的量化 Agent
如果你在团队中协作：
联系 [email protected] 了解 SKILL 协议的企业级部署方案，支持私有化部署和 SSO 集成。
本文不构成任何投资建议。市场有风险，投资需谨慎。