《打造自己的套利机器人》课程第 4 课。今天,机器人的第 1 阶段开始使用真实数据运行。
我们要做什么
第 1 课比较了两家交易所上的一种币。真正的机器人的第 1 阶段会比较每一对交易所和多种币,只报告扣除手续费后仍为正的价差。今天我们会扫描五家交易所上的一种币,你会看到有多少价差能留下来。
让价差变成假象的三个陷阱
看代码之前,先了解三个会制造虚假价差的因素。
陷阱 1:手续费。每家交易所都会从每笔交易中收取一定比例的费用;当你主动接受已有报价时,这笔费用叫吃单手续费。假设买入一侧的手续费是 0.1%,卖出一侧是 0.26%。那么价差必须高于 0.36% 才能刚好保本。
陷阱 2:报价货币。价格是用某种货币表示的。BTC/USDT 的价格以与美元挂钩的代币 USDT 计价。BTC/USD 的价格则以真正的美元计价。这两种价格并不相同。USDT 的交易价格可能略高于或低于一美元,有时差距还会大得多。如果你不做换算,就把一家交易所的 BTC/USDT 和另一家的 BTC/USD 进行比较,那么所谓的“价差”可能只是 USDT 与美元之间的差异。规则:只比较报价货币相同的价格。
陷阱 3:单位。一家交易所可能报一枚币的价格,另一家可能报一千枚币的价格,还有一些交易所报的是有自己合约规模的合约价格。一定要确认一个单位具体代表什么。
公式
在交易所 X 买入、在交易所 Y 卖出时:
百分比价差 = (Y 的买价 − X 的卖价) / X 的卖价 × 100
净价差 = 价差 − X 的手续费 − Y 的手续费
如果净价差大于零,这一交易对就通过第 1 阶段筛选。
计算示例。X 的卖价是 100.00,X 的手续费是 0.10%。Y 的买价是 100.50,Y 的手续费是 0.10%。 价差 = (100.50 − 100.00) / 100.00 × 100 = 0.50%。 净价差 = 0.50 − 0.10 − 0.10 = 0.30%。这个交易对通过筛选。
价格相同,但 Y 的手续费是 0.40%:净价差 = 0.50 − 0.10 − 0.40 = 0.00。这个交易对无法通过筛选。
程序
程序会向五家交易所查询 BTC/USDT 的最佳买价和卖价,尝试每一种有序组合,并打印结果最好的五组。
这是最直接的方法:两个循环,所有交易所两两比较,每次运行时每家交易所各发一次网络请求。我们是故意这样写的。它是展示问题的最短程序,你会在下一节感受到它的局限。不要在生产环境中使用这种设计。
import json, urllib.request
def get(url):
req = urllib.request.Request(url, headers={"User-Agent": "course-scanner"})
with urllib.request.urlopen(req, timeout=10) as r:
return json.load(r)
def binance(s): d = get(f"https://api.binance.com/api/v3/ticker/bookTicker?symbol={s}USDT"); return float(d["bidPrice"]), float(d["askPrice"])
def kraken(s): d = next(iter(get(f"https://api.kraken.com/0/public/Ticker?pair={s}USDT")["result"].values())); return float(d["b"][0]), float(d["a"][0])
def okx(s): d = get(f"https://www.okx.com/api/v5/market/ticker?instId={s}-USDT")["data"][0]; return float(d["bidPx"]), float(d["askPx"])
def bitstamp(s): d = get(f"https://www.bitstamp.net/api/v2/ticker/{s.lower()}usdt/"); return float(d["bid"]), float(d["ask"])
def coinbase(s): d = get(f"https://api.exchange.coinbase.com/products/{s}-USDT/ticker"); return float(d["bid"]), float(d["ask"])
# 名称:(函数,吃单手续费百分比)。这些手续费只是示例:请查看各交易所的手续费页面。
VENUES = {"binance": (binance, 0.10), "kraken": (kraken, 0.26), "okx": (okx, 0.10),
"bitstamp": (bitstamp, 0.30), "coinbase": (coinbase, 0.40)}
def scan(symbol):
quotes = {}
for name, (fn, fee) in VENUES.items():
try:
quotes[name] = fn(symbol)
except Exception as e:
print("skip", name, "-", type(e).__name__)
rows = []
for buy, (_, ask) in quotes.items():
for sell, (bid, _) in quotes.items():
if buy == sell:
continue
gap = (bid - ask) / ask * 100
fees = VENUES[buy][1] + VENUES[sell][1]
rows.append((gap - fees, gap, buy, sell))
rows.sort(reverse=True)
for net, gap, buy, sell in rows[:5]:
print(f"buy {buy:9} sell {sell:9} gap {gap:+.3f}% after fees {net:+.3f}%")
scan("BTC")
将它保存为 scan5.py,然后运行 python3 scan5.py。
你应该看到什么结果
会有五行。一次测试运行打印了以下结果:
buy binance sell okx gap +0.004% after fees -0.196%
buy okx sell binance gap -0.004% after fees -0.204%
buy binance sell kraken gap +0.003% after fees -0.357%
你的数字会有所不同,但整体情况会一样:价差只有百分之几千分之一,而手续费要大两个数量级,所以每个净价差都是负数。这就是在大型交易所交易大市值币种时,第 1 阶段得到的真实初步结果。如果你看到净价差大于零,先别庆祝。检查三件事:报价货币、单位,以及报价是否足够新(某家交易所一分钟没有更新,显示的就是旧价格)。
如果某家交易所被跳过,可能是它屏蔽了你所在的国家或地区。把它从 VENUES 中移除,或者换成另一家。
为什么不能在生产环境中这样做
这个直接运行的程序能用,处理五家交易所和一种币时也没问题。现在算一算当机器人扩大规模后要付出的成本。
- 比较的交易对:有 N 家交易所时,有 N × (N − 1) 个有顺序的交易对。5 家交易所有 20 个,30 家交易所有 870 个。每种币都要这样比较。
- 请求:程序每次运行都会下载最新报价,每家交易所、每种币各发一个网络请求。30 家交易所和 200 种币,每次扫描就有 6000 个请求。交易所会限制请求并可能封禁你,而且 6000 个请求需要几分钟,而一个价差可能只持续几秒钟。
- 过时价格:最后一个请求返回时,第一个报价已经过时,因此比较的是不同时间的价格。
简单说,生产环境中的机器人会这样做:
- 它和每家交易所保持实时连接(websocket),在每次价格变化时接收更新,因此不需要轮询。
- 它在内存中保存每家交易所、每种币的最佳买价和最佳卖价。
- 对一种币来说,它只需要找出所有交易所中的最高买价和最低卖价:遍历 N 个报价一次即可,而不是进行 N × (N − 1) 次比较。如果最高买价减去最低卖价,即使对手续费最低的交易对也无法覆盖手续费,就不需要再检查其他情况。
你今天不会把这些全部做出来。重点是了解这个简单程序隐藏了什么,这样当你的机器人很慢或被拦截时,你就知道该替换哪个部分。第 9 到第 12 课会在机器人变成完整系统时再次讲到这些内容。
真正出现价差的地方
小型、交易不活跃的币种、小型交易所和价格快速波动时,价差会更大。手续费越低,价差也越有机会留下来。许多交易所会给交易量大的用户或持有交易所自有代币的用户更低的手续费。专业机器人的费率比你刚开始时能拿到的低。这也是显而易见的交易对竞争如此激烈的原因之一。
让 AI 智能体编写并扩展程序
含糊的请求:
比较许多交易所的价格。
包含关键细节的请求:
扩展 scan5.py。添加一个函数:如果某条报价带有时间戳,而且已经超过 5 秒,就跳过它。把手续费和币种列表放在 config.py 中。扫描 BTC、ETH 和 SOL。只打印净价差大于 0 的交易对;如果没有,就打印净价差最好的三组,并标上 NEAR。不要添加任何交易代码。把输出展示给我。
然后手动检查:从打印的一行中取一组,用计算器根据程序显示的两个价格计算 (bid − ask) / ask × 100,再与价差列比较。结果必须精确到小数点后三位。
练习
- 每隔一分钟运行一次
scan5.py,共运行五次。有几次出现了净价差大于零的情况? - 把
VENUES中两边的手续费都改成 0.02%(超大型交易者可能拿到的费率)。现在有交易对能通过筛选吗?这说明谁能从这种价差中获利? - 再添加一种币,比如 ETH。它的最佳价差比 BTC 的大还是小?
- 把你的文件另存为
scan5_v1.py。这是本课要保存的版本。
接下来
第 4 课比较了不同交易所。第 5 课我们会换个问题:如果价差不是出现在两家交易所之间,而是出现在同一家交易所的一圈货币兑换中,会怎样?为此,我们需要把价格画成图。