获取加密货币历史价格

Instructions

Implementation Reference

• mcp_aktools/__init__.py:372-407 (handler)

  The handler function that executes the 'okx_prices' tool logic. It fetches OKX cryptocurrency historical K-line data, processes it with technical indicators (MACD, KDJ, RSI, BOLL), and returns CSV-formatted results.

  def okx_prices(
      instId: str = Field("BTC-USDT", description="产品ID，格式: BTC-USDT"),
      bar: str = Field("1H", description="K线时间粒度，仅支持: [1m/3m/5m/15m/30m/1H/2H/4H/6H/12H/1D/2D/3D/1W/1M/3M] 除分钟为小写m外,其余均为大写"),
      limit: int = Field(100, description="返回数量(int)，最大300，最小建议30", strict=False),
  ):
      if not bar.endswith("m"):
          bar = bar.upper()
      res = requests.get(
          f"{OKX_BASE_URL}/api/v5/market/candles",
          params={
              "instId": instId,
              "bar": bar,
              "limit": max(300, limit + 62),
          },
          timeout=20,
      )
      data = res.json() or {}
      dfs = pd.DataFrame(data.get("data", []))
      if dfs.empty:
          return pd.DataFrame()
      dfs.columns = ["时间", "开盘", "最高", "最低", "收盘", "成交量", "成交额", "成交额USDT", "K线已完结"]
      dfs.sort_values("时间", inplace=True)
      dfs["时间"] = pd.to_datetime(dfs["时间"], errors="coerce", unit="ms")
      dfs["开盘"] = pd.to_numeric(dfs["开盘"], errors="coerce")
      dfs["最高"] = pd.to_numeric(dfs["最高"], errors="coerce")
      dfs["最低"] = pd.to_numeric(dfs["最低"], errors="coerce")
      dfs["收盘"] = pd.to_numeric(dfs["收盘"], errors="coerce")
      dfs["成交量"] = pd.to_numeric(dfs["成交量"], errors="coerce")
      dfs["成交额"] = pd.to_numeric(dfs["成交额"], errors="coerce")
      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:]])

• mcp_aktools/__init__.py:368-376 (registration)

  The registration decorator (@mcp.tool) that registers 'okx_prices' as an MCP tool with title '获取加密货币历史价格' and description.

  @mcp.tool(
      title="获取加密货币历史价格",
      description="获取OKX加密货币的历史K线数据，包括价格、交易量和技术指标",
  )
  def okx_prices(
      instId: str = Field("BTC-USDT", description="产品ID，格式: BTC-USDT"),
      bar: str = Field("1H", description="K线时间粒度，仅支持: [1m/3m/5m/15m/30m/1H/2H/4H/6H/12H/1D/2D/3D/1W/1M/3M] 除分钟为小写m外,其余均为大写"),
      limit: int = Field(100, description="返回数量(int)，最大300，最小建议30", strict=False),
  ):

• mcp_aktools/__init__.py:372-376 (schema)

  Input schema for okx_prices: instId (str, default BTC-USDT), bar (str, default 1H), limit (int, default 100). Uses Pydantic Field for validation.

  def okx_prices(
      instId: str = Field("BTC-USDT", description="产品ID，格式: BTC-USDT"),
      bar: str = Field("1H", description="K线时间粒度，仅支持: [1m/3m/5m/15m/30m/1H/2H/4H/6H/12H/1D/2D/3D/1W/1M/3M] 除分钟为小写m外,其余均为大写"),
      limit: int = Field(100, description="返回数量(int)，最大300，最小建议30", strict=False),
  ):

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

  Helper function add_technical_indicators that computes MACD, KDJ, RSI, and Bollinger Bands technical indicators, used by okx_prices and other tools.

  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