当前位置：首页 > news >正文

mcp-server高级技巧：如何用Python脚本批量获取历史股价并生成分析报告

news 2026/8/3 23:20:19

mcp-server高级技巧：如何用Python脚本批量获取历史股价并生成分析报告

【免费下载链接】mcp-serverAn MCP server for interacting with the Financial Datasets stock market API.项目地址: https://gitcode.com/gh_mirrors/mcp/mcp-server

mcp-server是一个用于与Financial Datasets股票市场API交互的MCP服务器，通过它可以轻松获取各类金融数据。本文将分享使用Python脚本批量获取历史股价并生成分析报告的实用技巧，帮助投资者和开发者更高效地利用金融数据进行决策。

准备工作：快速搭建mcp-server环境

要开始使用mcp-server获取历史股价数据，首先需要搭建运行环境。按照以下步骤操作，只需几分钟即可完成配置：

一键安装步骤

克隆仓库：执行以下命令获取项目代码

git clone https://gitcode.com/gh_mirrors/mcp/mcp-server cd mcp-server

安装依赖：项目使用uv进行依赖管理，运行以下命令安装所需包
```
uv install
```
配置API密钥：创建.env文件并添加Financial Datasets API密钥
```
FINANCIAL_DATASETS_API_KEY=your_api_key_here
```

完成以上步骤后，即可启动mcp-server服务：

python server.py

核心功能解析：mcp-server的股价数据接口

mcp-server提供了丰富的金融数据接口，其中获取历史股价的核心函数是get_historical_stock_prices。该函数位于server.py文件中，支持灵活的参数配置：

ticker：股票代码（如AAPL、GOOGL）
start_date/end_date：数据时间范围（格式：YYYY-MM-DD）
interval：数据间隔（支持minute、hour、day、week、month）
interval_multiplier：间隔倍数（如2表示每2个单位间隔）

通过这些参数，可以精确获取所需的历史股价数据，为后续分析奠定基础。

实战教程：批量获取历史股价的Python脚本

下面将介绍如何编写Python脚本，利用mcp-server批量获取多个股票的历史数据，并保存为CSV格式以便进一步分析。

步骤1：创建批量请求函数

首先编写一个函数，用于批量获取指定股票列表的历史数据：

import asyncio import json from mcp.client import MCPClient async def batch_fetch_historical_prices(tickers, start_date, end_date): client = MCPClient(transport="stdio") results = {} for ticker in tickers: try: response = await client.get_historical_stock_prices( ticker=ticker, start_date=start_date, end_date=end_date, interval="day", interval_multiplier=1 ) results[ticker] = json.loads(response) except Exception as e: print(f"获取{ticker}数据失败: {str(e)}") return results

步骤2：数据保存与格式转换

将获取的JSON数据转换为CSV格式，方便使用Excel或Pandas进行分析：

import csv def save_to_csv(data, output_file): with open(output_file, 'w', newline='') as f: writer = csv.writer(f) # 写入表头 writer.writerow(["Ticker", "Date", "Open", "High", "Low", "Close", "Volume"]) for ticker, prices in data.items(): for price in prices: writer.writerow([ ticker, price["date"], price["open"], price["high"], price["low"], price["close"], price["volume"] ])

步骤3：执行批量获取并生成报告

组合上述函数，执行批量数据获取并生成简单的分析报告：

def generate_simple_report(csv_file): # 使用Pandas读取数据并生成基本统计信息 import pandas as pd df = pd.read_csv(csv_file) # 计算各股票的价格波动 volatility = df.groupby("Ticker")["Close"].std() # 生成报告 report = f"历史股价分析报告\n" report += f"数据时间范围: {df['Date'].min()} 至 {df['Date'].max()}\n" report += f"包含股票数量: {len(df['Ticker'].unique())}\n\n" report += "股价波动性统计:\n" report += volatility.to_string() with open("price_analysis_report.txt", "w") as f: f.write(report) print("分析报告已生成: price_analysis_report.txt") # 主函数 async def main(): tickers = ["AAPL", "GOOGL", "MSFT", "AMZN"] # 要获取的股票列表 start_date = "2023-01-01" end_date = "2023-12-31" # 批量获取数据 data = await batch_fetch_historical_prices(tickers, start_date, end_date) # 保存为CSV save_to_csv(data, "historical_prices.csv") # 生成分析报告 generate_simple_report("historical_prices.csv") if __name__ == "__main__": asyncio.run(main())

高级技巧：优化数据获取与分析效率

最快配置方法：并发请求处理

为提高批量获取数据的效率，可以使用并发请求。修改batch_fetch_historical_prices函数，利用asyncio.gather实现并行请求：

async def batch_fetch_historical_prices(tickers, start_date, end_date): client = MCPClient(transport="stdio") async def fetch_single(ticker): try: response = await client.get_historical_stock_prices( ticker=ticker, start_date=start_date, end_date=end_date, interval="day", interval_multiplier=1 ) return (ticker, json.loads(response)) except Exception as e: print(f"获取{ticker}数据失败: {str(e)}") return (ticker, None) # 并发执行所有请求 results = await asyncio.gather(*[fetch_single(ticker) for ticker in tickers]) return {ticker: data for ticker, data in results if data is not None}

数据可视化：使用Matplotlib生成价格走势图

结合Matplotlib库，可以快速生成股价走势图，使分析结果更加直观：

import matplotlib.pyplot as plt def plot_price_chart(csv_file): df = pd.read_csv(csv_file) df["Date"] = pd.to_datetime(df["Date"]) plt.figure(figsize=(12, 6)) for ticker in df["Ticker"].unique(): ticker_data = df[df["Ticker"] == ticker] plt.plot(ticker_data["Date"], ticker_data["Close"], label=ticker) plt.title("股票历史收盘价走势") plt.xlabel("日期") plt.ylabel("收盘价 (USD)") plt.legend() plt.grid(True) plt.savefig("price_chart.png") print("价格走势图已生成: price_chart.png")

常见问题解决：数据获取与分析中的痛点

问题1：API请求频率限制

如果获取大量股票数据时遇到API频率限制，可以添加请求间隔控制：

async def fetch_single(ticker): try: # 添加随机间隔避免触发频率限制 await asyncio.sleep(random.uniform(0.5, 1.5)) # 后续请求代码... except Exception as e: # 错误处理...

问题2：数据格式不一致

部分股票数据可能存在格式不一致的问题，可以在保存前进行数据清洗：

def clean_price_data(price_data): cleaned = [] for price in price_data: # 检查必要字段是否存在 required_fields = ["date", "open", "high", "low", "close", "volume"] if all(field in price for field in required_fields): cleaned.append(price) return cleaned