4、三角套利实战：使用CCXT库获取实时汇率，检测套利机会

三角套利，说白了就是利用三个币种之间的汇率不一致性来赚钱。你想想看，如果A→B→C→A这条路径走下来，最后手里的币变多了，那就是机会。我在刚接触这个策略时，总觉得它很神秘，后来发现核心就两件事：拿到实时汇率，然后快速计算。

4.1 为什么选CCXT？

我个人习惯用CCXT库来做这件事。原因很简单——它统一了上百家交易所的API接口。你不用去记每家交易所的奇葩签名方式，也不用管他们返回的数据格式有多乱。CCXT帮你把这些脏活累活都干了。

我记得第一次手动对接某小交易所时，光处理他们的时间戳格式就花了两小时。后来换了CCXT，一行代码搞定。嗯，这就是选它的理由。

4.2 获取实时汇率

先看一段最基础的代码。我们以币安为例，获取三个币种的交易对价格：

import ccxt
import time

# 初始化交易所
exchange = ccxt.binance({
    'apiKey': '你的API_KEY',
    'secret': '你的SECRET',
    'enableRateLimit': True,  # 我建议一定要开，防止被交易所封
})

# 获取三个交易对的订单簿
def get_prices(symbols):
    prices = {}
    for symbol in symbols:
        try:
            ticker = exchange.fetch_ticker(symbol)
            prices[symbol] = ticker['last']
            # 这里我习惯用last价格，实际交易时建议用bid/ask
        except Exception as e:
            print(f"获取{symbol}失败: {e}")
            prices[symbol] = None
    return prices

# 测试一下
symbols = ['BTC/USDT', 'ETH/BTC', 'ETH/USDT']
prices = get_prices(symbols)
print(prices)

4.3 三角套利的核心逻辑

检测套利机会，本质上就是算一条路径的隐含汇率，然后和实际汇率做对比。我画了一张图，帮你理解这个流程：

你看这张图，从USDT出发，经过BTC再到ETH，最后回到USDT。如果最后手里的USDT变多了，那就是套利机会。说白了就是汇率环的乘积大于1。

4.4 检测套利机会的完整代码

下面这段代码，是我在实际项目中用过的检测逻辑。你直接拿去改改就能用：

def detect_triangular_arbitrage():
    # 获取三个交易对的价格
    symbols = ['BTC/USDT', 'ETH/BTC', 'ETH/USDT']
    prices = get_prices(symbols)
    
    if None in prices.values():
        print("价格数据不完整，跳过")
        return None
    
    btc_usdt = prices['BTC/USDT']
    eth_btc = prices['ETH/BTC']
    eth_usdt = prices['ETH/USDT']
    
    # 路径1: USDT -> BTC -> ETH -> USDT
    # 先买BTC，再用BTC买ETH，最后卖ETH换USDT
    rate1 = (1 / btc_usdt) * eth_btc * eth_usdt
    
    # 路径2: USDT -> ETH -> BTC -> USDT
    rate2 = (1 / eth_usdt) * (1 / eth_btc) * btc_usdt
    
    # 判断是否有套利机会
    threshold = 1.001  # 千分之一的手续费门槛
    # 我习惯留0.1%的缓冲，因为交易所要收手续费
    
    result = {}
    if rate1 > threshold:
        result['path1'] = {
            'direction': 'USDT→BTC→ETH→USDT',
            'rate': rate1,
            'profit_pct': (rate1 - 1) * 100
        }
        print(f"发现套利机会！路径1收益率: {(rate1-1)*100:.4f}%")
    
    if rate2 > threshold:
        result['path2'] = {
            'direction': 'USDT→ETH→BTC→USDT',
            'rate': rate2,
            'profit_pct': (rate2 - 1) * 100
        }
        print(f"发现套利机会！路径2收益率: {(rate2-1)*100:.4f}%")
    
    return result

# 运行检测
result = detect_triangular_arbitrage()

4.5 实战中的坑

我在项目中遇到过几个坑，这里列出来给你参考：

• 价格延迟：三个价格不是同时获取的，中间可能有几百毫秒的延迟。我曾经因为这个吃了亏，后来改用WebSocket实时流才解决。
• 深度不足：检测到机会时，可能订单簿上的量不够你吃。我建议在计算时同时检查深度，别只看最后一笔成交价。
• 交易所限制：有些交易所对API调用频率有限制。我习惯在每次请求后加一个sleep(0.1)，虽然慢了点，但稳。

4.6 数据记录与监控

光检测还不够，你得把数据记下来。我习惯用CSV文件记录每次检测的结果，方便复盘：

import csv
from datetime import datetime

def log_arbitrage(result):
    with open('arbitrage_log.csv', 'a', newline='') as f:
        writer = csv.writer(f)
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        if result:
            for path, data in result.items():
                writer.writerow([timestamp, path, data['direction'], 
                               f"{data['rate']:.6f}", f"{data['profit_pct']:.4f}%"])
        else:
            writer.writerow([timestamp, '无机会', '-', '-', '-'])

嗯，到这里你已经掌握了三角套利检测的核心。记住，检测到机会只是第一步，真正执行时还有滑点、延迟、手续费这些现实问题。但先把检测逻辑跑通，后面的事就好办了。

好了，这一章的内容就这些。代码你拿去跑跑看，有问题随时调整参数。套利这行，细节决定成败。

---