get_stock_analysis
Generate data-driven stock analysis reports for A-share stocks, covering fundamental, technical, or comprehensive analysis based on stock code input.
Instructions
提供基于数据的股票分析报告，而非投资建议。
Args:
code: 股票代码，如'sh.600000'
analysis_type: 'fundamental'|'technical'|'comprehensive'
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | Yes | ||
| analysis_type | No | fundamental |
Implementation Reference
- src/tools/analysis.py:19-31 (handler)The handler function for the 'get_stock_analysis' tool. Defines input parameters with type hints and docstring schema. Logs invocation and delegates execution to the use case layer via run_tool_with_handling.
def get_stock_analysis(code: str, analysis_type: str = "fundamental") -> str: """ 提供基于数据的股票分析报告，而非投资建议。 Args: code: 股票代码，如'sh.600000' analysis_type: 'fundamental'|'technical'|'comprehensive' """ logger.info(f"Tool 'get_stock_analysis' called for {code}, type={analysis_type}") return run_tool_with_handling( lambda: build_stock_analysis_report(active_data_source, code=code, analysis_type=analysis_type), context=f"get_stock_analysis:{code}:{analysis_type}", ) - src/tools/analysis.py:15-32 (registration)The registration function that defines and registers the get_stock_analysis tool using the @app.tool() decorator within the FastMCP app.
def register_analysis_tools(app: FastMCP, active_data_source: FinancialDataSource): """Register analysis tools.""" @app.tool() def get_stock_analysis(code: str, analysis_type: str = "fundamental") -> str: """ 提供基于数据的股票分析报告，而非投资建议。 Args: code: 股票代码，如'sh.600000' analysis_type: 'fundamental'|'technical'|'comprehensive' """ logger.info(f"Tool 'get_stock_analysis' called for {code}, type={analysis_type}") return run_tool_with_handling( lambda: build_stock_analysis_report(active_data_source, code=code, analysis_type=analysis_type), context=f"get_stock_analysis:{code}:{analysis_type}", ) - mcp_server.py:57-57 (registration)Top-level call in the main MCP server file to register all analysis tools, including get_stock_analysis.
register_analysis_tools(app, active_data_source) - src/use_cases/analysis.py:8-79 (helper)Supporting use case function that implements the core logic of generating the stock analysis report by querying the data source and formatting results into a markdown string.
def build_stock_analysis_report(data_source: FinancialDataSource, *, code: str, analysis_type: str) -> str: basic_info = data_source.get_stock_basic_info(code=code) # Fundamental data if analysis_type in ["fundamental", "comprehensive"]: recent_year = datetime.now().strftime("%Y") recent_quarter = (datetime.now().month - 1) // 3 + 1 if recent_quarter < 1: recent_year = str(int(recent_year) - 1) recent_quarter = 4 profit_data = data_source.get_profit_data(code=code, year=recent_year, quarter=recent_quarter) growth_data = data_source.get_growth_data(code=code, year=recent_year, quarter=recent_quarter) balance_data = data_source.get_balance_data(code=code, year=recent_year, quarter=recent_quarter) dupont_data = data_source.get_dupont_data(code=code, year=recent_year, quarter=recent_quarter) else: profit_data = growth_data = balance_data = dupont_data = None # Technical data if analysis_type in ["technical", "comprehensive"]: end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=180)).strftime("%Y-%m-%d") price_data = data_source.get_historical_k_data(code=code, start_date=start_date, end_date=end_date) else: price_data = None report = f"# {basic_info['code_name'].values[0] if not basic_info.empty else code} 数据分析报告\n\n" report += "## 免责声明\n本报告基于公开数据生成，仅供参考，不构成投资建议。投资决策需基于个人风险承受能力和研究。\n\n" # Basic info if not basic_info.empty: report += "## 公司基本信息\n" report += f"- 股票代码: {code}\n" report += f"- 股票名称: {basic_info['code_name'].values[0]}\n" report += f"- 所属行业: {basic_info['industry'].values[0] if 'industry' in basic_info.columns else '未知'}\n" report += f"- 上市日期: {basic_info['ipoDate'].values[0] if 'ipoDate' in basic_info.columns else '未知'}\n\n" if analysis_type in ["fundamental", "comprehensive"] and profit_data is not None and not profit_data.empty: report += f"## 基本面指标分析 ({recent_year}年第{recent_quarter}季度)\n\n" report += "### 盈利能力指标\n" if 'roeAvg' in profit_data.columns: report += f"- ROE(净资产收益率): {profit_data['roeAvg'].values[0]}%\n" if 'npMargin' in profit_data.columns: report += f"- 销售净利率: {profit_data['npMargin'].values[0]}%\n" if growth_data is not None and not growth_data.empty: report += "\n### 成长能力指标\n" if 'YOYEquity' in growth_data.columns: report += f"- 净资产同比增长: {growth_data['YOYEquity'].values[0]}%\n" if 'YOYAsset' in growth_data.columns: report += f"- 总资产同比增长: {growth_data['YOYAsset'].values[0]}%\n" if 'YOYNI' in growth_data.columns: report += f"- 净利润同比增长: {growth_data['YOYNI'].values[0]}%\n" if balance_data is not None and not balance_data.empty: report += "\n### 偿债能力指标\n" if 'currentRatio' in balance_data.columns: report += f"- 流动比率: {balance_data['currentRatio'].values[0]}\n" if 'assetLiabRatio' in balance_data.columns: report += f"- 资产负债率: {balance_data['assetLiabRatio'].values[0]}%\n" if analysis_type in ["technical", "comprehensive"] and price_data is not None and not price_data.empty: report += "\n## 技术面简析（近180日）\n" latest_price = price_data['close'].iloc[-1] start_price = price_data['close'].iloc[0] price_change = ((latest_price - start_price) / start_price) * 100 if start_price else 0 report += f"- 区间涨跌幅: {price_change:.2f}%\n" if 'close' in price_data.columns and price_data.shape[0] >= 20: ma20 = price_data['close'].astype(float).rolling(window=20).mean().iloc[-1] report += f"- 20日均线: {ma20:.2f}\n" return report