简介OKX欧易交易辅助机器人是一套面向加密货币量化交易场景的自动化Bot源码工程适合有TS/Node基础、希望借助OKX API实现策略执行与仓位管理的开发者参考。压缩包共70个文件、约93KB以JSON配置、TS/TSX源码、Markdown文档及SQL/Prisma数据模型为主另有容器编排与工程配置。TS源码覆盖核心交易逻辑与Web界面JSON负责策略与模块配置文档和数据库模型便于快速理解项目也可快速定位前后端与定时任务的实现。工程按共享模块、Web前端、定时任务分层能清晰看到交易所接入、行情获取、订单管理和策略执行的协作方式。借助源码可系统学习量化机器人架构理解其在订单管理与仓位配置上的设计并以此为基础做二次开发与定制。目前已有1600人浏览学习适合对交易所API自动化交易感兴趣的进阶开发者。1. OKX 欧易交易辅助机器人它到底帮你省下哪一步一个人开着三四个网格、盯两套均线、手里还捏着止损单这种状态下点错一次按钮就要为情绪买单。量化交易自动化 Bot 干的事就是把“你事先想清楚的规则”变成一条流水线程序听行情、算信号、自动下单撤单你只在旁边看日志。它不是预测涨跌的黑匣子也不会让你的策略突然变神它最大的价值是消灭手动操作里的犹豫和延迟。适合手里已有一套明确规则、不想做高频、愿意先花一周跑模拟盘的人。它的边界也很清楚规则写得烂自动化只会让你更快亏完所以真正值得投入的是规则本身和规则外面的那层风控壳。2. 先定协议再写策略WebSocket 行情流与 REST 下单的职责划分2.1 为什么行情走 WebSocket、下单走 REST做这个 Bot 第一件事不是写策略是选定数据通道和下单通道。我的习惯是行情一律走 WebSocket交易操作一律走 REST。原因很直接WebSocket 是一条长连接服务器有行情就推给你延迟低、不用轮询适合订阅 ticker、K 线、盘口深度REST 是请求-响应模式每次请求都带独立签名适合下单、撤单、查余额这类低频但必须拿到明确返回结果的操作。这两条通道分开还有一个好处出问题好定位。如果行情延迟看 WebSocket 连接的心跳和重连逻辑如果下单失败看 REST 返回码和签名。混在一起用遇到行情断了连带着下单也异常排查时就会像无头苍蝇。下单频率天然不该高一个辅助机器人每秒发十次下单请求本身就是风控事故所以 REST 的延迟对它完全够用。通道职责切清楚了后面接什么策略都不会打架。2.2 最小可用的行情订阅与签名请求代码先看行情订阅。用 Python 的 websockets 库连接公共频道订阅一个交易对的 ticker这是整个 Bot 的地基。import json import asyncio import websockets async def subscribe_ticker(inst_id: str): # wss 是公共行情地址只推送市场数据不需要鉴权 ws_url wss://ws.okx.com:8443/ws/v5/public async with websockets.connect(ws_url) as ws: await ws.send(json.dumps({ op: subscribe, args: [{channel: ticker, instId: inst_id}] })) async for raw in ws: data json.loads(raw) # 订阅确认事件不带 data 字段先跳过 if data not in data: continue ticker data[data][0] print(ticker[instId], ticker[last], ticker[ts])逻辑说明连接建立后客户端要先发一条 subscribe 消息服务器才会开始推送。事件类型有两种第一种是订阅确认此时消息里没有 data 字段直接跳过第二种才是真正的行情推送最新成交价在 data[0].last 里ts 是交易所时间戳。这个死循环适合放进独立线程跑拿到价格后丢进队列别在回调里直接下单。再看下单。REST 下单的核心是签名头OKX v5 的签名规则是把时间戳、请求方法、请求路径、请求体拼成一个字符串用 secretKey 做 HMAC SHA256 后 base64 编码。import json import time import hmac import hashlib import base64 import requests API_KEY 你的 apiKey SECRET_KEY 你的 secretKey # 建议用环境变量注入不要写死在源码里 PASSPHRASE 你的 passphrase def sign(timestamp: str, method: str, path: str, body: str ): msg timestamp method path body mac hmac.new(SECRET_KEY.encode(), msg.encode(), hashlib.sha256) return base64.b64encode(mac.digest()).decode() def place_limit_order(inst_id: str, side: str, sz: float, px: float): path /api/v5/trade/order body json.dumps({ instId: inst_id, # 交易对如 BTC-USDT tdMode: cash, # 现货现货模式合约用 cross 或 isolated side: side, # buy 或 sell ordType: limit, # 限价单 sz: str(sz), # 数量字符串格式 px: str(px) # 价格字符串格式 }) timestamp str(int(time.time())) headers { OK-ACCESS-KEY: API_KEY, OK-ACCESS-SIGN: sign(timestamp, POST, path, body), OK-ACCESS-TIMESTAMP: timestamp, OK-ACCESS-PASSPHRASE: PASSPHRASE, Content-Type: application/json } resp requests.post(https://www.okx.com path, headersheaders, databody) return resp.json()逻辑说明签名串必须和请求头里的 OK-ACCESS-TIMESTAMP 完全一致差一秒就会返回签名错误这是新手最常踩的坑。sz 和 px 都要求传字符串直接用 float 序列化在某些版本里会丢精度这里显式 str() 转一下更稳。返回的 resp.json() 里有订单 ID建议立刻存到本地日志。2.3 参数说明为什么订单参数容易写错下面这张表是下单请求体里最容易出问题的字段我建议贴到代码注释旁边。参数可选值我一般怎么选常见翻车点instIdBTC-USDT 等与行情订阅保持一致大小写、全角冒号tdModecash / cross / isolated现货选 cash合约按账户模式选合约用 cash 直接报错sidebuy / sell按信号方向平多单和平空单用的 side 不同ordTypelimit / market / post_only网格用 limit趋势开仓用 marketpost_only 在极端行情下会一直挂不上sz字符串先查一次最小交易量再算低于最小量被拒且提示略隐晦px字符串限价单才需要市价单带了 px 反而报参一个重要提示如果你做的是合约下单前必须确认账户的持仓模式是单向还是双向。单向持仓不用传 posSide双向持仓必须显式传 long 或 short否则返回参数错误。这个字段的策略含义在第 4 章展开这里先记住一句话——模拟盘验证过的参数组合换到实盘账户之前再手查一遍账户设置别直接依赖代码注释。3. 策略引擎怎么挂从网格到趋势的两种常见落地写法3.1 网格策略的挂单逻辑与参数表网格是最适合做成机器人首战的策略因为它规则简单、边界清晰回测容易跑起来也直观。核心思路是在价格上下界之间划分 N 个格子价格跌到某个格位就挂买单涨到上一个格位就挂卖单靠震荡赚差价。它怕单边行情单边下跌会把网格买满然后一路套住所以网格参数里必须包含一个最大投入量。我的参数通常长这样参数示例值说明价格下界 lower58000网格最低买入位价格上界 upper72000网格最高卖出位格数 grid_count20划分段数越多越密每格投入50 USDT单格买入金额交易对BTC-USDT只做一个交易对生成网格价位的代码我习惯用等比网格而不是等差网格价格跨度大时等比分布更合理。def build_grid(lower: float, upper: float, grid_count: int, total_amount: float): # 等比网格相邻价位的比例相同低价区间格子更密 ratio (upper / lower) ** (1 / grid_count) prices [lower * (ratio ** i) for i in range(grid_count 1)] # 每格金额均分数量按当前价位换算保留 6 位小数 step_amount total_amount / grid_count orders [] for price in prices: qty round(step_amount / price, 6) orders.append({price: round(price, 2), qty: qty}) return orders逻辑说明ratio 是相邻网格的倍率比如下界 58000、上界 72000、20 格ratio 约等于 1.011。prices 是格位列表每格金额均分后除以格位价格得到该格的下单数量。实际运行时不是把这 21 个单全挂出去而是拿最新 ticker 价格在格位之间移动价格跌破当前格位就买入一格涨上一格就卖出一格。注意一个细节网格单的数量精度有上限不同交易对的最小交易量不一样BTC-USDT 和山寨币的精度要求差很多。生成订单列表后统一做一次精度检查否则真下单时会被交易所拒掉而且是到执行那一步才爆出来。3.2 趋势策略用简单均线判断方向后再下单网格吃震荡趋势策略就反过来明确方向后顺着方向追。最省事的信号是均线金叉死叉短期均线上穿长期均线做多下穿做空。代码不复杂但有个地方必须处理好——信号只能在拐点触发一次不能在上涨途中反复下单。import pandas as pd def signal_from_kline(df: pd.DataFrame, fast: int 7, slow: int 25): # df 至少包含 close 列按时间升序排列 df df.copy() df[ma_fast] df[close].rolling(fast).mean() df[ma_slow] df[close].rolling(slow).mean() # 当前快线在慢线上方视为多头状态 df[bull] df[ma_fast] df[ma_slow] # 从空翻多的那一根 K 线才返回 True避免持仓期间重复触发 df[cross_up] df[bull] (~df[bull].shift(1).fillna(False)) return df[cross_up].iloc[-1]逻辑说明rolling(fast).mean() 是最近 fast 根 K 线收盘价的均值fast 越小对价格越敏感。bull 列维护多头状态cross_up 判断的是“前一根还不在多头区、这一根刚进入”这样信号只会出现在金叉当根。实盘里拿到新 K 线就调一次这个函数返回 False 就不动作返回 True 才执行开仓。提示均线参数 fast7、slow25 是默认值不同交易对和不同 K 线周期差异很大。我一般会用第 5 章的短回测快速扫一遍参数组合而不是凭感觉拍脑袋。3.3 把策略和行情线程解耦用队列而不是共享变量很多人第一次写这类的 Bot会用一个全局变量存最新价格然后策略线程里去 sleep 轮询读这个变量。这种做法短时间能跑但行情更新和策略读取之间没有同步机制行情快了策略读到的价格跳帧行情慢了策略空转。更稳的做法是中间加一个 queue.Queue生产者是 WebSocket 线程消费者是策略线程。import queue import threading tick_queue queue.Queue(maxsize200) def ws_producer(): # 伪代码从 WebSocket 拿到最新价格后放入队列 while True: price get_tick_from_ws() # 你的订阅回调 try: tick_queue.put_nowait(price) except queue.Full: # 队列满了说明行情远快于策略丢最旧的数据保最新 tick_queue.get_nowait() tick_queue.put_nowait(price) def strategy_consumer(): while True: price tick_queue.get(timeout1) action decide_by_strategy(price) # 调用网格或均线逻辑 if action: execute_order(action)逻辑说明put_nowait 在队列满的时候抛 queue.Full这里捕获后先把最旧的价格丢掉再把最新价格放进去保证策略拿到的永远是最近行情。get(timeout1) 让策略线程最多空等 1 秒避免行情停止时线程完全卡死。两个线程之间没有共享变量没有锁自然也不会出现脏读。下单函数放消费者线程里即使下单慢也不会阻塞行情接收这是整个结构里最关键的解耦。4. 辅助机器人最容易翻车的五个坑与排查清单4.1 下单后查不到订单时间戳与签名过期现象接口返回成功日志里也有订单 ID但打开 App 看不到这笔单。原因本地系统时间和服务器时间偏差大你的 timestamp 比交易所时间早了半分钟签名串里用的时间戳和请求头里的又对不上服务器拒绝但错误码被忽略程序以为自己下单成功了。解决下单前先调用公共时间接口对一次时或者直接用服务器返回的时间戳去拼签名串。我见过最稳的写法是启动时对时一次之后每次签名都用同一个时间源不要用本地 time.time() 裸拼。4.2 重复下单键盘中断后旧线程没退干净现象程序跑得好好的你 CtrlC 杀掉再重启结果发现同一价位下了两笔单。原因第一次进程的 WebSocket 线程没退干净还在队列里继续消费第二次启动的进程又起了一个消费者两个消费者同时执行策略。解决在入口处注册信号处理收到 SIGINT 时先设置一个 stop_event所有线程循环里都检查这个事件置位后退出循环、关闭 WebSocket 连接再让主线程收尾。不要把 CtrlC 当成“进程立刻没了”线程不一定跟着没。4.3 持仓模式没对齐导致合约单被拒现象现货跑了一周正常切到合约后首次下单返回参数错误错误码提示很模糊。原因合约账户有单向持仓和双向持仓两种模式单向模式下不需要传 posSide双向模式下必须显式传 long 或 short你的下单代码在现货逻辑上加了合约字段两边对不上。解决下单前读一次账户配置确认当前持仓模式再把 posSide 塞进请求体。这个字段在不同模式下语义还不一样单向模式下传了会报错双向模式下不传也会报错所以不要在代码里写死要读账户设置。4.4 日志把私钥打出来了现象某天排查下单失败发现日志文件里 Secret Key 明文躺在那里。原因调试的时候为了看 header 内容print(headers) 直接发出去了顺手把 OK-ACCESS-SIGN 也打出来了。解决打日志之前先拷贝一份 header把 OK-ACCESS-SIGN 字段替换成脱敏字符串再输出。这个小函数值得全局统一用def redact_headers(headers: dict) - dict: safe dict(headers) if OK-ACCESS-SIGN in safe: safe[OK-ACCESS-SIGN] safe[OK-ACCESS-SIGN][:6] *** return safe4.5 模拟盘和实盘环境配混了现象模拟盘验证了一周没问题切到实盘后下单全部失败或者反过来实盘单跑到了模拟盘账户里。原因代码里的 base_url 是全局常量模拟盘和实盘切换时只改了配置文件的一部分WebSocket 地址和 REST 地址改反了。解决把环境信息收敛到一个配置文件全局只读一个 BASE_URL 和 WS_URL启动时打印当前环境标记强制自己确认一遍。养成习惯每次启动先看一行日志确认当前是 demo 还是 live再继续操作。5. 回测与模拟盘上线前至少做到这三步5.1 用历史 K 线先回测拉数据、算信号、看最大回撤策略写完别急着连模拟盘先用历史 K 线跑一遍回测。回测的目的不是证明策略赚钱而是帮你发现逻辑漏洞比如未来函数、信号重复触发、手续费没算。最小流程是拉历史 K 线、逐根算信号、模拟成交、统计收益率和最大回撤。import pandas as pd def backtest(df: pd.DataFrame, fee_rate: float 0.001): # df 至少包含 close 列按时间升序 df df.copy() # signal 列由你的策略生成1 表示持仓0 表示空仓 df[position] df[signal].shift(1).fillna(0) df[ret] df[close].pct_change().fillna(0) df[strategy_ret] df[position] * df[ret] - fee_rate df[cum] (1 df[strategy_ret]).cumprod() total_return df[cum].iloc[-1] - 1 max_drawdown (df[cum] / df[cum].cummax() - 1).min() return {total_return: total_return, max_drawdown: max_drawdown}逻辑说明position 用 signal.shift(1) 是回测里最重要的一行它表示“本根 K 线收盘后才产生信号下一根 K 线才执行”避免未来函数。很多回测结果虚高就是差了这个 shift。strategy_ret 里减掉 fee_rate每笔交易扣千一的费率不扣费的回测在震荡市里会严重失真。返回的最大回撤是负数比如 -0.18 表示从最高点回撤了 18%这个数字决定了你能接受的仓位大小。5.2 模拟盘验证demo 环境和实盘环境的差异回测通过后连模拟盘模拟盘的核心价值是验证“代码执行链路”没问题不是验证策略收益。模拟盘的行情和真实市场接近但撮合结果有参考级差异模拟盘的深度盘口不完全真实市价单成交价格可能比实盘好容易被回测收益误导。模拟盘验证期间我会顺手给策略函数配一组 pytest 用例把固定输入和预期输出写死防止后面调参数把逻辑改坏。举个例子import pandas as pd def test_signal_cross_up(): df pd.DataFrame({close: [10, 11, 12, 11, 12]}) assert signal_from_kline(df, fast2, slow3) is True这个用例的价值在于策略代码一旦重构跑一遍 pytest 就知道金叉判断是否还正确。这算是把自动化测试的思想挪到交易 Bot 上来不复杂但对长期维护特别有用。5.3 一个简单的风控守卫最大亏损比例和单笔仓位上限回测和模拟盘都过了最后挂一个风控守卫。我的原则是风控逻辑独立于策略所有下单入口统一过守卫不在每个策略里各写一份。这个类是整个 Bot 里最重要的一块代码。class RiskGuard: def __init__(self, max_loss_pct: float 0.05, max_position: float 100): self.max_loss_pct max_loss_pct self.max_position max_position self.initial_balance get_account_balance() def can_open(self, order_size: float): balance get_account_balance() loss_pct (self.initial_balance - balance) / self.initial_balance if loss_pct self.max_loss_pct: return False, 超过最大亏损线停止开仓 if order_size self.max_position: return False, 单笔仓位超过上限 return True, ok逻辑说明initial_balance 在启动时记录一次之后每次都对比当前余额亏损比例达到 5% 就拒绝一切开仓请求。max_position 限制单笔仓位上限防止策略在某个瞬间计算出过大的下单量。can_open 返回两个值第一个是布尔结果第二个是拒绝原因这个原因必须写进日志否则你不知道机器人为什么突然不动作了。6. 进阶把订单状态管住程序挂了也有后悔药走到这一步你的 Bot 已经能跑策略、能下单、有风控但还差最后一块拼图崩溃恢复。程序崩了不可怕可怕的是重启后不知道刚才哪些单已经提交、哪些单还没发出于是重复下单或者漏掉已有仓位。我的习惯是做一个“下单记录 重启对账”的最小机制。每下一笔单立刻往本地文件里追加一行 JSON记录时间、订单 ID、交易对、方向、数量、价格。程序启动时先读一遍当天这些记录再调一次“未成交订单查询”接口把已经成交的标记掉把还挂着的归到本地状态里。只有经过这一步对账策略引擎才会继续运行。这样做的好处是即使半夜机器重启你也永远不会对着一个空状态去重复建仓。这里有个我踩过的具体教训早期版本我下完单只打印日志没落地文件有次进程被杀重启策略不知道已有仓位又按网格逻辑补下一单结果仓位翻倍。后来我加了对账逻辑重启后的第一件事永远是查状态而不是跑策略。整套方案的核心其实就一句话把交易所当成唯一的事实来源本地只做缓存每次启动先同步再做决定。另一个进阶习惯是每笔操作都留审计日志字段至少包含时间、动作、订单 ID、原因。出问题回溯时有这个文件和没这个文件排查效率差一个量级。这个方向做到这个程度基本就到了“敢挂真金白银”的边界了。最后提醒一句小仓位跑起来观察自己的风控和运维习惯是不是真的能坚持再决定要不要加资金。希望帮到你。本文还有配套的精品资源点击获取