get_fund_flow
Retrieve stock fund flow data for Chinese stocks to analyze capital movements and investor behavior over recent trading periods.
Instructions
获取股票资金流向数据 (近100交易日)
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:373-385 (handler)The main handler function decorated with @mcp.tool for registration. Fetches stock fund flow data using akshare based on symbol, determines market, formats output as DataFrame in specified format.@mcp.tool(name="get_fund_flow", description="获取股票资金流向数据 (近100交易日)") def get_fund_flow( 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: market = _get_market_from_symbol(symbol) df = ak.stock_individual_fund_flow(stock=symbol, market=market) if df.empty: df = pd.DataFrame() return _format_dataframe_output(df, output_format)
- src/china_stock_mcp/server.py:73-87 (helper)Helper function used by get_fund_flow to determine the stock market (sh/sz/bj) from the symbol prefix.def _get_market_from_symbol(symbol: str) -> str: """ 根据股票代码判断所属市场。 上海证券交易所: sh (600, 601, 603, 605, 688 开头) 深圳证券交易所: sz (000, 001, 002, 300 开头) 北京证券交易所: bj (830, 870, 880 开头) 默认返回 "sh" """ if symbol.startswith(("600", "601", "603", "605", "688")): return "sh" elif symbol.startswith(("000", "001", "002", "300")): return "sz" elif symbol.startswith(("830", "870", "880")): return "bj" return "sh" # 默认上海市场
- src/china_stock_mcp/server.py:89-117 (helper)Helper function used by get_fund_flow to format the DataFrame output in various formats (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)