get_realtime_data

Fetch real-time stock quotes for Chinese A/B/H shares by entering the stock symbol to get current market data in your preferred format.

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:288-311 (handler)

The main handler function for the 'get_realtime_data' tool, including registration decorator (@mcp.tool), input schema (Annotated parameters), and execution logic. It uses fallback data sources to fetch realtime stock data via the external 'ako.get_realtime_data' function and formats the output.

@mcp.tool(
    name="get_realtime_data", description="获取指定的股票实时行情数据"
)
def get_realtime_data(
   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:
    """获取实时股票行情数据. 'eastmoney_direct' """

    # 定义内部 fetch_func
    def realtime_data_fetcher(source: str, **kwargs: Any) -> pd.DataFrame:
        return ako.get_realtime_data(source=source, **kwargs)
        

    df = _fetch_data_with_fallback(
        fetch_func=realtime_data_fetcher,
        primary_source="eastmoney",
        fallback_sources=["eastmoney_direct","xueqiu"],
        symbol=symbol,
    )
    return _format_dataframe_output(df, output_format)

src/china_stock_mcp/server.py:15-72 (helper)

Helper function that implements fallback logic for fetching data from multiple sources (eastmoney, eastmoney_direct, xueqiu), used by the get_realtime_data handler.

def _fetch_data_with_fallback(
    fetch_func: Callable[..., pd.DataFrame],
    primary_source: str,
    fallback_sources: List[str],
    **kwargs: Any,
) -> pd.DataFrame:
    """
    通用的数据源故障切换辅助函数。
    按优先级尝试数据源，直到获取到有效数据或所有数据源都失败。

    Args:
        fetch_func: 实际调用 akshare 或 akshare_one 获取数据的函数。
                    这个函数应该接受 'source' 参数，或者在内部处理 source 的映射。
        primary_source: 用户指定的首选数据源。
        fallback_sources: 备用数据源列表，按优先级排序。
        **kwargs: 传递给 fetch_func 的其他参数。

    Returns:
        pd.DataFrame: 获取到的数据。

    Raises:
        RuntimeError: 如果所有数据源都未能获取到有效数据。
    """
    if primary_source is None:
        return fetch_func(**kwargs)
    data_source_priority = [primary_source] + fallback_sources
    # 移除重复项并保持顺序
    seen = set()
    unique_data_source_priority = []
    for x in data_source_priority:
        if x not in seen:
            unique_data_source_priority.append(x)
            seen.add(x)

    df = None
    errors = []

    for current_source in unique_data_source_priority:
        try:
            # 假设 fetch_func 能够接受 source 参数
            # 或者 fetch_func 内部根据 kwargs 中的 source 参数进行逻辑判断
            temp_df = fetch_func(source=current_source, **kwargs)
            if temp_df is not None and not temp_df.empty:
                print(f"成功从数据源 '{current_source}' 获取数据。")
                df = temp_df
                break
            else:
                errors.append(f"数据源 '{current_source}' 返回空数据。")
        except Exception as e:
            errors.append(f"从数据源 '{current_source}' 获取数据失败: {str(e)}")

    if df is None or df.empty:
        raise RuntimeError(
            f"所有数据源都未能获取到有效数据。详细错误: {'; '.join(errors)}"
        )

    return df

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

Helper function to format the fetched DataFrame into various output formats (json, csv, etc.), called by the handler.

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)

src/china_stock_mcp/server.py:288-290 (registration)
The @mcp.tool decorator registers the get_realtime_data function as an MCP tool.
```
@mcp.tool(
    name="get_realtime_data", description="获取指定的股票实时行情数据"
)
```

China Stock MCP