import logging import time from typing import Dict, Any, List, Optional import requests import hmac import hashlib from urllib.parse import urlencode from .exchange_interface import ExchangeInterface logger = logging.getLogger(__name__) class MEXCInterface(ExchangeInterface): """MEXC Exchange API Interface""" def __init__(self, api_key: str = None, api_secret: str = None, test_mode: bool = True): """Initialize MEXC exchange interface. Args: api_key: MEXC API key api_secret: MEXC API secret test_mode: If True, use test/sandbox environment (Note: MEXC doesn't have a true sandbox) """ super().__init__(api_key, api_secret, test_mode) self.base_url = "https://api.mexc.com" self.api_version = "v3" def connect(self) -> bool: """Connect to MEXC API. This is a no-op for REST API.""" if not self.api_key or not self.api_secret: logger.warning("MEXC API credentials not provided. Running in read-only mode.") return False try: # Test connection by getting account info self.get_account_info() logger.info("Successfully connected to MEXC API") return True except Exception as e: logger.error(f"Failed to connect to MEXC API: {str(e)}") return False def _generate_signature(self, params: Dict[str, Any]) -> str: """Generate signature for authenticated requests.""" query_string = urlencode(params) signature = hmac.new( self.api_secret.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256 ).hexdigest() return signature def _send_public_request(self, method: str, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]: """Send public request to MEXC API.""" url = f"{self.base_url}/{self.api_version}/{endpoint}" try: if method.upper() == 'GET': response = requests.get(url, params=params) else: response = requests.post(url, json=params) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error in public request to {endpoint}: {str(e)}") raise def _send_private_request(self, method: str, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]: """Send private/authenticated request to MEXC API.""" if not self.api_key or not self.api_secret: raise ValueError("API key and secret are required for private requests") if params is None: params = {} # Add timestamp params['timestamp'] = int(time.time() * 1000) # Generate signature signature = self._generate_signature(params) params['signature'] = signature # Set headers headers = { 'X-MEXC-APIKEY': self.api_key } url = f"{self.base_url}/{self.api_version}/{endpoint}" try: if method.upper() == 'GET': response = requests.get(url, params=params, headers=headers) elif method.upper() == 'POST': response = requests.post(url, json=params, headers=headers) elif method.upper() == 'DELETE': response = requests.delete(url, params=params, headers=headers) else: raise ValueError(f"Unsupported HTTP method: {method}") response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error in private request to {endpoint}: {str(e)}") raise def get_account_info(self) -> Dict[str, Any]: """Get account information.""" return self._send_private_request('GET', 'account') def get_balance(self, asset: str) -> float: """Get balance of a specific asset. Args: asset: Asset symbol (e.g., 'BTC', 'USDT') Returns: float: Available balance of the asset """ try: account_info = self._send_private_request('GET', 'account') balances = account_info.get('balances', []) for balance in balances: if balance['asset'] == asset: return float(balance['free']) # Asset not found return 0.0 except Exception as e: logger.error(f"Error getting balance for {asset}: {str(e)}") return 0.0 def get_ticker(self, symbol: str) -> Dict[str, Any]: """Get current ticker data for a symbol. Args: symbol: Trading symbol (e.g., 'BTC/USDT') Returns: dict: Ticker data including price information """ mexc_symbol = symbol.replace('/', '') try: ticker = self._send_public_request('GET', 'ticker/24hr', {'symbol': mexc_symbol}) # Convert to a standardized format result = { 'symbol': symbol, 'bid': float(ticker['bidPrice']), 'ask': float(ticker['askPrice']), 'last': float(ticker['lastPrice']), 'volume': float(ticker['volume']), 'timestamp': int(ticker['closeTime']) } return result except Exception as e: logger.error(f"Error getting ticker for {symbol}: {str(e)}") raise def place_order(self, symbol: str, side: str, order_type: str, quantity: float, price: float = None) -> Dict[str, Any]: """Place an order on the exchange. Args: symbol: Trading symbol (e.g., 'BTC/USDT') side: Order side ('buy' or 'sell') order_type: Order type ('market', 'limit', etc.) quantity: Order quantity price: Order price (for limit orders) Returns: dict: Order information including order ID """ mexc_symbol = symbol.replace('/', '') params = { 'symbol': mexc_symbol, 'side': side.upper(), 'type': order_type.upper(), 'quantity': quantity, } if order_type.lower() == 'limit' and price is not None: params['price'] = price params['timeInForce'] = 'GTC' # Good Till Cancelled try: order_result = self._send_private_request('POST', 'order', params) return order_result except Exception as e: logger.error(f"Error placing {side} {order_type} order for {symbol}: {str(e)}") raise def cancel_order(self, symbol: str, order_id: str) -> bool: """Cancel an existing order. Args: symbol: Trading symbol (e.g., 'BTC/USDT') order_id: ID of the order to cancel Returns: bool: True if cancellation successful, False otherwise """ mexc_symbol = symbol.replace('/', '') params = { 'symbol': mexc_symbol, 'orderId': order_id } try: cancel_result = self._send_private_request('DELETE', 'order', params) return True except Exception as e: logger.error(f"Error cancelling order {order_id} for {symbol}: {str(e)}") return False def get_order_status(self, symbol: str, order_id: str) -> Dict[str, Any]: """Get status of an existing order. Args: symbol: Trading symbol (e.g., 'BTC/USDT') order_id: ID of the order Returns: dict: Order status information """ mexc_symbol = symbol.replace('/', '') params = { 'symbol': mexc_symbol, 'orderId': order_id } try: order_info = self._send_private_request('GET', 'order', params) return order_info except Exception as e: logger.error(f"Error getting order status for {order_id} on {symbol}: {str(e)}") raise def get_open_orders(self, symbol: str = None) -> List[Dict[str, Any]]: """Get all open orders, optionally filtered by symbol. Args: symbol: Trading symbol (e.g., 'BTC/USDT'), or None for all symbols Returns: list: List of open orders """ params = {} if symbol: params['symbol'] = symbol.replace('/', '') try: open_orders = self._send_private_request('GET', 'openOrders', params) return open_orders except Exception as e: logger.error(f"Error getting open orders: {str(e)}") return []