get_investor_sentiment
Analyze investor sentiment data for Chinese stocks including user attention index, daily market participation willingness, stock rating records, and institutional involvement to assess market sentiment trends.
Instructions
分析投资者情绪数据，包括用户关注指数、日度市场参与意愿、股票评级记录和机构参与度。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | 股票代码 (例如: '000001') | |
| output_format | No | 输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown | markdown |
Implementation Reference
- src/china_stock_mcp/server.py:586-586 (registration)Tool registration decorator defining the name and description for get_investor_sentiment.@mcp.tool(name="get_investor_sentiment", description="分析投资者情绪数据，包括用户关注指数、日度市场参与意愿、股票评级记录和机构参与度。")
- src/china_stock_mcp/server.py:587-639 (handler)Main handler function for get_investor_sentiment tool. Fetches investor sentiment data from multiple akshare sources, combines them, formats as DataFrame, and returns in specified output format.def get_investor_sentiment( symbol: Annotated[str, Field(description="股票代码 (例如: '000001')")], output_format: Annotated[ Literal["json", "csv", "xml", "excel", "markdown", "html"], Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"), ] = "markdown" ) -> str: """分析散户和机构投资者的投资情绪""" def get_investor_sentiment_fetcher( symbol: str, indicator: str, **kwargs: Any ) -> pd.DataFrame: """获取投资情绪数据""" if indicator == "用户关注指数": df = ak.stock_comment_detail_scrd_focus_em(symbol) elif indicator == "日度市场参与意愿": df = ak.stock_comment_detail_scrd_desire_daily_em(symbol) # elif indicator == "northbound_flow": # df = ak.stock_hsgt_fund_flow_summary_em() elif indicator == "股票评级记录": df = ak.stock_institute_recommend_detail(symbol) elif indicator == "机构参与度": df = ak.stock_comment_detail_zlkp_jgcyd_em(symbol) return df def get_all_investor_sentiment_fetcher( symbol: str, **kwargs: Any ) -> pd.DataFrame: df_list = [] indicators = [ "用户关注指数", "日度市场参与意愿", "股票评级记录", "机构参与度", ] for indicator in indicators: indicator_df = get_investor_sentiment_fetcher(symbol, indicator, **kwargs) if indicator_df is not None and not indicator_df.empty: # 为DataFrame添加指标名称列，以便区分不同指标的数据 indicator_df['indicator'] = indicator df_list.append(indicator_df) if df_list: # 使用 pd.concat 一次性合并所有DataFrame df = pd.concat(df_list, ignore_index=True) else: # 如果没有获取到任何数据，返回空的DataFrame df = pd.DataFrame() return df df = get_all_investor_sentiment_fetcher(symbol) if df.empty: df = pd.DataFrame() return _format_dataframe_output(df, output_format)
- Helper function to fetch specific investor sentiment indicator data using akshare functions.def get_investor_sentiment_fetcher( symbol: str, indicator: str, **kwargs: Any ) -> pd.DataFrame: """获取投资情绪数据""" if indicator == "用户关注指数": df = ak.stock_comment_detail_scrd_focus_em(symbol) elif indicator == "日度市场参与意愿": df = ak.stock_comment_detail_scrd_desire_daily_em(symbol) # elif indicator == "northbound_flow": # df = ak.stock_hsgt_fund_flow_summary_em() elif indicator == "股票评级记录": df = ak.stock_institute_recommend_detail(symbol) elif indicator == "机构参与度": df = ak.stock_comment_detail_zlkp_jgcyd_em(symbol) return df
- Helper function that fetches and combines all investor sentiment indicators for a given symbol.def get_all_investor_sentiment_fetcher( symbol: str, **kwargs: Any ) -> pd.DataFrame: df_list = [] indicators = [ "用户关注指数", "日度市场参与意愿", "股票评级记录", "机构参与度", ] for indicator in indicators: indicator_df = get_investor_sentiment_fetcher(symbol, indicator, **kwargs) if indicator_df is not None and not indicator_df.empty: # 为DataFrame添加指标名称列，以便区分不同指标的数据 indicator_df['indicator'] = indicator df_list.append(indicator_df) if df_list: # 使用 pd.concat 一次性合并所有DataFrame df = pd.concat(df_list, ignore_index=True) else: # 如果没有获取到任何数据，返回空的DataFrame df = pd.DataFrame() return df