获取股票历史价格

Instructions

Implementation Reference

• mcp_aktools/__init__.py:71-108 (handler)

  The `stock_prices` function is the main handler for the MCP tool. It retrieves historical stock prices and technical indicators (MACD, KDJ, RSI, BOLL) for stocks across markets (sh, sz, hk, us) and ETFs, using akshare library. It accepts symbol, market, period (daily/weekly), and limit parameters, returns CSV-formatted data.

  @mcp.tool(
      title="获取股票历史价格",
      description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币",
  )
  def stock_prices(
      symbol: str = field_symbol,
      market: str = field_market,
      period: str = Field("daily", description="周期，如: daily(日线), weekly(周线，不支持美股)"),
      limit: int = Field(30, description="返回数量(int)", strict=False),
  ):
      if period == "weekly":
          delta = {"weeks": limit + 62}
      else:
          delta = {"days": limit + 62}
      start_date = (datetime.now() - timedelta(**delta)).strftime("%Y%m%d")
      markets = [
          ["sh", ak.stock_zh_a_hist, {}],
          ["sz", ak.stock_zh_a_hist, {}],
          ["hk", ak.stock_hk_hist, {}],
          ["us", stock_us_daily, {}],
          ["sh", fund_etf_hist_sina, {"market": "sh"}],
          ["sz", fund_etf_hist_sina, {"market": "sz"}],
      ]
      for m in markets:
          if m[0] != market:
              continue
          kws = {"period": period, "start_date": start_date, **m[2]}
          dfs = ak_cache(m[1], symbol=symbol, ttl=3600, **kws)
          if dfs is None or dfs.empty:
              continue
          add_technical_indicators(dfs, dfs["收盘"], dfs["最低"], dfs["最高"])
          columns = [
              "日期", "开盘", "收盘", "最高", "最低", "成交量", "换手率",
              "MACD", "DIF", "DEA", "KDJ.K", "KDJ.D", "KDJ.J", "RSI", "BOLL.U", "BOLL.M", "BOLL.L",
          ]
          all = dfs.to_csv(columns=columns, index=False, float_format="%.2f").strip().split("\n")
          return "\n".join([all[0], *all[-limit:]])
      return f"Not Found for {symbol}.{market}"

• mcp_aktools/__init__.py:71-73 (schema)

  The decorator @mcp.tool with title '获取股票历史价格' and description defines the tool's metadata/schema for MCP registration. The function signature with Field annotations (symbol, market, period, limit) defines the input schema/validation.

  @mcp.tool(
      title="获取股票历史价格",
      description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币",

• mcp_aktools/__init__.py:71-73 (registration)

  The @mcp.tool() decorator on the `stock_prices` function registers this function as an MCP tool named 'stock_prices' with FastMCP, with title '获取股票历史价格' and description describing its functionality.

  @mcp.tool(
      title="获取股票历史价格",
      description="根据股票代码和市场获取股票历史价格及技术指标, 不支持加密货币",

• mcp_aktools/__init__.py:111-118 (helper)

  Helper function `stock_us_daily` is called by `stock_prices` to fetch US stock daily data from akshare, renaming columns to Chinese and filtering by start_date.

  def stock_us_daily(symbol, start_date="2025-01-01", period="daily"):
      dfs = ak.stock_us_daily(symbol=symbol)
      if dfs is None or dfs.empty:
          return None
      dfs.rename(columns={"date": "日期", "open": "开盘", "close": "收盘", "high": "最高", "low": "最低", "volume": "成交量"}, inplace=True)
      dfs["换手率"] = None
      dfs.index = pd.to_datetime(dfs["日期"], errors="coerce")
      return dfs[start_date:"2222-01-01"]

• mcp_aktools/__init__.py:120-127 (helper)

  Helper function `fund_etf_hist_sina` is called by `stock_prices` to fetch ETF historical data from akshare, renaming columns and filtering by start_date.

  def fund_etf_hist_sina(symbol, market="sh", start_date="2025-01-01", period="daily"):
      dfs = ak.fund_etf_hist_sina(symbol=f"{market}{symbol}")
      if dfs is None or dfs.empty:
          return None
      dfs.rename(columns={"date": "日期", "open": "开盘", "close": "收盘", "high": "最高", "low": "最低", "volume": "成交量"}, inplace=True)
      dfs["换手率"] = None
      dfs.index = pd.to_datetime(dfs["日期"], errors="coerce")
      return dfs[start_date:"2222-01-01"]