"构建你自己的套利机器人"课程第 12 课。本课结束世界 3。你的机器人已经有扫描器、数量计算器和现实检验。今天要让它可以安全运行。
最直接的做法
最直接的启动方式:把密钥放进脚本,在函数里硬编码所有数字,费率 0.001、数量 1 等,然后用真钱运行来 "看看会发生什么"。它会一直正常,直到第一次意外出现。之后三个问题都没有好答案:我改了哪个数字?我亏了多少?它为什么一直在交易?
本课用三个工具逐一解决这三个问题。
工具 1:模拟交易
模拟交易指的是整个机器人使用实时价格运行、作出决策,并将订单发送到一个只记录原本会发生什么的虚拟账户。资金不会移动。你会知道机器人如何运行,每个错误都不会花钱。
使它有用的关键是:虚拟账户和真实交易所客户端拥有相同的方法,此处叫作 trade。机器人不知道在和哪一个交互。之后切换到真实客户端时,其他任何东西都不用改。
模拟账户也会略微说谎,如第 11 课所示:它按你传入的盘口价格成交,没有延迟,也没有其他机器人。把它的利润视为上界。真实结果会更低。
工具 2:配置文件
所有可能需要更改的数字都放在一个位置,于启动时读取。下面的代码把它写成程序内的 JSON 文本,因此可以直接运行。你的机器人中,它会成为一个文件 config.json。
- 无需碰逻辑即可修改数字。
- 可在每次运行旁保存配置,之后知道哪套设置得到哪个结果。
- 数字错误只需在一行中定位。
密钥和秘密永远不要放到仓库中的配置文件。应放进环境变量。
工具 3:风险限额
风险限额是在某件事看起来不对时停止机器人的规则,不管机器人自身认为如何。四个基本限额:
- 单笔最大数量。要求 1000 枚而不是 1 枚的错误会在此处被拦住。
- 最小 edge:计划利润低于安全边际时不交易,原因见第 11 课。
- 一个周期内的最大交易次数。失控循环会撞上它。
- 每日最大亏损。当亏损越过限额时,机器人停止,只有人可以重新启动。它就是紧急停止开关。
代码
import json
CONFIG = json.loads("""{
"fee": 0.001,
"min_edge": 0.05,
"max_size": 1.0,
"max_trades": 20,
"max_daily_loss": 0.50
}""")
def walk(levels, amount):
money, left = 0.0, amount
for price, size in levels:
take = min(size, left)
money += take * price
left -= take
if left <= 0:
break
return money, amount - left
def plan_profit(asks, bids, q, fee):
"""Profit after fees for size q, or None if a book is too thin."""
paid, got_a = walk(asks, q)
received, got_b = walk(bids, q)
if got_a < q or got_b < q:
return None
return received * (1 - fee) - paid * (1 + fee)
class PaperAccount:
"""A fake account. A real exchange client would have the same trade() method."""
def __init__(self):
self.pnl, self.trades = 0.0, 0
def trade(self, asks, bids, q, fee):
profit = plan_profit(asks, bids, q, fee)
self.pnl += profit
self.trades += 1
return profit
class Guard:
def __init__(self, cfg):
self.cfg, self.stopped = cfg, False
def check(self, account, size, edge):
if self.stopped:
return "stopped"
if account.pnl <= -self.cfg["max_daily_loss"]:
self.stopped = True
return "daily loss limit"
if account.trades >= self.cfg["max_trades"]:
return "trade limit"
if size > self.cfg["max_size"]:
return "size above limit"
if edge is None or edge < self.cfg["min_edge"]:
return "edge below minimum"
return "ok"
asks = [(100.00, 2), (100.20, 3)]
bids = [(100.30, 1), (99.90, 5)]
account, guard = PaperAccount(), Guard(CONFIG)
for size in [1.0, 1.0, 1.5, 0.2]:
edge = plan_profit(asks, bids, size, CONFIG["fee"])
verdict = guard.check(account, size, edge)
profit = account.trade(asks, bids, size, CONFIG["fee"]) if verdict == "ok" else None
print(f"size {size:<4} edge {edge:+.4f} -> {verdict:20} profit {profit if profit is None else round(profit, 4)}")
print("pnl", round(account.pnl, 4), "trades", account.trades)
print("loss streak:")
account, guard = PaperAccount(), Guard(CONFIG)
for i in range(6):
account.pnl -= 0.20
print(i, round(account.pnl, 2), guard.check(account, 1.0, 0.1))
输出:
size 1.0 edge +0.0997 -> ok profit 0.0997
size 1.0 edge +0.0997 -> ok profit 0.0997
size 1.5 edge -0.0502 -> size above limit profit None
size 0.2 edge +0.0199 -> edge below minimum profit None
pnl 0.1994 trades 2
loss streak:
0 -0.2 ok
1 -0.4 ok
2 -0.6 daily loss limit
3 -0.8 stopped
4 -1.0 stopped
5 -1.2 stopped
阅读输出。
- 前两笔交易通过所有检查并被记录:每笔 0.0997,共计 0.1994。可以手动核算 0.0997,参见第 10 课。
- 数量 1.5 被数量限额阻止,它也会亏钱:-0.0502。
- 数量 0.2 的 edge 为正,0.0199,但低于最小 edge 0.05,因此机器人跳过它。微小利润不值得承担第 11 课里的风险。
- 在亏损序列中,机器人每步损失 0.20。保护机制在第三步触发,此时累计为 -0.6。限额是 0.50,亏损超出它 0.10,因为检查在每笔交易之前运行,只看到已经产生的亏损。限额可以止损,但不能阻止最后一笔交易。把它设得低于你实际承受得起的亏损。
- 在 "daily loss limit" 之后,保护机制会永远回答 "stopped"。只有人工重启可以清除它。这是有意设计的。
开始实盘前
合理的顺序,规则是只有当前一步已经变得无聊时才走到下一步:
- 至少一周的模拟交易,并保留日志。将它声称的利润与第 11 课的估计比较。
- 真实交易所、极小数量,交易所允许的最小值,几美元的亏损限额。目标是测试连接、订单和手续费,而不是赚钱。
- 只有真实结果持续与模拟结果一致时,才慢慢增加数量。
如果真实结果远差于模拟结果,差额就是你的真实成本:延迟、失败的一腿、其他机器人。在提高数量前找出是哪一项。
让 AI 代理编写并测试它
模糊的请求:
给我的机器人添加安全保护。
带检查的请求:
编写
safe.py,包含第 12 课的plan_profit、PaperAccount和Guard,从config.json读取设置(fee 0.001, min_edge 0.05, max_size 1.0, max_trades 20, max_daily_loss 0.50)。使用 asks [(100.00, 2), (100.20, 3)] 和 bids [(100.30, 1), (99.90, 5)] 测试:数量 1.0、1.0、1.5、0.2 必须得到 ok、ok、size above limit、edge below minimum,最终 pnl 必须为 0.1994。测试 2:每步损失 0.20 时,必须在第 2 步打印 "daily loss limit",之后打印 "stopped"。只从环境变量读取 API 密钥,绝不打印它们。不要连接任何交易所。
接受代理工作前,先用计算器核算 pnl 0.1994。
你的练习
- 运行代码并与上面的输出比较。
- 将配置移到真正的
config.json文件并用json.load读取。把min_edge改为 0.10,看看哪些交易被跳过。 - 让保护机制在每笔交易后检查亏损,以便它在第一笔越过限额的交易后停止。超出的部分消失了吗?不完全是:最后一笔交易仍然发生。
- 添加 "最大未平仓头寸" 限额:如果机器人在一次失败的一腿后同时持有超过 2 枚币,它就停止。
下一步
世界 3 已完成:你拥有一个完整的机器人,它能发现机会、计算数量、模拟并保护。世界 4 让它更快、更广:第 13 课从速度开始,解释 Python 为什么变慢、该怎么办,以及 Rust 如何进入画面。