黑科技软件合集网站app手机软件开发公司

黑科技软件合集网站,app手机软件开发公司,wordpress免插件代码高亮,聚诚商务做网站多少钱7.3.5 金融MCP服务器文件server.py是本项目的核心服务器实现#xff0c;基于模型上下文协议#xff08;MCP#xff09;搭建了一个金融数据服务平台。它整合了多个工具函数#xff0c;提供了全面的金融数据获取与分析能力#xff0c;包括市场动态#xff08;涨幅榜、跌幅…7.3.5 金融MCP服务器文件server.py是本项目的核心服务器实现基于模型上下文协议MCP搭建了一个金融数据服务平台。它整合了多个工具函数提供了全面的金融数据获取与分析能力包括市场动态涨幅榜、跌幅榜等、CNN 恐惧与贪婪指数、加密货币情绪指数、谷歌趋势数据、个股详细信息财务指标、新闻、分析师推荐等、期权数据、价格历史、财务报表、机构持股、收益历史、内幕交易等还支持在特定依赖如 TA-Lib、Playwright存在时提供技术指标计算和收益日历功能。这些工具通过 MCP 协议暴露可被客户端如大型语言模型调用以获取专业金融分析支持。1下面代码的功能是设置日志记录器和基本配置创建一个名为“Investor-Agent”的FastMCP实例该实例声明了其依赖的Python库以便于在MCP环境中正确初始化和运行。logger logging.getLogger(__name__) logging.basicConfig( levellogging.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 data4下面代码的功能是定义一个名为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(hlen-US, tz360) pytrends.build_payload(keywords, timeframefnow {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, limitmax_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 data7下面代码的功能是定义一个名为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, limitmax_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, limitmax_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 的默认移动平均线类型0SMA num_results: int 50 # 要返回的最近结果数量 ) - dict: 计算技术指标包含适当的日期对齐和结果限制。 import numpy as np history yfinance_utils.get_price_history(ticker, periodperiod, interval1d) 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, timeperiodtimeperiod)}, EMA: lambda: {ema: talib.EMA(close_prices, timeperiodtimeperiod)}, RSI: lambda: {rsi: talib.RSI(close_prices, timeperiodtimeperiod)}, MACD: lambda: dict(zip([macd, signal, histogram], talib.MACD(close_prices, fastperiodfastperiod, slowperiodslowperiod, signalperiodsignalperiod))), BBANDS: lambda: dict(zip([upper_band, middle_band, lower_band], talib.BBANDS(close_prices, timeperiodtimeperiod, nbdevupnbdev, nbdevdnnbdev, matypematype))) } 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)) ]