三角套利（CEX×DEX×跨链）

日期: 2026-08-05 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #套利 #三角套利 #跨所 #跨链

---

今日目标

---

一、理论与模型

1.1 三角套利数学

定义：在三个货币对中循环换汇，回到原币种获得净收益。

经典三角： $$ USD \to BTC \to ETH \to USD $$

无套利条件： $$ \frac{P_{BTC/USD}}{P_{ETH/USD} \cdot P_{BTC/ETH}} = 1 $$

套利信号： $$ \text{Profit} = \frac{1}{P_{USD/BTC} \cdot P_{BTC/ETH} \cdot P_{ETH/USD}} - 1 - \text{Fees} $$

如果 > 0，存在套利机会。

1.2 CEX-CEX 跨所套利

最简模型：在 CEX_A 买 BTC，在 CEX_B 卖 BTC。

$$ \text{Profit} = (P_B^{bid} - P_A^{ask}) \cdot Q - Q \cdot (\text{fee}_A + \text{fee}_B) $$

约束：

• 资金需事先在两所
• 提币时间（10-60 分钟）使实时套利不可能
• 大单滑点

1.3 CEX-DEX 套利

CEX 与 DEX 价差通常更大（DEX 流动性碎片化）。 常见组合：

• Binance BTC vs Uniswap WBTC：套利 BTC 与 WBTC 桥接
• Binance ETH vs Uniswap ETH：直接套利
• Binance USDC vs Curve USDC：稳定币池失衡

关键考量：

• gas 成本（动态）
• DEX 滑点（AMM 公式）
• MEV bot 抢跑
• block confirmation 时间

1.4 跨链套利

跨链桥（Stargate/Across/Synapse）使资产跨链流动有摩擦。

套利场景：ETH 在 Arbitrum 比 ETH 主网便宜 0.5%。

步骤：

1. 主网 ETH → Arbitrum（桥）
2. Arbitrum 卖 ETH → USDC
3. Arbitrum USDC → 主网（桥）
4. 主网 USDC 买 ETH

问题：

• 桥时间 1 秒（CCTP）到 7 天（Optimistic Rollup withdrawal）
• 桥费 0.05-0.30%
• 桥流动性限制（大额拆分）

1.5 套利的执行风险

1.6 套利容量

单笔容量（受滑点限制）： $$ Q_{\max} \approx \min\left(\frac{\text{spread}}{2 \cdot \sigma \cdot \sqrt{1/V}}, \text{order book depth at price}\right) $$

频率容量：

• CEX-CEX：每秒 1-10 次
• CEX-DEX：每分钟 5-20 次（gas 限制）
• 跨链：每天 10-100 次

---

二、直觉与陷阱

陷阱 1：手续费忽略

Spread 看起来 0.3%，实际：
- Maker fee: 0.05%
- Taker fee: 0.10%
- 两边各 0.10% taker
- 净 spread: 0.30% - 0.20% = 0.10%（高频做才划算）

许多新手只看 mid-mid spread 忽略 bid-ask 和 fee。

陷阱 2：账户余额时序问题

CEX-CEX 套利需要在两所都有资金：

• 在 A 卖 BTC，B 没 BTC 怎么办？只能事先存
• 大资金需要在 N 个所都备货 → 资金占用大
• 提币转账有最低额（如 ETH 0.05），小额套利不划算

陷阱 3：MEV 抢跑（DEX）

你看到 spread → 提交 swap tx → MEV bot 在你前面执行 → 你成交价更差或失败。 解法：

1. Flashbots / Eden 私有 mempool
2. CoW Protocol（batch auction，无 front-run）
3. 1inch Pathfinder（最优路径）

陷阱 4：跨链时间风险

桥 ETH 主网 → Arbitrum 大约 10-15 分钟（canonical bridge）或 1-2 分钟（Across/Stargate）。 桥的过程中价格变了，套利消失。

解法：

1. 用快桥（Across, Stargate）
2. 提前在两条链各备货
3. 双向同时执行（桥 + 反向 swap）

陷阱 5：稳定币 ≠ 1:1

USDC vs USDT 可能在 0.998-1.002 间漂移。 做"等价"假设的套利会有持续 noise。 2023-03 USDC 跌到 0.87，套利模型完全失效。

陷阱 6：闪电贷攻击使套利消失

DEX 套利机会出现 → 闪电贷 bot 借 $50M 一次性吃完 → 你的 swap 失败或亏损。

陷阱 7：API 限流和 rate limit

Binance API 限制 1200 req/min。不当请求被 ban → 错过套利机会。

---

三、代码实现

3.1 跨所价差监控

# arb.py
"""
Real-time Crypto Arbitrage Engine
- CEX-CEX spread monitor
- CEX-DEX with gas/slippage
- Cross-chain opportunity scoring
"""

import asyncio
import time
import requests
import json
import numpy as np
import pandas as pd
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass


# ----------------------------- CEX 价格 -----------------------------
class CEXPriceClient:
    """统一 CEX 价格接口"""

    @staticmethod
    def binance_orderbook(symbol: str) -> Dict:
        url = "https://api.binance.com/api/v3/depth"
        r = requests.get(url, params={'symbol': symbol, 'limit': 5})
        d = r.json()
        return {
            'best_bid': float(d['bids'][0][0]),
            'best_ask': float(d['asks'][0][0]),
            'bid_qty': float(d['bids'][0][1]),
            'ask_qty': float(d['asks'][0][1]),
            'mid': (float(d['bids'][0][0]) + float(d['asks'][0][0])) / 2,
        }

    @staticmethod
    def coinbase_ticker(symbol: str) -> Dict:
        # symbol: BTC-USD
        url = f"https://api.exchange.coinbase.com/products/{symbol}/ticker"
        r = requests.get(url)
        d = r.json()
        return {
            'best_bid': float(d['bid']),
            'best_ask': float(d['ask']),
            'mid': (float(d['bid']) + float(d['ask'])) / 2,
        }

    @staticmethod
    def kraken_ticker(symbol: str) -> Dict:
        url = f"https://api.kraken.com/0/public/Ticker?pair={symbol}"
        r = requests.get(url)
        d = r.json()['result']
        key = list(d.keys())[0]
        info = d[key]
        return {
            'best_bid': float(info['b'][0]),
            'best_ask': float(info['a'][0]),
            'mid': (float(info['b'][0]) + float(info['a'][0])) / 2,
        }


# ----------------------------- DEX 价格 (Uniswap V2 公式) -----------------------------
@dataclass
class AMMPool:
    reserve_x: float  # token0 reserves
    reserve_y: float  # token1 reserves
    fee: float = 0.003  # 0.3% Uniswap V2

    def quote_x_to_y(self, dx: float) -> float:
        """给定 dx 输入，计算输出 dy（恒定积公式）"""
        dx_after_fee = dx * (1 - self.fee)
        return (self.reserve_y * dx_after_fee) / (self.reserve_x + dx_after_fee)

    def quote_y_to_x(self, dy: float) -> float:
        dy_after_fee = dy * (1 - self.fee)
        return (self.reserve_x * dy_after_fee) / (self.reserve_y + dy_after_fee)

    def price(self) -> float:
        return self.reserve_y / self.reserve_x

    def slippage(self, dx: float) -> float:
        """Slippage = 1 - effective_price / spot_price"""
        spot = self.price()
        dy = self.quote_x_to_y(dx)
        effective = dy / dx
        return 1 - effective / spot


# ----------------------------- 三角套利 -----------------------------
@dataclass
class ArbOpportunity:
    path: List[str]
    profit_pct: float
    profit_usd: float
    notional: float
    confidence: float  # 0-1


def triangular_arb_check(prices: Dict[str, Dict], notional_usd: float = 10_000,
                          fee_per_leg: float = 0.001) -> Optional[ArbOpportunity]:
    """
    检查 USD -> BTC -> ETH -> USD 三角
    prices: {'BTCUSD': {...}, 'ETHUSD': {...}, 'BTCETH': {...}}
    """
    p_btc_usd = prices['BTCUSD']['best_ask']  # buy BTC
    p_eth_btc = 1 / prices['BTCETH']['best_ask']  # buy ETH with BTC
    p_eth_usd = prices['ETHUSD']['best_bid']  # sell ETH for USD

    # 1 USD -> BTC -> ETH -> USD
    btc_amount = notional_usd / p_btc_usd * (1 - fee_per_leg)
    eth_amount = btc_amount * p_eth_btc * (1 - fee_per_leg)
    final_usd = eth_amount * p_eth_usd * (1 - fee_per_leg)

    profit = (final_usd - notional_usd) / notional_usd
    if profit > 0.001:  # 至少 10 bps
        return ArbOpportunity(
            path=['USD', 'BTC', 'ETH', 'USD'],
            profit_pct=profit,
            profit_usd=final_usd - notional_usd,
            notional=notional_usd,
            confidence=0.8,
        )
    return None


# ----------------------------- CEX-CEX 套利 -----------------------------
def cex_cex_spread(symbol: str = 'BTC') -> pd.DataFrame:
    """对比多个所的同一币 spread"""
    rows = []
    try:
        b = CEXPriceClient.binance_orderbook(f'{symbol}USDT')
        rows.append({'exchange': 'Binance', 'bid': b['best_bid'],
                     'ask': b['best_ask'], 'mid': b['mid']})
    except Exception as e:
        print(f"Binance err: {e}")
    try:
        c = CEXPriceClient.coinbase_ticker(f'{symbol}-USD')
        rows.append({'exchange': 'Coinbase', 'bid': c['best_bid'],
                     'ask': c['best_ask'], 'mid': c['mid']})
    except Exception as e:
        print(f"Coinbase err: {e}")
    try:
        # Kraken 命名特殊 (XBTUSD, XETHUSD)
        k_sym = 'XBTUSD' if symbol == 'BTC' else f'{symbol}USD'
        k = CEXPriceClient.kraken_ticker(k_sym)
        rows.append({'exchange': 'Kraken', 'bid': k['best_bid'],
                     'ask': k['best_ask'], 'mid': k['mid']})
    except Exception as e:
        print(f"Kraken err: {e}")

    df = pd.DataFrame(rows)
    if len(df) >= 2:
        # 找最低 ask 和最高 bid
        min_ask = df.loc[df['ask'].idxmin()]
        max_bid = df.loc[df['bid'].idxmax()]
        df['spread_bps'] = (max_bid['bid'] - min_ask['ask']) / min_ask['ask'] * 10000
    return df


# ----------------------------- CEX-DEX 套利 (含 gas) -----------------------------
def cex_dex_opportunity(cex_price: float, pool: AMMPool, trade_size_usd: float,
                         gas_price_gwei: float = 30,
                         gas_limit: int = 200_000,
                         eth_price: float = 3000) -> Dict:
    """
    评估 CEX vs DEX 套利
    假设 token0 = USDC, token1 = ETH
    """
    # DEX 买 ETH 价格
    dex_buy_amount_in = trade_size_usd
    eth_received = pool.quote_x_to_y(dex_buy_amount_in)
    dex_effective_price = dex_buy_amount_in / eth_received
    dex_slippage = pool.slippage(dex_buy_amount_in)

    # 在 CEX 卖 ETH 拿 USDC
    cex_proceeds = eth_received * cex_price * (1 - 0.001)  # taker fee 0.1%

    # Gas
    gas_cost_eth = (gas_price_gwei * 1e-9) * gas_limit
    gas_cost_usd = gas_cost_eth * eth_price

    # 总损益
    gross_profit = cex_proceeds - dex_buy_amount_in
    net_profit = gross_profit - gas_cost_usd

    return {
        'eth_received': eth_received,
        'dex_effective_price': dex_effective_price,
        'cex_price': cex_price,
        'price_diff_pct': (cex_price - dex_effective_price) / dex_effective_price,
        'dex_slippage': dex_slippage,
        'cex_proceeds': cex_proceeds,
        'gas_cost_usd': gas_cost_usd,
        'gross_profit_usd': gross_profit,
        'net_profit_usd': net_profit,
        'profit_bps': net_profit / dex_buy_amount_in * 10000,
        'profitable': net_profit > 0,
    }


# ----------------------------- 跨链套利 -----------------------------
@dataclass
class CrossChainArb:
    src_chain: str
    dst_chain: str
    asset: str
    src_price: float
    dst_price: float
    bridge_fee_pct: float
    bridge_time_sec: float
    notional_usd: float

    def evaluate(self) -> Dict:
        # 1. Buy on src
        amount_asset = self.notional_usd / self.src_price * (1 - 0.001)  # CEX fee
        # 2. Bridge
        amount_after_bridge = amount_asset * (1 - self.bridge_fee_pct)
        # 3. Sell on dst
        proceeds = amount_after_bridge * self.dst_price * (1 - 0.001)
        gross_profit = proceeds - self.notional_usd

        # 时间风险（价格波动）
        # 假设 dst 资产 vol = 60% annualized = ~3% per hour
        time_hours = self.bridge_time_sec / 3600
        time_vol_pct = 0.6 / np.sqrt(365 * 24) * np.sqrt(time_hours)

        # 95% 风险调整
        risk_adjusted = gross_profit - 1.65 * time_vol_pct * self.notional_usd

        return {
            'gross_profit': gross_profit,
            'profit_pct': gross_profit / self.notional_usd,
            'time_vol_risk_pct': time_vol_pct,
            'risk_adjusted_profit': risk_adjusted,
            'recommended': risk_adjusted > 0,
        }


# ----------------------------- 执行引擎（模拟） -----------------------------
class ArbExecutionSim:
    """模拟订单执行"""

    def __init__(self, fail_rate: float = 0.05):
        self.fail_rate = fail_rate
        self.history = []

    def execute(self, opp: ArbOpportunity) -> Dict:
        # 模拟 5% 失败率（MEV/滑点）
        success = np.random.random() > self.fail_rate
        if not success:
            return {'success': False, 'realized_profit': 0,
                    'reason': 'execution failure (MEV/slippage)'}

        # 模拟实际滑点（比报价多 20%）
        realized = opp.profit_usd * np.random.uniform(0.7, 1.0)
        record = {'success': True, 'realized_profit': realized,
                   'opp': opp, 'time': time.time()}
        self.history.append(record)
        return record


# ----------------------------- 主程序 -----------------------------
if __name__ == '__main__':
    print("=" * 60)
    print("CEX-CEX BTC Spread Monitor")
    df = cex_cex_spread('BTC')
    print(df.to_string(index=False))
    if 'spread_bps' in df.columns:
        print(f"Max spread: {df['spread_bps'].iloc[0]:.2f} bps")

    print("\nCEX-CEX ETH Spread")
    df_eth = cex_cex_spread('ETH')
    print(df_eth.to_string(index=False))

    # 模拟 CEX-DEX
    print("\nCEX-DEX Arbitrage Simulation")
    pool = AMMPool(reserve_x=5_000_000, reserve_y=1500, fee=0.003)
    # ETH/USDC pool: 5M USDC + 1500 ETH, mid ≈ 3333
    cex_price = 3350  # CEX ETH 价
    result = cex_dex_opportunity(cex_price, pool, trade_size_usd=10000,
                                  gas_price_gwei=30, eth_price=cex_price)
    for k, v in result.items():
        if isinstance(v, float):
            print(f"  {k}: {v:.4f}")
        else:
            print(f"  {k}: {v}")

    # 跨链
    print("\nCross-Chain Arbitrage (ETH on Mainnet vs Arbitrum)")
    cc = CrossChainArb(
        src_chain='Ethereum', dst_chain='Arbitrum', asset='ETH',
        src_price=3300, dst_price=3320,
        bridge_fee_pct=0.0010, bridge_time_sec=120,  # Across 快桥
        notional_usd=50_000
    )
    eval_result = cc.evaluate()
    for k, v in eval_result.items():
        print(f"  {k}: {v}")

---

四、真实数据/案例

案例 1：BitMEX 套利时代（2017-2018）

BitMEX 永续 vs Bitstamp 现货曾长期 +5-15% 溢价。

• 简单套利：Bitstamp 买 BTC，BitMEX 开空 BTC perp
• 收 BitMEX funding（高时 0.5%/8h）
• 早期年化 100%+，但需要 large balance 在 BitMEX

案例 2：FTX 破产前的 GBTC 折价套利失败

GBTC 折价 -50% 看似机会：

• 买 GBTC，做空 BTC perp
• 假设折价回归到 0
• 但 GBTC 没有赎回机制（直到 2024 ETF 转换）
• 折价持续 2 年，套利者资金成本高昂

教训：套利需要明确的 convergence path。

案例 3：USDC depeg 套利（2023-03）

2023-03-11 USDC = 0.88 时：

• DEX (Curve 3pool) USDC vs USDT 套利
• 但 Coinbase 保证 USDC 1:1 兑 USD
• 操作：DEX 买 0.88 USDC → Coinbase 取出 0.88 美元 → 买 1 USDC（套到 12%）
• 但 Coinbase 实际限制提款，部分套利者卡住资金 48h
• 实际盈利 7-10%

案例 4：跨链桥黑客攻击

教训：跨链套利的"黑天鹅"是桥被黑，资金永久损失。风控：单桥仓位 < 总资本 5%。

案例 5：MEV bot 占据简单套利

2021 后 Ethereum 上简单 DEX 套利被专业 MEV bot 占据：

• bot 平均利润 $0.5-5/笔
• 普通用户根本抢不过（gas 战）
• Flashbots 调查：> 95% 的 atomic arbitrage 由 < 100 个地址完成

---

五、CEX vs DEX 策略差异

DeFi 原生套利机会：

1. Curve 池失衡：3pool 中 USDC/USDT/DAI 偏离 1:1:1
2. Pendle PT 折价收敛：到期前必然回归到 1
3. LST 折价：stETH/ETH、rETH/ETH 在压力期折价
4. 借贷利率套利：AAVE 借 USDC → Compound 存 USDC（利差时）
5. Stable swap：跨链稳定币池（Stargate）失衡

---

六、风险管理

6.1 套利特有风险

6.2 仓位 sizing 原则

def safe_arb_size(spread_pct, expected_slippage_pct, fee_pct,
                   confidence=0.8):
    """
    保守 sizing
    """
    net_edge = spread_pct - 2 * fee_pct - expected_slippage_pct
    if net_edge < 0.001:  # 至少 10 bps
        return 0
    # 保守：用 2σ 滑点估计
    safe_size = min_capital_for_edge(net_edge, confidence)
    return safe_size

6.3 熔断规则

• 单笔失败率 > 20% 持续 10 笔 → 暂停 30 分钟
• 累计 1 天损失 > 1% → 暂停一天
• gas > 500 gwei → 暂停 DEX 套利

---

七、关键速查

三角套利公式

Profit% = (1 / (P_AB × P_BC × P_CA)) - 1 - 3×fee

CEX-DEX 套利门槛

Net = (P_CEX - P_DEX) - slippage - 2×fee - gas/notional
要求 Net > 30 bps 才入场（保守）

桥时间表

---

八、面试题

Q1：为什么简单 DEX 套利已经不赚钱了？

答：

1. MEV bot 占据：> 95% atomic arb 被 < 100 个 bot 占据
2. Gas 战争：bot 用极高 gas 抢跑
3. Flashloan 套利：能借大量资金一次执行
4. 私有 mempool：bot 用 Flashbots 隐藏意图
5. AMM 优化：Uniswap V3 集中流动性减少了 V2 时代的明显 spread

• 普通玩家机会：复杂多腿、跨链、长尾 token、新协议初期

Q2：跨链套利怎么管理桥风险？

答：

1. 桥分散：用 3+ 个桥分散流量（Across、Stargate、LayerZero）
2. 限额：单桥仓位 < 总 5%；单笔 < $500K
3. 时间分散：避免短时大量桥
4. 保险：Nexus Mutual / InsurAce 桥保险（5-15% 年费）
5. 优先用 native bridge：CCTP（USDC 官方）、Polygon PoS bridge 等
6. 历史信誉：避免新桥（Ronin/Wormhole 都是高估值时被黑）

Q3：你怎么判断套利机会真实可执行？

答：

• 数据延迟：API 时间戳 vs 当前时间差 < 100ms
• 深度检查：top-of-book 价格 + 实际 size 可执行价
• 历史成功率：类似机会过去 100 笔实际成功 > 80%
• 滑点保守估计：实际滑点 = 模型 × 1.5
• gas 检查（DEX）：当前 gas 与平均比较，> 2× 平均时暂停
• MEV 风险评分：是否走 Flashbots（DEX）

Q4：CEX-CEX 套利的资金占用怎么优化？

答：

1. 动态再平衡：每天结束后用快桥/低费率所转移
2. 稳定币本位：用 USDC 在多所，BTC/ETH 用 perp 对冲
3. 借贷融资：AAVE 借稳定币部署套利，息差 < 套利收益时净赚
4. Cross-margin 在交易所内部：减少总占用
5. 分级管理：高确定性 mainnet 备货多，低确定性少量

Q5：稳定币套利的核心风险？

答：

• 脱钩风险：USDC 2023-03 脱钩 13%，套利模型崩
• 流动性单点：Curve 3pool 占多数流动性，被攻击或失衡
• 算法稳定币崩溃：UST 2022-05 归零
• 监管冻结：USDC 黑名单地址、提款限额
• Tether 储备质疑：每次 FUD 时 USDT 短暂折价
• 解法：分散稳定币（USDC + USDT + DAI + FRAX），单稳定币不超 30%

---

明日预告

Day 97: 资金费率套利 — Cash-and-carry 实战、对冲 delta、不同所 funding 差异。从今天的瞬时套利转到持续性 carry trade。