China Stock MCP

Instructions

Implementation Reference

• src/china_stock_mcp/server.py:128-286 (handler)

  The primary handler function for the 'get_hist_data' tool. Fetches historical stock data from multiple sources with fallback, computes and adds technical indicators based on user specification, and formats the output as a string in the requested format.
  def get_hist_data( symbol: Annotated[str, Field(description="股票代码 (例如: '000001')")], interval: Annotated[ Literal["minute", "hour", "day", "week", "month", "year"], Field(description="时间周期: minute, hour, day, week, month, year。默认:day"), ] = "day", interval_multiplier: Annotated[int, Field(description="时间周期乘数", ge=1)] = 1, start_date: Annotated[ str, Field(description="开始日期，格式为 YYYY-MM-DD") ] = "1970-01-01", end_date: Annotated[ str, Field(description="结束日期，格式为 YYYY-MM-DD") ] = "2030-12-31", adjust: Annotated[ Literal["none", "qfq", "hfq"], Field(description="复权类型: none, qfq(前复权), hfq(后复权)。默认：none"), ] = "none", indicators_list: Annotated[ str | list[ Literal[ "SMA", "EMA", "RSI", "MACD", "BOLL", "STOCH", "ATR", "CCI", "ADX", "WILLR", "AD", "ADOSC", "OBV", "MOM", "SAR", "TSF", "APO", "AROON", "AROONOSC", "BOP", "CMO", "DX", "MFI", "MINUS_DI", "MINUS_DM", "PLUS_DI", "PLUS_DM", "PPO", "ROC", "ROCP", "ROCR", "ROCR100", "TRIX", "ULTOSC", ] ]|None, Field( description="要添加的技术指标，可以是逗号分隔的字符串（例如: 'SMA,EMA'）或字符串列表（例如: ['SMA', 'EMA']）。支持的指标包括: SMA, EMA, RSI, MACD, BOLL, STOCH, ATR, CCI, ADX, WILLR, AD, ADOSC, OBV, MOM, SAR, TSF, APO, AROON, AROONOSC, BOP, CMO, DX, MFI, MINUS_DI, MINUS_DM, PLUS_DI, PLUS_DM, PPO, ROC, ROCP, ROCR, ROCR100, TRIX, ULTOSC。常用指标：SMA, EMA, RSI, MACD, BOLL, STOCH, OBV, MFI,建议不超过10个。" ), ] = "SMA, EMA, RSI, MACD, BOLL, STOCH, OBV, MFI", output_format: Annotated[ Literal["json", "csv", "xml", "excel", "markdown", "html"], Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"), ] = "markdown" ) -> str: """获取股票历史行情数据.""" # 定义内部 fetch_func def hist_data_fetcher(source: str, **kwargs: Any) -> pd.DataFrame: return ako.get_hist_data(source=source, **kwargs) df = _fetch_data_with_fallback( fetch_func=hist_data_fetcher, primary_source="eastmoney", fallback_sources=["eastmoney_direct", "sina"], symbol=symbol, interval=interval, interval_multiplier=interval_multiplier, start_date=start_date, end_date=end_date, adjust=adjust, ) indicator_map = { "SMA": (indicators.get_sma, {"window": 20}), "EMA": (indicators.get_ema, {"window": 20}), "RSI": (indicators.get_rsi, {"window": 14}), "MACD": (indicators.get_macd, {"fast": 12, "slow": 26, "signal": 9}), "BOLL": (indicators.get_bollinger_bands, {"window": 20, "std": 2}), "STOCH": ( indicators.get_stoch, {"window": 14, "smooth_d": 3, "smooth_k": 3}, ), "ATR": (indicators.get_atr, {"window": 14}), "CCI": (indicators.get_cci, {"window": 14}), "ADX": (indicators.get_adx, {"window": 14}), "WILLR": (indicators.get_willr, {"window": 14}), "AD": (indicators.get_ad, {}), "ADOSC": (indicators.get_adosc, {"fast_period": 3, "slow_period": 10}), "OBV": (indicators.get_obv, {}), "MOM": (indicators.get_mom, {"window": 10}), "SAR": (indicators.get_sar, {"acceleration": 0.02, "maximum": 0.2}), "TSF": (indicators.get_tsf, {"window": 14}), "APO": ( indicators.get_apo, {"fast_period": 12, "slow_period": 26, "ma_type": 0}, ), "AROON": (indicators.get_aroon, {"window": 14}), "AROONOSC": (indicators.get_aroonosc, {"window": 14}), "BOP": (indicators.get_bop, {}), "CMO": (indicators.get_cmo, {"window": 14}), "DX": (indicators.get_dx, {"window": 14}), "MFI": (indicators.get_mfi, {"window": 14}), "MINUS_DI": (indicators.get_minus_di, {"window": 14}), "MINUS_DM": (indicators.get_minus_dm, {"window": 14}), "PLUS_DI": (indicators.get_plus_di, {"window": 14}), "PLUS_DM": (indicators.get_plus_dm, {"window": 14}), "PPO": ( indicators.get_ppo, {"fast_period": 12, "slow_period": 26, "ma_type": 0}, ), "ROC": (indicators.get_roc, {"window": 10}), "ROCP": (indicators.get_rocp, {"window": 10}), "ROCR": (indicators.get_rocr, {"window": 10}), "ROCR100": (indicators.get_rocr100, {"window": 10}), "TRIX": (indicators.get_trix, {"window": 30}), "ULTOSC": ( indicators.get_ultosc, {"window1": 7, "window2": 14, "window3": 28}, ), } print("indicators_list: ", indicators_list) if indicators_list is None: indicators_list = [] elif isinstance(indicators_list, str): indicators_list = [ indicator.strip() for indicator in indicators_list.split(",") if indicator.strip() ] # 过滤掉无效的指标 if indicators_list: valid_indicators = [] for indicator in indicators_list: if indicator in indicator_map: valid_indicators.append(indicator) else: print(f"警告: 指标 '{indicator}' 不存在，将被忽略。") indicators_list = valid_indicators temp = [] for indicator in indicators_list: if indicator in indicator_map: func, params = indicator_map[indicator] indicator_df = func(df, **params) temp.append(indicator_df) if temp: df = df.join(temp) return _format_dataframe_output(df, output_format)
• src/china_stock_mcp/server.py:125-127 (registration)

  Registration of the 'get_hist_data' tool with the FastMCP server using the @mcp.tool decorator.
  @mcp.tool( name="get_hist_data", description="获取指定的股票历史行情数据及技术指标" )
• src/china_stock_mcp/server.py:128-192 (schema)

  Input schema definition using Pydantic's Annotated types and Field for validation and documentation of tool parameters.
  def get_hist_data( symbol: Annotated[str, Field(description="股票代码 (例如: '000001')")], interval: Annotated[ Literal["minute", "hour", "day", "week", "month", "year"], Field(description="时间周期: minute, hour, day, week, month, year。默认:day"), ] = "day", interval_multiplier: Annotated[int, Field(description="时间周期乘数", ge=1)] = 1, start_date: Annotated[ str, Field(description="开始日期，格式为 YYYY-MM-DD") ] = "1970-01-01", end_date: Annotated[ str, Field(description="结束日期，格式为 YYYY-MM-DD") ] = "2030-12-31", adjust: Annotated[ Literal["none", "qfq", "hfq"], Field(description="复权类型: none, qfq(前复权), hfq(后复权)。默认：none"), ] = "none", indicators_list: Annotated[ str | list[ Literal[ "SMA", "EMA", "RSI", "MACD", "BOLL", "STOCH", "ATR", "CCI", "ADX", "WILLR", "AD", "ADOSC", "OBV", "MOM", "SAR", "TSF", "APO", "AROON", "AROONOSC", "BOP", "CMO", "DX", "MFI", "MINUS_DI", "MINUS_DM", "PLUS_DI", "PLUS_DM", "PPO", "ROC", "ROCP", "ROCR", "ROCR100", "TRIX", "ULTOSC", ] ]|None, Field( description="要添加的技术指标，可以是逗号分隔的字符串（例如: 'SMA,EMA'）或字符串列表（例如: ['SMA', 'EMA']）。支持的指标包括: SMA, EMA, RSI, MACD, BOLL, STOCH, ATR, CCI, ADX, WILLR, AD, ADOSC, OBV, MOM, SAR, TSF, APO, AROON, AROONOSC, BOP, CMO, DX, MFI, MINUS_DI, MINUS_DM, PLUS_DI, PLUS_DM, PPO, ROC, ROCP, ROCR, ROCR100, TRIX, ULTOSC。常用指标：SMA, EMA, RSI, MACD, BOLL, STOCH, OBV, MFI,建议不超过10个。" ), ] = "SMA, EMA, RSI, MACD, BOLL, STOCH, OBV, MFI", 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:15-72 (helper)

  Helper function that attempts to fetch data from primary and fallback sources, used by get_hist_data to handle data source failures gracefully.
  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 pandas DataFrame output into various string formats (JSON, CSV, etc.) as specified by the tool.
  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)