get_macro_data
Retrieve macroeconomic indicator data for China including GDP, CPI, PMI, money supply, and stock market summaries in multiple output formats.
Instructions
获取宏观经济指标数据，支持多个指标
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| indicator | No | 要获取的宏观经济指标，可以是逗号分隔的字符串（例如: 'gdp,cpi'）或字符串列表（例如: ['gdp', 'cpi']）。支持的指标包括: money_supply, gdp, cpi, pmi, stock_summary。默认: 'gdp' | gdp |
| output_format | No | 输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown | markdown |
Implementation Reference
- src/china_stock_mcp/server.py:497-497 (registration)Tool registration decorator defining the name and description for get_macro_data.@mcp.tool(name="get_macro_data", description="获取宏观经济指标数据，支持多个指标")
- src/china_stock_mcp/server.py:498-584 (handler)Main handler function for get_macro_data tool. Parses indicators, fetches data using akshare functions via cached fetcher, cleans data, concatenates results, and formats output.def get_macro_data( indicator: Annotated[ str | list[ Literal["money_supply", "gdp", "cpi", "pmi", "stock_summary"] ] | None, Field( description="要获取的宏观经济指标，可以是逗号分隔的字符串（例如: 'gdp,cpi'）或字符串列表（例如: ['gdp', 'cpi']）。支持的指标包括: money_supply, gdp, cpi, pmi, stock_summary。默认: 'gdp'" ), ] = "gdp", output_format: Annotated[ Literal["json", "csv", "xml", "excel", "markdown", "html"], Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"), ] = "markdown" ) -> str: """获取宏观经济数据""" def _clean_macro_data(df: pd.DataFrame) -> pd.DataFrame: """ 通用数据清洗函数，删除全 null 列和全 null 行。 """ if df.empty: return df # 删除所有列值都为 null 的列 df = df.dropna(axis=1, how='all') # 删除所有行值都为 null 的行 df = df.dropna(axis=0, how='all') return df macro_indicator_map = { "money_supply": ak.macro_china_money_supply, "gdp": ak.macro_china_gdp_yearly, "cpi": ak.macro_china_cpi_yearly, "pmi": ak.macro_china_pmi_yearly, "stock_summary": ak.macro_china_stock_market_cap, } @cached_data_fetch() def get_macro_data_fetcher( indicator_name: str, **kwargs: Any ) -> pd.DataFrame: fetch_func = macro_indicator_map.get(indicator_name) if fetch_func: return fetch_func(**kwargs) else: raise ValueError(f"不支持的宏观经济指标: {indicator_name}") if indicator is None: indicator_list = ["gdp"] # 默认指标 elif isinstance(indicator, str): indicator_list = [ ind.strip() for ind in indicator.split(",") if ind.strip() ] else: # 假设是 list[Literal[...]] indicator_list = indicator # 过滤掉无效的指标 valid_indicators = [] for ind in indicator_list: if ind in macro_indicator_map: valid_indicators.append(ind) else: print(f"警告: 宏观经济指标 '{ind}' 不存在，将被忽略。") if not valid_indicators: print("警告: 未找到有效的宏观经济指标，将返回空数据。") return _format_dataframe_output(pd.DataFrame(), output_format) df_list = [] for ind in valid_indicators: try: temp_df = get_macro_data_fetcher(ind) if temp_df is not None and not temp_df.empty: temp_df = _clean_macro_data(temp_df) temp_df['indicator'] = ind # 添加指标名称列 df_list.append(temp_df) except Exception as e: print(f"获取宏观经济指标 '{ind}' 失败: {e}") if df_list: df = pd.concat(df_list, ignore_index=True) else: df = pd.DataFrame() # 如果没有获取到任何数据，返回空的DataFrame return _format_dataframe_output(df, output_format)
- Input schema definitions using Pydantic Annotated and Field for indicator parameter (supports multiple macro types) and output_format.indicator: Annotated[ str | list[ Literal["money_supply", "gdp", "cpi", "pmi", "stock_summary"] ] | None, Field( description="要获取的宏观经济指标，可以是逗号分隔的字符串（例如: 'gdp,cpi'）或字符串列表（例如: ['gdp', 'cpi']）。支持的指标包括: money_supply, gdp, cpi, pmi, stock_summary。默认: 'gdp'" ), ] = "gdp", output_format: Annotated[ Literal["json", "csv", "xml", "excel", "markdown", "html"], Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"), ] = "markdown" ) -> str:
- Cached helper function to fetch data for a specific macro indicator using the mapped akshare functions.def get_macro_data_fetcher( indicator_name: str, **kwargs: Any ) -> pd.DataFrame: fetch_func = macro_indicator_map.get(indicator_name) if fetch_func: return fetch_func(**kwargs) else: raise ValueError(f"不支持的宏观经济指标: {indicator_name}")
- Helper function to clean macro data by dropping all-null rows and columns.def _clean_macro_data(df: pd.DataFrame) -> pd.DataFrame: """ 通用数据清洗函数，删除全 null 列和全 null 行。 """ if df.empty: return df # 删除所有列值都为 null 的列 df = df.dropna(axis=1, how='all') # 删除所有行值都为 null 的行 df = df.dropna(axis=0, how='all') return df