Function test_spot_websocket

tests/test_binance.py:20–81

Test Binance Spot WebSocket.

()

def test_spot_websocket():
    """Test Binance Spot WebSocket."""
    print("=" * 60)
    print("Testing Binance Spot WebSocket (1m closed klines only)")
    print("=" * 60)
    
    message_count = 0
    should_stop = False
    
    def on_message(ws, data):
        nonlocal message_count, should_stop
        message_count += 1
        
        # Data is already processed by WebSocket class (only closed 1m klines)
        # timestamp is already formatted as 'YYYY-MM-DD HH:MM:SS'
        print(f"[Spot] {data['timestamp']} | {data['open_time']} | {data['close_time']} | {data['symbol']} | "
              f"Open: {data['open']} | Close: {data['close']} | "
              f"High: {data['high']} | Low: {data['low']} | "
              f"Volume: {data['volume']} | Interval: {data['interval']}")
        
        # Stop after receiving 5 closed klines
        if message_count >= 5:
            should_stop = True
            try:
                ws.close()
            except:
                pass
    
    def on_error(ws, error):
        print(f"[Spot] Error: {error}")
    
    def on_close(ws):
        print("[Spot] WebSocket closed")
    
    def on_open(ws):
        print("[Spot] WebSocket opened")
    
    # Create and start Spot WebSocket
    ws_client = BinanceSpotWebSocket(
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
        on_open=on_open,
        testnet=False
    )
    
    # Subscribe to 1-minute klines (only closed klines will be processed)
    ws_client.subscribe_kline("BTCUSDT", "1m")
    
    print("Starting Spot WebSocket...")
    ws_client.start()
    
    # Wait for messages
    try:
        while ws_client.is_running() and not should_stop:
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("\n[Spot] Interrupted by user")