#!/usr/bin/env python3
"""
Financial Data MCP Server
Provides tools for accessing stock prices, options data, and calculating financial metrics
"""
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
import time
from functools import wraps
import sys
import yfinance as yf
import pandas as pd
import numpy as np
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("financial-data-server")
cache: Dict[str, tuple[Any, float]] = {}
CACHE_DURATION = 300
last_request_time = 0
MIN_REQUEST_INTERVAL = 0.5
rate_limit_lock = asyncio.Lock()
VALID_PERIODS = {
"1d",
"5d",
"1mo",
"3mo",
"6mo",
"1y",
"2y",
"5y",
"10y",
"ytd",
"max",
}
VALID_INTERVALS = {
"1m",
"2m",
"5m",
"15m",
"30m",
"60m",
"90m",
"1h",
"1d",
"5d",
"1wk",
"1mo",
"3mo",
}
MAX_SYMBOLS_COMPARE = 5
OPTIONS_SORT_FIELDS = {
"volume",
"openInterest",
"impliedVolatility",
"lastPrice",
"bid",
"ask",
}
def rate_limit(func):
"""Simple rate limiting decorator"""
@wraps(func)
async def wrapper(*args, **kwargs):
global last_request_time
async with rate_limit_lock:
current_time = time.time()
time_since_last = current_time - last_request_time
if time_since_last < MIN_REQUEST_INTERVAL:
await asyncio.sleep(MIN_REQUEST_INTERVAL - time_since_last)
last_request_time = time.time()
return await func(*args, **kwargs)
return wrapper
def get_cache_key(func_name: str, *args, **kwargs) -> str:
"""Generate cache key from function name and arguments"""
return f"{func_name}:{str(args)}:{str(kwargs)}"
def cached(duration: int = CACHE_DURATION):
"""Cache decorator with configurable duration"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
cache_key = get_cache_key(func.__name__, *args, **kwargs)
if cache_key in cache:
result, timestamp = cache[cache_key]
if time.time() - timestamp < duration:
return result
result = await func(*args, **kwargs)
cache[cache_key] = (result, time.time())
return result
return wrapper
return decorator
def _normalize_symbol(symbol: str) -> str:
return symbol.strip().upper()
def _is_valid_symbol(symbol: str) -> bool:
if not symbol:
return False
allowed = set("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-^=")
return all(ch in allowed for ch in symbol.upper())
@mcp.tool()
@rate_limit
@cached(duration=60)
async def get_stock_price(symbol: str) -> dict:
"""Get current stock price and basic info.
Args:
symbol: Stock ticker symbol (e.g., 'AAPL', 'GOOGL')
Returns:
Dictionary with price data and basic info
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
ticker = yf.Ticker(symbol)
info = await asyncio.to_thread(lambda: ticker.info)
return {
"symbol": symbol,
"current_price": info.get(
"currentPrice", info.get("regularMarketPrice", "N/A")
),
"previous_close": info.get("previousClose", "N/A"),
"day_high": info.get("dayHigh", "N/A"),
"day_low": info.get("dayLow", "N/A"),
"volume": info.get("volume", "N/A"),
"market_cap": info.get("marketCap", "N/A"),
"pe_ratio": info.get("trailingPE", "N/A"),
"52_week_high": info.get("fiftyTwoWeekHigh", "N/A"),
"52_week_low": info.get("fiftyTwoWeekLow", "N/A"),
"name": info.get("longName", symbol),
}
except Exception as e:
return {"error": f"Failed to fetch data for {symbol}: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=300)
async def get_historical_data(
symbol: str, period: str = "1mo", interval: str = "1d"
) -> dict:
"""Get historical price data for a stock.
Args:
symbol: Stock ticker symbol
period: Time period (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)
Returns:
Dictionary with historical price data
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if period not in VALID_PERIODS:
return {"error": f"Invalid period. Allowed: {sorted(VALID_PERIODS)}"}
if interval not in VALID_INTERVALS:
return {"error": f"Invalid interval. Allowed: {sorted(VALID_INTERVALS)}"}
ticker = yf.Ticker(symbol)
hist = await asyncio.to_thread(ticker.history, period=period, interval=interval)
if hist.empty:
return {"error": f"No historical data available for {symbol}"}
data = []
for date, row in hist.iterrows():
data.append(
{
"date": date.strftime("%Y-%m-%d"),
"open": (
round(float(row["Open"]), 2) if pd.notna(row["Open"]) else None
),
"high": (
round(float(row["High"]), 2) if pd.notna(row["High"]) else None
),
"low": (
round(float(row["Low"]), 2) if pd.notna(row["Low"]) else None
),
"close": (
round(float(row["Close"]), 2)
if pd.notna(row["Close"])
else None
),
"volume": int(row["Volume"]) if pd.notna(row["Volume"]) else 0,
}
)
return {
"symbol": symbol,
"period": period,
"interval": interval,
"data": data,
}
except Exception as e:
return {"error": f"Failed to fetch historical data for {symbol}: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=300)
async def get_options_chain(
symbol: str,
expiration_date: Optional[str] = None,
sort_by: Optional[str] = "volume",
limit: int = 20,
descending: bool = True,
) -> dict:
"""Get options chain data for a stock.
Args:
symbol: Stock ticker symbol
expiration_date: Optional expiration date (YYYY-MM-DD format). If not provided, uses nearest expiration.
Returns:
Dictionary with calls and puts data
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if sort_by is not None and sort_by not in OPTIONS_SORT_FIELDS:
return {
"error": f"Invalid sort field. Allowed: {sorted(OPTIONS_SORT_FIELDS)}"
}
if limit <= 0:
return {"error": "Limit must be positive"}
ticker = yf.Ticker(symbol)
expirations = await asyncio.to_thread(lambda: ticker.options)
if not expirations:
return {"error": f"No options data available for {symbol}"}
if expiration_date:
if expiration_date not in expirations:
return {
"error": f"Expiration date {expiration_date} not available",
"available_dates": list(expirations),
}
exp_date = expiration_date
else:
exp_date = expirations[0]
opt = await asyncio.to_thread(ticker.option_chain, exp_date)
calls_df = opt.calls
puts_df = opt.puts
if sort_by is not None and sort_by in calls_df.columns:
calls_df = calls_df.sort_values(
by=sort_by, ascending=not descending, na_position="last"
)
if sort_by is not None and sort_by in puts_df.columns:
puts_df = puts_df.sort_values(
by=sort_by, ascending=not descending, na_position="last"
)
calls = []
for _, row in calls_df.head(limit).iterrows():
calls.append(
{
"strike": float(row["strike"]),
"last_price": (
float(row["lastPrice"]) if pd.notna(row["lastPrice"]) else 0
),
"bid": float(row["bid"]) if pd.notna(row["bid"]) else 0,
"ask": float(row["ask"]) if pd.notna(row["ask"]) else 0,
"volume": int(row["volume"]) if pd.notna(row["volume"]) else 0,
"open_interest": (
int(row["openInterest"]) if pd.notna(row["openInterest"]) else 0
),
"implied_volatility": (
float(row["impliedVolatility"])
if pd.notna(row["impliedVolatility"])
else 0
),
}
)
puts = []
for _, row in puts_df.head(limit).iterrows():
puts.append(
{
"strike": float(row["strike"]),
"last_price": (
float(row["lastPrice"]) if pd.notna(row["lastPrice"]) else 0
),
"bid": float(row["bid"]) if pd.notna(row["bid"]) else 0,
"ask": float(row["ask"]) if pd.notna(row["ask"]) else 0,
"volume": int(row["volume"]) if pd.notna(row["volume"]) else 0,
"open_interest": (
int(row["openInterest"]) if pd.notna(row["openInterest"]) else 0
),
"implied_volatility": (
float(row["impliedVolatility"])
if pd.notna(row["impliedVolatility"])
else 0
),
}
)
return {
"symbol": symbol,
"expiration_date": exp_date,
"available_expirations": list(expirations)[:10],
"calls": calls,
"puts": puts,
}
except Exception as e:
return {"error": f"Failed to fetch options data for {symbol}: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=120)
async def calculate_moving_average(
symbol: str, period: int = 20, ma_type: str = "SMA"
) -> dict:
"""Calculate moving average for a stock.
Args:
symbol: Stock ticker symbol
period: Number of periods for the moving average
ma_type: Type of moving average (SMA or EMA)
Returns:
Dictionary with moving average data
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if period <= 0:
return {"error": "Period must be positive"}
ticker = yf.Ticker(symbol)
hist = await asyncio.to_thread(ticker.history, period="3mo")
if hist.empty:
return {"error": f"No data available for {symbol}"}
close_prices = hist["Close"]
if ma_type.upper() == "SMA":
ma = close_prices.rolling(window=period).mean()
elif ma_type.upper() == "EMA":
ma = close_prices.ewm(span=period, adjust=False).mean()
else:
return {"error": "Invalid MA type. Use 'SMA' or 'EMA'"}
recent_data = []
for date, value in ma.tail(10).items():
if pd.notna(value):
recent_data.append(
{"date": date.strftime("%Y-%m-%d"), "value": round(value, 2)}
)
return {
"symbol": symbol,
"period": period,
"type": ma_type.upper(),
"current_value": round(ma.iloc[-1], 2) if pd.notna(ma.iloc[-1]) else None,
"recent_values": recent_data,
}
except Exception as e:
return {"error": f"Failed to calculate moving average: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=120)
async def calculate_rsi(symbol: str, period: int = 14) -> dict:
"""Calculate Relative Strength Index (RSI) for a stock.
Args:
symbol: Stock ticker symbol
period: Number of periods for RSI calculation (default: 14)
Returns:
Dictionary with RSI data
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if period <= 0:
return {"error": "Period must be positive"}
ticker = yf.Ticker(symbol)
hist = await asyncio.to_thread(ticker.history, period="3mo")
if hist.empty:
return {"error": f"No data available for {symbol}"}
close_prices = hist["Close"]
delta = close_prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
recent_data = []
for date, value in rsi.tail(10).items():
if pd.notna(value):
recent_data.append(
{"date": date.strftime("%Y-%m-%d"), "value": round(value, 2)}
)
current_rsi = rsi.iloc[-1] if pd.notna(rsi.iloc[-1]) else None
interpretation = "Neutral"
if current_rsi and current_rsi > 70:
interpretation = "Overbought"
elif current_rsi and current_rsi < 30:
interpretation = "Oversold"
return {
"symbol": symbol,
"period": period,
"current_rsi": round(current_rsi, 2) if current_rsi else None,
"interpretation": interpretation,
"recent_values": recent_data,
}
except Exception as e:
return {"error": f"Failed to calculate RSI: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=300)
async def calculate_sharpe_ratio(
symbol: str, period: str = "1y", risk_free_rate: float = 0.05
) -> dict:
"""Calculate Sharpe Ratio for a stock.
Args:
symbol: Stock ticker symbol
period: Time period for calculation (1mo, 3mo, 6mo, 1y, 2y, 5y)
risk_free_rate: Annual risk-free rate (default: 0.05 or 5%)
Returns:
Dictionary with Sharpe Ratio and related metrics
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if period not in VALID_PERIODS:
return {"error": f"Invalid period. Allowed: {sorted(VALID_PERIODS)}"}
if not (-1.0 <= risk_free_rate <= 1.0):
return {"error": "risk_free_rate should be a decimal (e.g., 0.05 for 5%)"}
ticker = yf.Ticker(symbol)
hist = await asyncio.to_thread(ticker.history, period=period)
if hist.empty or len(hist) < 20:
return {"error": f"Insufficient data for {symbol}"}
daily_returns = hist["Close"].pct_change().dropna()
trading_days = 252
avg_daily_return = daily_returns.mean()
daily_volatility = daily_returns.std()
annualized_return = avg_daily_return * trading_days
annualized_volatility = daily_volatility * np.sqrt(trading_days)
sharpe_ratio = None
if annualized_volatility > 0:
sharpe_ratio = (annualized_return - risk_free_rate) / annualized_volatility
return {
"symbol": symbol,
"period": period,
"sharpe_ratio": (
round(sharpe_ratio, 3) if sharpe_ratio is not None else None
),
"annualized_return": round(annualized_return * 100, 2),
"annualized_volatility": round(annualized_volatility * 100, 2),
"risk_free_rate": round(risk_free_rate * 100, 2),
"interpretation": (
"Excellent"
if (sharpe_ratio is not None and sharpe_ratio > 2)
else (
"Good"
if (sharpe_ratio is not None and sharpe_ratio > 1)
else (
"Acceptable"
if (sharpe_ratio is not None and sharpe_ratio > 0.5)
else "Poor"
)
)
),
}
except Exception as e:
return {"error": f"Failed to calculate Sharpe Ratio: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=120)
async def compare_stocks(symbols: List[str], metric: str = "performance") -> dict:
"""Compare multiple stocks by various metrics.
Args:
symbols: List of stock ticker symbols
metric: Comparison metric ('performance', 'volatility', 'volume', 'pe_ratio')
Returns:
Dictionary with comparison data
"""
try:
if not symbols:
return {"error": "At least one symbol is required"}
if len(symbols) > MAX_SYMBOLS_COMPARE:
return {
"error": f"Maximum {MAX_SYMBOLS_COMPARE} symbols allowed for comparison"
}
metric = metric.lower()
allowed_metrics = {"performance", "volatility", "volume", "pe_ratio"}
if metric not in allowed_metrics:
return {"error": f"Invalid metric. Allowed: {sorted(allowed_metrics)}"}
symbols_norm = []
seen = set()
for s in symbols:
ns = _normalize_symbol(s)
if not _is_valid_symbol(ns) or ns in seen:
continue
seen.add(ns)
symbols_norm.append(ns)
async def _fetch_one(sym: str) -> dict:
ticker = yf.Ticker(sym)
info = await asyncio.to_thread(lambda: ticker.info)
if metric == "performance":
hist = await asyncio.to_thread(ticker.history, period="1mo")
performance = None
if not hist.empty:
start_price = float(hist["Close"].iloc[0])
end_price = float(hist["Close"].iloc[-1])
if start_price != 0:
performance = ((end_price - start_price) / start_price) * 100
return {
"symbol": sym,
"current_price": info.get(
"currentPrice", info.get("regularMarketPrice", "N/A")
),
"1m_performance": (
round(performance, 2) if performance is not None else "N/A"
),
"market_cap": info.get("marketCap", "N/A"),
}
if metric == "volatility":
hist = await asyncio.to_thread(ticker.history, period="1mo")
volatility = None
if not hist.empty:
daily_returns = hist["Close"].pct_change().dropna()
volatility = float(daily_returns.std() * np.sqrt(252) * 100)
return {
"symbol": sym,
"volatility": (
round(volatility, 2) if volatility is not None else "N/A"
),
"beta": info.get("beta", "N/A"),
}
if metric == "pe_ratio":
return {
"symbol": sym,
"pe_ratio": info.get("trailingPE", "N/A"),
"forward_pe": info.get("forwardPE", "N/A"),
"peg_ratio": info.get("pegRatio", "N/A"),
}
avg_vol = info.get("averageVolume", 0) or 0
cur_vol = info.get("volume", 0) or 0
vol_ratio = round(cur_vol / avg_vol, 2) if avg_vol > 0 else "N/A"
return {
"symbol": sym,
"volume": info.get("volume", "N/A"),
"avg_volume": info.get("averageVolume", "N/A"),
"volume_ratio": vol_ratio,
}
results = await asyncio.gather(*[_fetch_one(sym) for sym in symbols_norm])
return {"metric": metric, "comparison": results}
except Exception as e:
return {"error": f"Failed to compare stocks: {str(e)}"}
@mcp.tool()
@rate_limit
@cached(duration=300)
async def get_company_news(symbol: str, limit: int = 5) -> dict:
"""Get recent company news headlines.
Args:
symbol: Stock ticker symbol
limit: Maximum number of news items to return (default: 5)
Returns:
Dictionary with a list of news items
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if limit <= 0:
return {"error": "Limit must be positive"}
ticker = yf.Ticker(symbol)
news_items = await asyncio.to_thread(lambda: getattr(ticker, "news", []))
if not news_items:
return {"symbol": symbol, "news": []}
formatted = []
for item in news_items[:limit]:
formatted.append(
{
"title": item.get("title"),
"publisher": item.get("publisher"),
"link": item.get("link"),
"published": item.get("providerPublishTime"),
}
)
return {"symbol": symbol, "news": formatted}
except Exception as e:
return {"error": f"Failed to fetch news for {symbol}: {str(e)}"}
def _period_start_ts(period: str) -> Optional[pd.Timestamp]:
try:
now = pd.Timestamp.utcnow()
if period == "ytd":
return pd.Timestamp(year=now.year, month=1, day=1, tz="UTC")
if period.endswith("mo"):
months = int(period[:-2])
return now - pd.DateOffset(months=months)
if period.endswith("y"):
years = int(period[:-1])
return now - pd.DateOffset(years=years)
if period == "max":
return None
except Exception:
return None
return None
@mcp.tool()
@rate_limit
@cached(duration=600)
async def get_dividends(symbol: str, period: str = "5y") -> dict:
"""Get dividend history for a stock.
Args:
symbol: Stock ticker symbol
period: Time period (e.g., 1y, 3y, 5y, 10y, ytd, max)
Returns:
Dictionary with dividend entries
"""
try:
symbol = _normalize_symbol(symbol)
if not _is_valid_symbol(symbol):
return {"error": "Invalid symbol format"}
if period not in VALID_PERIODS:
return {"error": f"Invalid period. Allowed: {sorted(VALID_PERIODS)}"}
ticker = yf.Ticker(symbol)
dividends = await asyncio.to_thread(lambda: ticker.dividends)
if dividends is None or dividends.empty:
return {"symbol": symbol, "period": period, "dividends": []}
start_ts = _period_start_ts(period)
if start_ts is not None:
dividends = dividends[dividends.index >= start_ts]
items: List[dict] = []
for dt, val in dividends.tail(50).items():
items.append(
{
"date": pd.Timestamp(dt).strftime("%Y-%m-%d"),
"dividend": round(float(val), 4),
}
)
return {"symbol": symbol, "period": period, "dividends": items}
except Exception as e:
return {"error": f"Failed to fetch dividends for {symbol}: {str(e)}"}
@mcp.tool()
async def clear_cache() -> str:
"""Clear the cache to force fresh data retrieval."""
global cache
cache.clear()
return "Cache cleared successfully"
if __name__ == "__main__":
print("Starting Financial Data MCP Server...", file=sys.stderr)
mcp.run(transport="stdio")
