웹소켓 자동 재접속과 침묵의 연결 단절 방어

[ 정상 상태 ]
바이낸스 거래소  ◄════════════ WebSocket 연결 유지 (양방향 통신) ════════════►  DashCloud 봇
                                                                                (매 10ms 틱 수신)
 
[ 라우터 및 방화벽의 연결 증발 현상 ]
바이낸스 거래소  ──X (네트워크 라우팅 테이블에서 연결 증발) X──► [파이썬 프로세스: ws.recv() 무한 대기]
                                                                  (에러도 안 뜨고 시세 수신 완전 중단)
                                                                  (거래소에서는 비트코인 10% 폭락 중)
                                                                  ➔ 봇은 시세 변동을 모른 채 강제 청산!

websockets>=12.0
asyncio>=3.4.3

import asyncio
import json
import time
import logging
import websockets
from websockets.exceptions import ConnectionClosed
 
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("SocketWatchdog")
 
# 바이낸스 실시간 미니 티커 웹소켓 엔드포인트
BINANCE_WS_URL = "wss://stream.binance.com:9443/ws/btcusdt@miniTicker"
 
class ResilientWebSocketClient:
    def __init__(self, url: str):
        self.url = url
        self.ws = None
        self.is_running = True
        self.last_packet_time = 0.0
        self.inactivity_timeout = 10.0 # 10초간 무패킷 시 강제 재접속
 
    async def start(self):
        retry_delay = 1.0
        max_retry_delay = 60.0
 
        while self.is_running:
            try:
                logger.info(f"거래소 웹소켓 접속 시도: {self.url}")
                # ping_interval=15: 15초마다 핑 발송, ping_timeout=5: 5초 무응답 시 단절 처리
                async with websockets.connect(
                    self.url,
                    ping_interval=15,
                    ping_timeout=5,
                    close_timeout=5
                ) as ws:
                    self.ws = ws
                    self.last_packet_time = time.time()
                    retry_delay = 1.0 # 접속 성공 시 재시도 딜레이 초기화
                    logger.info("✅ 웹소켓 연결 성공! 실시간 틱 수신 시작.")
 
                    # 인액티비티 워치독 태스크 동시 실행
                    watchdog_task = asyncio.create_task(self._inactivity_watchdog())
                    
                    try:
                        await self._receive_loop()
                    finally:
                        watchdog_task.cancel()
 
            except (ConnectionClosed, asyncio.TimeoutError, Exception) as e:
                logger.warning(f"⚠️ 소켓 단절 감지 ({e}). {retry_delay:.1f}초 후 재접속 시도...")
                await asyncio.sleep(retry_delay)
                # 지수 백오프 적용 (1초 ➔ 2초 ➔ 4초 ... 최대 60초)
                retry_delay = min(retry_delay * 2, max_retry_delay)
 
    async def _receive_loop(self):
        """실시간 메시지 수신 루프"""
        while self.is_running:
            message = await self.ws.recv()
            self.last_packet_time = time.time() # 수신 시각 갱신
            data = json.loads(message)
            
            # 실시간 시세 데이터 처리
            symbol = data.get("s")
            close_price = data.get("c")
            logger.debug(f"틱 수신: [{symbol}] {close_price}")
 
    async def _inactivity_watchdog(self):
        """10초간 패킷 미수신 시 침묵 단절로 판정하고 소켓 강제 폐기"""
        while self.is_running:
            await asyncio.sleep(2.0)
            elapsed = time.time() - self.last_packet_time
            if elapsed > self.inactivity_timeout:
                logger.critical(f"🚨 침묵의 연결 단절 감지! ({elapsed:.1f}초간 무패킷). 소켓을 강제 종료합니다.")
                if self.ws:
                    await self.ws.close()
                break
 
    def stop(self):
        self.is_running = False
 
if __name__ == "__main__":
    client = ResilientWebSocketClient(BINANCE_WS_URL)
    try:
        asyncio.run(client.start())
    except KeyboardInterrupt:
        client.stop()
        logger.info("클라이언트 종료.")