金融MCP服务器)
7.3.5 金融MCP服务器
文件server.py是本项目的核心服务器实现，基于模型上下文协议（MCP）搭建了一个金融数据服务平台。它整合了多个工具函数，提供了全面的金融数据获取与分析能力，包括市场动态（涨幅榜、跌幅榜等）、CNN 恐惧与贪婪指数、加密货币情绪指数、谷歌趋势数据、个股详细信息（财务指标、新闻、分析师推荐等）、期权数据、价格历史、财务报表、机构持股、收益历史、内幕交易等，还支持在特定依赖（如 TA-Lib、Playwright）存在时提供技术指标计算和收益日历功能。这些工具通过 MCP 协议暴露，可被客户端（如大型语言模型）调用以获取专业金融分析支持。
（1）下面代码的功能是设置日志记录器和基本配置，创建一个名为“Investor-Agent”的FastMCP实例，该实例声明了其依赖的Python库，以便于在MCP环境中正确初始化和运行。
logger = logging.getLogger(__name__) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler(sys.stderr)] ) mcp = FastMCP("Investor-Agent", dependencies=["yfinance", "httpx", "pandas", "pytrends", "beautifulsoup4"]) FearGreedIndicator = Literal[ "fear_and_greed", "fear_and_greed_historical", "put_call_options", "market_volatility_vix", "market_volatility_vix_50", "junk_bond_demand", "safe_haven_demand" ]（2）下面代码的功能是定义一个名为get_market_movers的异步函数，该函数通过调用yahoo_finance_utils模块中的get_market_movers_data函数，获取市场动态数据，如涨幅最大、跌幅最大或最活跃的股票，并支持不同的市场交易时段。
@mcp.tool() async def get_market_movers( category: Literal["gainers", "losers", "most-active"] = "most-active", count: int = 25, market_session: Literal["regular", "pre-market", "after-hours"] = "regular" ) -> dict: """ 获取市场动态：涨幅榜、跌幅榜或最活跃股票。 对“最活跃”类别支持不同的市场交易时段。 参数： category: 要获取的动态类型 count: 要返回的股票数量（最多 100 只） market_session: 市场交易时段（仅适用于“最活跃”类别） 返回： 包含市场动态数据的字典 """ return await yahoo_finance_utils.get_market_movers_data(category, count, market_session)（3）下面代码的功能是定义一个名为get_cnn_fear_greed_index的异步函数，用于获取CNN恐惧与贪婪指数的历史数据，支持指定返回指标和时间范围。
@mcp.tool() async def get_cnn_fear_greed_index( days: int = 0, indicators: list[FearGreedIndicator] | None = None ) -> dict: """最多 30 天的历史数据。""" data = await fetch_fng_data() if not data: raise RuntimeError("无法获取 CNN 恐惧与贪婪指数数据") if indicators: invalid_keys = set(indicators) - set(data.keys()) if invalid_keys: raise ValueError(f"无效指标：{list(invalid_keys)}。可用指标：{list(data.keys())}") data = {k: v for k, v in data.items() if k in indicators} # 当 days = 0 时排除 fear_and_greed_historical if days == 0: data = {k: v for k, v in data.items() if k != "fear_and_greed_historical"} # 根据 days 参数处理历史数据 max_days = min(days, 30) if days > 0 else 0 for key, value in data.items(): if isinstance(value, dict) and "data" in value: if days == 0: data[key] = {k: v for k, v in value.items() if k != "data"} elif len(value["data"]) > max_days: data[key] = {**value, "data": value["data"][:max_days]} return data（4）下面代码的功能是定义一个名为get_crypto_fear_greed_index的异步函数，用于获取加密货币恐惧与贪婪指数的历史数据。
@mcp.tool() async def get_crypto_fear_greed_index(days: int = 7) -> dict: """获取加密货币恐惧与贪婪指数的历史数据。""" async with yahoo_finance_utils.create_cached_async_client() as client: response = await client.get("https://api.alternative.me/fng/ ", params={"limit": days}) response.raise_for_status() return response.json()["data"]（5）下面代码的功能是定义一个名为get_google_trends的函数，用于获取指定关键词在Google趋势上的相对搜索热度。
@mcp.tool() def get_google_trends( keywords: list[str], period_days: int = 7 ) -> dict: """获取指定关键词的谷歌趋势相对搜索热度。""" from pytrends.request import TrendReq logger.info(f"获取 {period_days} 天的谷歌趋势数据") pytrends = TrendReq(hl='en-US', tz=360) pytrends.build_payload(keywords, timeframe=f'now {period_days}-d') data = pytrends.interest_over_time() if data.empty: raise ValueError("谷歌趋势未返回数据") return data[keywords].mean().to_dict()（6）下面代码的功能是定义一个名为get_ticker_data的函数，用于获取股票的基本信息，包括核心财务指标、即将到来的收益日期和股息日期、最新新闻文章、最新分析师推荐以及近期分析师评级变动。
@mcp.tool() def get_ticker_data( ticker: str, max_news: int = 5, max_recommendations: int = 5, max_upgrades: int = 5 ) -> dict: """ 返回： - info: 核心财务指标（市盈率、利润率、增长率、负债率、市值、每股收益等） - calendar: 即将到来的收益日期和股息日期 - news: 最新新闻文章（标题、日期、来源） - recommendations: 最新分析师推荐（买入/卖出/持有评级） - upgrades_downgrades: 近期分析师评级变动 """ info = yfinance_utils.get_ticker_info(ticker) if not info: raise ValueError(f"{ticker} 没有可用信息") # 只保留关键字段 essential_fields = { 'symbol', 'longName', 'currentPrice', 'marketCap', 'volume', 'trailingPE', 'forwardPE', 'dividendYield', 'beta', 'eps', 'totalRevenue', 'totalDebt', 'profitMargins', 'operatingMargins', 'returnOnEquity', 'returnOnAssets', 'revenueGrowth', 'earningsGrowth', 'bookValue', 'priceToBook', 'enterpriseValue', 'pegRatio', 'trailingEps', 'forwardEps' } filtered_info = {k: v for k, v in info.items() if k in essential_fields} data = {"info": filtered_info} if calendar := yfinance_utils.get_calendar(ticker): data["calendar"] = calendar if news := yfinance_utils.get_news(ticker, limit=max_news): data["news"] = news recommendations = yfinance_utils.get_analyst_data(ticker, "recommendations", max_recommendations) if recommendations is not None and not recommendations.empty: data["recommendations"] = recommendations.to_dict('split') upgrades = yfinance_utils.get_analyst_data(ticker, "upgrades", max_upgrades) if upgrades is not None and not upgrades.empty: data["upgrades_downgrades"] = upgrades.to_dict('split') return data（7）下面代码的功能是定义一个名为get_options的函数，用于获取特定股票的期权数据，支持过滤特定日期范围、行权价范围和期权类型。
@mcp.tool() def get_options( ticker_symbol: str, num_options: int = 10, start_date: str | None = None, end_date: str | None = None, strike_lower: float | None = None, strike_upper: float | None = None, option_type: Literal["C", "P"] | None = None, ) -> dict: """获取期权数据。日期格式：YYYY-MM-DD。类型：C=认购期权，P=认沽期权。""" df, error = yfinance_utils.get_filtered_options( ticker_symbol, start_date, end_date, strike_lower, strike_upper, option_type ) if error: raise ValueError(error) return df.head(num_options).to_dict('split')（8）下面代码的功能是定义一个名为get_price_history的函数，用于获取特定股票的历史价格数据（OHLCV），支持不同时间周期的数据获取。
@mcp.tool() def get_price_history( ticker: str, period: Literal["1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max"] = "1mo" ) -> dict: """获取历史 OHLCV 数据：≤1 年周期为日间隔，≥2 年周期为月间隔。""" interval = "1mo" if period in ["2y", "5y", "10y", "max"] else "1d" history = yfinance_utils.get_price_history(ticker, period, interval) if history is None or history.empty: raise ValueError(f"{ticker} 没有找到历史数据") return history.to_dict('split')（9）下面代码的功能是定义一个名为get_financial_statements的函数，用于获取特定股票的财务报表数据，包括损益表、资产负债表和现金流量表，并支持选择报表类型和频率。
@mcp.tool() def get_financial_statements( ticker: str, statement_type: Literal["income", "balance", "cash"] = "income", frequency: Literal["quarterly", "annual"] = "quarterly", max_periods: int = 8 ) -> dict: data = yfinance_utils.get_financial_statements(ticker, statement_type, frequency) if data is None or data.empty: raise ValueError(f"{ticker} 没有找到 {statement_type} 报表数据") if len(data.columns) > max_periods: data = data.iloc[:, :max_periods] return data.to_dict('split')（10）下面代码的功能是定义一个名为get_institutional_holders的函数，用于获取特定股票的主要机构和共同基金持股情况。
@mcp.tool() def get_institutional_holders(ticker: str, top_n: int = 20) -> dict: """获取主要机构和共同基金持股情况。""" inst_holders, fund_holders = yfinance_utils.get_institutional_holders(ticker, top_n) if (inst_holders is None or inst_holders.empty) and (fund_holders is None or fund_holders.empty): raise ValueError(f"{ticker} 没有找到机构持股数据") return { key: data.to_dict('split') for key, data in [ ("institutional_holders", inst_holders), ("mutual_fund_holders", fund_holders) ] if data is not None and not data.empty }（11）下面代码的功能是定义一个名为get_earnings_history的函数，用于获取特定股票的收益历史数据。
@mcp.tool() def get_earnings_history(ticker: str, max_entries: int = 8) -> dict: earnings_history = yfinance_utils.get_earnings_history(ticker, limit=max_entries) if earnings_history is None or earnings_history.empty: raise ValueError(f"{ticker} 没有找到收益历史数据") return earnings_history.to_dict('split')（12）下面代码的功能是定义一个名为get_insider_trades的函数，用于获取特定股票的内部交易数据。此外，如果Playwright库可用，还会注册一个名为get_earnings_calendar的异步函数，用于获取指定日期范围内的收益日历；如果TA-Lib库可用，还会注册一个名为calculate_technical_indicator的函数，用于计算技术指标。
@mcp.tool() def get_insider_trades(ticker: str, max_trades: int = 20) -> dict: trades = yfinance_utils.get_insider_trades(ticker, limit=max_trades) if trades is None or trades.empty: raise ValueError(f"{ticker} 没有找到内幕交易数据") return trades.to_dict('split') # 仅当 Playwright 可用时注册收益日历工具 if _playwright_available: @mcp.tool() async def get_earnings_calendar( start: str | None = None, end: str | None = None, limit: int = 100 ) -> dict: """获取指定日期范围内的收益日历。""" return await yahoo_finance_utils.get_earnings_calendar_data(start, end, limit) # 仅当 TA-Lib 可用时注册技术指标工具 if _ta_available: @mcp.tool() def calculate_technical_indicator( ticker: str, indicator: Literal["SMA", "EMA", "RSI", "MACD", "BBANDS"], period: Literal["1mo", "3mo", "6mo", "1y", "2y", "5y"] = "1y", timeperiod: int = 14, # SMA、EMA、RSI 的默认时间周期 fastperiod: int = 12, # MACD 快速 EMA 的默认值 slowperiod: int = 26, # MACD 慢速 EMA 的默认值 signalperiod: int = 9, # MACD 信号线的默认值 nbdev: int = 2, # BBANDS 的默认标准差 matype: int = 0, # BBANDS 的默认移动平均线类型（0=SMA） num_results: int = 50 # 要返回的最近结果数量 ) -> dict: """计算技术指标，包含适当的日期对齐和结果限制。""" import numpy as np history = yfinance_utils.get_price_history(ticker, period=period, interval="1d") if history is None or history.empty or 'Close' not in history.columns: raise ValueError(f"{ticker} 没有找到有效的历史数据") close_prices = history['Close'].values min_required = { "SMA": timeperiod, "EMA": timeperiod * 2, "RSI": timeperiod + 1, "MACD": slowperiod + signalperiod, "BBANDS": timeperiod }.get(indicator, timeperiod) if len(close_prices) < min_required: raise ValueError(f"{indicator} 数据不足（{len(close_prices)} 个点，需要 {min_required} 个）") # 使用映射计算指标 indicator_funcs = { "SMA": lambda: {"sma": talib.SMA(close_prices, timeperiod=timeperiod)}, "EMA": lambda: {"ema": talib.EMA(close_prices, timeperiod=timeperiod)}, "RSI": lambda: {"rsi": talib.RSI(close_prices, timeperiod=timeperiod)}, "MACD": lambda: dict(zip(["macd", "signal", "histogram"], talib.MACD(close_prices, fastperiod=fastperiod, slowperiod=slowperiod, signalperiod=signalperiod))), "BBANDS": lambda: dict(zip(["upper_band", "middle_band", "lower_band"], talib.BBANDS(close_prices, timeperiod=timeperiod, nbdevup=nbdev, nbdevdn=nbdev, matype=matype))) } result = indicator_funcs[indicator]() # 限制结果并准备数据 dates = history.index.strftime('%Y-%m-%d').tolist() start_idx = max(0, len(dates) - num_results) if num_results > 0 else 0 return [ { "date": dates[i], "price": history.iloc[i].to_dict(), "indicators": { key: None if np.isnan(val := values[i]) else float(val) for key, values in result.items() } } for i in range(start_idx, len(dates)) ]
