加密货币市场因为交易所众多、流动性分散，天然存在价差套利机会。本文介绍 CEX/DEX 价差套利和三角套利的原理，以及用 Python + ccxt 构建套利监控系统的实战框架。
套利类型
1. 跨交易所价差套利
同一币对在不同交易所的买卖价存在差异。例如 BTC/USDT 在 Binance 卖价为 60000，在 OKX 买价为 60050，价差 50 USDT。
流程：在低价所买入 -> 在高价所卖出（或提前在两个所都持有资金，同时挂单）。
关键约束：
- 手续费（maker/taker）
- 提币时间和费用（链上转账）
- 滑点（大单冲击）
- 资金占用成本
实际有效价差 = 卖出价 - 买入价 - 买入手续费 - 卖出手续费 - 提币费
2. 三角套利
在同一交易所内利用三个交易对的汇率不一致获利。例如：
BTC/USDT = 60000
ETH/BTC = 0.05
ETH/USDT = 3010
理论上 ETH/USDT 应该 = ETH/BTC × BTC/USDT = 0.05 × 60000 = 3000。实际价格 3010，存在 10 USDT 的套利空间（每 ETH）。
路径：USDT -> BTC -> ETH -> USDT
三角套利的优势：无需跨交易所转账，速度快；劣势：价差通常很小，需要大资金量。
3. CEX-DEX 套利
中心化交易所（CEX）和去中心化交易所（DEX）之间的价差。DEX 使用 AMM（自动做市商），价格由流动性池比例决定，大额交易会产生明显滑点，与 CEX 订单簿价格产生偏差。
这类套利需要与链上合约交互，复杂度更高，本文主要聚焦 CEX 间套利。
ccxt 库调用交易所 API
ccxt 统一了 100+ 交易所的 API 接口：
import ccxt
# 初始化交易所
binance = ccxt.binance({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'options': {'defaultType': 'spot'},
})
okx = ccxt.okx({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'password': 'YOUR_PASSPHRASE',
})
# 获取订单簿
book = binance.fetch_order_book('BTC/USDT', limit=5)
best_ask = book['asks'][0][0] # 最低卖价
best_bid = book['bids'][0][0] # 最高买价
ask_volume = book['asks'][0][1] # 卖一量
print(f"Binance BTC/USDT: bid={best_bid}, ask={best_ask}")
获取多所行情
import asyncio
import ccxt.async_support as ccxt_async
async def fetch_ticker(exchange, symbol):
ticker = await exchange.fetch_ticker(symbol)
return {
'exchange': exchange.id,
'bid': ticker['bid'],
'ask': ticker['ask'],
'timestamp': ticker['timestamp'],
}
async def scan_spread(symbol):
exchanges = [
ccxt_async.binance(),
ccxt_async.okx(),
ccxt_async.bybit(),
ccxt_async.gate(),
]
try:
tasks = [fetch_ticker(ex, symbol) for ex in exchanges]
results = await asyncio.gather(*tasks, return_exceptions=True)
tickers = [r for r in results if isinstance(r, dict)]
if len(tickers) < 2:
return None
# 找最低卖价和最高买价
best_buy = min(tickers, key=lambda t: t['ask']) # 买入方
best_sell = max(tickers, key=lambda t: t['bid']) # 卖出方
spread = best_sell['bid'] - best_buy['ask']
spread_pct = spread / best_buy['ask'] * 100
return {
'symbol': symbol,
'buy_at': best_buy['exchange'],
'buy_price': best_buy['ask'],
'sell_at': best_sell['exchange'],
'sell_price': best_sell['bid'],
'spread': spread,
'spread_pct': spread_pct,
}
finally:
for ex in exchanges:
await ex.close()
# 运行
result = asyncio.run(scan_spread('BTC/USDT'))
if result and result['spread_pct'] > 0.1:
print(f"套利机会: 在{result['buy_at']}买入@{result['buy_price']}, "
f"在{result['sell_at']}卖出@{result['sell_price']}, "
f"价差{result['spread_pct']:.3f}%")
三角套利检测
async def find_triangular(exchange, base='USDT', min_profit_pct=0.1):
'''在单个交易所内寻找三角套利机会'''
markets = await exchange.load_markets()
tickers = await exchange.fetch_tickers()
# 构建交易对图
pairs = {}
for symbol, ticker in tickers.items():
if ticker['bid'] and ticker['ask'] and ticker['bid'] > 0:
pairs[symbol] = {
'bid': ticker['bid'],
'ask': ticker['ask'],
}
# 找所有以 base 开始和结束的三角路径
# 路径: BASE -> A -> B -> BASE
base_pairs = [s for s in pairs if s.endswith(f'/{base}')]
opportunities = []
for pair1_symbol in base_pairs:
coin_a = pair1_symbol.split('/')[0] # 第一步: BASE -> A
buy_a_price = pairs[pair1_symbol]['ask']
for pair2_symbol in pairs:
if not pair2_symbol.startswith(f'{coin_a}/'):
continue
coin_b = pair2_symbol.split('/')[1] # 第二步: A -> B
pair3_symbol = f'{coin_b}/{base}' # 第三步: B -> BASE
if pair3_symbol not in pairs:
continue
# 计算收益
# 1 BASE -> (1/ask1) A -> (1/ask1 * bid2) B -> (1/ask1 * bid2 * bid3) BASE
step1 = 1.0 / buy_a_price # BASE -> A
step2 = step1 * pairs[pair2_symbol]['bid'] # A -> B
step3 = step2 * pairs[pair3_symbol]['bid'] # B -> BASE
# 扣除手续费 (假设 taker 0.1%)
fee_rate = 0.001
final = step3 * (1 - fee_rate) ** 3
profit_pct = (final - 1) * 100
if profit_pct > min_profit_pct:
opportunities.append({
'path': f'{base}->{coin_a}->{coin_b}->{base}',
'profit_pct': profit_pct,
'pairs': [pair1_symbol, pair2_symbol, pair3_symbol],
})
opportunities.sort(key=lambda x: x['profit_pct'], reverse=True)
return opportunities
风险控制
套利看起来低风险，但实际操作中有很多坑：
滑点
订单簿的 best bid/ask 只是表面价格，实际成交要看深度。用 order book 计算实际成交均价：
def calc_fill_price(orderbook_side, amount):
'''计算实际成交均价'''
filled = 0.0
cost = 0.0
for price, volume in orderbook_side:
can_fill = min(volume, amount - filled)
cost += can_fill * price
filled += can_fill
if filled >= amount:
break
if filled < amount:
return None # 深度不足
return cost / filled
手续费
不同交易所、不同 VIP 等级费率不同。必须精确计算：
FEE_CONFIG = {
'binance': {'maker': 0.001, 'taker': 0.001}, # BNB 折扣后更低
'okx': {'maker': 0.0008, 'taker': 0.001},
'bybit': {'maker': 0.001, 'taker': 0.001},
}
延迟
价格在毫秒级变化，从发现机会到成交之间的延迟可能导致价差消失。应对措施：
- 使用 WebSocket 实时推送，不要用 REST 轮询
- 服务器部署在交易所数据中心附近
- 预签名订单，减少下单延迟
资金管理
class RiskManager:
def __init__(self, max_position_pct=0.1, max_daily_loss=500):
self.max_position_pct = max_position_pct # 单次最大仓位比例
self.max_daily_loss = max_daily_loss # 日最大亏损 (USDT)
self.daily_pnl = 0.0
def check_trade(self, balance, trade_amount):
if trade_amount > balance * self.max_position_pct:
return False, "Position too large"
if self.daily_pnl < -self.max_daily_loss:
return False, "Daily loss limit reached"
return True, "OK"
def record_pnl(self, pnl):
self.daily_pnl += pnl
整体架构
一个生产级套利系统的模块划分：
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Data Feed │───>│ Strategy │───>│ Execution │
│ WebSocket │ │ 价差检测 │ │ 下单/对冲 │
│ 订单簿/行情 │ │ 三角扫描 │ │ 滑点保护 │
└──────────────┘ └──────────────┘ └──────────────┘
│ │ │
└──────────┬────────┘ │
v v
┌──────────────┐ ┌──────────────┐
│ Risk Mgmt │ │ Monitoring │
│ 仓位控制 │ │ PnL 追踪 │
│ 止损/日限额 │ │ 告警/日志 │
└──────────────┘ └──────────────┘
关键原则：
- 先模拟后实盘 — 用历史数据回测，用模拟盘验证，确认稳定后再投入实盘
- 小资金开始 — 初始投入控制在可承受亏损范围内
- 持续监控 — 套利策略的利润空间会随竞争加剧而收缩
- 多重保险 — 网络断线、API 限频、交易所维护都要有处理逻辑
加密货币套利的窗口期越来越短，2025 年的市场比 2020 年竞争激烈得多。但对于技术能力强、基础设施到位的团队，仍然存在可观的机会。
