get_profit_forecast

Retrieve profit forecast data for Chinese stocks, including projected annual net profit and earnings per share, to support investment analysis and decision-making.

Instructions

获取股票的业绩预测数据，包括预测年报净利润和每股收益

Input Schema

TableJSON Schema

Name	Required	Description	Default
`symbol`	Yes	股票代码 (例如: '600519')
`output_format`	No	输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown	markdown

Implementation Reference

src/china_stock_mcp/server.py:695-717 (handler)

The main handler function for the 'get_profit_forecast' tool. It takes a stock symbol and optional output format, fetches profit forecast data for '预测年报净利润' (predicted annual net profit) and '预测年报每股收益' (predicted annual EPS) using ak.stock_profit_forecast_ths from akshare, concatenates the DataFrames, and returns the formatted output.

def get_profit_forecast(
    symbol: Annotated[str, Field(description="股票代码 (例如: '600519')")],   
    output_format: Annotated[
        Literal["json", "csv", "xml", "excel", "markdown", "html"],
        Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"),
    ] = "markdown"
) -> str:
    """
    获取股票的业绩预测数据。
    """
    supported_indicators = ["预测年报净利润", "预测年报每股收益"]
    df_list = []
    for ind in supported_indicators:
            temp_df = ak.stock_profit_forecast_ths(symbol=symbol, indicator=ind)
            if not temp_df.empty:
                temp_df["indicator"] = ind  # 添加指标列以便区分
                df_list.append(temp_df)
        
    if df_list:
            df = pd.concat(df_list, ignore_index=True)
    else:
        df = pd.DataFrame()
    return _format_dataframe_output(df, output_format)

src/china_stock_mcp/server.py:694-694 (registration)
Registers the 'get_profit_forecast' tool using the @mcp.tool decorator with the specified name and description.
```
@mcp.tool(name="get_profit_forecast", description="获取股票的业绩预测数据，包括预测年报净利润和每股收益")
```

src/china_stock_mcp/server.py:696-701 (schema)

Pydantic-based input schema definition using Annotated types and Field descriptions for the tool parameters: symbol (stock code) and output_format.

    symbol: Annotated[str, Field(description="股票代码 (例如: '600519')")],   
    output_format: Annotated[
        Literal["json", "csv", "xml", "excel", "markdown", "html"],
        Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"),
    ] = "markdown"
) -> str:

src/china_stock_mcp/server.py:89-117 (helper)

Helper function _format_dataframe_output used by the tool to format the resulting DataFrame into the requested output format (json, csv, etc.).

def _format_dataframe_output(
    df: pd.DataFrame,
    output_format: Literal["json", "csv", "xml", "excel", "markdown", "html"],
) -> str:
    """
    根据指定的格式格式化 DataFrame 输出。
    """
    if df.empty:
        return json.dumps([])

    if output_format == "json":
        return df.to_json(orient="records", force_ascii=False)
    elif output_format == "csv":
        return df.to_csv(index=False)
    elif output_format == "xml":
        return df.to_xml(index=False)
    elif output_format == "excel":
        # 使用 BytesIO 将 Excel 写入内存
        output = io.BytesIO()
        df.to_excel(output, index=False, engine='openpyxl')
        # 返回 base64 编码的二进制数据，或者直接返回字节流
        # 为了兼容性，这里尝试返回 utf-8 编码的字符串，但对于二进制文件，通常直接传输字节流更合适
        return output.getvalue().decode("utf-8", errors="ignore")
    elif output_format == "markdown":
        return df.to_markdown(index=False)
    elif output_format == "html":
        return df.to_html(index=False)
    else:
        return df.to_json(orient="records", force_ascii=False)

China Stock MCP