高频套利机器人开发实战：核心逻辑与避坑指南

核心算法实现

三角套利计算

假设存在交易对：BTC/USDT, ETH/BTC, ETH/USDT

套利条件：(1 / BTC/USDT_ask) * (1 / ETH/BTC_ask) * ETH/USDT_bid > 1 + fee

Python 实现：

def check_triangular_arb(btc_usdt, eth_btc, eth_usdt, fee=0.002):
    # 注意：所有价格都应从订单簿获取最优报价
    theoretical = (1 / btc_usdt['ask']) * (1 / eth_btc['ask']) * eth_usdt['bid']
    return theoretical - 1 > fee

异步下单系统

关键要点：

1. 使用订单 ID 跟踪交易状态
2. 实现超时重试机制
3. 严格处理部分成交情况

async def safe_order(exchange, symbol, side, amount, price):
    try:
        # NOTE: 使用 postOnly 避免吃单手续费
        order = await exchange.create_order(
            symbol=symbol, type='limit', side=side, amount=amount, price=price,
            params={'postOnly': True}
        )
        logger.info(f"Order created: {order['id']}")
        return await track_order(exchange, order['id'])
    except ccxt.NetworkError as e:
        logger.error(f"Network error: {e}")
        await asyncio.sleep(1)
        return await safe_order(...) # 指数退避重试

生产环境关键设计

熔断机制实现

当检测到连续亏损时自动停止交易：

class CircuitBreaker:
    def __init__(self, max_loss=0.05):
        self.max_loss = max_loss
        self.reset()

    def reset(self):
        self._consecutive_losses = 0
        self._active = True

    def update(self, pnl):
        if pnl < 0:
            self._consecutive_losses += 1
            if self._consecutive_losses >= 3:
                self._active = False
        else:
            self.reset()

心跳检测方案

防止僵尸进程的守护线程：

import threading
class Heartbeat:
    def __init__(self, timeout=10):
        self.last_beat = time.time()
        self.timeout = timeout
        self._monitor()

    def _monitor(self):
        def _run():
            while True:
                if time.time() - self.last_beat > self.timeout:
                    os._exit(1) # 强制退出
                time.sleep(1)
        threading.Thread(target=_run, daemon=True).start()

    def beat(self):
        self.last_beat = time.time()

常见陷阱与解决方案

浮点精度问题案例

手续费忽略：

# 错误：未计入手续费
profit = (sell_price - buy_price) * amount

# 正确：考虑 maker/taker 区别
fee = exchange.market(symbol)['taker'] if is_market_order else ...
profit = (sell_price*(1-fee) - buy_price*(1+fee)) * amount

数量舍入错误：

# 错误：未考虑交易所最小交易单位
amount = 0.123456789

# 正确：遵守 lot size 规则
amount = exchange.amount_to_precision(symbol, 0.123456789)

价格计算错误：

# 错误：使用浮点数直接比较
if price_a / price_b > 1.001:

# 正确：使用 Decimal 或固定精度
from decimal import Decimal, getcontext
getcontext().prec = 8
if Decimal(price_a) / Decimal(price_b) > Decimal('1.001'):

进阶思考

当扩展到多个交易所时，如何设计动态权重分配系统？考虑以下因素：

• 各交易所的 API 可靠性历史数据
• 当前网络延迟监测
• 资金利用率与风控平衡
• 交易所的流动性深度

（提示：可参考强化学习中的 Multi-Armed Bandit 算法）