server.py

#!/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")