获取股票历史价格

Instructions

Implementation Reference

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

  Registration of 'stock_prices' tool via @mcp.tool decorator with title '获取股票历史价格' and description about getting historical prices and technical indicators

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

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

  Main handler function for stock_prices tool. Fetches historical stock prices from akshare for different markets (SH/SZ/HK/US/ETF), adds technical indicators (MACD, KDJ, RSI, Bollinger Bands), and returns CSV-formatted data limited to specified number of rows.

  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:111-118 (helper)

  Helper function for fetching US stock daily data via akshare, with column renaming for consistency with other market data

  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 for fetching ETF fund historical data via akshare/sina, used by stock_prices for ETF market types

  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"]

• mcp_aktools/__init__.py:581-610 (helper)

  Helper function that adds MACD, KDJ, RSI, and Bollinger Bands technical indicators to the dataframe, used by stock_prices and okx_prices

  def add_technical_indicators(df, clos, lows, high):
      # 计算MACD指标
      ema12 = clos.ewm(span=12, adjust=False).mean()
      ema26 = clos.ewm(span=26, adjust=False).mean()
      df["DIF"] = ema12 - ema26
      df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean()
      df["MACD"] = (df["DIF"] - df["DEA"]) * 2
  
      # 计算KDJ指标
      low_min  = lows.rolling(window=9, min_periods=1).min()
      high_max = high.rolling(window=9, min_periods=1).max()
      rsv = (clos - low_min) / (high_max - low_min) * 100
      df["KDJ.K"] = rsv.ewm(com=2, adjust=False).mean()
      df["KDJ.D"] = df["KDJ.K"].ewm(com=2, adjust=False).mean()
      df["KDJ.J"] = 3 * df["KDJ.K"] - 2 * df["KDJ.D"]
  
      # 计算RSI指标
      delta = clos.diff()
      gain = delta.where(delta > 0, 0)
      loss = -delta.where(delta < 0, 0)
      avg_gain = gain.rolling(window=14).mean()
      avg_loss = loss.rolling(window=14).mean()
      rs = avg_gain / avg_loss
      df["RSI"] = 100 - (100 / (1 + rs))
  
      # 计算布林带指标
      df["BOLL.M"] = clos.rolling(window=20).mean()
      std = clos.rolling(window=20).std()
      df["BOLL.U"] = df["BOLL.M"] + 2 * std
      df["BOLL.L"] = df["BOLL.M"] - 2 * std

• mcp_aktools/__init__.py:564-579 (helper)

  Generic caching helper used by stock_prices to cache akshare API results with dual-layer cache (in-memory TTL cache + disk cache)

  def ak_cache(fun, *args, **kwargs) -> pd.DataFrame | None:
      key = kwargs.pop("key", None)
      if not key:
          key = f"{fun.__name__}-{args}-{kwargs}"
      ttl1 = kwargs.pop("ttl", 86400)
      ttl2 = kwargs.pop("ttl2", None)
      cache = CacheKey.init(key, ttl1, ttl2)
      all = cache.get()
      if all is None:
          try:
              _LOGGER.info("Request akshare: %s", [key, args, kwargs])
              all = fun(*args, **kwargs)
              cache.set(all)
          except Exception as exc:
              _LOGGER.exception(str(exc))
      return all