股票分析MCP服务器开发教程
本指南将向你展示如何构建一个MCP 服务器，它可以获取股市数据、分析趋势并生成交易建议。
当我开始探索如何将实时股票市场数据与 AI 驱动的见解集成时，我想要一个既强大又灵活的设置。这就是我为什么构建了自己的 模型上下文协议 (MCP) 服务器 。
我使用 Python、 AlphaVantage 获取市场数据和 Claude AI 作为我的分析助手。这个设置允许我获取股票数据、应用技术分析（如 移动平均线和 RSI），并生成交易建议—所有这些都以自动化和结构化的方式进行。
在这篇指南中，我将一步步向您展示我是如何构建它的，这样您也可以创建自己的智能股票市场分析系统。您还可以将其扩展到任何其他用例。这只是我的一个副项目。*
1、简介
为什么我要为股票市场分析构建一个 MCP 服务器？
作为一个喜欢金融市场的 AI 人士，我想创建一个可以帮助我更高效地分析股票的系统。虽然有很多工具可以做到这一点，但我想要一个可定制、自动化并且直接访问 AI 助手的系统——一个可以即时回答我关于股票趋势的问题的系统。
这就是MCP（模型上下文协议）服务器的用武之地。（当前热门话题）。它们允许我创建工具，LLM 如 Claude 可以调用这些工具，这意味着我不再需要自己做计算了。相反，我可以简单地问：
"AAPL 当前的 RSI 是多少？"
然后我的MCP 服务器会获取最新的股票数据，计算 RSI，并立即返回结果。这使得做出明智的交易决策变得更加容易，而无需在多个应用程序和网站之间跳转。
我选择了使用 AlphaVantage API（免费层） 来提取实时股票数据，并将其集成到我的 MCP 服务器中，以便使用自定义构建的 AI 工具分析股票。本指南将向您展示我是如何构建它的，这样您也可以这样做！
本指南涵盖的内容
本指南将向您展示如何构建一个MCP 服务器，它可以获取股市数据、分析趋势并生成交易建议。具体来说，您将学习：
✅ MCP 服务器的工作原理及其用途
✅ 设置您的 Python 环境以开发 MCP
✅ 使用 AlphaVantage API 获取股市数据
✅ 创建分析股票趋势的 MCP 工具（例如，移动平均线、RSI）
✅ 使用 MCP 提示自动执行常见的股票分析任务
✅ 运行和测试您的 MCP 服务器
到结束时，你将拥有一个功能齐全的AI 驱动的股票市场分析工具，您可以使用并扩展它！
必备知识
即使你不是 AI 或金融专家，也可以跟随本指南，但一些基础知识会有所帮助：
💡 Python 基础知识——您应该知道如何安装包、编写函数以及使用 API。
💡 LLMs & MCP——对大型语言模型的工作原理以及它们如何使用外部工具有一个大致了解（我会在讲解 MCP 时解释）。
💡 股票市场概念——您至少应该熟悉“移动平均线”和“RSI”等术语。如果不熟悉，不用担心——我会在实现时解释它们。
2、理解 MCP 服务器
模型上下文协议 (MCP) 服务器是一种专门的系统，它以标准化的方式为大型语言模型 (LLMs) 提供结构化的数据和功能。您可以将其视为AI 模型与外部数据源或工具之间的桥梁。与其仅仅给 LLM 提供静态提示，MCP 服务器允许 AI动态获取实时数据、执行计算并访问存储资源。
例如，MCP 服务器可以用于：
- 获取股票市场数据并计算趋势（就像我们在本教程中做的那样）。
- 处理用户输入，通过运行执行计算或检索特定数据的工具。
- 存储和管理上下文信息，LLM 在对话期间可以使用这些信息。
MCP 服务器提供三个核心功能：
资源：存储和提供文件类数据
MCP 资源充当只读数据源，为 LLM 提供结构化信息。您可以将它们视为REST API GET 请求——它们公开数据但不执行计算。
- 示例：资源可以存储用户配置文件、股票价格历史记录或配置设置，LLM 可以按需加载。
- 代码示例：
@mcp.resource("config://app")
def get_config() -> str:
return "App configuration here"资源：存储和提供文件类数据
工具：LLM 可以调用的功能
MCP 工具让 AI 执行特定任务——类似于API POST 请求。这些函数可以执行计算、与数据库交互或甚至调用外部 API。
- 示例：计算BMI、获取股票价格或处理图像的工具。
- 代码示例：
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
return weight_kg / (height_m ** 2)
提示：特定任务的预定义模板
MCP 提示是可重用的指令模板，有助于 AI 执行结构化交互。这些提示指导 LLM一致响应特定类型请求。
- 示例：代码审查、错误调试或财务分析的提示。
- 代码示例：
@mcp.prompt()
def review_code(code: str) -> str:
return f"Please review this code:\n\n{code}"
虽然 MCP 资源和提示很有用，但MCP 工具是系统中最强大的部分——特别是对于股票市场分析。工具允许 AI：
✅ 从 API 获取实时股票价格。
✅ 对实时数据进行计算，如移动平均线和 RSI。 ✅ 自动化复杂工作流，例如生成股票报告或警报。
由于我们想构建一个AI 驱动的股票分析系统，我们将专注于创建和使用 MCP 工具，这些工具可以获取金融数据、处理股票趋势并提供 AI 驱动的见解。
3、设置开发环境
系统要求：
- 安装了 Python 3.10 或更高版本。
- 您必须使用 Python MCP SDK 1.2.0 或更高版本。
首先，让我们安装 uv 并设置我们的 Python 项目和环境：
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
确保在之后重新启动终端以确保 uv 命令被识别。
现在，让我们创建并设置我们的项目：
# 创建一个新的项目目录
uv init finance
cd finance
# 创建虚拟环境并激活它
uv venv
.venv\Scripts\activate
# 安装依赖项
uv add mcp[cli] httpx
# 创建我们的服务器文件
new-item finance_server.py
4、使用 AlphaVantage API 获取股票市场数据
Alpha Vantage 是一家提供实时和历史金融数据的服务提供商，其 API 提供有关股票、货币、加密货币等的信息。
要访问 Alpha Vantage 的服务，您需要获得一个 API 密钥。首先在他们的网站上注册，之后您将收到一个免费 API 密钥，该密钥每天最多允许 25 次请求。如果您的需求超过此限制，Alpha Vantage 提供高级会员资格以容纳更高的使用量。
一旦您获得了 API 密钥，就可以使用 TIME_SERIES_INTRADAY 端点获取股票的盘中价格数据。此 API 提供盘中时间序列数据，包括开盘价、最高价、最低价、收盘价和成交量信息，并实时更新。您需要指定参数，例如股票符号、数据点之间的所需间隔（选项包括 1 分钟、5 分钟、15 分钟、30 分钟和 60 分钟）以及您的 API 密钥。例如，要获取 NVDA 的 5 分钟间隔数据，您的 API 调用将如下所示：
https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol=NVDA&interval=5min&apikey=YOUR_API_KEY
5、实现用于股票分析的 MCP 工具
5.1 工具 1：移动平均线计算
移动平均线是通过计算某个证券在特定时间段内的平均价格得出的，提供了一个连续更新的平均价格。
短期与长期移动平均线
短期移动平均线，如 5 日或 10 日平均线，反应迅速，能够捕捉近期的价格变化，适合捕捉短期市场趋势。相比之下，长期移动平均线，如 50 日或 200 日平均线，反应较慢，更适合识别长时间跨度的趋势。选择短期还是长期移动平均线取决于投资策略。
实现 calculate_moving_averages 函数
# 技术分析工具
@mcp.tool()
def calculate_moving_averages(symbol: str, short_period: int = 20, long_period: int = 50) -> Dict[str, Any]:
"""
计算符号的短期和长期移动平均线
参数:
symbol: 要分析的股票代码
short_period: 短期移动平均周期（分钟）
long_period: 长期移动平均周期（分钟）
返回:
包含移动平均数据和分析的字典
"""
cache_key = f"{symbol}_1min"
if cache_key not in market_data_cache:
df = AlphaVantageAPI.get_intraday_data(symbol, "1min", outputsize="full")
market_data_cache[cache_key] = MarketData(
symbol=symbol,
interval="1min",
data=df,
last_updated=datetime.now()
)
data = market_data_cache[cache_key].data
# 计算移动平均线
data[f'SMA{short_period}'] = data['close'].rolling(window=short_period).mean()
data[f'SMA{long_period}'] = data['close'].rolling(window=long_period).mean()
# 获取最新值
latest = data.iloc[-1]
current_price = latest['close']
short_ma = latest[f'SMA{short_period}']
long_ma = latest[f'SMA{long_period}']
# 确定信号
if short_ma > long_ma:
signal = "看涨（短期均线上穿长期均线）"
elif short_ma < long_ma:
signal = "看跌（短期均线下穿长期均线）"
else:
signal = "中性（均线相等）"
# 检查最近5个周期内的交叉
last_5 = data.iloc[-5:]
crossover = False
crossover_type = ""
for i in range(1, len(last_5)):
prev = last_5.iloc[i-1]
curr = last_5.iloc[i]
# 黄金交叉（短期均线上穿长期均线）
if prev[f'SMA{short_period}'] <= prev[f'SMA{long_period}'] and curr[f'SMA{short_period}'] > curr[f'SMA{long_period}']:
crossover = True
crossover_type = "黄金交叉（看涨）"
break
# 死亡交叉（短期均线下穿长期均线）
if prev[f'SMA{short_period}'] >= prev[f'SMA{long_period}'] and curr[f'SMA{short_period}'] < curr[f'SMA{long_period}']:
crossover = True
crossover_type = "死亡交叉（看跌）"
break
return {
"symbol": symbol,
"current_price": current_price,
f"SMA{short_period}": short_ma,
f"SMA{long_period}": long_ma,
"signal": signal,
"crossover_detected": crossover,
"crossover_type": crossover_type if crossover else "无",
"analysis": f"""{symbol}的移动平均分析：
当前价格: ${current_price:.2f}
{short_period}-周期SMA: ${short_ma:.2f}
{long_period}-周期SMA: ${long_ma:.2f}
信号: {signal}
近期交叉: {"是 - " + crossover_type if crossover else "否"}
建议: {
"强烈买入" if crossover and crossover_type == "黄金交叉（看涨）" else
"买入" if signal == "看涨（短期均线上穿长期均线）" else
"强烈卖出" if crossover and crossover_type == "死亡交叉（看跌）" else
"卖出" if signal == "看跌（短期均线下穿长期均线）" else
"持有"
}"""
}
检测趋势和交叉：黄金交叉和死亡交叉
移动平均交叉是市场趋势变化的重要事件。当短期移动平均线穿过长期移动平均线时，称为“黄金交叉”，这可能预示着潜在的看涨势头。相反，当短期移动平均线穿过长期移动平均线下方时，称为“死亡交叉”，这可能表明可能出现看跌趋势。这些交叉事件受到交易者的密切关注，因为它们可以提供对未来资产价格走势的洞察。
5.2 工具2：相对强弱指数（RSI）计算
什么是RSI及其重要性？
相对强弱指数（RSI） 是一种技术分析中的动量指标，用于衡量价格变动的速度和幅度。它在0到100之间波动，帮助交易者识别股票或其他金融资产的超买和超卖状态。
RSI通常基于特定时间段（通常是14天）的平均涨幅与平均跌幅之比来计算。如果RSI高于70，则表示该资产可能被高估，可能会出现回调，交易者可能考虑卖出。如果RSI低于30，则表示该资产可能被低估，可能存在买入机会。
实现 calculate_rsi 函数
@mcp.tool()
def calculate_rsi(symbol: str, period: int = 14) -> Dict[str, Any]:
"""
计算股票的相对强弱指数（RSI）
参数:
symbol: 要分析的股票代码
period: RSI计算周期（分钟）
返回:
包含RSI数据和分析的字典
"""
cache_key = f"{symbol}_1min"
if cache_key not in market_data_cache:
df = AlphaVantageAPI.get_intraday_data(symbol, "1min", outputsize="full")
market_data_cache[cache_key] = MarketData(
symbol=symbol,
interval="1min",
data=df,
last_updated=datetime.now()
)
data = market_data_cache[cache_key].data.copy()
# 计算价格变化
delta = data['close'].diff()
# 创建收益和损失序列
gain = delta.copy()
loss = delta.copy()
gain[gain < 0] = 0
loss[loss > 0] = 0
loss = abs(loss)
# 计算平均收益和平均损失
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
# 计算RS和RSI
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
# 获取最新的RSI
latest_rsi = rsi.iloc[-1]
# 确定信号
if latest_rsi < 30:
signal = "超卖（潜在买入机会）"
elif latest_rsi > 70:
signal = "超买（潜在卖出机会）"
else:
signal = "中性"
return {
"symbol": symbol,
"period": period,
"rsi": latest_rsi,
"signal": signal,
"analysis": f"""{symbol}的RSI分析：
{period}-周期RSI: {latest_rsi:.2f}
信号: {signal}
建议: {
"买入" if latest_rsi < 30 else
"卖出" if latest_rsi > 70 else
"持有"
}"""
}
确定超买/超卖条件
- RSI > 70 → 超买：股票可能被高估，可能会出现价格回调。交易者可能考虑卖出。
- RSI < 30 → 超卖：股票可能被低估，可能存在买入机会。
5.3 工具3：交易建议
此工具结合其他两个工具的信息，给出是否应该卖出、持有还是买入资产的建议。
@mcp.tool()
def trade_recommendation(symbol: str) -> Dict[str, Any]:
"""
基于多个指标提供全面的交易建议
参数:
symbol: 要分析的股票代码
返回:
包含交易建议和支持数据的字典
"""
# 计算各个指标
ma_data = calculate_moving_averages(symbol)
rsi_data = calculate_rsi(symbol)
# 提取信号
ma_signal = ma_data["signal"]
ma_crossover = ma_data["crossover_detected"]
ma_crossover_type = ma_data["crossover_type"]
rsi_value = rsi_data["rsi"]
rsi_signal = rsi_data["signal"]
# 确定总体信号强度
signal_strength = 0
# 移动平均贡献
if "看涨" in ma_signal:
signal_strength += 1
elif "看跌" in ma_signal:
signal_strength -= 1
# 交叉贡献
if ma_crossover:
if "黄金" in ma_crossover_type:
signal_strength += 2
elif "死亡" in ma_crossover_type:
signal_strength -= 2
# RSI贡献
if "超卖" in rsi_signal:
signal_strength += 1.5
elif "超买" in rsi_signal:
signal_strength -= 1.5
# 确定最终建议
if signal_strength >= 2:
recommendation = "强烈买入"
elif signal_strength > 0:
recommendation = "买入"
elif signal_strength <= -2:
recommendation = "强烈卖出"
elif signal_strength < 0:
recommendation = "卖出"
else:
recommendation = "持有"
# 计算风险水平（简化版）
risk_level = "中等"
if abs(signal_strength) > 3:
risk_level = "低" # 强信号，风险较低
elif abs(signal_strength) < 1:
risk_level = "高" # 弱信号，风险较高
analysis = f"""#{symbol}的交易建议
## 概要
建议: {recommendation}
风险水平: {risk_level}
信号强度: {signal_strength:.1f} / 4.5
## 技术指标
移动平均线: {ma_signal}
近期交叉: {"是 - " + ma_crossover_type if ma_crossover else "否"}
RSI ({rsi_data["period"]}): {rsi_value:.2f} - {rsi_signal}
## 推理
此建议基于移动平均线分析和RSI指标的组合。
{
f"{ma_crossover_type}提供了明确的方向信号。 " if ma_crossover else ""
}{
f"RSI显示股票处于{rsi_signal.split(' ')[0].lower()}状态。 " if "中性" not in rsi_signal else ""
}
## 行动计划
{
"考虑立即入场并设置止损点在近期低点。目标为下一个阻力位。 " if recommendation in ["买入", "强烈买入"] else ""
}{
"考虑立即离场并设置止盈点在近期高点。目标为下一个支撑位。 " if recommendation in ["卖出", "强烈卖出"] else ""
}{
"继续持有并观察市场动态。 " if recommendation == "持有" else ""
}
``````python
if recommendation == "STRONG BUY" else
"Look for a good entry point on small dips. Set reasonable stop loss." if recommendation == "BUY" else
"Consider immediate exit or setting tight stop losses to protect gains." if recommendation == "STRONG SELL" else
"Start reducing position on strength or set trailing stop losses." if recommendation == "SELL" else
"Monitor the position but no immediate action needed."
}
``````markdown
se
"STRONG SELL" if crossover and crossover_type == "DEATH CROSS (Bearish)" else
"SELL" if signal == "BEARISH (Short MA below Long MA)" else
"HOLD"
}"""
}
@mcp.tool()
def calculate_rsi(symbol: str, period: int = 14) -> Dict[str, Any]:
"""
计算相对强弱指数（RSI）
参数:
symbol: 要分析的股票代码
period: RSI 计算周期（分钟）
返回:
包含 RSI 数据和分析的字典
"""
cache_key = f"{symbol}_1min"
if cache_key not in market_data_cache:
df = AlphaVantageAPI.get_intraday_data(symbol, "1min", outputsize="full")
market_data_cache[cache_key] = MarketData(
symbol=symbol,
interval="1min",
data=df,
last_updated=datetime.now()
)
data = market_data_cache[cache_key].data.copy()
# 计算价格变化
delta = data['close'].diff()
# 创建收益和损失序列
gain = delta.copy()
loss = delta.copy()
gain[gain < 0] = 0
loss[loss > 0] = 0
loss = abs(loss)
# 计算平均收益和损失
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
# 计算 RS 和 RSI
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
# 获取最新的 RSI
latest_rsi = rsi.iloc[-1]
# 确定信号
if latest_rsi < 30:
signal = "超卖（潜在买入机会）"
elif latest_rsi > 70:
signal = "超买（潜在卖出机会）"
else:
signal = "中性"
return {
"symbol": symbol,
"period": period,
"rsi": latest_rsi,
"signal": signal,
"analysis": f"""{symbol} 的 RSI 分析：
{period}-周期 RSI: {latest_rsi:.2f}
信号: {signal}
建议: {
"买入" if latest_rsi < 30 else
"卖出" if latest_rsi > 70 else
"持有"
}"""
}
@mcp.tool()
def trade_recommendation(symbol: str) -> Dict[str, Any]:
"""
根据多个指标提供全面的交易建议
参数:
symbol: 要分析的股票代码
返回:
包含交易建议和支持数据的字典
"""
# 计算各个指标
ma_data = calculate_moving_averages(symbol)
rsi_data = calculate_rsi(symbol)
# 提取信号
ma_signal = ma_data["signal"]
ma_crossover = ma_data["crossover_detected"]
ma_crossover_type = ma_data["crossover_type"]
rsi_value = rsi_data["rsi"]
rsi_signal = rsi_data["signal"]
# 确定总体信号强度
signal_strength = 0
# MA 贡献
if "看涨" in ma_signal:
signal_strength += 1
elif "看跌" in ma_signal:
signal_strength -= 1
# 交叉贡献
if ma_crossover:
if "金叉" in ma_crossover_type:
signal_strength += 2
elif "死叉" in ma_crossover_type:
signal_strength -= 2
# RSI 贡献
if "超卖" in rsi_signal:
signal_strength += 1.5
elif "超买" in rsi_signal:
signal_strength -= 1.5
# 确定最终建议
if signal_strength >= 2:
recommendation = "强烈买入"
elif signal_strength > 0:
recommendation = "买入"
elif signal_strength <= -2:
recommendation = "强烈卖出"
elif signal_strength < 0:
recommendation = "卖出"
else:
recommendation = "持有"
# 计算风险水平（简单版本）
risk_level = "中等"
if abs(signal_strength) > 3:
risk_level = "低" # 强信号，较低风险
elif abs(signal_strength) < 1:
risk_level = "高" # 弱信号，较高风险
analysis = f"""# {symbol} 的交易建议
## 概要
建议: {recommendation}
风险水平: {risk_level}
信号强度: {signal_strength:.1f} / 4.5
## 技术指标
移动平均线: {ma_signal}
最近交叉: {"是 - " + ma_crossover_type if ma_crossover else "否"}
RSI ({rsi_data["period"]}): {rsi_value:.2f} - {rsi_signal}
## 推理
此建议基于移动平均线分析和 RSI 指标。
{
f"{ma_crossover_type} 提供了强烈的趋势信号。 " if ma_crossover else ""
}{
f"RSI 表示该股票 {rsi_signal.split(' ')[0].lower()}。 " if "中性" not in rsi_signal else ""
}
## 行动计划
{
"考虑立即入场，并在近期低点设置止损。目标是下一个阻力位。" if recommendation == "强烈买入" else
"寻找小回调时的良好入场点。设置合理的止损。" if recommendation == "买入" else
"考虑立即退出或设置紧的止损以保护收益。" if recommendation == "强烈卖出" else
"在强势时开始减少仓位或设置跟踪止损。" if recommendation == "卖出" else
"监控头寸但无需立即采取行动。"
}
"""
return {
"symbol": symbol,
"recommendation": recommendation,
"risk_level": risk_level,
"signal_strength": signal_strength,
"ma_signal": ma_signal,
"rsi_signal": rsi_signal,
"current_price": ma_data["current_price"],
"analysis": analysis
}
# 提示
@mcp.prompt()
def analyze_ticker(symbol: str) -> str:
"""
分析股票代码以发现交易机会
"""
return f"""你是一名专业的股票市场分析师。我想让你分析股票 {symbol} 并提供交易见解。
首先检查 {symbol} 的当前市场数据。
然后使用 calculate_moving_averages 工具计算移动平均线。
接着使用 calculate_rsi 工具计算 RSI。
最后使用 trade_recommendation 工具生成全面的交易建议。
基于所有这些信息，请提供你的专业分析，突出显示：
- 当前市场位置
- 关键技术指标及其暗示
- 潜在的交易机会和风险
- 你的推荐操作（买入、卖出或持有）以及简要解释
请以适合专业交易员的清晰结构化格式组织你的回复。"""
@mcp.prompt()
def compare_tickers(symbols: str) -> str:
"""
比较多个股票代码以找到最佳交易机会
参数:
symbols: 逗号分隔的股票代码列表
"""
symbol_list = [s.strip() for s in symbols.split(",")]
symbol_section = "\n".join([f"- {s}" for s in symbol_list])
return f"""你是一名专业的股票市场分析师。我想让你比较这些股票并识别最佳交易机会：
{symbol_section}
对于列表中的每只股票，请执行以下步骤：
1. 使用适当的资源检查当前市场数据
2. 使用 trade_recommendation 工具生成全面的交易建议
3. 基于以下方面比较所有股票：
- 当前趋势方向和强度
- 技术指标信号
- 风险/回报概况
- 交易建议强度
在分析每只股票后，按从最有可能到最不可能的交易机会进行排名。解释你的排名标准并说明为什么你认为排名最高的股票代表了当前的最佳交易机会。
最后，提出一个具体的建议，包括要交易的股票和要采取的操作（买入、卖出或持有）。"""
@mcp.prompt()
def intraday_strategy_builder(symbol: str) -> str:
"""
为特定股票代码构建自定义日内交易策略
"""
return f"""你是一位专注于日内策略的专家算法交易员。我希望你为 {symbol} 开发一个自定义的日内交易策略。
请遵循以下步骤：
1. 首先使用 market-data 资源分析 {symbol} 的当前市场数据
2. 计算相关技术指标：
- 移动平均线（短期和长期）
- RSI
3. 基于你的分析，设计一个日内交易策略，包括：
- 具体的进场条件（触发买入/卖出的技术设置）
- 退出条件（包括盈利目标和止损水平）
- 仓位大小建议
- 日内交易时间的最佳选择
- 风险管理规则
使你的策略具体针对 {symbol} 的当前市场状况，而不仅仅是通用建议。尽可能包含具体的指标值和价格水平。
最后总结你的策略，并说明交易者应如何实施它以应对今天的交易会话。"""
6、创建 MCP 提示以进行自动分析
提示 1：分析单个股票代码
@mcp.prompt()
def analyze_ticker(symbol: str) -> str:
"""
Analyze a ticker symbol for trading opportunities
"""
return f"""You are a professional stock market analyst. I would like you to analyze the stock {symbol} and provide trading insights.
Start by examining the current market data and technical indicators. Here are the specific tasks:
1. First, check the current market data for {symbol}
2. Calculate the moving averages using the calculate_moving_averages tool
3. Calculate the RSI using the calculate_rsi tool
4. Generate a comprehensive trade recommendation using the trade_recommendation tool
5. Based on all this information, provide your professional analysis, highlighting:
- The current market position
- Key technical indicators and what they suggest
- Potential trading opportunities and risks
- Your recommended action (buy, sell, or hold) with a brief explanation
Please organize your response in a clear, structured format suitable for a professional trader."""提示2：比较多支股票：
@mcp.prompt()
def compare_tickers(symbols: str) -> str:
"""
Compare multiple ticker symbols for the best trading opportunity
Args:
symbols: Comma-separated list of ticker symbols
"""
symbol_list = [s.strip() for s in symbols.split(",")]
symbol_section = "\n".join([f"- {s}" for s in symbol_list])
return f"""You are a professional stock market analyst. I would like you to compare these stocks and identify the best trading opportunity:
{symbol_section}
For each stock in the list, please:
1. Check the current market data using the appropriate resource
2. Generate a comprehensive trade recommendation using the trade_recommendation tool
3. Compare all stocks based on:
- Current trend direction and strength
- Technical indicator signals
- Risk/reward profile
- Trading recommendation strength
After analyzing each stock, rank them from most promising to least promising trading opportunity. Explain your ranking criteria and why you believe the top-ranked stock represents the best current trading opportunity.
Conclude with a specific recommendation on which stock to trade and what action to take (buy, sell, or hold)."""提示3：构建日间交易策略：
@mcp.prompt()
def intraday_strategy_builder(symbol: str) -> str:
"""
Build a custom intraday trading strategy for a specific ticker
"""
return f"""You are an expert algorithmic trader specializing in intraday strategies. I want you to develop a custom intraday trading strategy for {symbol}.
Please follow these steps:
1. First, analyze the current market data for {symbol} using the market-data resource
2. Calculate relevant technical indicators:
- Moving averages (short and long periods)
- RSI
3. Based on your analysis, design an intraday trading strategy that includes:
- Specific entry conditions (technical setups that would trigger a buy/sell)
- Exit conditions (both take-profit and stop-loss levels)
- Position sizing recommendations
- Optimal trading times during the day
- Risk management rules
Make your strategy specific to the current market conditions for {symbol}, not just generic advice. Include exact indicator values and price levels where possible.
Conclude with a summary of the strategy and how a trader should implement it for today's trading session."""7、将 MCP 服务器连接到客户端
完整代码文件：
# finance_server.py
from mcp.server.fastmcp import FastMCP
import requests
import pandas as pd
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Any
# Create the MCP server
mcp = FastMCP("AlphaVantageTrader", dependencies=["requests", "pandas", "tabulate"])
# Constants and configurations
API_KEY = "Your API KEY" # Replace with your actual AlphaVantage API key
@dataclass
class MarketData:
symbol: str
interval: str
data: pd.DataFrame
last_updated: datetime
class AlphaVantageAPI:
@staticmethod
def get_intraday_data(symbol: str, interval: str = "1min", outputsize: str = "compact") -> pd.DataFrame:
"""Fetch intraday data from AlphaVantage API"""
url = f"https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol={symbol}&interval={interval}&outputsize={outputsize}&apikey={API_KEY}"
response = requests.get(url)
data = response.json()
# Check for error responses
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
if "Note" in data:
print(f"API Note: {data['Note']}")
# Extract time series data
time_series_key = f"Time Series ({interval})"
if time_series_key not in data:
raise ValueError(f"No time series data found for {symbol} with interval {interval}")
time_series = data[time_series_key]
# Convert to DataFrame
df = pd.DataFrame.from_dict(time_series, orient="index")
df.index = pd.to_datetime(df.index)
df = df.sort_index()
# Rename columns and convert to numeric
df.columns = [col.split(". ")[1] for col in df.columns]
for col in df.columns:
df[col] = pd.to_numeric(df[col])
return df
# In-memory cache for market data
market_data_cache: Dict[str, MarketData] = {}
# Resources
@mcp.resource("config://app")
def get_config() -> str:
"""Static configuration data"""
return "App configuration here"
# Technical Analysis Tools
@mcp.tool()
def calculate_moving_averages(symbol: str, short_period: int = 20, long_period: int = 50) -> Dict[str, Any]:
"""
Calculate short and long moving averages for a symbol
Args:
symbol: The ticker symbol to analyze
short_period: Short moving average period in minutes
long_period: Long moving average period in minutes
Returns:
Dictionary with moving average data and analysis
"""
cache_key = f"{symbol}_1min"
if cache_key not in market_data_cache:
df = AlphaVantageAPI.get_intraday_data(symbol, "1min", outputsize="full")
market_data_cache[cache_key] = MarketData(
symbol=symbol,
interval="1min",
data=df,
last_updated=datetime.now()
)
data = market_data_cache[cache_key].data
# Calculate moving averages
data[f'SMA{short_period}'] = data['close'].rolling(window=short_period).mean()
data[f'SMA{long_period}'] = data['close'].rolling(window=long_period).mean()
# Get latest values
latest = data.iloc[-1]
current_price = latest['close']
short_ma = latest[f'SMA{short_period}']
long_ma = latest[f'SMA{long_period}']
# Determine signal
if short_ma > long_ma:
signal = "BULLISH (Short MA above Long MA)"
elif short_ma < long_ma:
signal = "BEARISH (Short MA below Long MA)"
else:
signal = "NEUTRAL (MAs are equal)"
# Check for crossover in the last 5 periods
last_5 = data.iloc[-5:]
crossover = False
crossover_type = ""
for i in range(1, len(last_5)):
prev = last_5.iloc[i-1]
curr = last_5.iloc[i]
# Golden Cross (short crosses above long)
if prev[f'SMA{short_period}'] <= prev[f'SMA{long_period}'] and curr[f'SMA{short_period}'] > curr[f'SMA{long_period}']:
crossover = True
crossover_type = "GOLDEN CROSS (Bullish)"
break
# Death Cross (short crosses below long)
if prev[f'SMA{short_period}'] >= prev[f'SMA{long_period}'] and curr[f'SMA{short_period}'] < curr[f'SMA{long_period}']:
crossover = True
crossover_type = "DEATH CROSS (Bearish)"
break
return {
"symbol": symbol,
"current_price": current_price,
f"SMA{short_period}": short_ma,
f"SMA{long_period}": long_ma,
"signal": signal,
"crossover_detected": crossover,
"crossover_type": crossover_type if crossover else "None",
"analysis": f"""Moving Average Analysis for {symbol}:
Current Price: ${current_price:.2f}
{short_period}-period SMA: ${short_ma:.2f}
{long_period}-period SMA: ${long_ma:.2f}
Signal: {signal}
Recent Crossover: {"Yes - " + crossover_type if crossover else "No"}
Recommendation: {
"STRONG BUY" if crossover and crossover_type == "GOLDEN CROSS (Bullish)" else
"BUY" if signal == "BULLISH (Short MA above Long MA)" else
"STRONG SELL" if crossover and crossover_type == "DEATH CROSS (Bearish)" else
"SELL" if signal == "BEARISH (Short MA below Long MA)" else
"HOLD"
}"""
}
@mcp.tool()
def calculate_rsi(symbol: str, period: int = 14) -> Dict[str, Any]:
"""
Calculate Relative Strength Index (RSI) for a symbol
Args:
symbol: The ticker symbol to analyze
period: RSI calculation period in minutes
Returns:
Dictionary with RSI data and analysis
"""
cache_key = f"{symbol}_1min"
if cache_key not in market_data_cache:
df = AlphaVantageAPI.get_intraday_data(symbol, "1min", outputsize="full")
market_data_cache[cache_key] = MarketData(
symbol=symbol,
interval="1min",
data=df,
last_updated=datetime.now()
)
data = market_data_cache[cache_key].data.copy()
# Calculate price changes
delta = data['close'].diff()
# Create gain and loss series
gain = delta.copy()
loss = delta.copy()
gain[gain < 0] = 0
loss[loss > 0] = 0
loss = abs(loss)
# Calculate average gain and loss
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
# Calculate RS and RSI
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
# Get latest RSI
latest_rsi = rsi.iloc[-1]
# Determine signal
if latest_rsi < 30:
signal = "OVERSOLD (Potential buy opportunity)"
elif latest_rsi > 70:
signal = "OVERBOUGHT (Potential sell opportunity)"
else:
signal = "NEUTRAL"
return {
"symbol": symbol,
"period": period,
"rsi": latest_rsi,
"signal": signal,
"analysis": f"""RSI Analysis for {symbol}:
{period}-period RSI: {latest_rsi:.2f}
Signal: {signal}
Recommendation: {
"BUY" if latest_rsi < 30 else
"SELL" if latest_rsi > 70 else
"HOLD"
}"""
}
@mcp.tool()
def trade_recommendation(symbol: str) -> Dict[str, Any]:
"""
Provide a comprehensive trade recommendation based on multiple indicators
Args:
symbol: The ticker symbol to analyze
Returns:
Dictionary with trading recommendation and supporting data
"""
# Calculate individual indicators
ma_data = calculate_moving_averages(symbol)
rsi_data = calculate_rsi(symbol)
# Extract signals
ma_signal = ma_data["signal"]
ma_crossover = ma_data["crossover_detected"]
ma_crossover_type = ma_data["crossover_type"]
rsi_value = rsi_data["rsi"]
rsi_signal = rsi_data["signal"]
# Determine overall signal strength
signal_strength = 0
# MA contribution
if "BULLISH" in ma_signal:
signal_strength += 1
elif "BEARISH" in ma_signal:
signal_strength -= 1
# Crossover contribution
if ma_crossover:
if "GOLDEN" in ma_crossover_type:
signal_strength += 2
elif "DEATH" in ma_crossover_type:
signal_strength -= 2
# RSI contribution
if "OVERSOLD" in rsi_signal:
signal_strength += 1.5
elif "OVERBOUGHT" in rsi_signal:
signal_strength -= 1.5
# Determine final recommendation
if signal_strength >= 2:
recommendation = "STRONG BUY"
elif signal_strength > 0:
recommendation = "BUY"
elif signal_strength <= -2:
recommendation = "STRONG SELL"
elif signal_strength < 0:
recommendation = "SELL"
else:
recommendation = "HOLD"
# Calculate risk level (simple version)
risk_level = "MEDIUM"
if abs(signal_strength) > 3:
risk_level = "LOW" # Strong signal, lower risk
elif abs(signal_strength) < 1:
risk_level = "HIGH" # Weak signal, higher risk
analysis = f"""# Trading Recommendation for {symbol}
## Summary
Recommendation: {recommendation}
Risk Level: {risk_level}
Signal Strength: {signal_strength:.1f} / 4.5
## Technical Indicators
Moving Averages: {ma_signal}
Recent Crossover: {"Yes - " + ma_crossover_type if ma_crossover else "No"}
RSI ({rsi_data["period"]}): {rsi_value:.2f} - {rsi_signal}
## Reasoning
This recommendation is based on a combination of Moving Average analysis and RSI indicators.
{
f"The {ma_crossover_type} provides a strong directional signal. " if ma_crossover else ""
}{
f"The RSI indicates the stock is {rsi_signal.split(' ')[0].lower()}. " if "NEUTRAL" not in rsi_signal else ""
}
## Action Plan
{
"Consider immediate entry with a stop loss at the recent low. Target the next resistance level." if recommendation == "STRONG BUY" else
"Look for a good entry point on small dips. Set reasonable stop loss." if recommendation == "BUY" else
"Consider immediate exit or setting tight stop losses to protect gains." if recommendation == "STRONG SELL" else
"Start reducing position on strength or set trailing stop losses." if recommendation == "SELL" else
"Monitor the position but no immediate action needed."
}
"""
return {
"symbol": symbol,
"recommendation": recommendation,
"risk_level": risk_level,
"signal_strength": signal_strength,
"ma_signal": ma_signal,
"rsi_signal": rsi_signal,
"current_price": ma_data["current_price"],
"analysis": analysis
}
# Prompts
@mcp.prompt()
def analyze_ticker(symbol: str) -> str:
"""
Analyze a ticker symbol for trading opportunities
"""
return f"""You are a professional stock market analyst. I would like you to analyze the stock {symbol} and provide trading insights.
Start by examining the current market data and technical indicators. Here are the specific tasks:
1. First, check the current market data for {symbol}
2. Calculate the moving averages using the calculate_moving_averages tool
3. Calculate the RSI using the calculate_rsi tool
4. Generate a comprehensive trade recommendation using the trade_recommendation tool
5. Based on all this information, provide your professional analysis, highlighting:
- The current market position
- Key technical indicators and what they suggest
- Potential trading opportunities and risks
- Your recommended action (buy, sell, or hold) with a brief explanation
Please organize your response in a clear, structured format suitable for a professional trader."""
@mcp.prompt()
def compare_tickers(symbols: str) -> str:
"""
Compare multiple ticker symbols for the best trading opportunity
Args:
symbols: Comma-separated list of ticker symbols
"""
symbol_list = [s.strip() for s in symbols.split(",")]
symbol_section = "\n".join([f"- {s}" for s in symbol_list])
return f"""You are a professional stock market analyst. I would like you to compare these stocks and identify the best trading opportunity:
{symbol_section}
For each stock in the list, please:
1. Check the current market data using the appropriate resource
2. Generate a comprehensive trade recommendation using the trade_recommendation tool
3. Compare all stocks based on:
- Current trend direction and strength
- Technical indicator signals
- Risk/reward profile
- Trading recommendation strength
After analyzing each stock, rank them from most promising to least promising trading opportunity. Explain your ranking criteria and why you believe the top-ranked stock represents the best current trading opportunity.
Conclude with a specific recommendation on which stock to trade and what action to take (buy, sell, or hold)."""
@mcp.prompt()
def intraday_strategy_builder(symbol: str) -> str:
"""
Build a custom intraday trading strategy for a specific ticker
"""
return f"""You are an expert algorithmic trader specializing in intraday strategies. I want you to develop a custom intraday trading strategy for {symbol}.
Please follow these steps:
1. First, analyze the current market data for {symbol} using the market-data resource
2. Calculate relevant technical indicators:
- Moving averages (short and long periods)
- RSI
3. Based on your analysis, design an intraday trading strategy that includes:
- Specific entry conditions (technical setups that would trigger a buy/sell)
- Exit conditions (both take-profit and stop-loss levels)
- Position sizing recommendations
- Optimal trading times during the day
- Risk management rules
Make your strategy specific to the current market conditions for {symbol}, not just generic advice. Include exact indicator values and price levels where possible.
Conclude with a summary of the strategy and how a trader should implement it for today's trading session."""配置 Claude AI Desktop 运行 MCP 服务器
导航至以下目录：
“C:\Users\username\AppData\Roaming\Claude\claude_desktop_config.json”
{
"mcpServers": {
"AlphaVantageTrader": {
"command": "uv",
"args": [
"run",
"--with",
"mcp[cli]",
"--with",
"pandas",
"--with",
"requests",
"--with",
"tabulate",
"mcp",
"run",
"ABSOLUTE PATH TO YOUR SERVER.PY FILE"
],
"capabilities": {
"resources": {
"subscribe": true,
"listChanged": true
}
}
}
}
}
或者在终端运行如下命令：
mcp install finance_server.py现在，先退出claude AI桌面应用程序并重新启动。您应该能够在claude桌面应用程序中看到工具图标。
输入内容后即可开始使用。
8、结束语
我们已经构建了一个提供股票市场洞察的模型上下文协议（MCP）服务器，使用了AlphaVantage数据和技术指标，如移动平均线和RSI。我们的MCP服务器提供了：
- 资源以获取股票价格数据。
- 工具以计算移动平均线和RSI等指标。
- 提示允许与LLM进行交互以获得见解。
通过将此MCP服务器与Claude AI桌面集成，我们创建了一个强大的系统，其中LLM可以获取实时股票数据、分析趋势并协助交易决策。
原文链接：Step-by-Step Guide: Building an MCP Server using Python-SDK, AlphaVantage & Claude AI
汇智网翻译整理，转载请标明出处
