加密货币套利机器人的开发实战，分析了 API 速率限制、网络延迟及资金安全等痛点。方案对比了 REST 与 WebSocket、同步与异步 IO 的优劣，推荐使用异步模式结合 CCXT 库。核心部分涵盖三角套利计算逻辑、异步下单系统及熔断机制、心跳检测等生产环境设计。同时指出了浮点精度、手续费忽略等常见陷阱，并提供了基于 Decimal 的解决方案及动态权重分配的进阶思路。
开发加密货币套利机器人时，新手常会遇到几个致命问题：
代码示例（CCXT 连接）：
import ccxt
exchange = ccxt.binance({
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
})
await exchange.load_markets()
while True:
orderbook = await exchange.watch_order_book('BTC/USDT')
# 处理订单簿数据...
典型架构：
import asyncio
async def monitor_exchange(exchange, symbol):
while True:
ob = await exchange.watch_order_book(symbol)
process_orderbook(ob)
async def main():
tasks = [
monitor_exchange(binance, 'BTC/USDT'),
monitor_exchange(ftx, 'BTC/USDT')
]
await asyncio.gather(*tasks)
假设存在交易对：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
关键要点：
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'):
当扩展到多个交易所时，如何设计动态权重分配系统？考虑以下因素：
（提示：可参考强化学习中的 Multi-Armed Bandit 算法）
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online