okx_prices
Retrieve cryptocurrency candlestick data from OKX exchange for technical analysis and market monitoring. Access price history across multiple timeframes to inform trading decisions.
Instructions
获取OKX加密货币K线数据
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bar | No | K线时间粒度，仅支持: [1m/3m/5m/15m/30m/1H/2H/4H/6H/12H/1D/2D/3D/1W/1M/3M] 注意大小写，仅分钟为小写m | 1h |
| instId | No | 产品ID，格式: BTC-USDT | BTC-USDT |
| limit | No | 返回数量(int)，最大300，最小建议30 |
Implementation Reference
- mcp_aktools/__init__.py:368-407 (handler)Handler function for 'okx_prices' tool. Fetches OKX candlestick data via API, processes with pandas, adds technical indicators using add_technical_indicators helper, and returns formatted recent data as CSV text. Input schema defined via Pydantic Field in parameters. Registered via @mcp.tool decorator.@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:]])