How to Build a Powerful Stock Analysis MCP Server: 4 Advanced Features
Update: This tutorial has evolved into MaverickMCP, a production-ready trading platform with 29 professional tools, 4x faster performance, and one-command setup. Read about the complete transformation here.
Welcome to my guide on building an advanced MCP stock analysis server! In Part 1, we laid the foundation for our Model Context Protocol stock analysis system by creating a basic framework to fetch historical data and calculate essential technical indicators. I'm now ready to transform this stock analysis server from its functional but basic state into a sophisticated trading tool with powerful analytical capabilities. Like upgrading a car's engine from basic factory specs to Daytona 500 performance, this MCP stock analysis server will gain four critical features that serious traders need for data-driven decision-making.
Today, I will soup up that engine by adding the promised features: relative strength calculations, volume profile analysis, pattern recognition, and risk management tools. Think of it as transforming our basic sedan into a high-performance trading machine. Let's dive in!
Reviewing Our MCP Architecture
Before we add new features, let's remind ourselves of our project structure:
mcp-trader/
├── pyproject.toml
├── README.md
├── .env
└── src/
└── mcp-trader/
├── __init__.py
├── server.py # Our core MCP server
├── indicators.py # Technical analysis functions
└── data.py # Data fetching layerI'll expand the indicators.py file with new analysis tools and update our server.py to expose these capabilities to Claude or any other MCP-compatible AI. These days, you can even rig this server to fetch stock recommendations without even leaving your IDE, if you like. We live in interesting times, friends. If you haven't already set up the project from Part 1, make sure to check it out first.
Feature 1: Relative Strength Calculations
Any experienced trader knows that it's not just how a stock performs in isolation, but how it performs compared to the broader market or its sector that matters. That's what relative strength is all about – measuring outperformance or underperformance.
Let's add this crucial metric to our indicators.py file:
class RelativeStrength:
"""Tools for calculating relative strength metrics."""
@staticmethod
async def calculate_rs(
market_data,
symbol: str,
benchmark: str = "SPY",
lookback_periods: List[int] = [21, 63, 126, 252],
) -> Dict[str, float]:
"""
Calculate relative strength compared to a benchmark across multiple timeframes.
Args:
market_data: Our market data fetcher instance
symbol (str): The stock symbol to analyze
benchmark (str): The benchmark symbol (default: SPY for S&P 500 ETF)
lookback_periods (List[int]): Periods in trading days to calculate RS
Returns:
Dict[str, float]: Relative strength scores for each timeframe
"""
try:
# Get data for both the stock and benchmark
stock_df = await market_data.get_historical_data(
symbol, max(lookback_periods) + 10
)
benchmark_df = await market_data.get_historical_data(
benchmark, max(lookback_periods) + 10
)
# Calculate returns for different periods
rs_scores = {}
for period in lookback_periods:
# Check if we have enough data for this period
if len(stock_df) <= period or len(benchmark_df) <= period:
continue
# Calculate the percent change for both
stock_return = (
stock_df["close"].iloc[-1] / stock_df["close"].iloc[-period] - 1
) * 100
benchmark_return = (
benchmark_df["close"].iloc[-1] / benchmark_df["close"].iloc[-period]
- 1
) * 100
# Calculate relative strength (stock return minus benchmark return)
relative_performance = stock_return - benchmark_return
# Convert to a 1-100 score
rs_score = min(max(50 + relative_performance, 1), 99)
rs_scores[f"RS_{period}d"] = round(rs_score, 2)
rs_scores[f"Return_{period}d"] = round(stock_return, 2)
rs_scores[f"Benchmark_{period}d"] = round(benchmark_return, 2)
rs_scores[f"Excess_{period}d"] = round(relative_performance, 2)
return rs_scores
except Exception as e:
raise Exception(f"Error calculating relative strength: {str(e)}")This method calculates the stock's performance versus a benchmark (typically the S&P 500) over different time periods. The result is a standardized score that helps identify market leaders (scores above 80) and laggards.
Feature 2: Volume Profile Analysis
Volume profile analysis helps us understand where most trading activity occurs. It can identify key support and resistance levels based on trading volume at specific price points – areas where traders have shown significant interest.
Let's add this to our indicators.py file:
class VolumeProfile:
"""Tools for analyzing volume distribution by price."""
@staticmethod
def analyze_volume_profile(df: pd.DataFrame, num_bins: int = 10) -> Dict[str, Any]:
"""
Create a volume profile analysis by price level.
Args:
df (pd.DataFrame): Historical price and volume data
num_bins (int): Number of price bins to create (default: 10)
Returns:
Dict[str, Any]: Volume profile analysis
"""
try:
if len(df) < 20:
raise ValueError("Not enough data for volume profile analysis")
# Find the price range for the period
price_min = df["low"].min()
price_max = df["high"].max()
# Create price bins
bin_width = (price_max - price_min) / num_bins
# Initialize the profile
profile = {
"price_min": price_min,
"price_max": price_max,
"bin_width": bin_width,
"bins": [],
}
# Calculate volume by price bin
for i in range(num_bins):
bin_low = price_min + i * bin_width
bin_high = bin_low + bin_width
bin_mid = (bin_low + bin_high) / 2
# Filter data in this price range
mask = (df["low"] <= bin_high) & (df["high"] >= bin_low)
volume_in_bin = df.loc[mask, "volume"].sum()
# Calculate percentage of total volume
volume_percent = (
(volume_in_bin / df["volume"].sum()) * 100
if df["volume"].sum() > 0
else 0
)
profile["bins"].append({
"price_low": round(bin_low, 2),
"price_high": round(bin_high, 2),
"price_mid": round(bin_mid, 2),
"volume": int(volume_in_bin),
"volume_percent": round(volume_percent, 2),
})
# Find the Point of Control (POC) - highest volume price level
poc_bin = max(profile["bins"], key=lambda x: x["volume"])
profile["point_of_control"] = round(poc_bin["price_mid"], 2)
# Find Value Area (70% of volume)
sorted_bins = sorted(
profile["bins"], key=lambda x: x["volume"], reverse=True
)
cumulative_volume = 0
value_area_bins = []
for bin_data in sorted_bins:
value_area_bins.append(bin_data)
cumulative_volume += bin_data["volume_percent"]
if cumulative_volume >= 70:
break
if value_area_bins:
profile["value_area_low"] = round(
min([b["price_low"] for b in value_area_bins]), 2
)
profile["value_area_high"] = round(
max([b["price_high"] for b in value_area_bins]), 2
)
return profile
except Exception as e:
raise Exception(f"Error analyzing volume profile: {str(e)}")The key outputs here are the Point of Control (POC) – the price level with the highest trading activity – and the Value Area, which encompasses 70% of the total volume. These are crucial for understanding where traders have shown the most interest.
Feature 3: Pattern Recognition
Pattern recognition is where things get interesting. We'll implement detection for some of the most reliable chart patterns that traders look for.
class PatternRecognition:
"""Tools for detecting chart patterns."""
@staticmethod
def detect_patterns(df: pd.DataFrame) -> Dict[str, Any]:
"""
Detect common chart patterns in price data.
Args:
df (pd.DataFrame): Historical price data
Returns:
Dict[str, Any]: Detected patterns and their characteristics
"""
try:
patterns = {"detected": [], "summary": {}}
# Calculate some basic metrics we'll need
df = df.copy()
df["sma_20"] = df["close"].rolling(window=20).mean()
df["sma_50"] = df["close"].rolling(window=50).mean()
df["high_20"] = df["high"].rolling(window=20).max()
df["low_20"] = df["low"].rolling(window=20).min()
# Detect Double Bottom
recent_lows = df["low"].tail(60)
if len(recent_lows) >= 60:
first_half_low = recent_lows.iloc[:30].min()
second_half_low = recent_lows.iloc[30:].min()
middle_high = df["high"].iloc[-45:-15].max()
# Check if lows are within 3% of each other
if abs(first_half_low - second_half_low) / first_half_low < 0.03:
if middle_high > first_half_low * 1.05:
patterns["detected"].append({
"pattern": "Double Bottom",
"type": "bullish",
"confidence": "medium",
"first_low": round(first_half_low, 2),
"second_low": round(second_half_low, 2),
"neckline": round(middle_high, 2),
})
# Detect Breakout
current_price = df["close"].iloc[-1]
resistance_20 = df["high"].iloc[-21:-1].max()
if current_price > resistance_20:
breakout_percent = ((current_price - resistance_20) / resistance_20) * 100
if breakout_percent > 0.5:
patterns["detected"].append({
"pattern": "Breakout",
"type": "bullish",
"confidence": "high" if breakout_percent > 2 else "medium",
"breakout_level": round(resistance_20, 2),
"current_price": round(current_price, 2),
"breakout_percent": round(breakout_percent, 2),
})
# Detect Golden Cross (50 SMA crossing above 200 SMA)
if len(df) >= 200:
df["sma_200"] = df["close"].rolling(window=200).mean()
if (df["sma_50"].iloc[-1] > df["sma_200"].iloc[-1] and
df["sma_50"].iloc[-2] <= df["sma_200"].iloc[-2]):
patterns["detected"].append({
"pattern": "Golden Cross",
"type": "bullish",
"confidence": "high",
"sma_50": round(df["sma_50"].iloc[-1], 2),
"sma_200": round(df["sma_200"].iloc[-1], 2),
})
# Summary
patterns["summary"] = {
"total_patterns": len(patterns["detected"]),
"bullish_patterns": len([p for p in patterns["detected"] if p["type"] == "bullish"]),
"bearish_patterns": len([p for p in patterns["detected"] if p["type"] == "bearish"]),
}
return patterns
except Exception as e:
raise Exception(f"Error detecting patterns: {str(e)}")This implementation detects three important patterns: Double Bottoms (a bullish reversal pattern), Breakouts (momentum signals), and Golden Crosses (long-term bullish signals).
Feature 4: Risk Management Tools
No trading system is complete without proper risk management. Let's add tools to help traders size their positions appropriately and manage their risk.
class RiskManagement:
"""Tools for position sizing and risk management."""
@staticmethod
def calculate_position_size(
account_size: float,
risk_percent: float,
entry_price: float,
stop_loss: float,
) -> Dict[str, Any]:
"""
Calculate the appropriate position size based on risk parameters.
Args:
account_size (float): Total account value
risk_percent (float): Percentage of account to risk (e.g., 1.0 for 1%)
entry_price (float): Planned entry price
stop_loss (float): Stop loss price
Returns:
Dict[str, Any]: Position sizing details
"""
try:
# Calculate the dollar risk
risk_amount = account_size * (risk_percent / 100)
# Calculate the per-share risk
per_share_risk = abs(entry_price - stop_loss)
if per_share_risk == 0:
raise ValueError("Stop loss cannot be equal to entry price")
# Calculate position size
shares = int(risk_amount / per_share_risk)
position_value = shares * entry_price
return {
"shares": shares,
"position_value": round(position_value, 2),
"risk_amount": round(risk_amount, 2),
"per_share_risk": round(per_share_risk, 2),
"risk_reward_at_2r": round(entry_price + (2 * per_share_risk), 2),
"risk_reward_at_3r": round(entry_price + (3 * per_share_risk), 2),
"position_percent": round((position_value / account_size) * 100, 2),
}
except Exception as e:
raise Exception(f"Error calculating position size: {str(e)}")
@staticmethod
def suggest_stop_loss(df: pd.DataFrame, method: str = "atr") -> Dict[str, float]:
"""
Suggest stop loss levels based on different methods.
Args:
df (pd.DataFrame): Historical price data
method (str): Method to use ('atr', 'swing', 'percent')
Returns:
Dict[str, float]: Suggested stop loss levels
"""
try:
current_price = df["close"].iloc[-1]
suggestions = {}
# ATR-based stop (2x ATR)
high_low = df["high"] - df["low"]
high_close = abs(df["high"] - df["close"].shift())
low_close = abs(df["low"] - df["close"].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
atr = tr.rolling(window=14).mean().iloc[-1]
suggestions["atr_stop"] = round(current_price - (2 * atr), 2)
suggestions["atr_value"] = round(atr, 2)
# Swing low stop (lowest low of last 10 days)
swing_low = df["low"].tail(10).min()
suggestions["swing_stop"] = round(swing_low * 0.99, 2)
# Percentage-based stops
suggestions["percent_3"] = round(current_price * 0.97, 2)
suggestions["percent_5"] = round(current_price * 0.95, 2)
suggestions["percent_8"] = round(current_price * 0.92, 2)
return suggestions
except Exception as e:
raise Exception(f"Error suggesting stop loss: {str(e)}")These risk management tools help traders: - Size positions appropriately based on their risk tolerance - Set stop losses using different methodologies (ATR, swing lows, or fixed percentages) - Calculate risk/reward ratios for potential trades
Wiring It All Up in server.py
Now let's expose these new features through our MCP server. We'll add new tools that Claude can call:
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""Handle tool calls from the MCP client."""
if name == "analyze-relative-strength":
symbol = arguments.get("symbol", "").upper()
benchmark = arguments.get("benchmark", "SPY").upper()
rs_data = await RelativeStrength.calculate_rs(
market_data, symbol, benchmark
)
return [types.TextContent(
type="text",
text=json.dumps(rs_data, indent=2)
)]
elif name == "analyze-volume-profile":
symbol = arguments.get("symbol", "").upper()
days = arguments.get("days", 60)
df = await market_data.get_historical_data(symbol, days)
profile = VolumeProfile.analyze_volume_profile(df)
return [types.TextContent(
type="text",
text=json.dumps(profile, indent=2)
)]
elif name == "detect-patterns":
symbol = arguments.get("symbol", "").upper()
df = await market_data.get_historical_data(symbol, 252)
patterns = PatternRecognition.detect_patterns(df)
return [types.TextContent(
type="text",
text=json.dumps(patterns, indent=2)
)]
elif name == "calculate-position-size":
result = RiskManagement.calculate_position_size(
account_size=arguments.get("account_size"),
risk_percent=arguments.get("risk_percent", 1.0),
entry_price=arguments.get("entry_price"),
stop_loss=arguments.get("stop_loss"),
)
return [types.TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
elif name == "suggest-stop-loss":
symbol = arguments.get("symbol", "").upper()
df = await market_data.get_historical_data(symbol, 30)
suggestions = RiskManagement.suggest_stop_loss(df)
return [types.TextContent(
type="text",
text=json.dumps(suggestions, indent=2)
)]Testing Our Enhanced Server
With all these features in place, we can now ask Claude sophisticated trading questions like:
- •"What's the relative strength of
NVDA compared to the QQQ over the last quarter?"
- •"Show me the volume profile for
AAPL and identify the point of control"
- •"Detect any chart patterns forming on
TSLA"
- •"I have a $50,000 account and want to risk 1% on a trade. Entry at $150, stop at $145. How many shares should I buy?"
Wrapping Up
We've transformed our basic MCP stock analysis server into a serious trading tool. The four features we added today – relative strength, volume profile, pattern recognition, and risk management – are staples in any professional trader's toolkit.
In the next part of this series, we'll explore how to add real-time data streaming and alert capabilities. Until then, happy trading!
Remember: These tools are for educational purposes. Always do your own research and consider consulting with a financial advisor before making trading decisions.