A股实时行情MCP服务器 v2.0

这是一个基于Model Context Protocol (MCP) 的A股实时行情查询服务器，采用多数据源架构，支持查询A股实时价格、历史数据、财务信息等。专为AI助手和金融分析工具设计。

🚀 新版本特性 (v2.0)

• 多数据源架构: 支持故障转移和负载均衡
• 模块化设计: 工具和数据源完全解耦
• 智能缓存: 提高响应速度和减少API调用
• 增强错误处理: 完善的异常处理和重试机制
• 完整测试覆盖: 单元测试和集成测试

功能特性

🔥 核心功能

• 实时价格查询: 获取A股实时价格、涨跌幅、成交量等
• 股票基本信息: 查询股票的基本信息和公司概况
• 市场概况: 获取上证、深证等主要指数信息
• 历史数据: 查询股票的历史K线数据
• 财务数据: 获取利润表、资产负债表、现金流量表

🛠️ 支持的工具

1. get_realtime_price - 获取实时价格
2. get_stock_info - 获取股票基本信息
3. get_market_summary - 获取市场概况
4. get_stock_history - 获取历史数据
5. get_financial_data - 获取财务数据

安装和使用

🚀 PyPI安装（推荐）

pip install a-stock-mcp-server

📦 从源码安装

# 克隆仓库
git clone https://github.com/Llldmiao/a-stock-mcp-server.git
cd a-stock-mcp-server

# 使用Poetry安装（推荐）
poetry install

# 或者使用pip安装
pip install -r requirements.txt
pip install -e .

🤖 MCP服务器使用

快速配置

在你的MCP客户端配置文件中添加：

{
  "mcpServers": {
    "a-stock-realtime": {
      "command": "python",
      "args": ["-m", "a_stock_mcp_server"]
    }
  }
}

可用工具

1. get_realtime_price - 获取实时价格
   • 参数: symbol (股票代码，如"000001")
2. get_stock_info - 获取股票基本信息
   • 参数: symbol (股票代码)
3. get_market_summary - 获取市场概况
   • 参数: 无
4. get_stock_history - 获取历史数据
   • 参数: symbol, period(daily/weekly/monthly), start_date, end_date
5. get_financial_data - 获取财务数据
   • 参数: symbol, report_type(income/balance/cashflow)

使用示例

// 查询平安银行实时价格
{
  "name": "get_realtime_price",
  "arguments": {
    "symbol": "000001"
  }
}

// 查询市场概况
{
  "name": "get_market_summary",
  "arguments": {}
}

详细使用说明请参考 MCP使用指南

🛠️ 命令行工具使用

安装后可以使用命令行工具：

# 查询股票价格
a-stock-cli price -s 000001

# 查询股票信息
a-stock-cli info -s 000001

# 查询市场概况
a-stock-cli market

🧪 本地测试

# 运行完整测试
python3 -m pytest tests/ -v

# 运行测试客户端
python3 -m a_stock_mcp_server.test_client

# 运行旧版测试（兼容性）
python3 local_test.py

📚 Python代码使用

新版本 (推荐)

import asyncio
from a_stock_mcp_server.test_client import AStockTestClient

async def main():
    client = AStockTestClient()
    
    # 查询平安银行实时价格
    result = await client.call_tool("get_realtime_price", {"symbol": "000001"})
    print(result)
    
    # 健康检查
    health = await client.health_check()
    print("数据源状态:", health)

asyncio.run(main())

旧版本 (兼容性)

import asyncio
from a_stock_mcp_server.local_test import AStockLocalTest

async def main():
    server = AStockLocalTest()
    
    # 查询平安银行实时价格
    result = await server.call_tool("get_realtime_price", {"symbol": "000001"})
    print(result)

asyncio.run(main())

输出示例：

股票代码: 000001
股票名称: 平安银行
当前价格: ¥11.45
涨跌额: +0.04
涨跌幅: +0.35%
成交量: 834,651.0
成交额: ¥955,004,096.91
最高价: ¥11.51
最低价: ¥11.37
开盘价: ¥11.41
昨收价: ¥11.41
换手率: 0.43%
市盈率: 4.47
市净率: 0.5

使用示例

查询实时价格

{
  "tool": "get_realtime_price",
  "arguments": {
    "symbol": "000001"
  }
}

查询历史数据

{
  "tool": "get_stock_history", 
  "arguments": {
    "symbol": "000001",
    "period": "daily",
    "start_date": "20240101",
    "end_date": "20241231"
  }
}

查询财务数据

{
  "tool": "get_financial_data",
  "arguments": {
    "symbol": "000001",
    "report_type": "income"
  }
}

🔧 常用股票代码

架构设计

多数据源架构

数据源层 (Data Sources)
├── BaseDataSource (抽象基类)
├── AKShareDataSource (AKShare实现)
├── DataSourceManager (数据源管理器)
└── CacheManager (缓存管理器)

工具层 (Tools)
├── BaseTool (工具基类)
├── RealtimePriceTool (实时价格)
├── StockInfoTool (股票信息)
├── MarketSummaryTool (市场概况)
├── StockHistoryTool (历史数据)
└── FinancialDataTool (财务数据)

服务器层 (Server)
├── AStockMCPServer (主服务器)
├── AStockMCPServerWithAKShare (兼容包装)
└── AStockTestClient (测试客户端)

核心特性

• 故障转移: 自动切换到备用数据源
• 智能缓存: 减少API调用，提高响应速度
• 模块化设计: 易于扩展和维护
• 完整测试: 单元测试和集成测试覆盖

数据源

本MCP服务器使用 AKShare 作为主要数据源：

• 免费、开源
• 数据更新及时
• 支持多种金融数据
• 社区活跃

扩展数据源

架构支持添加新的数据源，如：

• 新浪财经API
• 腾讯财经API
• 其他金融数据提供商

详细扩展指南请参考 架构文档

扩展建议

1. 多数据源支持

• 集成新浪财经API
• 集成腾讯财经API
• 添加数据源故障转移

2. 缓存机制

• 添加Redis缓存
• 减少API调用频率
• 提高响应速度

3. 数据验证

• 添加数据有效性检查
• 异常数据处理
• 错误重试机制

4. 更多功能

• 技术指标计算
• 股票筛选器
• 实时推送
• 历史回测

注意事项

1. 数据延迟: AKShare数据可能有15-20分钟延迟
2. API限制: 注意API调用频率限制
3. 数据准确性: 仅供参考，投资决策请谨慎
4. 网络依赖: 需要稳定的网络连接

故障排除

常见问题

问题1: 导入错误

ModuleNotFoundError: No module named 'akshare'

解决方案: 安装AKShare

pip3 install akshare

问题2: 网络超时

获取数据失败: timeout

解决方案: 检查网络连接，可能需要代理

问题3: 股票代码不存在

未找到股票代码: XXXXXX

解决方案: 检查股票代码格式，确保是6位数字

问题4: 数据为空

解决方案: 检查股票代码格式是否正确

日志调试

import logging
logging.basicConfig(level=logging.DEBUG)

📝 完整使用示例

#!/usr/bin/env python3
import asyncio
from a_stock_mcp_server.local_test import AStockLocalTest

async def main():
    # 创建服务器实例
    server = AStockLocalTest()
    
    # 查询平安银行实时价格
    print("=== 平安银行实时价格 ===")
    price = await server.call_tool("get_realtime_price", {"symbol": "000001"})
    print(price)
    
    # 查询股票基本信息
    print("\n=== 平安银行基本信息 ===")
    info = await server.call_tool("get_stock_info", {"symbol": "000001"})
    print(info)
    
    # 查询市场概况（只显示前10个）
    print("\n=== 市场概况（前10个）===")
    market = await server.call_tool("get_market_summary", {})
    lines = market.split('\n')
    for line in lines[:12]:  # 标题 + 前10个指数
        print(line)

if __name__ == "__main__":
    asyncio.run(main())

贡献

欢迎提交Issue和Pull Request来改进这个项目！

许可证

MIT License