MCP Server for stock and crypto

Instructions

Implementation Reference

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

  The core handler function for the 'okx_prices' tool. It fetches historical candlestick data from OKX API, processes it into a DataFrame, adds technical indicators (MACD, KDJ, RSI, Bollinger Bands), and returns a formatted CSV snippet.
  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-371 (registration)

  The @mcp.tool decorator that registers the 'okx_prices' function as an MCP tool with its title and description.
  @mcp.tool( title="获取加密货币历史价格", description="获取OKX加密货币的历史K线数据，包括价格、交易量和技术指标", )
• mcp_aktools/__init__.py:372-376 (schema)

  Pydantic Field definitions for the input parameters: instId (product ID), bar (time granularity), limit (number of records). These define the tool's input schema.
  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-611 (helper)

  Helper function that adds technical indicators (MACD, DIF, DEA, KDJ.K/D/J, RSI, BOLL.U/M/L) to the price DataFrame, used by 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