1. 项目背景与问题定义为什么要把股票数据能力接进 AI做量化或者做投资分析的朋友这两年应该都有一种很强烈的体感模型越来越聪明但数据获取依然是最让人头疼的环节。你让模型帮你写一个均线策略它能写得像模像样但你要是让它自己去拉行情、算指标、看财务数据它就只能干瞪眼——因为模型本身不具备访问外部数据的能力。我之前在做一个自己的投研分析工具时也撞上了这堵墙。每天凌晨要手动把一堆行情数据导出来再清洗、格式化喂给模型做复盘。换一个标的又要重跑一遍。中间还有各种数据源接口差异、频率限制、字段命名不统一的问题整个链路又脆又慢。后来我开始想如果能让模型直接通过一个标准接口去拿数据像调函数一样自然那整个工作流是不是就能盘活这正是 stock-sdk-mcp 要解决的问题。它本质上是一个把股票数据能力封装成 MCPModel Context Protocol服务的中间层。有了它AI 可以直接通过对话问出“帮我查一下最近20个交易日的成交量变化”或者“算一下这只股票的年化波动率”而不需要我手动写代码、导 CSV、再拼接上下文。这个项目适合谁适合三类人一是做量化策略研究的开发者想用 AI 加速数据探索二是做投资数据分析的从业者日常要频繁查询行情和财务信息三是想把 AI 接入自己投研流程的独立开发者。这个项目卡的位置非常关键——它不解决投资决策本身而是解决“AI 如何按时按量拿到靠谱数据”这个前置问题。2. 整体架构与设计思路拆解2.1 MCP 在中间层扮演的角色要理解 stock-sdk-mcp 的设计思路必须先讲清楚 MCP 到底是什么。你可以把它理解成 AI 模型的 USB 接口——模型通过这个标准协议去插拔各种外部工具。以前我们写 AI 应用要给模型定制各种函数调用Function Calling每个模型厂商的格式还不一样MCP 出现之后工具提供商只需要实现一份标准服务任何支持 MCP 的客户端都能对接。stock-sdk-mcp 就是站在这个位置上的服务端。它对上承接 Claude、或其他支持 MCP 的 AI 客户端对下对接各种行情数据源。用户问一句人话AI 把它翻译成标准工具调用再通过 MCP 协议路由到 stock-sdk-mcp由它去真实数据源拉取数据、做预处理、把结果返回给模型。这个中间层设计有三个明显优势第一隔离了数据源差异。我换数据源、加数据源对 AI 客户端完全透明。之前用某个免费接口后来换成了另一个付费接口客户端侧一行配置都没改模型照样拿数据。第二控制了权限边界。我不会把数据库账号、API Token 直接暴露给模型而是通过 MCP 服务统一管控。模型只能调用我预设好的工具不能自己乱连外网。第三可以做数据清洗和标准化。数据源返回的格式千奇百怪有的用 ts_code有的用 symbol有的日期是字符串有的是时间戳。这些脏活累活在 MCP 服务端统一处理掉模型拿到的永远是一份干净、结构化的数据。2.2 工具集的划分逻辑我把 stock-sdk-mcp 的工具按业务场景拆成几个模块这是整个项目最关键的设计决策。工具数量不是越多越好而是要贴合模型的调用习惯。第一类是行情快照类工具负责取当下或历史某一时刻的价格、成交量、涨跌幅。这类工具频率很高参数简单适合让模型快速判断“现在是个什么状况”。第二类是K线历史类工具按日线、周线、分钟线取一段历史区间。这类工具要重点处理复权因子不然取出来的价格和实际可交易价格对不上。第三类是财务指标类工具处理资产负债表、利润表、现金流量表里的关键字段。这类工具字段最多数据更新也最慢需要做缓存策略避免重复请求浪费额度。第四类是技术指标计算类工具。严格说这类工具不是拿数据而是在数据基础上做计算。我的做法是让 MCP 服务端直接算好 MA、MACD、RSI 这些常用指标省得模型每次都要自己写公式、容易出错。工具分得清楚模型在判定该调用哪个函数时就不会左右为难。我踩过一个坑最开始把所有数据能力揉在一个工具里参数有十几二十个模型经常传错参数名后来拆细之后调用准确率明显提上来了。3. 核心环节的协议设计与数据流说明3.1 MCP 服务的协议格式MCP 协议的精髓在于“工具即函数”。服务端向客户端声明一组工具每个工具有名称、描述、输入参数 JSON Schema。AI 客户端拿到这些声明后在对话中自动判断何时调用哪个工具。以 stock-sdk-mcp 为例一个获取K线工具的核心协议定义长这样{ name: get_kline_history, description: 获取指定股票的历史K线数据支持日线、周线、月线以及分钟线, inputSchema: { type: object, properties: { symbol: { type: string, description: 股票代码如 600519 }, start_date: { type: string, description: 开始日期格式 YYYYMMDD }, end_date: { type: string, description: 结束日期格式 YYYYMMDD }, period: { type: string, enum: [daily, weekly, monthly, 60m, 30m, 5m], description: K线周期 }, adjust: { type: string, enum: [none, qfq, hfq], description: 复权类型不复权、前复权、后复权 } }, required: [symbol, start_date, end_date, period] } }这段 JSON 描述看起来简单里面每一行都有讲究。Symbol 我直接用了不带交易所前缀的写法目的是兼容模型训练语料里最常见的表达方式复权参数单独拎出来枚举类型因为很多新手模型会自己发明参数值给它一个枚举范围能大大降低出错概率。3.2 一次完整的数据请求链路实际跑一次请求你就能体会中间层的重要了。假设我在客户端里问模型“请分析一下某只股票最近一个月的走势并计算20日均线。”第一步模型理解这句话后判定需要调用 get_kline_history 工具自动传入 symbol、start_date、end_date、periodadjust 参数它会默认选 qfq前复权这个倾向在描述里已经写清楚了。第二步MCP 客户端把这次调用意图发给服务端服务端检查参数合法性校验日期范围、代码格式。第三步服务端拿着这些参数去底层数据源拉原始数据。这里面的数据源可能是多个——日线走一个接口分钟线走另一个接口财务数据走第三个接口。stock-sdk-mcp 负责把不同接口返回的字段统一映射为内部标准格式。第四步服务端做必要的数据加工补全缺失交易日交易日历必须剔除节假日和周末、按时间排序、处理停牌导致的空数据。第五步服务端把处理好的数据以结构化文本或表格形式返回给模型。模型基于这份数据再做二次分析比如计算20日均线或者直接调用技术指标工具。流程图我不画了用文字描述就是这个闭环。关键点在于模型在整个环节里只做两件事根据用户意图决定调哪个工具以及解读返回的数据。至于数据从哪来、怎么清洗、用什么语法解析模型一概不用关心。3.3 数据返回的上下文控制这里有一个非常容易被忽略的坑MCP 返回数据要控制体积。如果你请求了一年的分钟级K线返回给模型的数据可能有几万行直接爆掉上下文窗口。我的做法是分层返回。默认只返回最近30根K线如果想看全量可以显式传一个 limit 参数。在工具描述里明确写清楚“默认返回最近30条记录如需更多请指定limit”。这样既保住了大多数对话场景的即时性也给了深度分析时的扩容空间。另外返回字段也做了裁剪。K线数据我只返回 open、high、low、close、volume 五个核心字段市净率、换手率这些指标类字段放到财务工具的返回值里。模型拿到的数据干净分析才不容易跑偏。4. 实操落地从零搭建并接入你自己的 stock-sdk-mcp4.1 开发环境与依赖准备我做这个项目用的环境是 Python 3.11 FastAPIMCP SDK 用的是官方 Python 包。如果你要复刻第一步先把环境搞定pip install mcp fastapi uvicorn pandas requests为什么选 Python因为金融数据生态里 Python 的库最全Pandas 处理时间序列数据几乎是标配数据源 SDK 也基本是 Python 优先。MCP 官方提供了 Python SDK所以技术选型上没什么好纠结的。项目目录结构我建议这样组织stock-sdk-mcp/ ├── server.py ├── mcp_server.py ├── tools/ │ ├── market.py │ ├── kline.py │ ├── financial.py │ └── indicator.py ├── datasource/ │ ├── base.py │ ├── source_a.py │ └── source_b.py ├── utils/ │ ├── validator.py │ └── formatter.py └── config.py把数据源和工具逻辑分开这是长期维护的关键。我最初贪图省事把数据源调用直接写进工具函数里后来加第二个数据源的时候差点重构到崩溃。4.2 服务端核心实现服务端骨架其实不长用 MCP 装饰器就能快速声明工具我先把最核心的注册代码贴出来from mcp.server import Server, Tool from mcp.types import TextContent, ToolResult import json app Server(stock-sdk-mcp) # 声明市场快照工具 tool_snapshot Tool( nameget_market_snapshot, description获取指定股票的最新行情快照包括最新价、涨跌幅、成交量、成交额等, inputSchema{ type: object, properties: { symbol: {type: string, description: 股票代码}, }, required: [symbol] } ) # 声明K线历史工具 tool_kline Tool( nameget_kline_history, description获取指定股票的历史K线数据支持日线、周线、月线和分钟线自动处理复权、停牌和缺失交易日, inputSchema{ type: object, properties: { symbol: {type: string, description: 股票代码}, start_date: {type: string, description: 开始日期 YYYYMMDD}, end_date: {type: string, description: 结束日期 YYYYMMDD}, period: {type: string, enum: [daily, weekly, monthly, 60m, 30m, 5m]}, adjust: {type: string, enum: [none, qfq, hfq], description: 默认qfq前复权} }, required: [symbol, start_date, end_date, period] } ) # 声明财务指标工具 tool_financial Tool( nameget_financial_indicators, description获取指定股票的主要财务指标如营收、净利润、市盈率、市净率、ROE等适合做基本面分析, inputSchema{ type: object, properties: { symbol: {type: string, description: 股票代码}, report_date: {type: string, description: 报告期留空则返回最近一期} }, required: [symbol] } ) # 注册工具处理器 app.entity_handler() async def handle_tool(name: str, arguments: dict): if name get_market_snapshot: result await fetch_market_snapshot(arguments[symbol]) return ToolResult(contents[TextContent(typetext, textformat_snapshot(result))]) elif name get_kline_history: result await fetch_kline_history(arguments) return ToolResult(contents[TextContent(typetext, textformat_kline(result))]) elif name get_financial_indicators: result await fetch_financial_indicators(arguments) return ToolResult(contents[TextContent(typetext, textformat_financial(result))]) else: return ToolResult(contents[TextContent(typetext, textf未知工具: {name})])这段代码大家看个骨架就行真正生产环境还需要加参数校验、错误处理、数据缓存、限流逻辑。但核心思路就是每个工具都是独立函数输入是结构化参数输出是文本/结构化内容。4.3 数据源适配与字段映射不同数据源的字段差异是接数据时最烦的问题。A 数据源返回的成交额字段叫 amountB 数据源叫 amount_totalC 数据源干脆叫 turnover。如果不做统一映射模型面对这些混乱字段根本没法稳定工作。我的做法是在 datasource/base.py 里定义一套内部标准字段名每个数据源适配器只负责把原始字段转换成这套标准STANDARD_ALIASES { symbol: [code, ts_code, symbol], open: [open, open_price, opening], high: [high, high_price, max_price], low: [low, low_price, min_price], close: [close, close_price, closing], volume: [volume, vol, trade_volume], amount: [amount, amount_total, turnover, total_turnover], }适配器写好后后端换数据源就变成改一个配置项的事。我甚至在适配层加了一个“方言自动识别”的功能根据返回字段的 key 集合自动匹配最接近的标准字段不需要硬编码映射表。这一步投入的工时很值。因为很多时候你自己也搞不清到底用哪个字段比如“成交额”在不同数据源里可能是 trade_amount、amount_sum、business_amount有自动识别能省很多手工补丁。4.4 交易日历与时间窗口处理股票数据坑最多的地方就是时间。节假日休市、临时停牌、下午盘和早盘规则不同这些都是数据「正确性」的隐形杀手。我在服务端内置了一份交易日历表覆盖最近十年。用户请求的日期区间如果跨过五一、国庆、春节服务端会自动把非交易日剔除。以前试过直接拿自然日过滤结果模型拿着一堆空数据瞎分析还分析得头头是道那叫一个荒唐。另外分钟线数据还涉及集合竞价、收盘集合竞价的特殊时段。我默认只抓取 09:30 到 15:00 的连续竞价数据把 9:15-9:25 的集合竞价数据单独标记为 auction 类型避免模型误把竞价单当成成交价。5. 常见问题与排查技巧实录5.1 模型调用工具时参数格式不匹配这是接入初期最频发的问题。模型经常把日期写成 20240101 没问题但偶尔会写成 2024-01-01或者把股票代码后缀加了 .SH / .SZ。我一开始靠描述约束后来发现最可靠的办法是在服务端做宽容解析。日期解析我写了正则容错先匹配 YYYYMMDD再匹配 YYYY-MM-DD两边都失败就尝试用日期解析库兜底。代码后缀我直接 strip 掉点号交易所缩写前缀只保留纯数字部分。这样模型传 600519.SH、SH600519、600519 都能正常识别。5.2 请求频率过高触发限流免费数据源基本都有每分钟请求次数限制。模型在分析多个股票时可能会短时间内密集调用同一个工具一不留神就触发了 429 状态码。我的应对策略是两级缓存首先是请求级缓存同一个 symbol 同一个时间窗口 同一周期10 分钟内不走网络直接读内存其次是原始数据文件缓存把当天获取的数据落到本地磁盘重启服务也能即时复用。实测把数据源请求量降到了原来的五分之一限流问题基本绝迹。5.3 返回数据被截断导致上下文爆掉如果你按我前面说的默认返回最近 30 条K线这个问题几乎不会出现。但如果你手贱加了一个全量返回的工具或者对着一年分钟线做策略分析上下文窗口瞬间就会被撑爆。我建议在客户端侧加一层输出摘要。大结果返回后模型可以直接用“数据已获取共 4800 条已汇总为最近 5 日均值趋势”这样的摘要作为工作上下文需要细节时再回查具体行。这样既保证了模型有全局视野又不至于把所有原始数据都塞进对话。5.4 复权数据算错前复权、后复权、不复权三种口径算同一根K线结果完全不同。我见过不少模型因为默认拿不复权数据做趋势判断遇到除权除息时突然出现断崖式“跌停”误判为重大利空。解决方案是在工具描述里明确标注默认前复权并在返回数据的表头加上 adjust 字段。模型拿到前复权数据后计算涨跌幅和均线才有意义。对于分红送转比较频繁的股票务必在复权口径下做历史回测。5.5 工具描述写得含糊导致调用率低有些开发者觉得工具描述随便写写就行反正模型能猜。但实际上描述里少了“返回最近30条”这句话模型可能默认请求全量数据少了“默认前复权”模型可能传一个联合查询再自己算。每一个缺省语义最后都会变成一次错误调用或一次超长返回。我把工具描述当成一份“使用说明书”来写把所有默认值、返回格式、限制条件全部写进去。写完之后自己当使用者跑一遍看描述能不能覆盖所有可能的问题。这是提高模型工具调用准确率最便宜的手段。6. 经验总结与后续扩展方向这个项目从最初我自己用的一个私有脚本慢慢长成了一个标准化的 MCP 服务中间踩的坑比我预想的多得多。但我最深的体会有两个。第一个体会是AI 接入数据服务真正的难点不在模型能力而在工程化的围墙。你要让模型稳定拿到正确数据就需要在工具描述、参数校验、数据清洗、缓存策略这些地方下功夫。模型就像一个新员工你把工具教好了它才能干得漂亮你给它一把电钻去拧螺丝那它只会把墙打穿。第二个体会是模块化拆解永远没有过早。我重构了三轮才把工具定义、数据源适配、服务注册这三件互相纠缠的事彻底拆开。现在每次加新数据源只需要写一个适配器加新工具只需要写一个处理函数。维护成本降到了以前的四分之一。如果你也想做一个类似的 MCP 数据服务我建议不要一开始就铺太大。先接一个数据源、实现两三个最常用的工具让模型跑通一条完整的链路。等你自己真的开始依赖这个服务了再去扩展财务指标、分钟线、技术指标计算这些进阶能力步步为营比一步到位稳得多。后续的扩展方向上我目前在重点尝试两个方向一个是把自然语言查询直接翻译成工具组合调用让用户直接说“帮我筛选出最近5日连续放量且涨幅超过3%的股票”服务端自动串联行情、K线、指标多个工具另一个是把回测引擎也封装成 MCP 工具让模型不仅能看数据还能直接跑策略回测并把结果拿回来。这些方向已经不只是数据接口的问题而是把整个科研工作流都搬进对话里。股票数据接入 AI 是第一步接完之后的世界比想象中复杂也比想象中有趣。