MCP Server for stock and crypto

Instructions

Implementation Reference

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

  The implementation of the 'stock_prices' tool. This is the handler function decorated with @mcp.tool(), which registers the tool and defines its schema via parameter annotations. It fetches historical stock price data using akshare for various markets (sh, sz, hk, us, etf), caches results, computes technical indicators (MACD, KDJ, RSI, BOLL), and returns the last 'limit' rows in CSV format.

  @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-74 (registration)

  The @mcp.tool decorator registers the 'stock_prices' tool with FastMCP, specifying title and description. The schema is defined inline via Field descriptions in parameters.

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

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

  Helper function stock_us_daily used by stock_prices for US 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 fund_etf_hist_sina used by stock_prices for ETF data in sh/sz markets.

  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-611 (helper)

  Helper function to compute and add technical indicators used by stock_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