Hummingbot 第10章：套利策略

套利策略利用不同市场之间的价格差异来获取无风险或低风险收益。Hummingbot 提供多种套利策略模板。

跨交易所套利

Arbitrage 策略

Arbitrage 策略监控两个交易所之间的价格差异，当价差超过阈值时进行低买高卖。

策略配置

# conf_arbitrage.yml
strategy: arbitrage
connector_1: binance         # 交易所 A
connector_2: okx              # 交易所 B
trading_pair: BTC-USDT

# 套利参数
min_profitability: 0.5       # 最低盈利阈值（%）
max_order_size: 0.1          # 最大订单量
max_concurrent_orders: 1     # 最大并发订单数

# 滑点保护
slippage_buffer: 0.1         # 滑点缓冲（%）

# 限速
rate_limits_enabled: true    # 启用限速

# 日志
log_level: INFO

策略工作原理

监控步骤:
1. 同时获取交易所 A 和 B 的订单簿
2. 计算理论套利利润（扣除手续费）
3. 如果利润 > min_profitability，执行交易
4. 在低价交易所买入，在高价交易所卖出

示例:
Binance BTC 卖一价: 50,000 USDT
OKX BTC 买一价:    50,300 USDT

套利空间: (50,300 - 50,000) / 50,000 = 0.6%

扣除手续费 (0.1% × 2 = 0.2%):
净收益: 0.6% - 0.2% = 0.4%

如果 min_profitability = 0.3%，执行套利

自动套利脚本

"""
cross_exchange_arbitrage.py - 跨交易所套利脚本
"""
from decimal import Decimal
from hummingbot.strategy_v2.scripts import ScriptBase
from hummingbot.core.data_type.common import OrderType, TradeType


class CrossExchangeArbitrage(ScriptBase):
    """跨交易所套利脚本"""

    def __init__(self, connectors: list, markets: dict):
        super().__init__(connectors, markets)
        self.exchange_a = "binance"
        self.exchange_b = "okx"
        self.pair = "BTC-USDT"
        self.min_profitability = Decimal("0.003")  # 0.3%
        self.max_order_size = Decimal("0.01")
        self.min_order_size = Decimal("0.001")

    async def check_arbitrage_opportunity(self):
        """检查套利机会"""
        conn_a = self.connectors[self.exchange_a]
        conn_b = self.connectors[self.exchange_b]

        # 获取订单簿
        orderbook_a = await conn_a.get_order_book(self.pair)
        orderbook_b = await conn_b.get_order_book(self.pair)

        # 最佳买卖价
        best_bid_a = orderbook_a.bid_prices[0]
        best_ask_a = orderbook_a.ask_prices[0]
        best_bid_b = orderbook_b.bid_prices[0]
        best_ask_b = orderbook_b.ask_prices[0]

        # 计算套利机会
        opportunities = []

        # 方向 1: A 买 B 卖
        if best_ask_a > 0 and best_bid_b > best_ask_a:
            profit_pct = (best_bid_b - best_ask_a) / best_ask_a
            if profit_pct > self.min_profitability:
                opportunities.append({
                    "buy_exchange": self.exchange_a,
                    "sell_exchange": self.exchange_b,
                    "buy_price": best_ask_a,
                    "sell_price": best_bid_b,
                    "profit_pct": profit_pct
                })

        # 方向 2: B 买 A 卖
        if best_ask_b > 0 and best_bid_a > best_ask_b:
            profit_pct = (best_bid_a - best_ask_b) / best_ask_b
            if profit_pct > self.min_profitability:
                opportunities.append({
                    "buy_exchange": self.exchange_b,
                    "sell_exchange": self.exchange_a,
                    "buy_price": best_ask_b,
                    "sell_price": best_bid_a,
                    "profit_pct": profit_pct
                })

        return opportunities

    async def execute_arbitrage(self, opportunity: dict):
        """执行套利交易"""
        buy_conn = self.connectors[opportunity["buy_exchange"]]
        sell_conn = self.connectors[opportunity["sell_exchange"]]

        amount = min(
            self.max_order_size,
            opportunity.get("available_amount", self.max_order_size)
        )

        if amount < self.min_order_size:
            self.logger().info("Order size too small, skipping")
            return

        # 执行买入
        buy_order = await buy_conn.place_order(
            trading_pair=self.pair,
            order_type=OrderType.LIMIT,
            side=TradeType.BUY,
            price=opportunity["buy_price"],
            amount=amount
        )

        # 执行卖出
        sell_order = await sell_conn.place_order(
            trading_pair=self.pair,
            order_type=OrderType.LIMIT,
            side=TradeType.SELL,
            price=opportunity["sell_price"],
            amount=amount
        )

        self.logger().info(
            f"Arbitrage executed: "
            f"Buy {amount} @ {opportunity['buy_price']} on "
            f"{opportunity['buy_exchange']}, "
            f"Sell @ {opportunity['sell_price']} on "
            f"{opportunity['sell_exchange']}, "
            f"Profit: {opportunity['profit_pct'] * 100:.3f}%"
        )

    async def on_tick(self):
        """每秒检查套利机会"""
        try:
            opportunities = await self.check_arbitrage_opportunity()
            for opp in opportunities:
                self.logger().info(
                    f"Arbitrage opportunity found: "
                    f"{opp['profit_pct'] * 100:.3f}%"
                )
                await self.execute_arbitrage(opp)
        except Exception as e:
            self.logger().error(f"Arbitrage check failed: {str(e)}")

三角套利

三角套利原理

利用三个交易对之间的价格不一致性进行套利：

三角套利路径:
BTC → ETH → USDT → BTC

1. 用 BTC 买 ETH (BTC-ETH)
2. 用 ETH 买 USDT (ETH-USDT)
3. 用 USDT 买 BTC (USDT-BTC)

如果最终 BTC 数量 > 初始 BTC 数量，存在套利机会

三角套利脚本

class TriangularArbitrage(ScriptBase):
    """三角套利脚本"""

    def __init__(self, connectors: list, markets: dict):
        super().__init__(connectors, markets)
        self.exchange = "binance"
        self.triangles = [
            # 三角形路径: (base, quote, intermediate)
            ("BTC", "ETH", "USDT"),
            ("ETH", "SOL", "USDT"),
            ("BTC", "BNB", "USDT"),
        ]

    async def check_triangle(self, base: str, quote: str, intermediate: str):
        """检查三角套利机会"""
        conn = self.connectors[self.exchange]

        # 获取三个交易对的价格
        pair_1 = f"{base}-{quote}"
        pair_2 = f"{quote}-{intermediate}"
        pair_3 = f"{intermediate}-{base}"

        price_1 = conn.get_price(pair_1)
        price_2 = conn.get_price(pair_2)
        price_3 = conn.get_price(pair_3)

        if not all([price_1, price_2, price_3]):
            return None

        # 计算套利比率
        profit = (1 / price_1) * (1 / price_2) * price_3

        return {
            "path": f"{base} -> {quote} -> {intermediate} -> {base}",
            "profit_pct": (profit - 1) * 100
        }

CROSS-EXCHANGE-MARKET-MAKING 策略

策略说明

Cross Exchange Market Making（XEMM）在两个交易所同时做市，利用其中一个交易所的流动性对冲在另一个交易所的做市操作。

配置示例

# conf_xemm.yml
strategy: cross_exchange_market_making
maker_exchange: binance        # 做市交易所
taker_exchange: okx            # 对冲交易所
trading_pair: BTC-USDT

# 做市参数
bid_spread: 0.2                # 买单做市价差
ask_spread: 0.2                # 卖单做市价差
order_amount: 0.01             # 做市订单量

# 对冲参数
min_profitability: 0.1         # 最低对冲盈利（%）
hedge_ratio: 1.0               # 对冲比例（1.0 = 完全对冲）

# 限价订单对冲
limit_orders_hedging: true     # 使用限价单对冲
adjust_order_enabled: true     # 自动调整订单

XEMM 与其他套利策略对比

套利风险控制

主要风险类型

风险参数配置

# 风控参数
risk_controls:
  # 每日最大交易次数
  max_daily_trades: 50

  # 单笔最大亏损
  max_loss_per_trade: 10.0  # USDT

  # 每日最大亏损
  max_daily_loss: 50.0  # USDT

  # 最大持仓
  max_position_size: 1000.0  # USDT 等值

  # 冷却时间（交易后暂停）
  cooldown_after_trade: 5  # 秒

  # 价差收窄时暂停
  pause_on_spread_narrowing: true
  min_spread_to_resume: 0.3  # %