okx_prices
Retrieve historical cryptocurrency price data from OKX exchange including candlestick patterns, trading volume, and technical indicators for market analysis.
Instructions
获取OKX加密货币的历史K线数据，包括价格、交易量和技术指标
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| instId | No | 产品ID，格式: BTC-USDT | BTC-USDT |
| bar | No | K线时间粒度，仅支持: [1m/3m/5m/15m/30m/1H/2H/4H/6H/12H/1D/2D/3D/1W/1M/3M] 除分钟为小写m外,其余均为大写 | 1H |
| limit | No | 返回数量(int)，最大300，最小建议30 |
Implementation Reference
- mcp_aktools/__init__.py:368-407 (handler)Primary handler for 'okx_prices' tool: decorated with @mcp.tool (includes registration and schema via parameters), fetches OKX candles API, processes DataFrame, adds technical indicators, returns formatted CSV output.@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), ): 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:581-611 (helper)Helper function called by okx_prices to compute technical indicators: MACD, DIF, DEA, KDJ.K/D/J, RSI, BOLL.U/M/L.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-580 (helper)Caching helper used in the codebase, though not directly in okx_prices, supports data fetching efficiency.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