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

TickDB Research ·

适用场景 LangChain 金融 Agent 开发
技术栈 Python / LangChain / REST API
覆盖市场 A 股 / 港股 / 美股 / 加密货币 / 外汇

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

你的问题本文解法
LangChain Agent 需要获取实时行情，但不知道该用什么 API封装 TickDB 为 LangChain Tool，统一 A 股/港股/美股/加密/外汇数据
不同市场数据格式各异，接入成本高统一字段、统一鉴权，只需写一次请求层
API 限流总让 Agent 崩溃指数退避重试逻辑一次性封装，所有 Tool 通用
不知道 K 线字段是 close 还是 last_price实测确认：K 线用 close，Ticker 用 last_price

你正在用 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 封装逻辑不变。示例：

from langchain.agents import create_tool_calling_agent, AgentExecutor
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
result = agent_executor.invoke({"input": "..."})

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

坑原因后果正确处理
K线收盘价用了 last_priceK线端点的收盘价字段是 close，而 ticker 的才是 last_priceAgent 拿到的永远是 None，导致分析结果异常永远用 close 取 K 线收盘价，用 last_price 取实时价
忘记了时间单位已经是毫秒TickDB 所有时间戳返回的都是毫秒 UTC如果你又 * 1000，时间会变成 2070 年直接使用返回的时间戳，不做额外转换
没有处理 3001 限流免费额度下请求频率有限制Agent 连续调用时突然报错中断用指数退避封装所有请求（上面已给出）

如果想做得更“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 聪明的部分。

相关文章

不写代码，能不能用 AI 同时查 A 股、港股和比特币？我试了 TickDB Skill

适用人群 个人投资者 / 不想写代码的行情关注者 阅读收获 了解一条“接近零代码”的 AI 查行情路径，以及它的前提条件和边界 免责声明 本文仅为工具体验记录，不构成任何投资建议 --- 5 秒速览：这篇文章写给你 | 你的困境 | 本文解法 | |---------|---------| | 几个行情 App 来回切，想看全市场太麻烦 | 一个对话窗口，覆盖 A 股/港股/美股/加密 |

TickDB Research ·

当你的A股策略需要同时盯着美股和比特币，一个WebSocket就够了

适用场景 跨市场量化策略开发 技术栈 WebSocket + REST 覆盖市场 A股 / 美股 / 加密货币 --- 凌晨两点，监控脚本准时唤醒。三个信号几乎同时出现： - 上证50隐含波动率跳升 - 纳斯达克期货盘前成交量异常放大 - BTCUSDT永续合约资金费率转负 但三个信号来自三个不同的数据源：A股用本地数据服务，美股接Polygon.io的WebSocket，加密货币靠轮

TickDB Research ·

AI Agent 框架接金融行情数据前，先检查这 7 个工程风险

> 本文不是框架排名，而是一份金融行情数据接入前的工程风险清单。每个风险点都附了检查方法和修正示例。 --- 一、不同框架，同一个坑 假设你用三个不同的 Agent 框架跑同一个任务——“每 30 分钟查一次价格，超过阈值时汇总分析”。 其中一个 Agent 把 ticker 快照的 （24 小时成交量）当成了单根 K 线的成交量，量级差了几千倍。另一个在 API 限流后陷入重试死循环，两分钟烧掉

TickDB Research ·

LangGraph 多 Agent 行情接入：节点之间传递实时数据的 3 种姿势

目录 - 一、当你的金融 Agent 节点“各说各话” - 二、姿势 1：State 共享——最简单，但 reducer 行为容易忽略 - 三、姿势 2：Tool 封装——最灵活，但 API 调用量是隐藏成本 - 四、姿势 3：Subgraph 组合——最工程，但状态映射是调试重点 - 五、三种姿势权衡矩阵与数据一致性策略 - 六、完整示例：盘前-盘中-盘后多 Agent 协同 - 七、FAQ -

TickDB Research ·

10道题测出5家国产大模型金融数据接入的真实差距——DeepSeek、Kimi、豆包、通义千问、智谱GLM横评

> 本文不评“谁最强”，只展示“差距在哪”。 > > 用同一套TickDB行情工具集（13个标准化工具，端点：），对5个国产大模型进行了10道中文金融查询题的测试。覆盖三类最易翻车的场景：中文简称映射、工具选择错误、错误恢复不足。所有测试基于2026年5月公开API，可复现。 > > 本文仅讨论技术接入和工程实践，不构成任何投资建议或模型推荐。 一、一个翻车现场，三种不同反应 你问大模型“帮我看看

TickDB Research ·