用 LangChain 打造金融 Agent，却被行情数据接入逼疯？一个统一 API 搞定全市场

适用场景 LangChain 金融 Agent 开发

技术栈 Python / LangChain / REST API

覆盖市场 A 股 / 港股 / 美股 / 加密货币 / 外汇

5 秒速览：这篇文章能帮你解决什么？

你正在用 LangChain 构建一个金融分析 Agent。你设想它能像资深研究员一样工作：你问它“对比一下茅台和腾讯近期的走势”，它应该自己拉出两支股票的日线、算出涨跌幅、给出一段简要分析。

开始动手后你很快发现，LangChain 的推理能力没问题，但 Agent 的手和眼不够用——它需要能获取实时行情和历史数据的工具。你得自己写 Tool。

于是你打开文档，准备接入数据源。先试了 Yahoo Finance，A 股覆盖不全，港股数据延迟严重。换 AKShare 接 A 股，但美股和加密得另找接口。再接入 Polygon.io 拿美股，最后还要写一套逻辑把三个来源的数据格式对齐。更头疼的是，每个接口的限流策略都不一样，你不得不在每个 Tool 里单独写重试逻辑。等把这堆数据管道搭好，你已经不想再去优化 Agent 的推理逻辑了。

痛点不在 Agent 本身，而在于让它“看到”市场数据的成本太高。

把 TickDB 封装成 LangChain Tool

如果有一个 API，能用同一套鉴权、同一种字段名、同一个重试逻辑覆盖 A 股、港股、美股、加密、外汇等市场，那你的 Tool 代码会简单多少？

TickDB 做的就是这件事。它的 REST 接口提供统一 REST + WebSocket 接口、统一字段、统一鉴权，你只需要封装一次限流处理和字段解析，所有市场的数据就全通了。

下面直接把 TickDB 封装成两个 LangChain Tool：一个查实时行情，一个查历史 K 线。Agent 拿到它们，就能像查字典一样获取全市场数据。

安装依赖

在项目环境中执行：

pip install langchain langchain-openai requests

版本提示：本示例基于 langchain ≥0.1.0。若你已升级到 langchain 0.3+，推荐使用 create_tool_calling_agent，但 initialize_agent 在 0.3 中仍可用。文末给出了兼容写法。

基础请求层（含限流处理）

TickDB 返回标准错误码，其中 3001 表示请求频率超限，会在响应头中附带 Retry-After 字段。下面这个函数对所有 API 调用做统一的指数退避处理，写一次，所有 Tool 共用：

# 替换为你的 API Key：访问 docs.tickdb.ai 获取
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.tickdb.ai"

import time
import requests

def tickdb_request(endpoint: str, params: dict = None) -> dict:
    """发送 GET 请求，自动处理 3001 限流（指数退避）"""
    headers = {"X-API-Key": API_KEY}
    retry = 0
    while True:
        resp = requests.get(f"{BASE_URL}{endpoint}", headers=headers, params=params)
        if resp.status_code == 200:
            return resp.json()
        elif resp.status_code == 429:          # 对应 TickDB 错误码 3001
            retry_after = int(resp.headers.get("Retry-After", 1))
            wait = min(retry_after * (2 ** retry), 60)
            time.sleep(wait)
            retry += 1
        else:
            resp.raise_for_status()

封装两个 LangChain Tool

#### Tool 1：实时行情快照

• 端点：GET /v1/market/ticker
• 参数：symbols（复数，最多 50 个），例如 600519.SH,700.HK
• 返回字段：last_price（最新价）、timestamp（毫秒 UTC）、volume_24h 等

from langchain.tools import tool

@tool
def get_ticker(symbols: str) -> str:
    """
    获取一个或多个品种的实时行情快照。
    symbols: 逗号分隔的品种代码，如 "600519.SH,700.HK,AAPL.US,BTCUSDT"
    返回 JSON 字符串，包含每个品种的最新价、成交量等信息。
    """
    params = {"symbols": symbols}
    data = tickdb_request("/v1/market/ticker", params)
    results = []
    for item in data.get("data", []):
        results.append({
            "symbol": item.get("symbol"),
            "last_price": item.get("last_price"),
            "volume_24h": item.get("volume_24h"),
            "timestamp": item.get("timestamp")
        })
    return str(results)

#### Tool 2：历史 K 线

• 端点：GET /v1/market/kline
• 参数：symbol（单数）和 interval（1m, 5m, 15m, 30m, 1h, 4h, 1d 等）
• 返回字段：time（毫秒 UTC）、close（注意：K线用 close 不是 last_price）

@tool
def get_kline(symbol: str, interval: str = "1d", limit: int = 10) -> str:
    """
    获取指定品种的历史 K 线。
    symbol: 品种代码，如 "600519.SH"
    interval: K线周期，如 "1d" (日线), "1h" (小时线)
    limit: 返回条数，默认 10
    返回 JSON 字符串，包含时间、开高低收、成交量。
    """
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    data = tickdb_request("/v1/market/kline", params)
    klines = data.get("data", {}).get("klines", [])
    simplified = []
    for k in klines:
        simplified.append({
            "time": k.get("time"),
            "open": k.get("open"),
            "high": k.get("high"),
            "low": k.get("low"),
            "close": k.get("close"),
            "volume": k.get("volume")
        })
    return str(simplified)

将 Tool 装进 Agent

现在把这两个工具交给 LangChain，构建一个可以分析全市场数据的 Agent：

from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_ticker, get_kline]

agent = initialize_agent(
    tools, llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

# 让 Agent 对比茅台和腾讯近期走势
response = agent.invoke(
    "帮我查一下贵州茅台(600519.SH)和腾讯控股(700.HK)最近5天的日线走势，"
    "并简单对比它们的涨跌情况。"
)
print(response)

当你执行这段代码时，Agent 内部会依次调用 get_kline(symbol='600519.SH', interval='1d', limit=5) 和 get_kline(symbol='700.HK', interval='1d', limit=5)，将返回的真实 K 线数据交给 LLM 进行分析。无需你手动拼接任何行情文本。

兼容性提醒：如果你使用的 langchain 版本 ≥0.3，建议将 initialize_agent 替换为 create_tool_calling_agent + AgentExecutor 的组合，核心 Tool 封装逻辑不变。示例：

%%CB_5%%

关键点速查：LangChain 集成中容易踩的 3 个坑

如果想做得更“AI 原生”

如果你已经在用 Cursor 或 Windsurf 进行开发，TickDB 还可以通过 MCP 方式直接接入。MCP 端点 https://mcp.tickdb.ai 提供了 13 个行情工具，配置后你的 AI 编程助手自己就能理解并调用，你甚至不需要手动写上面那些 Tool 代码。对话用 Skill，编码用 MCP，自动化用 CLI——选择最适合你当前工作流的方式。

从两个 Tool 到完整的金融 Agent

今天你只封装了实时行情和 K 线两个 Tool，但这条路可以延伸得很远：

• 需要看盘口深度？加一个 get_depth Tool，调 GET /v1/market/depth
• 需要判断美股是否还在夜盘？加一个 get_trading_sessions Tool，调 GET /v1/market/trading-sessions?market=US
• 需要分析北向资金？加一个 get_capital_flow Tool，调 GET /v1/market/capital-flow?symbol=000001.SH

所有 Tool 共享同一个 tickdb_request 基函数，接入一个新市场的增量成本趋近于零。TickDB 在 GitHub 完全开源——GitHub 开源，文档可查，代码可跑，让你的 Agent 从一开始就建在一个可审计、可持续的基础设施上。

下一步

1. 访问 TickDB 官方文档 docs.tickdb.ai 获取 API Key 和完整 API 参考。
2. 将上面的两个 Tool 代码粘贴进你的 LangChain 项目，替换掉那些零散的数据源调用。
3. 跑一遍茅台 vs 腾讯的对比查询，观察 Agent 如何自动拉取全市场数据并给出分析。

当数据接入不再占用你 80% 的开发时间，你才有精力去打磨真正让 Agent 聪明的部分。

通过 TickDB API 获取实时行情数据

一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。