实时监控加密货币现货市场的三角套利机会

嘿，大家好！最近我在研究加密货币市场的套利策略，发现了一个特别有意思的玩法——三角套利。这个策略原本是外汇市场的经典玩法，但放在加密货币市场里也特别适用。今天就来聊聊这个策略，顺便分享一下我开发的实时监控工具。

三角套利：外汇市场的老朋友

先说说三角套利是什么吧。简单来说，它就是利用三种货币对之间的汇率失衡来实现无风险利润。比如，A/B、B/C 和 C/A 三种货币对的价格如果出现偏差，你就可以通过 A→B→C→A 的路径完成交易，赚取差价。

这玩意儿在外汇市场里已经玩了很多年了，但加密货币市场的玩法其实也差不多。比如，BTC/USDT、ETH/BTC 和 ETH/USDT 这三个交易对的价格如果出现偏差，理论上你就能通过类似的路径完成套利。

不过，这里有个问题：加密货币市场比外汇市场复杂得多，交易对数量爆炸式增长，很多情况下你根本不可能手动找到机会。这就需要自动化工具来帮忙。

一个简单的案例：USDT、BTC 和 ETH 的套利路径

先来个简单的案例吧。假设某交易所的 BTC/USDT 价格是 30000，ETH/BTC 是 0.05，ETH/USDT 是 1520。假设没有滑点，每次交易手续费是 0.1%。我们按 USDT→BTC→ETH→USDT 的路径来算：

1. USDT → BTC：用 10000 USDT 买入 BTC，得到 10000 / 30000 ≈ 0.3333 BTC；
2. BTC → ETH：用 0.3333 BTC 兑换 ETH，得到 0.3333 / 0.05 ≈ 6.666 ETH；
3. ETH → USDT：卖出 6.666 ETH，得到 6.666 × 1520 ≈ 10132 USDT。

三次交易后，扣除手续费（10000 × 0.001 × 3 = 30 USDT），净收益是 10132 - 10000 - 30 = 102 USDT。看起来很美吧？

但别高兴太早。随着加密货币市场逐渐成熟，这种纯理论上的套利机会其实不多了。而且，这种操作通常需要自动化工具来瞬时完成，否则机会稍纵即逝。

我的实时监控工具：如何捕捉机会？

为了捕捉这些机会，我开发了一个简单的实时监控工具。这个工具的核心是实时监控订单簿数据，计算三角汇率，看看是否存在套利机会。

工具的代码放在了 GitHub 上：github.com/poloxue/seekoptrader。运行方式也很简单：

$ export PYTHONPATH=`pwd`$ python seekoptrader/__main__.py triangle --exchange-name okx

运行后，工具会显示六个汇率值。这是因为不同的币种作为起点，以及不同的兑换路径，会产生六种汇率计算方式。比如，如果你的费率是 0.1%，发现汇率大于 1.003 的机会，那就有机会获利。

不过，这里有个小插曲：我一开始想在币安上跑这个工具，但发现币安的币种太多了，跑不动。后来我优化了一下，发现 OKX 和其他交易所的表现还不错。如果你有质押借贷的需求，还要考虑借贷利息。

工具实现的思路：从匹配到计算

这个工具的实现其实和跨市场价差监控有点类似，主要分为三个步骤：

1. 匹配三角交易对：找到可以形成闭环的交易对组合；
2. 实时监控订单簿：批量获取订单簿数据，提高实时性；
3. 计算三角汇率：根据不同的兑换路径和计价币种，计算出六组汇率。

不过，三角套利的匹配比跨交易所的匹配复杂得多，这里有一些有趣的细节。

三角交易对的匹配：有向图的魔法

匹配三角交易对时，我用了一个网络拓扑工具库 networkx 来构建有向图。这个图能帮我找到可以形成闭环的交易对组合。

比如，交易对 BTC/USDT、ETH/BTC 和 ETH/USDT 就能形成一个闭环。代码片段如下：

deffind_triangles(self, markets):    G = nx.DiGraph()for m in markets:        base, quote, symbol = m["base"], m["quote"], m["symbol"]        G.add_edge(base, quote, symbol=symbol)    triangles = {}    currencies = G.nodes()for b in currencies:for a in G[b]:for c in currencies:if c == b or c == a:continueif c in G and b in G[c]:if a in G[c]:if self.valid_currencies([b, a, c]):                            triangles[(a, b, c)] = (                                G[b][a]["symbol"],                                G[c][b]["symbol"],                                G[c][a]["symbol"],                            )return triangles

这里还有一个小细节：我过滤掉了一些法币和稳定币的交易对，因为这些交易对的价格变化太小，监控起来意义不大。

兑换汇率的计算：路径和基点的多样性

兑换汇率的计算是整个工具的核心。为了捕捉更多机会，我不限制兑换路径和计价币种。比如，除了 USDT→BTC→ETH→USDT 的路径，还可以是 USDT→ETH→BTC→USDT，甚至是 ETH→USDT→BTC→ETH。

每种路径的计算公式都不一样。比如：

• USDT→BTC→ETH→USDT：1 / ask(BTC/USDT) / ask(ETH/BTC) * bid(ETH/USDT)
• BTC→ETH→USDT→BTC：1 / ask(ETH/BTC) * bid(ETH/USDT) / ask(BTC/USDT)

这就是为什么工具中会显示六种汇率的原因。

代码中的小彩蛋：灵活推算汇率公式

为了进一步优化工具，我还写了一段代码，可以根据起点币种和三角交易对，灵活推算汇率公式。这个功能虽然没集成到工具里，但我觉得特别有意思，就把它保存下来了。

代码如下：

from collections import defaultdictdefdetect_arbitrage_ops(start_currency, symbols):    graph = defaultdict(dict)for symbol in symbols:        base, quote = symbol.split("/")        graph[base] = symbol  # 正向交易 base→quote        graph[base] = symbol  # 反向交易 quote→base    chains = []    max_depth = 3defdfs(current, path, ops=[], depth=0):if depth == max_depth:if current == start_currency:                chains.append(ops)returnfor next_currency in graph[current]:            symbol = graph[current][next_currency]            base, quote = symbol.split("/")            new_op = f"*bid({symbol})"if current == base elsef"/ask({symbol})"            dfs(next_currency, path + [next_currency], ops + [new_op], depth + 1)    dfs(start_currency, [start_currency])return ["1" + "".join(chain) for chain in chains]

测试一下：

base = "ETH"pairs = ["BTC/USDT", "ETH/BTC", "ETH/USDT"]operations = detect_arbitrage_ops(base, pairs)print("ETH 兑换起点:", operations)

输出结果：

ETH 兑换起点: ['1*bid(ETH/BTC)*bid(BTC/USDT)/ask(ETH/USDT)','1*bid(ETH/USDT)/ask(BTC/USDT)/ask(ETH/BTC)']

这个功能虽然没用上，但我觉得特别适合未来扩展工具时用。

总结：工具的局限与未来的改进

虽然这个工具能实时监控三角套利机会，但它也有一些局限性。比如，币安的币种太多，运行效率不高；稳定币的交易对太多，监控起来意义不大；还有，借贷利息和滑点的影响也没完全考虑进去。

不过，我觉得这个工具已经足够简单好用了。如果你对代码实现感兴趣，可以看看 triangle/monitor.py。

最后，希望这篇文章对你有帮助。如果你有任何建议或者想交流，可以加我的微信 polo-xue 加群。最近感冒了，脑子有点不清楚，希望文章里没太多错误！