☰
MCP协议实战:将股票数据服务封装成AI可调用的标准化接口
2026/10/20 17:09:17 网站建设 项目流程

1. 项目背景与问题定义:为什么要把股票数据能力接进 AI

做量化或者做投资分析的朋友,这两年应该都有一种很强烈的体感:模型越来越聪明,但数据获取依然是最让人头疼的环节。你让模型帮你写一个均线策略,它能写得像模像样;但你要是让它自己去拉行情、算指标、看财务数据,它就只能干瞪眼——因为模型本身不具备访问外部数据的能力。

我之前在做一个自己的投研分析工具时,也撞上了这堵墙。每天凌晨要手动把一堆行情数据导出来,再清洗、格式化,喂给模型做复盘。换一个标的,又要重跑一遍。中间还有各种数据源接口差异、频率限制、字段命名不统一的问题,整个链路又脆又慢。后来我开始想:如果能让模型直接通过一个标准接口去拿数据,像调函数一样自然,那整个工作流是不是就能盘活?

这正是 stock-sdk-mcp 要解决的问题。它本质上是一个把股票数据能力封装成 MCP(Model 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、period,adjust 参数它会默认选 qfq(前复权),这个倾向在描述里已经写清楚了。

第二步,MCP 客户端把这次调用意图发给服务端,服务端检查参数合法性,校验日期范围、代码格式。

第三步,服务端拿着这些参数去底层数据源拉原始数据。这里面的数据源可能是多个——日线走一个接口,分钟线走另一个接口,财务数据走第三个接口。stock-sdk-mcp 负责把不同接口返回的字段统一映射为内部标准格式。

第四步,服务端做必要的数据加工:补全缺失交易日(交易日历必须剔除节假日和周末)、按时间排序、处理停牌导致的空数据。

第五步,服务端把处理好的数据以结构化文本或表格形式返回给模型。模型基于这份数据再做二次分析,比如计算20日均线,或者直接调用技术指标工具。

流程图我不画了,用文字描述就是这个闭环。关键点在于,模型在整个环节里只做两件事:根据用户意图决定调哪个工具,以及解读返回的数据。至于数据从哪来、怎么清洗、用什么语法解析,模型一概不用关心。

3.3 数据返回的上下文控制

这里有一个非常容易被忽略的坑:MCP 返回数据要控制体积。如果你请求了一年的分钟级K线,返回给模型的数据可能有几万行,直接爆掉上下文窗口。

我的做法是分层返回。默认只返回最近30根K线,如果想看全量,可以显式传一个 limit 参数。在工具描述里明确写清楚:“默认返回最近30条记录,如需更多请指定limit”。这样既保住了大多数对话场景的即时性,也给了深度分析时的扩容空间。

另外,返回字段也做了裁剪。K线数据我只返回 open、high、low、close、volume 五个核心字段,市净率、换手率这些指标类字段放到财务工具的返回值里。模型拿到的数据干净,分析才不容易跑偏。

4. 实操落地:从零搭建并接入你自己的 stock-sdk-mcp

4.1 开发环境与依赖准备

我做这个项目用的环境是 Python 3.11 + FastAPI,MCP 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( name="get_market_snapshot", description="获取指定股票的最新行情快照,包括最新价、涨跌幅、成交量、成交额等", inputSchema={ "type": "object", "properties": { "symbol": {"type": "string", "description": "股票代码"}, }, "required": ["symbol"] } ) # 声明K线历史工具 tool_kline = Tool( name="get_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( name="get_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(type="text", text=format_snapshot(result))]) elif name == "get_kline_history": result = await fetch_kline_history(arguments) return ToolResult(contents=[TextContent(type="text", text=format_kline(result))]) elif name == "get_financial_indicators": result = await fetch_financial_indicators(arguments) return ToolResult(contents=[TextContent(type="text", text=format_financial(result))]) else: return ToolResult(contents=[TextContent(type="text", text=f"未知工具: {name}")])

这段代码大家看个骨架就行,真正生产环境还需要加参数校验、错误处理、数据缓存、限流逻辑。但核心思路就是:每个工具都是独立函数,输入是结构化参数,输出是文本/结构化内容。

4.3 数据源适配与字段映射

不同数据源的字段差异是接数据时最烦的问题。A 数据源返回的成交额字段叫 amount,B 数据源叫 amount_total,C 数据源干脆叫 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 是第一步,接完之后的世界,比想象中复杂,也比想象中有趣。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询