""" Gate.io exchange connector implementation. Supports WebSocket connections to Gate.io with their WebSocket v4 API. """ import json import hmac import hashlib import time from typing import Dict, List, Optional, Any from datetime import datetime, timezone from ..models.core import OrderBookSnapshot, TradeEvent, PriceLevel from ..utils.logging import get_logger, set_correlation_id from ..utils.exceptions import ValidationError, ConnectionError from ..utils.validation import validate_symbol, validate_price, validate_volume from .base_connector import BaseExchangeConnector logger = get_logger(__name__) class GateIOConnector(BaseExchangeConnector): """ Gate.io WebSocket connector implementation. Supports: - WebSocket v4 API - Order book streams - Trade streams - Symbol normalization - Authentication for private channels """ # Gate.io WebSocket URLs WEBSOCKET_URL = "wss://api.gateio.ws/ws/v4/" TESTNET_URL = "wss://fx-api-testnet.gateio.ws/ws/v4/" API_URL = "https://api.gateio.ws" def __init__(self, use_testnet: bool = False, api_key: str = None, api_secret: str = None): """ Initialize Gate.io connector. Args: use_testnet: Whether to use testnet environment api_key: API key for authentication (optional) api_secret: API secret for authentication (optional) """ websocket_url = self.TESTNET_URL if use_testnet else self.WEBSOCKET_URL super().__init__("gateio", websocket_url) # Authentication credentials (optional) self.api_key = api_key self.api_secret = api_secret self.use_testnet = use_testnet # Gate.io-specific message handlers self.message_handlers.update({ 'spot.order_book_update': self._handle_orderbook_update, 'spot.trades': self._handle_trade_update, 'spot.pong': self._handle_pong, 'error': self._handle_error_message }) # Subscription tracking self.subscribed_channels = set() self.request_id = 1 logger.info(f"Gate.io connector initialized ({'testnet' if use_testnet else 'mainnet'})") def _get_message_type(self, data: Dict) -> str: """ Determine message type from Gate.io message data. Args: data: Parsed message data Returns: str: Message type identifier """ # Gate.io v4 API message format if 'method' in data: return data['method'] # 'spot.order_book_update', 'spot.trades', etc. elif 'error' in data: return 'error' elif 'result' in data: return 'result' return 'unknown' def normalize_symbol(self, symbol: str) -> str: """ Normalize symbol to Gate.io format. Args: symbol: Standard symbol format (e.g., 'BTCUSDT') Returns: str: Gate.io symbol format (e.g., 'BTC_USDT') """ # Gate.io uses underscore-separated format if symbol.upper() == 'BTCUSDT': return 'BTC_USDT' elif symbol.upper() == 'ETHUSDT': return 'ETH_USDT' elif symbol.upper().endswith('USDT'): base = symbol[:-4].upper() return f"{base}_USDT" elif symbol.upper().endswith('USD'): base = symbol[:-3].upper() return f"{base}_USD" else: # Assume it's already in correct format or add underscore if '_' not in symbol: # Try to split common patterns if len(symbol) >= 6: # Assume last 4 chars are quote currency base = symbol[:-4].upper() quote = symbol[-4:].upper() return f"{base}_{quote}" else: return symbol.upper() else: return symbol.upper() def _denormalize_symbol(self, gateio_symbol: str) -> str: """ Convert Gate.io symbol back to standard format. Args: gateio_symbol: Gate.io symbol format (e.g., 'BTC_USDT') Returns: str: Standard symbol format (e.g., 'BTCUSDT') """ if '_' in gateio_symbol: return gateio_symbol.replace('_', '') return gateio_symbol async def subscribe_orderbook(self, symbol: str) -> None: """ Subscribe to order book updates for a symbol. Args: symbol: Trading symbol (e.g., 'BTCUSDT') """ try: set_correlation_id() gateio_symbol = self.normalize_symbol(symbol) # Create subscription message subscription_msg = { "method": "spot.order_book", "params": [gateio_symbol, 20, "0"], # symbol, limit, interval "id": self.request_id } self.request_id += 1 # Send subscription success = await self._send_message(subscription_msg) if success: # Track subscription if symbol not in self.subscriptions: self.subscriptions[symbol] = [] if 'orderbook' not in self.subscriptions[symbol]: self.subscriptions[symbol].append('orderbook') self.subscribed_channels.add(f"spot.order_book:{gateio_symbol}") logger.info(f"Subscribed to order book for {symbol} ({gateio_symbol}) on Gate.io") else: logger.error(f"Failed to subscribe to order book for {symbol} on Gate.io") except Exception as e: logger.error(f"Error subscribing to order book for {symbol}: {e}") raise async def subscribe_trades(self, symbol: str) -> None: """ Subscribe to trade updates for a symbol. Args: symbol: Trading symbol (e.g., 'BTCUSDT') """ try: set_correlation_id() gateio_symbol = self.normalize_symbol(symbol) # Create subscription message subscription_msg = { "method": "spot.trades", "params": [gateio_symbol], "id": self.request_id } self.request_id += 1 # Send subscription success = await self._send_message(subscription_msg) if success: # Track subscription if symbol not in self.subscriptions: self.subscriptions[symbol] = [] if 'trades' not in self.subscriptions[symbol]: self.subscriptions[symbol].append('trades') self.subscribed_channels.add(f"spot.trades:{gateio_symbol}") logger.info(f"Subscribed to trades for {symbol} ({gateio_symbol}) on Gate.io") else: logger.error(f"Failed to subscribe to trades for {symbol} on Gate.io") except Exception as e: logger.error(f"Error subscribing to trades for {symbol}: {e}") raise async def unsubscribe_orderbook(self, symbol: str) -> None: """ Unsubscribe from order book updates for a symbol. Args: symbol: Trading symbol (e.g., 'BTCUSDT') """ try: gateio_symbol = self.normalize_symbol(symbol) # Create unsubscription message unsubscription_msg = { "method": "spot.unsubscribe", "params": [f"spot.order_book", gateio_symbol], "id": self.request_id } self.request_id += 1 # Send unsubscription success = await self._send_message(unsubscription_msg) if success: # Remove from tracking if symbol in self.subscriptions and 'orderbook' in self.subscriptions[symbol]: self.subscriptions[symbol].remove('orderbook') if not self.subscriptions[symbol]: del self.subscriptions[symbol] self.subscribed_channels.discard(f"spot.order_book:{gateio_symbol}") logger.info(f"Unsubscribed from order book for {symbol} ({gateio_symbol}) on Gate.io") else: logger.error(f"Failed to unsubscribe from order book for {symbol} on Gate.io") except Exception as e: logger.error(f"Error unsubscribing from order book for {symbol}: {e}") raise async def unsubscribe_trades(self, symbol: str) -> None: """ Unsubscribe from trade updates for a symbol. Args: symbol: Trading symbol (e.g., 'BTCUSDT') """ try: gateio_symbol = self.normalize_symbol(symbol) # Create unsubscription message unsubscription_msg = { "method": "spot.unsubscribe", "params": ["spot.trades", gateio_symbol], "id": self.request_id } self.request_id += 1 # Send unsubscription success = await self._send_message(unsubscription_msg) if success: # Remove from tracking if symbol in self.subscriptions and 'trades' in self.subscriptions[symbol]: self.subscriptions[symbol].remove('trades') if not self.subscriptions[symbol]: del self.subscriptions[symbol] self.subscribed_channels.discard(f"spot.trades:{gateio_symbol}") logger.info(f"Unsubscribed from trades for {symbol} ({gateio_symbol}) on Gate.io") else: logger.error(f"Failed to unsubscribe from trades for {symbol} on Gate.io") except Exception as e: logger.error(f"Error unsubscribing from trades for {symbol}: {e}") raise async def get_symbols(self) -> List[str]: """ Get list of available trading symbols from Gate.io. Returns: List[str]: List of available symbols in standard format """ try: import aiohttp api_url = "https://fx-api-testnet.gateio.ws" if self.use_testnet else self.API_URL async with aiohttp.ClientSession() as session: async with session.get(f"{api_url}/api/v4/spot/currency_pairs") as response: if response.status == 200: data = await response.json() symbols = [] for pair_info in data: if pair_info.get('trade_status') == 'tradable': pair_id = pair_info.get('id', '') # Convert to standard format standard_symbol = self._denormalize_symbol(pair_id) symbols.append(standard_symbol) logger.info(f"Retrieved {len(symbols)} symbols from Gate.io") return symbols else: logger.error(f"Failed to get symbols from Gate.io: HTTP {response.status}") return [] except Exception as e: logger.error(f"Error getting symbols from Gate.io: {e}") return [] async def get_orderbook_snapshot(self, symbol: str, depth: int = 20) -> Optional[OrderBookSnapshot]: """ Get current order book snapshot from Gate.io REST API. Args: symbol: Trading symbol depth: Number of price levels to retrieve Returns: OrderBookSnapshot: Current order book or None if unavailable """ try: import aiohttp gateio_symbol = self.normalize_symbol(symbol) api_url = "https://fx-api-testnet.gateio.ws" if self.use_testnet else self.API_URL # Gate.io supports various depths api_depth = min(depth, 100) url = f"{api_url}/api/v4/spot/order_book" params = { 'currency_pair': gateio_symbol, 'limit': api_depth } async with aiohttp.ClientSession() as session: async with session.get(url, params=params) as response: if response.status == 200: data = await response.json() return self._parse_orderbook_snapshot(data, symbol) else: logger.error(f"Failed to get order book for {symbol}: HTTP {response.status}") return None except Exception as e: logger.error(f"Error getting order book snapshot for {symbol}: {e}") return None def _parse_orderbook_snapshot(self, data: Dict, symbol: str) -> OrderBookSnapshot: """ Parse Gate.io order book data into OrderBookSnapshot. Args: data: Raw Gate.io order book data symbol: Trading symbol Returns: OrderBookSnapshot: Parsed order book """ try: # Parse bids and asks bids = [] for bid_data in data.get('bids', []): price = float(bid_data[0]) size = float(bid_data[1]) if validate_price(price) and validate_volume(size): bids.append(PriceLevel(price=price, size=size)) asks = [] for ask_data in data.get('asks', []): price = float(ask_data[0]) size = float(ask_data[1]) if validate_price(price) and validate_volume(size): asks.append(PriceLevel(price=price, size=size)) # Create order book snapshot orderbook = OrderBookSnapshot( symbol=symbol, exchange=self.exchange_name, timestamp=datetime.now(timezone.utc), # Gate.io doesn't provide timestamp in snapshot bids=bids, asks=asks, sequence_id=data.get('id') ) return orderbook except Exception as e: logger.error(f"Error parsing order book snapshot: {e}") raise ValidationError(f"Invalid order book data: {e}", "PARSE_ERROR") async def _handle_orderbook_update(self, data: Dict) -> None: """ Handle order book update from Gate.io. Args: data: Order book update data """ try: set_correlation_id() params = data.get('params', []) if len(params) < 2: logger.warning("Invalid order book update format") return # Gate.io format: [symbol, order_book_data] gateio_symbol = params[0] symbol = self._denormalize_symbol(gateio_symbol) book_data = params[1] # Parse bids and asks bids = [] for bid_data in book_data.get('bids', []): price = float(bid_data[0]) size = float(bid_data[1]) if validate_price(price) and validate_volume(size): bids.append(PriceLevel(price=price, size=size)) asks = [] for ask_data in book_data.get('asks', []): price = float(ask_data[0]) size = float(ask_data[1]) if validate_price(price) and validate_volume(size): asks.append(PriceLevel(price=price, size=size)) # Create order book snapshot orderbook = OrderBookSnapshot( symbol=symbol, exchange=self.exchange_name, timestamp=datetime.fromtimestamp(int(book_data.get('t', 0)) / 1000, tz=timezone.utc), bids=bids, asks=asks, sequence_id=book_data.get('id') ) # Notify callbacks self._notify_data_callbacks(orderbook) logger.debug(f"Processed order book update for {symbol}") except Exception as e: logger.error(f"Error handling order book update: {e}") async def _handle_trade_update(self, data: Dict) -> None: """ Handle trade update from Gate.io. Args: data: Trade update data """ try: set_correlation_id() params = data.get('params', []) if len(params) < 2: logger.warning("Invalid trade update format") return # Gate.io format: [symbol, [trade_data]] gateio_symbol = params[0] symbol = self._denormalize_symbol(gateio_symbol) trades_data = params[1] # Process each trade for trade_data in trades_data: price = float(trade_data.get('price', 0)) amount = float(trade_data.get('amount', 0)) # Validate data if not validate_price(price) or not validate_volume(amount): logger.warning(f"Invalid trade data: price={price}, amount={amount}") continue # Determine side (Gate.io uses 'side' field) side = trade_data.get('side', 'unknown').lower() # Create trade event trade = TradeEvent( symbol=symbol, exchange=self.exchange_name, timestamp=datetime.fromtimestamp(int(trade_data.get('time', 0)), tz=timezone.utc), price=price, size=amount, side=side, trade_id=str(trade_data.get('id', '')) ) # Notify callbacks self._notify_data_callbacks(trade) logger.debug(f"Processed trade for {symbol}: {side} {amount} @ {price}") except Exception as e: logger.error(f"Error handling trade update: {e}") async def _handle_pong(self, data: Dict) -> None: """ Handle pong response from Gate.io. Args: data: Pong response data """ logger.debug("Received Gate.io pong") async def _handle_error_message(self, data: Dict) -> None: """ Handle error message from Gate.io. Args: data: Error message data """ error_info = data.get('error', {}) code = error_info.get('code', 'unknown') message = error_info.get('message', 'Unknown error') logger.error(f"Gate.io error {code}: {message}") def _get_auth_signature(self, method: str, url: str, query_string: str, payload: str, timestamp: str) -> str: """ Generate authentication signature for Gate.io. Args: method: HTTP method url: Request URL query_string: Query string payload: Request payload timestamp: Request timestamp Returns: str: Authentication signature """ if not self.api_key or not self.api_secret: return "" try: # Create signature string message = f"{method}\n{url}\n{query_string}\n{hashlib.sha512(payload.encode()).hexdigest()}\n{timestamp}" # Generate signature signature = hmac.new( self.api_secret.encode('utf-8'), message.encode('utf-8'), hashlib.sha512 ).hexdigest() return signature except Exception as e: logger.error(f"Error generating auth signature: {e}") return "" async def _send_ping(self) -> None: """Send ping to keep connection alive.""" try: ping_msg = { "method": "spot.ping", "params": [], "id": self.request_id } self.request_id += 1 await self._send_message(ping_msg) logger.debug("Sent ping to Gate.io") except Exception as e: logger.error(f"Error sending ping: {e}") def get_gateio_stats(self) -> Dict[str, Any]: """Get Gate.io-specific statistics.""" base_stats = self.get_stats() gateio_stats = { 'subscribed_channels': list(self.subscribed_channels), 'use_testnet': self.use_testnet, 'authenticated': bool(self.api_key and self.api_secret), 'next_request_id': self.request_id } base_stats.update(gateio_stats) return base_stats