Python实现API长连接：BitMEX交易程序自动重连方案问询

Hey there! Let’s tackle this connection monitoring and auto-reconnect issue for your BitMEX trading bot—this is a super common pain point with crypto exchange APIs, especially when dealing with high load scenarios where connections drop silently. Here’s a practical, battle-tested approach to keep your bot connected without interrupting user input flow:

核心思路 #

Instead of only detecting disconnects when a user sends a command, we’ll:

• Run a background monitoring thread to check connection health periodically
• Use heartbeat/ping mechanisms to validate if the connection is alive
• Implement a robust reconnection workflow that restores state after reconnecting
• Add fail-safe checks in your command handlers to catch disconnects early

Step 1: 实现连接存活检测 #

BitMEX’s API (both WebSocket and REST) supports ping/pong or lightweight status checks. Here’s how to implement it for your B connection class:

import time
import threading

class BitMEXAutoReconnectBot:
    def __init__(self):
        self.client = B()  # 初始化你的BitMEX连接类
        self.is_connected = False
        # 启动后台监控线程（daemon=True让线程随主程序退出）
        self.monitor_thread = threading.Thread(target=self._connection_monitor_loop, daemon=True)
        self.monitor_thread.start()
        self._get_initial_user_input()  # 你的初始输入逻辑

    def _get_initial_user_input(self):
        # 这里处理用户初始设置（比如交易对、杠杆等）
        print("请完成初始设置...")
        # 你的代码逻辑

    def _is_connection_alive(self):
        """检测连接是否存活的核心方法"""
        try:
            # 针对WebSocket连接：发送ping并等待pong响应
            if hasattr(self.client, 'ws') and self.client.ws:
                self.client.ws.ping()
                # 等待5秒获取pong响应（根据BitMEX的超时调整）
                start_time = time.time()
                while time.time() - start_time < 5:
                    if getattr(self.client.ws, 'last_pong', False):
                        self.client.ws.last_pong = False  # 重置标记
                        return True
                    time.sleep(0.1)
            # 针对REST API：发送轻量请求（比如获取服务器时间）
            # response = self.client.get('/api/v1/time')
            # return response.status_code == 200
            return False
        except (ConnectionResetError, BrokenPipeError, TimeoutError) as e:
            print(f"连接检测失败: {str(e)}")
            return False

    def _connection_monitor_loop(self):
        """后台线程循环：检测连接，断开时自动重连"""
        while True:
            if not self.is_connected:
                # 尝试建立连接
                try:
                    self.client.connect()  # 调用你的连接方法
                    self.is_connected = True
                    print("✅ 成功连接到BitMEX")
                    self._post_reconnect_setup()  # 重连后恢复状态
                except Exception as e:
                    print(f"❌ 重连失败，5秒后重试: {str(e)}")
                    time.sleep(5)
            else:
                # 定期检测连接状态（每10秒一次，可调整）
                if not self._is_connection_alive():
                    print("⚠️ 连接已断开，启动重连流程")
                    self.is_connected = False
                time.sleep(10)

    def _post_reconnect_setup(self):
        """重连后必须执行的状态恢复操作"""
        # 示例：重新订阅行情、同步账户持仓/订单数据
        # self.client.subscribe_orderbook('XBTUSD')
        # self.current_position = self.client.get_position()
        print("🔄 完成重连后的状态恢复")

Step 2: 在用户指令处理中添加安全检查 #

即使有后台监控，在处理用户指令时也要快速捕获断开异常，避免无效操作：

def _handle_user_commands(self):
    """用户指令处理循环"""
    while True:
        command = input("\n请输入指令（买入/卖出/撤单/退出）: ").strip().lower()
        if not self.is_connected:
            print("⚠️ 当前未连接到交易所，请等待重连完成")
            continue
        
        try:
            if command == "买入":
                # 你的买入逻辑
                print("执行买入操作")
            elif command == "卖出":
                # 你的卖出逻辑
                print("执行卖出操作")
            elif command == "撤单":
                # 你的撤单逻辑
                print("执行撤单操作")
            elif command == "退出":
                print("👋 退出程序")
                break
            else:
                print("❌ 无效指令，请重新输入")
        except (ConnectionResetError, BrokenPipeError) as e:
            print(f"⚠️ 执行指令时连接断开: {str(e)}")
            self.is_connected = False  # 立即标记断开，触发监控线程重连

关键注意事项 #

• 限流控制: 不要把重连间隔设得太短（建议5-10秒），避免触发BitMEX的API限流机制
• 线程安全: 如果你的B类不是线程安全的，在访问共享资源时要加锁（比如threading.Lock()）
• 指令缓存: 如果用户在重连时发送指令，可以把指令缓存起来，等重连完成后自动执行
• 日志记录: 建议添加详细的日志（比如用logging模块），方便排查连接问题

内容的提问来源于stack exchange，提问作者Guilherme Lima