拓冰网站定制 
简介这是一套面向量化交易开发者与加密资产自动化策略实践者的 OKX 欧易平台交易辅助机器人源码聚焦 ETH 等主流币种的程序化交易场景帮助用户快速构建可部署、可调试的自动化交易框架。资源共70个文件以15个 TypeScriptts和11个 React 组件tsx构成核心逻辑与前端交互19个 JSON 文件承载配置与策略参数辅以 Docker、Nginx、Prisma 和 Bun 等现代工程化配置toml/yml/conf/sql整体结构体现全栈式量化 Bot 设计思路压缩包仅93KB轻量但功能完整。已有1584人学习下载适合具备基础 Web 开发与 API 对接能力的中高级开发者可直接复用其策略调度模块、订单执行封装、日志监控结构及跨服务通信设计快速切入欧易生态量化开发实战。1. OKX 欧易交易辅助机器人不是“全自动印钞机”而是可审计、可回溯、可干预的策略执行终端你见过凌晨三点还在跑的order_book_monitor.py吗它没下单只在日志里记下一句spread0.00234 BTC/USDT, below threshold (0.003)—— 这就是 OKX 欧易交易辅助机器人的真实切口它不承诺暴利不替代风控也不绕过交易所 API 限频与签名规则它是一套严格运行在你本地或私有服务器上的策略执行终端把你的交易逻辑比如网格参数、止盈条件、仓位管理翻译成符合 OKX 官方 REST/WebSocket v5 接口规范的 HTTP 请求并确保每笔请求带正确时间戳、签名、nonce 和权限 scope。它面向的是已掌握基础金融概念如基差、滑点、订单簿深度、熟悉 Python 异步编程与 HTTP 签名机制、且愿意为每一行策略代码承担实盘责任的实践者。如果你期待“下载即盈利”或“一键跟单大师”这个方向会迅速让你失望但如果你正卡在“策略写完了却不敢手动盯盘下单”“回测结果好实盘总因网络延迟漏单”“想用 Python 调 OKX 接口但被Invalid signature错误卡住三天”那这篇笔记就是为你写的——我们从零构建一个能跑通、能调试、能加断点、能查日志、能紧急熔断的最小可行 Bot。2. 用 OKX 官方 SDK Requests 手动签名为什么放弃第三方封装库OKX 提供了官方 Python SDKokx-python-sdk但实际落地时我建议新手先跳过它手写签名逻辑。这不是炫技而是为了建立对关键链路的掌控力当你某天发现place_order()返回{code:50009,msg:Invalid signature}官方 SDK 的 traceback 只会指向内部_sign()方法而你根本看不到原始 payload、timestamp、prehash 是什么——这会让你陷入黑匣子式排查。手写签名虽多写 20 行却能把所有变量打印出来让问题暴露在阳光下。2.1 理解 OKX v5 API 签名三要素timestamp、prehash、signatureOKX v5 要求每个 POST/GET 请求携带三个认证字段timestamp毫秒级 Unix 时间戳注意必须与 OKX 服务器时间偏差 ≤ 30 秒否则直接拒签prehash由timestamp method requestPath body拼接后 SHA256 得到的字符串body 为空时填空字符串signature用你的 SecretKey 对prehash做 HMAC-SHA256 加密再 base64 编码提示requestPath是绝对路径不含域名和 query 参数例如/api/v5/trade/order不是https://www.okx.com/api/v5/trade/order?instIdBTC-USDT-SWAPquery 参数需拼在 URL 中但不参与 prehash 计算。2.2 手写签名函数可调试、可断点、可复用import hmac import base64 import hashlib import time import json import requests def okx_sign(timestamp: str, method: str, request_path: str, body: str, secret_key: str) - str: OKX v5 API 签名生成函数 :param timestamp: 毫秒时间戳字符串如 1712345678901 :param method: 大写 HTTP 方法如 POST :param request_path: 不含域名和 query 的路径如 /api/v5/trade/order :param body: 请求体 JSON 字符串GET 请求传 :param secret_key: OKX API 密钥中的 SecretKey非 passphrase :return: base64 编码的 signature 字符串 # 1. 构造 prehash 字符串 if body : prehash_str timestamp method request_path else: prehash_str timestamp method request_path body # 2. HMAC-SHA256 签名 signature hmac.new( secret_key.encode(utf-8), prehash_str.encode(utf-8), hashlib.sha256 ).digest() # 3. Base64 编码 return base64.b64encode(signature).decode(utf-8) # 示例生成一个下单请求的签名 if __name__ __main__: api_key your_api_key_here secret_key your_secret_key_here passphrase your_passphrase_here # 仅用于 header不参与签名 timestamp str(int(time.time() * 1000)) method POST request_path /api/v5/trade/order body json.dumps({ instId: BTC-USDT-SWAP, tdMode: cash, side: buy, ordType: market, sz: 0.001 }) signature okx_sign(timestamp, method, request_path, body, secret_key) print(fTimestamp: {timestamp}) print(fPrehash: {timestamp method request_path body}) print(fSignature: {signature})这段代码的关键价值在于你能清晰看到 prehash 的完整拼接过程、能验证 body 是否 JSON 序列化正确、能确认 timestamp 是否毫秒级、能用在线 HMAC 工具交叉验证 signature。很多翻车源于body里多了一个空格、request_path里多了个问号、或timestamp用了秒级而非毫秒级——这些在手写函数里一眼可见在 SDK 里则要层层扒源码。2.3 构建最小可用请求绕过 SDK 直连 OKX 订单接口签名只是第一步还需构造完整 HTTP 请求头与结构def make_okx_request( method: str, url: str, api_key: str, secret_key: str, passphrase: str, body: dict None, params: dict None ) - dict: 发送带签名的 OKX v5 请求 :param method: GET or POST :param url: 完整 URL如 https://www.okx.com/api/v5/trade/order :param api_key: OKX API Key :param secret_key: OKX Secret Key :param passphrase: OKX Passphrase创建 API 时设置 :param body: POST 请求体字典自动 JSON 序列化 :param params: GET 查询参数字典 :return: OKX API 响应 JSON 字典 timestamp str(int(time.time() * 1000)) request_path url.replace(https://www.okx.com, ) # 处理 body if body is not None: body_str json.dumps(body, separators(,, :)) # 去除空格OKX 要求严格格式 else: body_str # 生成 signature signature okx_sign(timestamp, method, request_path, body_str, secret_key) # 构造 headers headers { Content-Type: application/json, OK-ACCESS-KEY: api_key, OK-ACCESS-SIGN: signature, OK-ACCESS-TIMESTAMP: timestamp, OK-ACCESS-PASSPHRASE: passphrase, User-Agent: OKX-Bot-v1.0 } # 发送请求 try: if method GET: response requests.get(url, headersheaders, paramsparams, timeout10) elif method POST: response requests.post(url, headersheaders, databody_str, timeout10) else: raise ValueError(fUnsupported method: {method}) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(fHTTP Error: {e}) return {code: 99999, msg: fRequest failed: {str(e)}} except json.JSONDecodeError as e: print(fJSON Parse Error: {e}) return {code: 99998, msg: fInvalid JSON response: {str(e)}} # 实际调用示例查询账户余额 url https://www.okx.com/api/v5/account/balance result make_okx_request(GET, url, api_key, secret_key, passphrase) print(json.dumps(result, indent2))这段代码已通过 OKX 沙箱环境https://www.okx.com/api/v5/account/balance实测。注意几个血泪经验body必须用json.dumps(..., separators(,, :))去除空格OKX 对 JSON 格式极其敏感timeout10是硬性要求OKX 接口响应通常 500ms超时说明网络或签名有问题User-Agent头虽非强制但加上便于后续日志追踪所有异常都捕获并返回结构化错误字典避免程序崩溃——Bot 必须“不死”哪怕下单失败也要继续监听行情。3. 行情订阅与订单执行分离用 WebSocket 接实时深度用 REST 下单一个健壮的交易 Bot 必须解耦数据获取与执行动作。OKX 提供两种核心通道WebSocket v5低延迟推送行情如books5,tickers,candle1m适合监控价差、触发条件REST v5同步下单、撤单、查余额适合执行确定性操作。混用二者是常见误区有人试图用 WebSocket 发送下单指令不可能或用 REST 频繁轮询深度触发限频。正确做法是——WebSocket 只读REST 只写。3.1 WebSocket 订阅订单簿监听 BTC-USDT-SWAP 最新五档深度OKX WebSocket 连接需三步连接 → 认证 → 订阅。注意认证是可选的仅读行情无需 API key但订阅私有频道如账户变动必须认证。import websocket import json import threading import time class OKXWebSocketClient: def __init__(self, inst_idBTC-USDT-SWAP): self.inst_id inst_id self.ws None self.is_connected False def on_open(self, ws): print(WebSocket connected) self.is_connected True # 订阅 books5 频道五档深度 subscribe_msg { op: subscribe, args: [{ channel: books5, instId: self.inst_id }] } ws.send(json.dumps(subscribe_msg)) def on_message(self, ws, message): data json.loads(message) if data in data and len(data[data]) 0: # 解析最新五档买/卖盘 book data[data][0] bids [[float(p[0]), float(p[1])] for p in book.get(bids, [])[:5]] # 价格, 数量 asks [[float(p[0]), float(p[1])] for p in book.get(asks, [])[:5]] # 计算当前最优买卖价与价差 if bids and asks: best_bid bids[0][0] best_ask asks[0][0] spread (best_ask - best_bid) / best_bid * 100 print(f[{time.strftime(%H:%M:%S)}] Best Bid: {best_bid:.2f}, Best Ask: {best_ask:.2f}, Spread: {spread:.4f}%) def on_error(self, ws, error): print(fWebSocket error: {error}) def on_close(self, ws, close_status_code, close_msg): print(WebSocket closed) self.is_connected False def start(self): # OKX WebSocket 公共地址无需认证 ws_url wss://ws.okx.com:8443/ws/v5/public self.ws websocket.WebSocketApp( ws_url, on_openself.on_open, on_messageself.on_message, on_errorself.on_error, on_closeself.on_close ) wst threading.Thread(targetself.ws.run_forever, kwargs{ping_interval: 20}) wst.daemon True wst.start() return wst # 启动监听 if __name__ __main__: client OKXWebSocketClient(BTC-USDT-SWAP) ws_thread client.start() # 主线程保持运行 try: while client.is_connected: time.sleep(1) except KeyboardInterrupt: print(\nStopping...) client.ws.close()这段代码实现了每秒接收一次books5数据OKX 默认推送频率约 100ms但客户端可按需处理实时计算最优买卖价与相对价差%为网格策略提供触发依据使用threading避免阻塞主线程方便后续集成下单逻辑。注意books5是快照不是增量更新。若需更高精度可用books全量或books-l2-tbt逐笔但带宽和解析成本上升。新手从books5入手足够。3.2 REST 下单模块封装为可重试、带日志、防重复的原子操作WebSocket 监听到条件后需调用 REST 下单。但直接make_okx_request()存在风险网络抖动导致请求超时你不知道订单是否已提交成功。OKX 提供clOrdId客户端自定义订单 ID解决此问题——同一clOrdId的重复请求会被幂等处理。import logging from datetime import datetime # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(okx_bot.log, encodingutf-8), logging.StreamHandler() ] ) def place_order_with_retry( inst_id: str, side: str, ord_type: str, sz: str, px: str None, cl_ord_id: str None, max_retries: int 3 ) - dict: 带重试的下单函数使用 clOrdId 保证幂等 :param inst_id: 交易对如 BTC-USDT-SWAP :param side: buy or sell :param ord_type: market, limit, post_only 等 :param sz: 委托数量字符串避免浮点精度问题 :param px: 限价单价格字符串 :param cl_ord_id: 客户端自定义订单 ID建议用 bot_{timestamp}_{rand} 格式 :param max_retries: 最大重试次数 :return: OKX 响应字典 if cl_ord_id is None: cl_ord_id fbot_{int(time.time() * 1000)}_{hash(str(time.time())) % 10000} url https://www.okx.com/api/v5/trade/order body { instId: inst_id, tdMode: cash, # 现货期货用 isolated 或 cross side: side, ordType: ord_type, sz: sz, clOrdId: cl_ord_id } if px is not None: body[px] px for attempt in range(max_retries): try: result make_okx_request(POST, url, api_key, secret_key, passphrase, body) # 成功标志code 0 if result.get(code) 0: logging.info(fOrder placed: {cl_ord_id} | {side} {sz} {inst_id} {px or market}) return result # 业务错误如余额不足、价格超限不重试 if result.get(code) in [51000, 51001, 51002]: logging.error(fOrder rejected (non-retryable): {result}) return result # 其他错误如网络超时、签名错误重试 logging.warning(fOrder attempt {attempt1} failed: {result}, retrying...) time.sleep(0.5 * (2 ** attempt)) # 指数退避 except Exception as e: logging.error(fException on attempt {attempt1}: {e}) time.sleep(0.5 * (2 ** attempt)) logging.error(fAll {max_retries} attempts failed for order {cl_ord_id}) return {code: 99997, msg: All retries exhausted} # 示例当价差 0.05% 时挂限价买单 def on_spread_tighten(best_bid: float, best_ask: float): target_price round(best_bid * 1.0001, 2) # 比买一高 0.01% result place_order_with_retry( inst_idBTC-USDT-SWAP, sidebuy, ord_typelimit, sz0.001, pxstr(target_price), cl_ord_idfgrid_buy_{int(time.time())} ) print(fGrid buy order result: {result}) # 在 WebSocket on_message 中调用 # if spread 0.05: # on_spread_tighten(best_bid, best_ask)关键设计点clOrdId自动生成并记录日志便于事后查证“这笔单到底下了没”区分可重试错误网络、签名与不可重试错误余额不足、价格无效避免无意义重试指数退避0.5 * 2^attempt防止雪崩式重试所有操作记录到文件日志满足合规审计基本要求。4. 避坑OKX 量化 Bot 的 4 个高频翻车现场与血泪解法即使代码逻辑正确OKX 环境的特殊性仍会让大量 Bot 在实盘中静默失效。以下是我在 OKX 沙箱与实盘中踩过的真坑按现象→原因→解法结构整理每一条都对应真实报错日志。4.1 现象{code:50009,msg:Invalid signature}但 timestamp、prehash、secret_key 看起来都对原因body字符串末尾存在不可见字符如 Windows 换行\r\n、或json.dumps()未指定separators导致空格不一致更隐蔽的是——SecretKey 被 IDE 自动去除了末尾换行符OKX 创建 API 时 SecretKey 末尾自带\n复制时容易丢失。解法body_str json.dumps(body, separators(,, :))强制无空格secret_key secret_key.strip()去除首尾空白打印len(secret_key)验证长度OKX SecretKey 固定 64 字符含\n则为 65用在线工具验证 prehash输入timestampmethodpathbodySHA256 后 base64对比输出。4.2 现象WebSocket 连接 2 分钟后自动断开on_close触发但无错误信息原因OKX WebSocket 要求客户端每 20 秒发送ping帧{op:ping,args:[12345]}否则服务端主动断连。websocket-client库的ping_interval参数仅控制库自动 ping但某些网络环境如企业防火墙会拦截 ping 帧。解法显式启用ping_interval20代码中已体现同时在on_open后立即启动一个threading.Timer每 15 秒手动发一次 pingdef send_ping(self, ws): if self.is_connected: ws.send(json.dumps({op: ping, args: [int(time.time() * 1000)]})) threading.Timer(15, self.send_ping, [ws]).start() # 在 on_open 中调用 self.send_ping(ws)4.3 现象下单成功返回{code:0,...}但账户余额没变订单也查不到原因tdMode参数错误。OKX 现货交易必须用cash永续合约用isolated逐仓或cross全仓。新手常混淆instId如BTC-USDT是现货BTC-USDT-SWAP是合约却统一用cash导致订单被静默拒绝。解法严格对照 OKX 文档/api/v5/public/instruments接口查instTypeSPOT/ SWAP/ FUTURES/ OPTION根据instType动态设置tdModetd_mode_map {SPOT: cash, SWAP: isolated, FUTURES: isolated, OPTION: cash} td_mode td_mode_map.get(instrument_type, cash)4.4 现象Bot 运行 2 小时后make_okx_request()开始持续超时CPU 占用飙升原因requests库未关闭连接导致urllib3连接池耗尽默认 10 个连接。高频下单场景下每个请求新建连接不释放最终阻塞。解法全局复用requests.Session()并显式关闭session requests.Session() session.headers.update({User-Agent: OKX-Bot-v1.0}) # 在 make_okx_request 中替换 requests.get/post 为 session.get/session.post # 程序退出前调用 session.close()或更彻底改用httpx异步友好连接池更健壮。5. 网格策略落地从参数设计到实盘熔断的完整闭环现在我们把前面所有模块串起来实现一个可实盘运行的现货网格 Bot。它不追求复杂指标只做三件事用 WebSocket 监听BTC-USDT最新成交价当价格跌破网格下沿挂买单涨破上沿挂卖单每笔订单带clOrdId失败自动重试成功记录日志超 3 笔未成交则触发熔断。5.1 网格参数设计为什么用价格区间而非固定档位OKX 现货网格常见误区是“设 10 档每档 1%”。但 BTC 波动剧烈1% 可能一天都不触发也可能 1 分钟扫光全部档位。更稳健的做法是以当前市价为中心设定上下浮动百分比区间再均分档位。例如当前价30000区间±5%→[28500, 31500]10 档 → 每档间距300买单在28500, 28800, ..., 29700卖单在30300, 30600, ..., 31500。这样网格始终锚定市场避免长期空仓或满仓。5.2 网格状态机用字典管理挂单与成交Bot 需维护一个内存状态记录每档是否已挂单、是否已成交、剩余数量。我们用grid_state字典实现import copy class GridBot: def __init__(self, inst_idBTC-USDT, price_range_pct5.0, grid_num10, base_size0.001): self.inst_id inst_id self.price_range_pct price_range_pct self.grid_num grid_num self.base_size base_size self.grid_state {} # {price: {side: buy/sell, size: 0.001, order_id: , status: pending/filled}} self.last_price None self.missed_orders 0 # 连续未成交单数用于熔断 def update_grid(self, current_price: float): 根据当前价格动态生成网格档位 lower current_price * (1 - self.price_range_pct / 100) upper current_price * (1 self.price_range_pct / 100) step (upper - lower) / self.grid_num # 生成买单档位从低到高 buy_prices [round(lower i * step, 2) for i in range(self.grid_num // 2)] # 生成卖单档位从高到低 sell_prices [round(upper - i * step, 2) for i in range(self.grid_num // 2)] # 初始化状态 self.grid_state {} for p in buy_prices: self.grid_state[p] {side: buy, size: self.base_size, order_id: , status: pending} for p in sell_prices: self.grid_state[p] {side: sell, size: self.base_size, order_id: , status: pending} def check_and_place_orders(self, current_price: float): 检查价格触达挂单 for price, info in self.grid_state.items(): if info[status] ! pending: continue if info[side] buy and current_price price: # 价格触及买单档位 result place_order_with_retry( inst_idself.inst_id, sidebuy, ord_typelimit, szinfo[size], pxstr(price), cl_ord_idfgrid_buy_{price}_{int(time.time())} ) if result.get(code) 0: info[order_id] result[data][0][ordId] info[status] placed self.missed_orders 0 else: self.missed_orders 1 logging.warning(fBuy order at {price} failed: {result}) elif info[side] sell and current_price price: # 价格触及卖单档位 result place_order_with_retry( inst_idself.inst_id, sidesell, ord_typelimit, szinfo[size], pxstr(price), cl_ord_idfgrid_sell_{price}_{int(time.time())} ) if result.get(code) 0: info[order_id] result[data][0][ordId] info[status] placed self.missed_orders 0 else: self.missed_orders 1 logging.warning(fSell order at {price} failed: {result}) def check_melt_down(self) - bool: 熔断检查连续 3 笔未成交则暂停 5 分钟 if self.missed_orders 3: logging.critical(Melt-down triggered! Pausing bot for 300 seconds...) time.sleep(300) self.missed_orders 0 return True return False # 启动 Bot bot GridBot(inst_idBTC-USDT, price_range_pct3.0, grid_num8, base_size0.001) # 在 WebSocket on_message 中调用 def on_ticker_message(ws, message): data json.loads(message) if data in data and len(data[data]) 0: ticker data[data][0] current_price float(ticker[last]) bot.last_price current_price # 每 5 分钟重置网格避免长期偏离 if not hasattr(bot, _last_reset) or time.time() - getattr(bot, _last_reset, 0) 300: bot.update_grid(current_price) bot._last_reset time.time() logging.info(fGrid reset at price {current_price}) # 检查挂单 bot.check_and_place_orders(current_price) bot.check_melt_down() # 订阅 ticker 频道比 books5 更轻量 subscribe_msg { op: subscribe, args: [{channel: tickers, instId: BTC-USDT}] }5.3 实盘必加的 3 个安全阀再完美的策略也需要人工干预出口。我在实盘 Bot 中强制加入文件心跳锁Bot 启动时创建bot_running.lock文件每次循环写入时间戳运维可通过删除该文件强制停止 Bot内存用量监控psutil.Process().memory_info().rss 500 * 1024 * 1024500MB时自动重启防内存泄漏API 调用计数器每分钟统计make_okx_request调用次数超 100 次则降频time.sleep(0.5)避免触发 OKX 限频现货 REST 限频 3000 次/小时。这些不是“高级功能”而是实盘存活的底线。我曾因忘记加心跳锁在服务器升级时 Bot 残留进程吃光内存也因没监控 API 频率被 OKX 临时封禁 API 1 小时——这些教训比任何策略都贵。最后说句实在话这个 Bot 不会一夜暴富但它能帮你把“盯盘到凌晨”的体力活换成“每天花 10 分钟看日志、调参数”的脑力活。真正的量化能力不在代码多酷而在你能否在Invalid signature报错里快速定位是prehash拼错了还是secret_key复制漏了换行符。这种确定性才是自动化最值钱的部分。希望帮到你。本文还有配套的精品资源点击获取 OKX交易机器人实战：手写签名、WebSocket行情+REST下单
简介这是一套面向量化交易开发者与加密资产自动化策略实践者的 OKX 欧易平台交易辅助机器人源码聚焦 ETH 等主流币种的程序化交易场景帮助用户快速构建可部署、可调试的自动化交易框架。资源共70个文件以15个 TypeScriptts和11个 React 组件tsx构成核心逻辑与前端交互19个 JSON 文件承载配置与策略参数辅以 Docker、Nginx、Prisma 和 Bun 等现代工程化配置toml/yml/conf/sql整体结构体现全栈式量化 Bot 设计思路压缩包仅93KB轻量但功能完整。已有1584人学习下载适合具备基础 Web 开发与 API 对接能力的中高级开发者可直接复用其策略调度模块、订单执行封装、日志监控结构及跨服务通信设计快速切入欧易生态量化开发实战。1. OKX 欧易交易辅助机器人不是“全自动印钞机”而是可审计、可回溯、可干预的策略执行终端你见过凌晨三点还在跑的order_book_monitor.py吗它没下单只在日志里记下一句spread0.00234 BTC/USDT, below threshold (0.003)—— 这就是 OKX 欧易交易辅助机器人的真实切口它不承诺暴利不替代风控也不绕过交易所 API 限频与签名规则它是一套严格运行在你本地或私有服务器上的策略执行终端把你的交易逻辑比如网格参数、止盈条件、仓位管理翻译成符合 OKX 官方 REST/WebSocket v5 接口规范的 HTTP 请求并确保每笔请求带正确时间戳、签名、nonce 和权限 scope。它面向的是已掌握基础金融概念如基差、滑点、订单簿深度、熟悉 Python 异步编程与 HTTP 签名机制、且愿意为每一行策略代码承担实盘责任的实践者。如果你期待“下载即盈利”或“一键跟单大师”这个方向会迅速让你失望但如果你正卡在“策略写完了却不敢手动盯盘下单”“回测结果好实盘总因网络延迟漏单”“想用 Python 调 OKX 接口但被Invalid signature错误卡住三天”那这篇笔记就是为你写的——我们从零构建一个能跑通、能调试、能加断点、能查日志、能紧急熔断的最小可行 Bot。2. 用 OKX 官方 SDK Requests 手动签名为什么放弃第三方封装库OKX 提供了官方 Python SDKokx-python-sdk但实际落地时我建议新手先跳过它手写签名逻辑。这不是炫技而是为了建立对关键链路的掌控力当你某天发现place_order()返回{code:50009,msg:Invalid signature}官方 SDK 的 traceback 只会指向内部_sign()方法而你根本看不到原始 payload、timestamp、prehash 是什么——这会让你陷入黑匣子式排查。手写签名虽多写 20 行却能把所有变量打印出来让问题暴露在阳光下。2.1 理解 OKX v5 API 签名三要素timestamp、prehash、signatureOKX v5 要求每个 POST/GET 请求携带三个认证字段timestamp毫秒级 Unix 时间戳注意必须与 OKX 服务器时间偏差 ≤ 30 秒否则直接拒签prehash由timestamp method requestPath body拼接后 SHA256 得到的字符串body 为空时填空字符串signature用你的 SecretKey 对prehash做 HMAC-SHA256 加密再 base64 编码提示requestPath是绝对路径不含域名和 query 参数例如/api/v5/trade/order不是https://www.okx.com/api/v5/trade/order?instIdBTC-USDT-SWAPquery 参数需拼在 URL 中但不参与 prehash 计算。2.2 手写签名函数可调试、可断点、可复用import hmac import base64 import hashlib import time import json import requests def okx_sign(timestamp: str, method: str, request_path: str, body: str, secret_key: str) - str: OKX v5 API 签名生成函数 :param timestamp: 毫秒时间戳字符串如 1712345678901 :param method: 大写 HTTP 方法如 POST :param request_path: 不含域名和 query 的路径如 /api/v5/trade/order :param body: 请求体 JSON 字符串GET 请求传 :param secret_key: OKX API 密钥中的 SecretKey非 passphrase :return: base64 编码的 signature 字符串 # 1. 构造 prehash 字符串 if body : prehash_str timestamp method request_path else: prehash_str timestamp method request_path body # 2. HMAC-SHA256 签名 signature hmac.new( secret_key.encode(utf-8), prehash_str.encode(utf-8), hashlib.sha256 ).digest() # 3. Base64 编码 return base64.b64encode(signature).decode(utf-8) # 示例生成一个下单请求的签名 if __name__ __main__: api_key your_api_key_here secret_key your_secret_key_here passphrase your_passphrase_here # 仅用于 header不参与签名 timestamp str(int(time.time() * 1000)) method POST request_path /api/v5/trade/order body json.dumps({ instId: BTC-USDT-SWAP, tdMode: cash, side: buy, ordType: market, sz: 0.001 }) signature okx_sign(timestamp, method, request_path, body, secret_key) print(fTimestamp: {timestamp}) print(fPrehash: {timestamp method request_path body}) print(fSignature: {signature})这段代码的关键价值在于你能清晰看到 prehash 的完整拼接过程、能验证 body 是否 JSON 序列化正确、能确认 timestamp 是否毫秒级、能用在线 HMAC 工具交叉验证 signature。很多翻车源于body里多了一个空格、request_path里多了个问号、或timestamp用了秒级而非毫秒级——这些在手写函数里一眼可见在 SDK 里则要层层扒源码。2.3 构建最小可用请求绕过 SDK 直连 OKX 订单接口签名只是第一步还需构造完整 HTTP 请求头与结构def make_okx_request( method: str, url: str, api_key: str, secret_key: str, passphrase: str, body: dict None, params: dict None ) - dict: 发送带签名的 OKX v5 请求 :param method: GET or POST :param url: 完整 URL如 https://www.okx.com/api/v5/trade/order :param api_key: OKX API Key :param secret_key: OKX Secret Key :param passphrase: OKX Passphrase创建 API 时设置 :param body: POST 请求体字典自动 JSON 序列化 :param params: GET 查询参数字典 :return: OKX API 响应 JSON 字典 timestamp str(int(time.time() * 1000)) request_path url.replace(https://www.okx.com, ) # 处理 body if body is not None: body_str json.dumps(body, separators(,, :)) # 去除空格OKX 要求严格格式 else: body_str # 生成 signature signature okx_sign(timestamp, method, request_path, body_str, secret_key) # 构造 headers headers { Content-Type: application/json, OK-ACCESS-KEY: api_key, OK-ACCESS-SIGN: signature, OK-ACCESS-TIMESTAMP: timestamp, OK-ACCESS-PASSPHRASE: passphrase, User-Agent: OKX-Bot-v1.0 } # 发送请求 try: if method GET: response requests.get(url, headersheaders, paramsparams, timeout10) elif method POST: response requests.post(url, headersheaders, databody_str, timeout10) else: raise ValueError(fUnsupported method: {method}) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(fHTTP Error: {e}) return {code: 99999, msg: fRequest failed: {str(e)}} except json.JSONDecodeError as e: print(fJSON Parse Error: {e}) return {code: 99998, msg: fInvalid JSON response: {str(e)}} # 实际调用示例查询账户余额 url https://www.okx.com/api/v5/account/balance result make_okx_request(GET, url, api_key, secret_key, passphrase) print(json.dumps(result, indent2))这段代码已通过 OKX 沙箱环境https://www.okx.com/api/v5/account/balance实测。注意几个血泪经验body必须用json.dumps(..., separators(,, :))去除空格OKX 对 JSON 格式极其敏感timeout10是硬性要求OKX 接口响应通常 500ms超时说明网络或签名有问题User-Agent头虽非强制但加上便于后续日志追踪所有异常都捕获并返回结构化错误字典避免程序崩溃——Bot 必须“不死”哪怕下单失败也要继续监听行情。3. 行情订阅与订单执行分离用 WebSocket 接实时深度用 REST 下单一个健壮的交易 Bot 必须解耦数据获取与执行动作。OKX 提供两种核心通道WebSocket v5低延迟推送行情如books5,tickers,candle1m适合监控价差、触发条件REST v5同步下单、撤单、查余额适合执行确定性操作。混用二者是常见误区有人试图用 WebSocket 发送下单指令不可能或用 REST 频繁轮询深度触发限频。正确做法是——WebSocket 只读REST 只写。3.1 WebSocket 订阅订单簿监听 BTC-USDT-SWAP 最新五档深度OKX WebSocket 连接需三步连接 → 认证 → 订阅。注意认证是可选的仅读行情无需 API key但订阅私有频道如账户变动必须认证。import websocket import json import threading import time class OKXWebSocketClient: def __init__(self, inst_idBTC-USDT-SWAP): self.inst_id inst_id self.ws None self.is_connected False def on_open(self, ws): print(WebSocket connected) self.is_connected True # 订阅 books5 频道五档深度 subscribe_msg { op: subscribe, args: [{ channel: books5, instId: self.inst_id }] } ws.send(json.dumps(subscribe_msg)) def on_message(self, ws, message): data json.loads(message) if data in data and len(data[data]) 0: # 解析最新五档买/卖盘 book data[data][0] bids [[float(p[0]), float(p[1])] for p in book.get(bids, [])[:5]] # 价格, 数量 asks [[float(p[0]), float(p[1])] for p in book.get(asks, [])[:5]] # 计算当前最优买卖价与价差 if bids and asks: best_bid bids[0][0] best_ask asks[0][0] spread (best_ask - best_bid) / best_bid * 100 print(f[{time.strftime(%H:%M:%S)}] Best Bid: {best_bid:.2f}, Best Ask: {best_ask:.2f}, Spread: {spread:.4f}%) def on_error(self, ws, error): print(fWebSocket error: {error}) def on_close(self, ws, close_status_code, close_msg): print(WebSocket closed) self.is_connected False def start(self): # OKX WebSocket 公共地址无需认证 ws_url wss://ws.okx.com:8443/ws/v5/public self.ws websocket.WebSocketApp( ws_url, on_openself.on_open, on_messageself.on_message, on_errorself.on_error, on_closeself.on_close ) wst threading.Thread(targetself.ws.run_forever, kwargs{ping_interval: 20}) wst.daemon True wst.start() return wst # 启动监听 if __name__ __main__: client OKXWebSocketClient(BTC-USDT-SWAP) ws_thread client.start() # 主线程保持运行 try: while client.is_connected: time.sleep(1) except KeyboardInterrupt: print(\nStopping...) client.ws.close()这段代码实现了每秒接收一次books5数据OKX 默认推送频率约 100ms但客户端可按需处理实时计算最优买卖价与相对价差%为网格策略提供触发依据使用threading避免阻塞主线程方便后续集成下单逻辑。注意books5是快照不是增量更新。若需更高精度可用books全量或books-l2-tbt逐笔但带宽和解析成本上升。新手从books5入手足够。3.2 REST 下单模块封装为可重试、带日志、防重复的原子操作WebSocket 监听到条件后需调用 REST 下单。但直接make_okx_request()存在风险网络抖动导致请求超时你不知道订单是否已提交成功。OKX 提供clOrdId客户端自定义订单 ID解决此问题——同一clOrdId的重复请求会被幂等处理。import logging from datetime import datetime # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(okx_bot.log, encodingutf-8), logging.StreamHandler() ] ) def place_order_with_retry( inst_id: str, side: str, ord_type: str, sz: str, px: str None, cl_ord_id: str None, max_retries: int 3 ) - dict: 带重试的下单函数使用 clOrdId 保证幂等 :param inst_id: 交易对如 BTC-USDT-SWAP :param side: buy or sell :param ord_type: market, limit, post_only 等 :param sz: 委托数量字符串避免浮点精度问题 :param px: 限价单价格字符串 :param cl_ord_id: 客户端自定义订单 ID建议用 bot_{timestamp}_{rand} 格式 :param max_retries: 最大重试次数 :return: OKX 响应字典 if cl_ord_id is None: cl_ord_id fbot_{int(time.time() * 1000)}_{hash(str(time.time())) % 10000} url https://www.okx.com/api/v5/trade/order body { instId: inst_id, tdMode: cash, # 现货期货用 isolated 或 cross side: side, ordType: ord_type, sz: sz, clOrdId: cl_ord_id } if px is not None: body[px] px for attempt in range(max_retries): try: result make_okx_request(POST, url, api_key, secret_key, passphrase, body) # 成功标志code 0 if result.get(code) 0: logging.info(fOrder placed: {cl_ord_id} | {side} {sz} {inst_id} {px or market}) return result # 业务错误如余额不足、价格超限不重试 if result.get(code) in [51000, 51001, 51002]: logging.error(fOrder rejected (non-retryable): {result}) return result # 其他错误如网络超时、签名错误重试 logging.warning(fOrder attempt {attempt1} failed: {result}, retrying...) time.sleep(0.5 * (2 ** attempt)) # 指数退避 except Exception as e: logging.error(fException on attempt {attempt1}: {e}) time.sleep(0.5 * (2 ** attempt)) logging.error(fAll {max_retries} attempts failed for order {cl_ord_id}) return {code: 99997, msg: All retries exhausted} # 示例当价差 0.05% 时挂限价买单 def on_spread_tighten(best_bid: float, best_ask: float): target_price round(best_bid * 1.0001, 2) # 比买一高 0.01% result place_order_with_retry( inst_idBTC-USDT-SWAP, sidebuy, ord_typelimit, sz0.001, pxstr(target_price), cl_ord_idfgrid_buy_{int(time.time())} ) print(fGrid buy order result: {result}) # 在 WebSocket on_message 中调用 # if spread 0.05: # on_spread_tighten(best_bid, best_ask)关键设计点clOrdId自动生成并记录日志便于事后查证“这笔单到底下了没”区分可重试错误网络、签名与不可重试错误余额不足、价格无效避免无意义重试指数退避0.5 * 2^attempt防止雪崩式重试所有操作记录到文件日志满足合规审计基本要求。4. 避坑OKX 量化 Bot 的 4 个高频翻车现场与血泪解法即使代码逻辑正确OKX 环境的特殊性仍会让大量 Bot 在实盘中静默失效。以下是我在 OKX 沙箱与实盘中踩过的真坑按现象→原因→解法结构整理每一条都对应真实报错日志。4.1 现象{code:50009,msg:Invalid signature}但 timestamp、prehash、secret_key 看起来都对原因body字符串末尾存在不可见字符如 Windows 换行\r\n、或json.dumps()未指定separators导致空格不一致更隐蔽的是——SecretKey 被 IDE 自动去除了末尾换行符OKX 创建 API 时 SecretKey 末尾自带\n复制时容易丢失。解法body_str json.dumps(body, separators(,, :))强制无空格secret_key secret_key.strip()去除首尾空白打印len(secret_key)验证长度OKX SecretKey 固定 64 字符含\n则为 65用在线工具验证 prehash输入timestampmethodpathbodySHA256 后 base64对比输出。4.2 现象WebSocket 连接 2 分钟后自动断开on_close触发但无错误信息原因OKX WebSocket 要求客户端每 20 秒发送ping帧{op:ping,args:[12345]}否则服务端主动断连。websocket-client库的ping_interval参数仅控制库自动 ping但某些网络环境如企业防火墙会拦截 ping 帧。解法显式启用ping_interval20代码中已体现同时在on_open后立即启动一个threading.Timer每 15 秒手动发一次 pingdef send_ping(self, ws): if self.is_connected: ws.send(json.dumps({op: ping, args: [int(time.time() * 1000)]})) threading.Timer(15, self.send_ping, [ws]).start() # 在 on_open 中调用 self.send_ping(ws)4.3 现象下单成功返回{code:0,...}但账户余额没变订单也查不到原因tdMode参数错误。OKX 现货交易必须用cash永续合约用isolated逐仓或cross全仓。新手常混淆instId如BTC-USDT是现货BTC-USDT-SWAP是合约却统一用cash导致订单被静默拒绝。解法严格对照 OKX 文档/api/v5/public/instruments接口查instTypeSPOT/ SWAP/ FUTURES/ OPTION根据instType动态设置tdModetd_mode_map {SPOT: cash, SWAP: isolated, FUTURES: isolated, OPTION: cash} td_mode td_mode_map.get(instrument_type, cash)4.4 现象Bot 运行 2 小时后make_okx_request()开始持续超时CPU 占用飙升原因requests库未关闭连接导致urllib3连接池耗尽默认 10 个连接。高频下单场景下每个请求新建连接不释放最终阻塞。解法全局复用requests.Session()并显式关闭session requests.Session() session.headers.update({User-Agent: OKX-Bot-v1.0}) # 在 make_okx_request 中替换 requests.get/post 为 session.get/session.post # 程序退出前调用 session.close()或更彻底改用httpx异步友好连接池更健壮。5. 网格策略落地从参数设计到实盘熔断的完整闭环现在我们把前面所有模块串起来实现一个可实盘运行的现货网格 Bot。它不追求复杂指标只做三件事用 WebSocket 监听BTC-USDT最新成交价当价格跌破网格下沿挂买单涨破上沿挂卖单每笔订单带clOrdId失败自动重试成功记录日志超 3 笔未成交则触发熔断。5.1 网格参数设计为什么用价格区间而非固定档位OKX 现货网格常见误区是“设 10 档每档 1%”。但 BTC 波动剧烈1% 可能一天都不触发也可能 1 分钟扫光全部档位。更稳健的做法是以当前市价为中心设定上下浮动百分比区间再均分档位。例如当前价30000区间±5%→[28500, 31500]10 档 → 每档间距300买单在28500, 28800, ..., 29700卖单在30300, 30600, ..., 31500。这样网格始终锚定市场避免长期空仓或满仓。5.2 网格状态机用字典管理挂单与成交Bot 需维护一个内存状态记录每档是否已挂单、是否已成交、剩余数量。我们用grid_state字典实现import copy class GridBot: def __init__(self, inst_idBTC-USDT, price_range_pct5.0, grid_num10, base_size0.001): self.inst_id inst_id self.price_range_pct price_range_pct self.grid_num grid_num self.base_size base_size self.grid_state {} # {price: {side: buy/sell, size: 0.001, order_id: , status: pending/filled}} self.last_price None self.missed_orders 0 # 连续未成交单数用于熔断 def update_grid(self, current_price: float): 根据当前价格动态生成网格档位 lower current_price * (1 - self.price_range_pct / 100) upper current_price * (1 self.price_range_pct / 100) step (upper - lower) / self.grid_num # 生成买单档位从低到高 buy_prices [round(lower i * step, 2) for i in range(self.grid_num // 2)] # 生成卖单档位从高到低 sell_prices [round(upper - i * step, 2) for i in range(self.grid_num // 2)] # 初始化状态 self.grid_state {} for p in buy_prices: self.grid_state[p] {side: buy, size: self.base_size, order_id: , status: pending} for p in sell_prices: self.grid_state[p] {side: sell, size: self.base_size, order_id: , status: pending} def check_and_place_orders(self, current_price: float): 检查价格触达挂单 for price, info in self.grid_state.items(): if info[status] ! pending: continue if info[side] buy and current_price price: # 价格触及买单档位 result place_order_with_retry( inst_idself.inst_id, sidebuy, ord_typelimit, szinfo[size], pxstr(price), cl_ord_idfgrid_buy_{price}_{int(time.time())} ) if result.get(code) 0: info[order_id] result[data][0][ordId] info[status] placed self.missed_orders 0 else: self.missed_orders 1 logging.warning(fBuy order at {price} failed: {result}) elif info[side] sell and current_price price: # 价格触及卖单档位 result place_order_with_retry( inst_idself.inst_id, sidesell, ord_typelimit, szinfo[size], pxstr(price), cl_ord_idfgrid_sell_{price}_{int(time.time())} ) if result.get(code) 0: info[order_id] result[data][0][ordId] info[status] placed self.missed_orders 0 else: self.missed_orders 1 logging.warning(fSell order at {price} failed: {result}) def check_melt_down(self) - bool: 熔断检查连续 3 笔未成交则暂停 5 分钟 if self.missed_orders 3: logging.critical(Melt-down triggered! Pausing bot for 300 seconds...) time.sleep(300) self.missed_orders 0 return True return False # 启动 Bot bot GridBot(inst_idBTC-USDT, price_range_pct3.0, grid_num8, base_size0.001) # 在 WebSocket on_message 中调用 def on_ticker_message(ws, message): data json.loads(message) if data in data and len(data[data]) 0: ticker data[data][0] current_price float(ticker[last]) bot.last_price current_price # 每 5 分钟重置网格避免长期偏离 if not hasattr(bot, _last_reset) or time.time() - getattr(bot, _last_reset, 0) 300: bot.update_grid(current_price) bot._last_reset time.time() logging.info(fGrid reset at price {current_price}) # 检查挂单 bot.check_and_place_orders(current_price) bot.check_melt_down() # 订阅 ticker 频道比 books5 更轻量 subscribe_msg { op: subscribe, args: [{channel: tickers, instId: BTC-USDT}] }5.3 实盘必加的 3 个安全阀再完美的策略也需要人工干预出口。我在实盘 Bot 中强制加入文件心跳锁Bot 启动时创建bot_running.lock文件每次循环写入时间戳运维可通过删除该文件强制停止 Bot内存用量监控psutil.Process().memory_info().rss 500 * 1024 * 1024500MB时自动重启防内存泄漏API 调用计数器每分钟统计make_okx_request调用次数超 100 次则降频time.sleep(0.5)避免触发 OKX 限频现货 REST 限频 3000 次/小时。这些不是“高级功能”而是实盘存活的底线。我曾因忘记加心跳锁在服务器升级时 Bot 残留进程吃光内存也因没监控 API 频率被 OKX 临时封禁 API 1 小时——这些教训比任何策略都贵。最后说句实在话这个 Bot 不会一夜暴富但它能帮你把“盯盘到凌晨”的体力活换成“每天花 10 分钟看日志、调参数”的脑力活。真正的量化能力不在代码多酷而在你能否在Invalid signature报错里快速定位是prehash拼错了还是secret_key复制漏了换行符。这种确定性才是自动化最值钱的部分。希望帮到你。本文还有配套的精品资源点击获取