IntentKit cn_stock 工具集实战指南：为 AI Agent 接入中国 A 股实时与历史行情数据

IntentKit cn_stock 工具集实战指南：为 AI Agent 接入中国 A 股实时与历史行情数据

【免费下载链接】intentkitIntentKit is an open-source, self-hosted cloud agent cluster that manages a collaborative team of AI agents for you.项目地址: https://gitcode.com/GitHub_Trending/int/intentkit

导读：本文围绕 IntentKit 仓库中intentkit/tools/cn_stock工具包展开，系统讲解其 9 个 A 股（沪/深/京交易所）数据工具的能力边界、启用配置、底层实现原理与 Agent 编排模式。读完本文，你将掌握如何在 IntentKit 的 Agent 配置中按需启用行情、K 线、指数、板块、资金流向、新闻、公告、财务与交易日历工具，理解其基于 akshare + Redis 缓存 + 限流的架构设计，并学会参照仓库中的 leaf agent 示例搭建自己的 A 股数据 Agent。

一、工具集概览：9 个覆盖 A 股数据全场景的工具

cn_stock是 IntentKit 中面向中国 A 股市场（上海 / 深圳 / 北京交易所）的行情数据工具包，数据源由 akshare，对外暴露 9 个工具，各自职责如下：

每个工具的运行时类均继承自CNStockBaseTool（定义在 intentkit/tools/cn_stock/base.py），并统一以cn_stock_作为工具名前缀，例如cn_stock_get_quote。这一点由测试 tests/tools/test_cn_stock.py 逐一对 9 个工具类断言验证。

二、启用与配置：schema 中的 enabled 与 states 三态控制

工具包的 JSON Schema 定义在 intentkit/tools/cn_stock/schema.json，它是 Agent 配置校验与前端表单生成的依据，核心结构如下：

{ "type": "object", "title": "China A-Share", "description": "Real-time and historical market data for Chinese A-shares ...", "x-tags": ["Analytics", "Stocks", "China"], "properties": { "enabled": { "type": "boolean", "title": "Enabled", "default": false }, "states": { "type": "object", "properties": { "get_quote": { "type": "string", "enum": ["disabled", "public", "private"], "x-enum-title": ["Disabled", "Agent Owner + All Users", "Agent Owner Only"], "default": "disabled" }, "get_kline": { "...": "..." }, "get_index": { "...": "..." }, "get_board": { "...": "..." }, "get_capital_flow": { "...": "..." }, "get_news": { "...": "..." }, "get_announcement": { "...": "..." }, "get_financials": { "...": "..." }, "is_trading_day": { "...": "..." } } } }, "required": ["states", "enabled"] }

要点说明：

• enabled：工具集总开关，默认false，需显式开启；
• states：每个工具独立三态控制——
  • disabled：禁用，不注入任何 Agent；
  • public：Agent 所有者 + 所有用户可用；
  • private：仅 Agent 所有者可用；
• 所有工具默认均为disabled，按需开启，避免为 Agent 注入用不到的工具。

这套三态过滤逻辑在 intentkit/tools/cn_stock/init.py 的get_tools中实现：遍历config["states"]，跳过disabled，当状态为private且is_private=False时同样跳过；工具实例在模块导入时一次性构建为无状态单例（_TOOLS字典），每次调用直接复用。对应行为由测试 tests/tools/test_cn_stock.py 验证：配置中get_quote为public时，私有请求与公开请求都能拿到；get_kline为private时仅is_private=True的请求可见。

三、底层实现原理：阻塞调用、Redis 缓存与全局限流

README 的 "Operational notes" 揭示了该工具包最关键的三条工程约束，它们全部实现在CNStockBaseTool.run_blocking（intentkit/tools/cn_stock/base.py）：

1. 阻塞调用在线程池中执行

akshare 的所有接口都是同步阻塞调用，直接放在事件循环中会卡死整个 Agent 进程。run_blocking统一通过asyncio.to_thread(func, *args, **kwargs)将其放入线程池执行，调用方（各工具的_arun）仍是异步接口，天然适配 LangChain 工具体系。

2. Redis 短 TTL 缓存吸收重试风暴

akshare 爬取的是免费公开端点，按 IP 限流，因此工具包在 Redis 中做了两级缓存：调用前先以cn_stock:{cache_key}为键GET，命中则直接反序列化返回；未命中则执行真实请求后以ex=cache_ttl写入。各工具的缓存键与 TTL 各不相同：

其中get_quote是唯一显式不缓存旧数据的特殊场景：它缓存的是全市场行情快照（ak.stock_zh_a_spot_em()），TTL 仅 10 秒，调用时再按请求代码过滤，属于 README 所述"live quotes 5–60 秒"的范畴。K 线与财务数据则采用更长 TTL，因为历史数据变化极慢。缓存行为由测试 tests/tools/test_cn_stock.py 双路验证：缓存命中时run_blocking内的执行函数不会被调用；cache_key=None时完全不触碰 Redis。

3. 分类级全局限流保护共享基础设施

每次真实请求前都会执行await self.global_rate_limit_by_category(limit=60, seconds=60)（intentkit/tools/cn_stock/base.py），即整个cn_stock分类下所有工具共享每分钟 60 次的调用配额，防止某个 Agent 的循环查询拖垮 akshare 免费端点。该方法定义在 intentkit/tools/base.py 的IntentKitTool基类中。

此外还有两个值得注意的细节：

• 时区固定为 Asia/Shanghai：today_cn()使用pytz.timezone("Asia/Shanghai")计算"今天"，即使服务器运行在 UTC 时区，默认日期参数也遵循 A 股交易所的墙钟时间；
• 空结果显式报错：未上市或已退市的代码可能返回空数据，各工具统一抛出ToolException并附清晰原因（如"No quotes returned for ...; codes may be invalid or market is closed."），绝不静默返回空结果。akshare 调用失败时也会包装为ToolException(f"akshare call failed: {e}")，保证错误在 LangChain 工具层可被 Agent 捕获理解。

四、股票代码规范化：四种格式一视同仁

README 明确支持四种代码书写方式：600519、sh600519、SH600519、600519.SH。这一能力由 intentkit/tools/cn_stock/base.py 的normalize_a_share_symbol实现：先strip().upper()，剥离SH/SZ/BJ前缀，再按.切掉交易所后缀，最后校验必须是 6 位纯数字，否则抛出ToolException(f"Invalid A-share code: ...")。测试用例覆盖了全部合法格式与非法输入（""、foo、12345、1234567、60051X等）。

配套的market_of函数（intentkit/tools/cn_stock/base.py）负责从 6 位代码推断交易所，规则如下：

该函数被get_capital_flow用于将个股资金流请求路由到对应市场的 akshare 接口，分类断言见测试 tests/tools/test_cn_stock.py。

五、九个工具逐一详解：参数、默认值与数据来源

5.1cn_stock_get_quote— 实时行情快照

• 入参：symbols: list[str]，一次最多 50 个代码（max_length=50，至少 1 个），例如["600519", "000001"]；
• 实现：get_quote.py 调用ak.stock_zh_a_spot_em()拉取全市场快照，再用QUOTE_COLUMNS（17 个中文字段）裁剪后按请求代码过滤；所有请求代码都无匹配时抛ToolException；
• 返回字段：代码、名称、最新价、涨跌幅、涨跌额、成交量、成交额、振幅、最高、最低、今开、昨收、换手率、市盈率-动态、市净率、总市值、流通市值；
• 注意：单次调用即拉全市场再过滤，因此依赖 10 秒缓存与限流，避免高频重复触发。

5.2cn_stock_get_kline— 历史 K 线

• 入参：get_kline.py 中symbol（6 位代码）、period（daily/weekly/monthly，默认daily）、days_back（回溯自然日数，1–1825，默认 90）、adjust（""不复权 /qfq前复权 /hfq后复权，默认qfq）；
• 实现：以today_cn()为终点向前推days_back天，调用ak.stock_zh_a_hist，缓存键包含代码、周期、复权方式与日期区间，TTL 900 秒；无数据时提示"symbol may be delisted"；
• 用途：README 定位为趋势、波动率与形态分析的基础数据。

5.3cn_stock_get_index— 主要指数点位与历史

• 入参：get_index.py 的indices（指数中文名列表，默认四个头部指数）、history（spot仅实时 /30d附加 30 个日 K 柱，默认spot）；
• 内置指数映射INDEX_CODES：上证指数000001、沪深300000300、中证500000905、中证1000000852、深证成指399001、创业板指399006、科创50000688；默认请求["上证指数", "深证成指", "创业板指", "沪深300"]，未知名称直接抛ToolException；
• 实现细节：spot模式调用ak.stock_zh_index_spot_em(symbol="沪深重要指数")；30d模式用asyncio.gather并发拉取每个指数的日 K（回溯 45 个自然日再截取最后 30 个交易日），返回{"spot": [...], "history": {...}}。

5.4cn_stock_get_board— 行业 / 概念板块快照

• 入参：get_board.py 的kind（industry行业板块 /concept概念板块，默认industry）、top（涨跌幅前 N 与后 N，1–50，默认 20）；
• 实现：分别调用ak.stock_board_industry_name_em或ak.stock_board_concept_name_em，按涨跌幅排序后返回{"kind": ..., "gainers": [...], "losers": [...]}，用于识别热点板块与轮动。

5.5cn_stock_get_capital_flow— 资金流向

• 入参：get_capital_flow.py 的scope（stock个股 /market全市场，必填）、symbol（scope="stock"时必填，运行时校验，缺失抛ToolException）、days（最近交易日数，1–30，默认 5）；
• 实现：个股模式按market_of推断市场后调ak.stock_individual_fund_flow，全市场模式调ak.stock_market_fund_flow，均取末尾days行返回，用于观察主力、散户与北向资金动向。

5.6cn_stock_get_news— 个股新闻与宏观头条

• 入参：get_news.py 的scope（stock/macro，必填）、symbol（scope="stock"时必填）、limit（1–50，默认 10）；
• 实现：个股调ak.stock_news_em(symbol=code)，宏观调ak.stock_info_global_em，在深入分析前为 Agent 提供情绪上下文。

5.7cn_stock_get_announcement— 上市公司公告

• 入参：get_announcement.py 的on_date（YYYYMMDD，默认今天 Asia/Shanghai，公告通常在收盘后发布）、limit（1–100，默认 20）；
• 实现：调用ak.stock_notice_report(symbol="全部", date=d)，覆盖财报、重大事项、增减持、停复牌等可能影响股价的披露信息。

5.8cn_stock_get_financials— 财务基本面

• 入参：get_financials.py 的indicator（按报告期/按年度/按单季度，默认按报告期）、limit（报告期数，1–40，默认 8）；
• 实现：调用ak.stock_financial_abstract_ths返回 EPS、ROE、利润、营收、利润率与增长率等指标，缓存 6 小时，无数据时抛ToolException。

5.9cn_stock_is_trading_day— 交易日判断（调度任务必用）

• 入参：is_trading_day.py 的on_date（支持YYYY-MM-DD、YYYYMMDD、YYYY/MM/DD三种格式，默认今天）；
• 实现：调用ak.tool_trade_date_hist_sina获取全量交易日历（缓存 24 小时），将目标日期isoformat()与集合比对，返回{"date": "2026-06-16", "is_trading_day": true}；
• README 特别强调：cron 触发器不会跳过节假日，因此任何定时行情任务的第一步都必须是is_trading_day检查，避免在休市日拉取到前一日收盘价产生误导。

六、Agent 编排模式：leaf 数据 Agent + 监控 Agent 编排

README 的 Composition 一节指出，这些工具被设计为叶子（leaf）公开 Agent使用——每个数据领域一个 Agent，再由团队构建的监控 Agent 通过lead_call_agent编排调用。仓库提供了四个可工作的参考配置，位于public_agents/base/目录：

• public_agents/base/cn-stock-quote.yaml — 实时行情、K 线、指数、交易日
• public_agents/base/cn-stock-fundamentals.yaml — 财务基本面
• public_agents/base/cn-stock-news.yaml — 新闻与公告
• public_agents/base/cn-market-overview.yaml — 市场总览

以 cn-stock-quote.yaml 为例，其工具段配置清晰展示了上文所述的启用方式：

name: "China A-Share Quote Bot" slug: "cn-stock-quote" description: "Real-time spot quotes, K-line history and major-index snapshots for Chinese A-shares (Shanghai/Shenzhen/Beijing)." tags: ["Base", "China", "Stocks"] model: google/gemini-3-flash-preview temperature: 0.1 tools: cn_stock: enabled: true states: get_quote: public get_kline: public get_index: public is_trading_day: public

该 YAML 还示范了 leaf Agent 的几条关键操作纪律，可作为编写自有 A 股 Agent 的模板：

1. 交易日检查先行：当调用方意图是"今日行情"时，必须先调cn_stock_is_trading_day，若返回false则立即如实返回该事实、不再拉行情，避免把上一交易日收盘数据当实时数据；
2. 代码不预处理：600519、sh600519、SH600519、600519.SH均可直接传入，工具内部规范化；
3. 按需请求：只请求调用方明确要求的代码列表，绝不下发全市场数据（浪费且慢）；
4. 保守输出：只返回工具真实产出的数字，禁止编造行情或用训练数据补缺口，保留中文字段名以便下游确定性解析；
5. 数据新鲜度标注：当休市日仍被要求提供数据时，允许取数但必须在输出中附带"data_freshness": "previous_close"，让下游 Agent 知晓上下文。

七、质量保障：单元测试覆盖的关键契约

工具包配套测试集中在 tests/tools/test_cn_stock.py，从四个层面锁定了行为契约：

• 元数据一致性：9 个工具类的name与category逐一断言（cn_stock_*/cn_stock）；
• 输入校验：代码规范化接受四种合法格式、拒绝各类垃圾输入；GetQuoteInput拒绝空列表与超过 50 个代码；GetKLineInput校验days_back上下界（1–1825）；GetCapitalFlowInput在scope="stock"缺symbol时于运行时抛ToolException；GetIndexInput拒绝未知指数名；
• 状态过滤：get_tools按public/private/disabled正确筛选工具集合；
• 缓存语义：缓存命中时不再执行底层函数；cache_key=None时完全不触碰 Redis。

这些测试无需真实网络即可运行（全部通过 mock / patch 隔离 akshare），新接入其他行情数据源时可直接复用同套模式。

八、使用建议与限制

• 适用前提：cn_stock 依赖 Redis（用于缓存）与 akshare（数据源），部署环境需保证二者可用；akshare 免费公开端点按 IP 限流，故全分类每分钟 60 次上限是硬约束，设计 Agent 循环查询时应预估配额；
• 数据口径：返回字段保留中文列名（如涨跌幅、市盈率-动态），下游解析需按此口径处理；
• 边界情况：停牌、未上市、退市股票可能返回空数据，工具以ToolException显式报错而非静默吞掉，Agent 应在错误处理分支中如实转述原因；
• 组合建议：将行情类（quote/kline/index）、基本面类（financials）、事件类（news/announcement）拆分为独立 leaf Agent，由上层监控 Agent 统一编排，既符合仓库给出的cn-*.yaml设计，也有利于各自独立复用与限流配额管理。

从配置 Schema、源码实现到 leaf Agent 编排与测试，cn_stock工具包为在 IntentKit 中构建中国 A 股数据能力提供了完整、可复用的范式——阅读 intentkit/tools/cn_stock/base.py 与 public_agents/base/cn-stock-quote.yaml 两份核心文件，即可快速上手扩展自己的行情 Agent。

【免费下载链接】intentkitIntentKit is an open-source, self-hosted cloud agent cluster that manages a collaborative team of AI agents for you.项目地址: https://gitcode.com/GitHub_Trending/int/intentkit