☰
用DeepSeek、MCP和AKShare搭建智能金融问答系统:技术方案与落地实践
2026/10/8 12:47:09 网站建设 项目流程

1. 从一次“答非所问”说起:智能金融问答系统到底难在哪

你问“腾讯控股最近一周股价走势如何”,模型一本正经地回了一段“一般来说港股受外围市场影响较大”的套话,数据一个没有。这不是模型不聪明,而是它压根没拿到真实行情。金融问答和普通闲聊最大的区别在于:答案必须建立在可验证的实时数据上,而不是语言模型的记忆。

我试过把行情数据直接塞进提示词,问题立刻来了:数据每天变,你不可能手动更新;股票几千只,你不可能全量塞进去;用户问法千奇百怪,“腾讯”“00700”“鹅厂”指的都是同一只票。所以真正要解决的是三件事——让模型知道什么时候该查数据、让它能真的查到数据、让它把查回来的数据讲成人话。

这套智能金融问答系统的分工很清晰:DeepSeek 负责理解意图和生成回答,MCP 负责把“查数据”这个动作标准化成模型可调用的工具,AKShare 负责真正去取行情。三者拼起来,就是一条“提问 → 识别意图 → 调用工具 → 取数 → 组织答案”的完整链路。适合谁?适合想做一个能跑通、能验证、能自己改数据源的开发者,不需要你有量化背景,但需要你会写一点 Python、能看懂 JSON。

下面我按“先跑通再优化”的顺序,把 MCP Server 配置、AKShare 接口封装、DeepSeek 提示词模板全部给出来,最后附一轮完整的问答链路验证动作,确认数据取得到、答案答得准。

2. 前置准备:DeepSeek API Key、MCP 协议与 AKShare 环境

2.1 DeepSeek 的接入方式与 Key 获取

DeepSeek 提供 OpenAI 兼容的接口,这意味着你原来写 OpenAI 的代码,改个 base_url 和 model 就能用。对智能金融问答来说,它的价值在于推理稳定、支持工具调用(Function Calling),能把“用户这句话要不要查数据”判断得比较准。

获取 Key 的路径:进入模型对话页面先确认模型可用,再到 API Keys 页面创建密钥。地址如下:

  • 模型对话:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite
  • API Keys:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite
  • 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite

拿到 Key 后,接口地址统一用https://taotoken.net/api(注意这个地址不加 UTM 参数,直接作为 base_url 使用)。模型 ID 用deepseek-chat即可。这里要提醒一句:Key 只放在服务端环境变量里,别写进前端代码,也别提交到 Git。

2.2 MCP 协议:把“查数据”变成模型能调用的工具

MCP(Model Context Protocol)你可以理解成 AI 世界的 USB-C 接口。以前每接一个数据源,你都要为某个模型单独写一套 Function Calling 的胶水代码;换了模型就得重写。MCP 把这层标准化了:Server 端只负责暴露工具,Client 端负责把工具转成模型认识的格式,模型换了,Server 不用动。

核心就三个角色:

  • MCP Server:提供具体能力,比如我们的 AKShare 行情服务,暴露get_stock_hist、get_stock_info这类工具。
  • MCP Client:连接模型和 Server 的中间层,负责把 Server 的工具列表转成 OpenAI 格式的 tools 参数。
  • 传输层:本地进程通信用 Stdio,跨进程或远程用 SSE。本文用 SSE,方便你后面部署成服务。

2.3 AKShare 环境与依赖安装

AKShare 是基于 Python 的金融数据接口库,股票、基金、期货、宏观经济都能取,免费开源,数据源丰富。它的返回基本是 pandas DataFrame,转 JSON 很方便。

Python 建议 3.11。安装命令:

pip install akshare mcp openai uvicorn starlette python-dotenv

如果你打算用 LangChain 那套做 Agent 编排,再补上:

pip install langchain langgraph langchain-mcp-adapters langchain-deepseek

装完先验证 AKShare 能取到数据,这一步别跳过:

import akshare as ak df = ak.stock_zh_a_hist(symbol="000001", period="daily", start_date="20250101", end_date="20250110", adjust="hfq") print(df.head())

能打印出带日期、开盘、收盘的表格,说明数据链路通了。取不到通常是网络或数据源临时波动,重试一次即可。

3. 可复制配置:AKShare MCP Server 与 DeepSeek 工具调用

3.1 用 FastMCP 写一个行情 MCP Server

新建akshare_server.py。核心思路:每个@mcp.tool()装饰的函数就是一个模型可调用的工具,函数签名和 docstring 就是给模型看的说明书,写得越清楚,模型选工具越准。

from mcp.server.fastmcp import FastMCP import akshare as ak import logging from datetime import datetime logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) mcp = FastMCP("AKShareFinance") @mcp.tool() def get_stock_hist(stock_code: str, start_date: str, end_date: str) -> str: """获取A股股票指定日期区间的日频历史行情。 Args: stock_code: 股票代码,如 '000001' start_date: 开始日期,格式 YYYYMMDD end_date: 结束日期,格式 YYYYMMDD Returns: 包含日期、开盘、收盘、最高、最低、成交量的 JSON 字符串 """ try: df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="hfq") if df.empty: return "未找到该股票在指定区间的行情数据。" return df.to_json(orient="records", force_ascii=False) except Exception as e: return f"查询股票行情时出现错误: {e}" @mcp.tool() def get_stock_bid_ask(stock_code: str) -> str: """获取指定A股股票的实时行情报价,含最新价与涨跌幅。 Args: stock_code: 股票代码,如 '000001' Returns: 实时行情 JSON 字符串 """ try: df = ak.stock_bid_ask_em(symbol=stock_code) if df.empty: return "未找到该股票的实时行情。" return df.to_json(orient="records", force_ascii=False) except Exception as e: return f"查询实时行情时出现错误: {e}" @mcp.tool() def get_stock_news(stock_code: str) -> str: """获取指定个股最近的新闻资讯。 Args: stock_code: 股票代码,如 '000001' Returns: 新闻资讯 JSON 字符串 """ try: df = ak.stock_news_em(symbol=stock_code) if df.empty: return "未找到该股票的新闻。" return df.head(20).to_json(orient="records", force_ascii=False) except Exception as e: return f"查询新闻时出现错误: {e}" if __name__ == "__main__": logger.info("Starting AKShare Finance MCP Server") mcp.run(transport="sse", port=9000)

三个工具覆盖了历史行情、实时报价、个股新闻,足够支撑大部分“某股票最近怎么样”类问题。注意 docstring 里参数格式写死成YYYYMMDD,因为 AKShare 的stock_zh_a_hist就吃这个格式,模型看到说明后会照着生成。

3.2 MCP Client 配置:把工具转成 DeepSeek 认识的格式

新建client.py。这里的关键动作是:从 MCP Server 拉取工具列表,把每个工具的inputSchema转成 OpenAI 的tools格式,再交给 DeepSeek。

import asyncio import json import sys from contextlib import AsyncExitStack from mcp import ClientSession from mcp.client.sse import sse_client from openai import OpenAI API_KEY = "你的Key" BASE_URL = "https://taotoken.net/api" MODEL = "deepseek-chat" class MCPClient: def __init__(self): self.session = None self.exit_stack = AsyncExitStack() self.llm = OpenAI(api_key=API_KEY, base_url=BASE_URL) async def connect_to_sse_server(self, server_url: str): self._streams_context = sse_client(url=server_url) streams = await self._streams_context.__aenter__() self._session_context = ClientSession(*streams) self.session = await self._session_context.__aenter__() await self.session.initialize() async def process_query(self, query: str) -> str: messages = [{"role": "user", "content": query}] resp = await self.session.list_tools() available_tools = [] for tool in resp.tools: schema = getattr(tool, "inputSchema", {"type": "object", "properties": {}, "required": []}) available_tools.append({ "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": schema, }, }) first = self.llm.chat.completions.create( model=MODEL, max_tokens=1000, messages=messages, tools=available_tools) msg = first.choices[0].message messages.append(msg.model_dump()) if msg.tool_calls: call = msg.tool_calls[0] args = json.loads(call.function.arguments) result = await self.session.call_tool(call.function.name, args) messages.append({ "role": "tool", "content": f"{result}", "tool_call_id": call.id, }) second = self.llm.chat.completions.create( model=MODEL, max_tokens=1000, messages=messages) return second.choices[0].message.content return msg.content async def chat_loop(self): print("MCP Client Started. 输入 quit 退出。") while True: query = input("\nQuery: ").strip() if query.lower() == "quit": break print("\n" + await self.process_query(query)) async def cleanup(self): await self.exit_stack.aclose() async def main(): client = MCPClient() try: await client.connect_to_sse_server("http://localhost:9000/sse") await client.chat_loop() finally: await client.cleanup() if __name__ == "__main__": asyncio.run(main())

3.3 DeepSeek 提示词模板与工具描述规范

工具描述就是提示词的一部分。模型靠它判断“这个问题要不要调工具、调哪个”。所以 docstring 里要写清楚:这个工具能回答什么问题、参数是什么格式、返回什么。上面三个工具的 docstring 已经按这个原则写了。

如果你想让回答更稳,可以在 system 提示里加一段约束:

{ "role": "system", "content": "你是金融数据助手。涉及具体股票价格、涨跌幅、成交量、新闻的问题,必须先调用工具获取真实数据,禁止凭记忆回答。拿到数据后用简洁中文总结,给出关键数字和时间区间。数据缺失时如实说明,不要编造。" }

这段 system 提示配合工具描述,能显著减少“模型自己编数据”的情况。实测下来,加了这段之后,问“某股票最近一周走势”,模型基本都会先调get_stock_hist。

4. 验证请求:跑通一轮“腾讯控股最近一周股价”问答

4.1 启动 Server 与 Client

开两个终端。第一个终端启动 MCP Server:

python akshare_server.py

看到Starting AKShare Finance MCP Server和监听 9000 端口就对了。第二个终端启动 Client:

python client.py

出现MCP Client Started后,输入问题:

Query: 腾讯控股最近一周的股价走势如何?

4.2 观察工具调用与数据返回

正常链路是这样的:DeepSeek 先判断需要查历史行情,生成工具调用参数,比如:

{"stock_code": "00700", "start_date": "20250413", "end_date": "20250420"}

Client 把这个调用转发给 MCP Server,Server 用 AKShare 取数,返回 JSON 数组,里面是每天的日期、开盘、收盘、最高、最低、成交量。然后 DeepSeek 拿到这批数据,生成自然语言总结,类似“腾讯控股在 4 月 13 日至 4 月 20 日区间,周初开盘约 320 港元,周内最高触及 328 港元,周末收盘约 325 港元,整体小幅上行,成交量在周中有所放大”。

4.3 确认数据取得到、答案答得准

验证分两步。第一步看 Server 日志,有没有Fetching stock history这类记录,有就说明工具被调用了。第二步核对答案里的数字和 AKShare 原始返回是否一致——你可以单独跑一次ak.stock_zh_a_hist对比收盘价。数字对得上,说明整条链路没有丢数据、没有串参数。

如果模型直接回答没调工具,检查两点:system 提示有没有加、工具 docstring 是否足够明确。如果调了工具但报错,看 Server 日志里的异常信息,多半是股票代码格式或日期格式不对。

5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth

5.1 401 与鉴权失败

报401 Unauthorized,先查 Key 有没有写错、有没有多余空格。再确认 base_url 是https://taotoken.net/api,不要自己拼/v1之外的路径。如果 Key 是从环境变量读的,打印一下确认读到了。还有一种情况是 Key 被禁用或额度用尽,去 API Keys 页面确认状态。

5.2 local proxy failed 与连接问题

出现local proxy failed或连接超时,通常是本地网络到接口地址不通,或者你本机开了某些网络工具导致请求被拦。先确认能正常访问接口域名,再检查防火墙有没有拦 9000 端口。MCP Server 和 Client 在同一台机器时,Client 连http://localhost:9000/sse;跨机器要换成 Server 的实际 IP,并确保端口放行。

5.3 reading choices 报错与返回解析

reading 'choices'这类报错,一般是接口返回的不是预期结构,比如返回了错误信息而不是正常 completion。打印完整响应体看error字段。常见原因是模型 ID 写错、tools 格式不合法、或者 messages 里 tool 消息缺少tool_call_id。检查messages.append那几行,tool_call_id必须和模型返回的call.id一致。

5.4 OAuth 与 Claude Code 场景

如果你是在 Claude Code 里接 MCP,遇到 OAuth 相关提示,注意 Claude Code 的 MCP 配置需要写全三件套:Base URL、Key、Model ID。配置文件里command、args、env三段都要对。Base URL 用https://taotoken.net/api,Key 放 env 里,Model ID 用deepseek-chat。配置改完重启 Claude Code 生效。Cline 的 MCP 配置同理,mcpServers下每个 server 的url指向你的 SSE 地址即可。

6. 把系统跑稳:从能答到答得好的下一步

链路跑通只是起点。想让这套智能金融问答系统真正好用,有几个方向可以继续做。一是扩充工具集,现在只有历史行情、实时报价、新闻,可以再加财务指标、板块资金流、宏观经济数据,AKShare 里都有对应接口,照着@mcp.tool()的写法加就行。二是加对话记忆,把历史问答存下来,用户追问“那成交量呢”时能接上上下文,LangChain 的ConversationBufferMemory或自己维护 messages 列表都可以。三是做结果缓存,同一只股票同一天的行情没必要反复请求,加个带过期时间的本地缓存能明显降延迟。

如果你打算长期跑编码和 Agent 类任务,可以了解下 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite 。需要管理多个 Key 或看调用量,控制台在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite 。遇到接入问题,接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite 里有各语言的示例,API Keys 页面 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=deepseek_mcp_akshare&utm_campaign=rewrite 随时可以新建和吊销密钥。

最后提醒一句:金融数据接口有频率限制,别在循环里无脑刷;模型生成的回答只作信息整理,不构成投资建议,涉及真实决策时务必自己核对原始数据。

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

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

立即咨询