""" MEXC Futures Web Client This module implements a web-based client for MEXC futures trading since their official API doesn't support futures (leverage) trading. It mimics browser behavior by replicating the exact HTTP requests that the web interface makes. """ import logging import requests import time import json import hmac import hashlib import base64 from typing import Dict, List, Optional, Any from datetime import datetime import uuid from urllib.parse import urlencode import glob import os logger = logging.getLogger(__name__) class MEXCSessionManager: def __init__(self): self.captcha_token = None def get_captcha_token(self) -> str: return self.captcha_token if self.captcha_token else "" def save_captcha_token(self, token: str): self.captcha_token = token logger.info("MEXC: Captcha token saved in session manager") class MEXCFuturesWebClient: """ MEXC Futures Web Client that mimics browser behavior for futures trading. Since MEXC's official API doesn't support futures, this client replicates the exact HTTP requests made by their web interface. """ def __init__(self, api_key: str, api_secret: str, user_id: str, base_url: str = 'https://www.mexc.com', headless: bool = True): """ Initialize the MEXC Futures Web Client Args: api_key: API key for authentication api_secret: API secret for authentication user_id: User ID for authentication base_url: Base URL for the MEXC website headless: Whether to run the browser in headless mode """ self.api_key = api_key self.api_secret = api_secret self.user_id = user_id self.base_url = base_url self.is_authenticated = False self.headless = headless self.session = requests.Session() self.session_manager = MEXCSessionManager() # Adding session_manager attribute self.captcha_url = f'{base_url}/ucgateway/captcha_api' self.futures_api_url = "https://futures.mexc.com/api/v1" # Setup default headers that mimic a real browser self.setup_browser_headers() def setup_browser_headers(self): """Setup default headers that mimic Chrome browser""" self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36', 'Accept': '*/*', 'Accept-Language': 'en-GB,en-US;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'sec-ch-ua': '"Chromium";v="136", "Google Chrome";v="136", "Not.A/Brand";v="99"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'same-origin', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', 'Referer': f'{self.base_url}/en-GB/futures/ETH_USDT?type=linear_swap', 'Language': 'English', 'X-Language': 'en-GB', 'trochilus-trace-id': f"{uuid.uuid4()}-{int(time.time() * 1000) % 10000:04d}", 'trochilus-uid': str(self.user_id) if self.user_id is not None else '' }) def load_session_cookies(self, cookies: Dict[str, str]): """ Load session cookies from browser Args: cookies: Dictionary of cookie name-value pairs """ for name, value in cookies.items(): self.session.cookies.set(name, value) # Extract important session info from cookies self.auth_token = cookies.get('uc_token') self.user_id = cookies.get('u_id') self.fingerprint = cookies.get('x-mxc-fingerprint') self.visitor_id = cookies.get('mexc_fingerprint_visitorId') if self.auth_token and self.user_id: self.is_authenticated = True logger.info("MEXC: Loaded authenticated session") else: logger.warning("MEXC: Session cookies incomplete - authentication may fail") def extract_cookies_from_browser(self, cookie_string: str) -> Dict[str, str]: """ Extract cookies from a browser cookie string Args: cookie_string: Raw cookie string from browser (copy from Network tab) Returns: Dictionary of parsed cookies """ cookies = {} cookie_pairs = cookie_string.split(';') for pair in cookie_pairs: if '=' in pair: name, value = pair.strip().split('=', 1) cookies[name] = value return cookies def verify_captcha(self, symbol: str, side: str, leverage: str) -> bool: """ Verify captcha for robot trading protection Args: symbol: Trading symbol (e.g., 'ETH_USDT') side: 'openlong', 'closelong', 'openshort', 'closeshort' leverage: Leverage string (e.g., '200X') Returns: bool: True if captcha verification successful """ if not self.is_authenticated: logger.error("MEXC: Cannot verify captcha - not authenticated") return False # Build captcha endpoint URL endpoint = f"robot.future.{side}.{symbol}.{leverage}" url = f"{self.captcha_url}/{endpoint}" # Attempt to get captcha token from session manager captcha_token = self.session_manager.get_captcha_token() if not captcha_token: logger.warning("MEXC: No captcha token available, attempting to fetch from browser") captcha_token = self._extract_captcha_token_from_browser() if captcha_token: self.session_manager.save_captcha_token(captcha_token) else: logger.error("MEXC: Failed to extract captcha token from browser") return False headers = { 'Content-Type': 'application/json', 'Language': 'en-GB', 'Referer': f'{self.base_url}/en-GB/futures/{symbol}?type=linear_swap', 'trochilus-uid': self.user_id if self.user_id else '', 'trochilus-trace-id': f"{uuid.uuid4()}-{int(time.time() * 1000) % 10000:04d}", 'captcha-token': captcha_token } logger.info(f"MEXC: Verifying captcha for {endpoint}") try: response = self.session.get(url, headers=headers, timeout=10) if response.status_code == 200: data = response.json() if data.get('success'): logger.info(f"MEXC: Captcha verified successfully for {endpoint}") return True else: logger.error(f"MEXC: Captcha verification failed for {endpoint}: {data}") return False else: logger.error(f"MEXC: Captcha verification request failed with status {response.status_code}: {response.text}") return False except Exception as e: logger.error(f"MEXC: Captcha verification error for {endpoint}: {str(e)}") return False def _extract_captcha_token_from_browser(self) -> str: """ Extract captcha token from browser session using stored cookies or requests. This method looks for the most recent mexc_captcha_tokens JSON file to retrieve a token. """ try: # Look for the most recent mexc_captcha_tokens file captcha_files = glob.glob("mexc_captcha_tokens_*.json") if not captcha_files: logger.error("MEXC: No CAPTCHA token files found") return "" # Sort files by timestamp (most recent first) latest_file = max(captcha_files, key=os.path.getctime) logger.info(f"MEXC: Using CAPTCHA token file {latest_file}") with open(latest_file, 'r') as f: captcha_data = json.load(f) if captcha_data and isinstance(captcha_data, list) and len(captcha_data) > 0: # Return the most recent token return captcha_data[0].get('token', '') else: logger.error("MEXC: No valid CAPTCHA tokens found in file") return "" except Exception as e: logger.error(f"MEXC: Error extracting captcha token from browser data: {str(e)}") return "" def generate_signature(self, method: str, path: str, params: Dict[str, Any], timestamp: int, nonce: int) -> str: """ Generate signature for MEXC futures API requests This is reverse-engineered from the browser requests """ # This is a placeholder - the actual signature generation would need # to be reverse-engineered from the browser's JavaScript # For now, return empty string and rely on cookie authentication return "" def open_long_position(self, symbol: str, volume: float, leverage: int = 200, price: Optional[float] = None) -> Dict[str, Any]: """ Open a long futures position Args: symbol: Trading symbol (e.g., 'ETH_USDT') volume: Position size (contracts) leverage: Leverage multiplier (default 200) price: Limit price (None for market order) Returns: dict: Order response with order ID """ if not self.is_authenticated: logger.error("MEXC: Cannot open position - not authenticated") return {'success': False, 'error': 'Not authenticated'} # First verify captcha if not self.verify_captcha(symbol, 'openlong', f'{leverage}X'): logger.error("MEXC: Captcha verification failed for opening long position") return {'success': False, 'error': 'Captcha verification failed'} # Prepare order parameters based on the request dump timestamp = int(time.time() * 1000) nonce = timestamp order_data = { 'symbol': symbol, 'side': 1, # 1 = long, 2 = short 'openType': 2, # Open position 'type': '5', # Market order (might be '1' for limit) 'vol': volume, 'leverage': leverage, 'marketCeiling': False, 'priceProtect': '0', 'ts': timestamp, 'mhash': self._generate_mhash(), # This needs to be implemented 'mtoken': self.visitor_id } # Add price for limit orders if price is not None: order_data['price'] = price order_data['type'] = '1' # Limit order # Add encrypted parameters (these would need proper implementation) order_data['p0'] = self._encrypt_p0(order_data) # Placeholder order_data['k0'] = self._encrypt_k0(order_data) # Placeholder order_data['chash'] = self._generate_chash(order_data) # Placeholder # Setup headers for the order request headers = { 'Authorization': self.auth_token, 'Content-Type': 'application/json', 'Language': 'English', 'x-language': 'en-GB', 'x-mxc-nonce': str(nonce), 'x-mxc-sign': self.generate_signature('POST', '/private/order/create', order_data, timestamp, nonce), 'trochilus-uid': self.user_id, 'trochilus-trace-id': f"{uuid.uuid4()}-{int(time.time() * 1000) % 10000:04d}", 'Referer': 'https://www.mexc.com/' } # Make the order request url = f"{self.futures_api_url}/private/order/create" try: # First make OPTIONS request (preflight) options_response = self.session.options(url, headers=headers, timeout=10) if options_response.status_code == 200: # Now make the actual POST request response = self.session.post(url, json=order_data, headers=headers, timeout=15) if response.status_code == 200: data = response.json() if data.get('success') and data.get('code') == 0: order_id = data.get('data', {}).get('orderId') logger.info(f"MEXC: Long position opened successfully - Order ID: {order_id}") return { 'success': True, 'order_id': order_id, 'timestamp': data.get('data', {}).get('ts'), 'symbol': symbol, 'side': 'long', 'volume': volume, 'leverage': leverage } else: logger.error(f"MEXC: Order failed: {data}") return {'success': False, 'error': data.get('msg', 'Unknown error')} else: logger.error(f"MEXC: Order request failed with status {response.status_code}") return {'success': False, 'error': f'HTTP {response.status_code}'} else: logger.error(f"MEXC: OPTIONS preflight failed with status {options_response.status_code}") return {'success': False, 'error': f'Preflight failed: HTTP {options_response.status_code}'} except Exception as e: logger.error(f"MEXC: Order execution error: {e}") return {'success': False, 'error': str(e)} def close_long_position(self, symbol: str, volume: float, leverage: int = 200, price: Optional[float] = None) -> Dict[str, Any]: """ Close a long futures position Args: symbol: Trading symbol (e.g., 'ETH_USDT') volume: Position size to close (contracts) leverage: Leverage multiplier price: Limit price (None for market order) Returns: dict: Order response """ if not self.is_authenticated: logger.error("MEXC: Cannot close position - not authenticated") return {'success': False, 'error': 'Not authenticated'} # First verify captcha if not self.verify_captcha(symbol, 'closelong', f'{leverage}X'): logger.error("MEXC: Captcha verification failed for closing long position") return {'success': False, 'error': 'Captcha verification failed'} # Similar to open_long_position but with closeType instead of openType timestamp = int(time.time() * 1000) nonce = timestamp order_data = { 'symbol': symbol, 'side': 2, # Close side is opposite 'closeType': 1, # Close position 'type': '5', # Market order 'vol': volume, 'leverage': leverage, 'marketCeiling': False, 'priceProtect': '0', 'ts': timestamp, 'mhash': self._generate_mhash(), 'mtoken': self.visitor_id } if price is not None: order_data['price'] = price order_data['type'] = '1' order_data['p0'] = self._encrypt_p0(order_data) order_data['k0'] = self._encrypt_k0(order_data) order_data['chash'] = self._generate_chash(order_data) return self._execute_order(order_data, 'close_long') def open_short_position(self, symbol: str, volume: float, leverage: int = 200, price: Optional[float] = None) -> Dict[str, Any]: """Open a short futures position""" if not self.verify_captcha(symbol, 'openshort', f'{leverage}X'): return {'success': False, 'error': 'Captcha verification failed'} order_data = { 'symbol': symbol, 'side': 2, # 2 = short 'openType': 2, 'type': '5', 'vol': volume, 'leverage': leverage, 'marketCeiling': False, 'priceProtect': '0', 'ts': int(time.time() * 1000), 'mhash': self._generate_mhash(), 'mtoken': self.visitor_id } if price is not None: order_data['price'] = price order_data['type'] = '1' order_data['p0'] = self._encrypt_p0(order_data) order_data['k0'] = self._encrypt_k0(order_data) order_data['chash'] = self._generate_chash(order_data) return self._execute_order(order_data, 'open_short') def close_short_position(self, symbol: str, volume: float, leverage: int = 200, price: Optional[float] = None) -> Dict[str, Any]: """Close a short futures position""" if not self.verify_captcha(symbol, 'closeshort', f'{leverage}X'): return {'success': False, 'error': 'Captcha verification failed'} order_data = { 'symbol': symbol, 'side': 1, # Close side is opposite 'closeType': 1, 'type': '5', 'vol': volume, 'leverage': leverage, 'marketCeiling': False, 'priceProtect': '0', 'ts': int(time.time() * 1000), 'mhash': self._generate_mhash(), 'mtoken': self.visitor_id } if price is not None: order_data['price'] = price order_data['type'] = '1' order_data['p0'] = self._encrypt_p0(order_data) order_data['k0'] = self._encrypt_k0(order_data) order_data['chash'] = self._generate_chash(order_data) return self._execute_order(order_data, 'close_short') def _execute_order(self, order_data: Dict[str, Any], action: str) -> Dict[str, Any]: """Common order execution logic""" timestamp = order_data['ts'] nonce = timestamp headers = { 'Authorization': self.auth_token, 'Content-Type': 'application/json', 'Language': 'English', 'x-language': 'en-GB', 'x-mxc-nonce': str(nonce), 'x-mxc-sign': self.generate_signature('POST', '/private/order/create', order_data, timestamp, nonce), 'trochilus-uid': self.user_id, 'trochilus-trace-id': f"{uuid.uuid4()}-{int(time.time() * 1000) % 10000:04d}", 'Referer': 'https://www.mexc.com/' } url = f"{self.futures_api_url}/private/order/create" try: response = self.session.post(url, json=order_data, headers=headers, timeout=15) if response.status_code == 200: data = response.json() if data.get('success') and data.get('code') == 0: order_id = data.get('data', {}).get('orderId') logger.info(f"MEXC: {action} executed successfully - Order ID: {order_id}") return { 'success': True, 'order_id': order_id, 'timestamp': data.get('data', {}).get('ts'), 'action': action } else: logger.error(f"MEXC: {action} failed: {data}") return {'success': False, 'error': data.get('msg', 'Unknown error')} else: logger.error(f"MEXC: {action} request failed with status {response.status_code}") return {'success': False, 'error': f'HTTP {response.status_code}'} except Exception as e: logger.error(f"MEXC: {action} execution error: {e}") return {'success': False, 'error': str(e)} # Placeholder methods for encryption/hashing - these need proper implementation def _generate_mhash(self) -> str: """Generate mhash parameter (needs reverse engineering)""" return "a0015441fd4c3b6ba427b894b76cb7dd" # Placeholder from request dump def _encrypt_p0(self, order_data: Dict[str, Any]) -> str: """Encrypt p0 parameter (needs reverse engineering)""" return "placeholder_p0_encryption" # This needs proper implementation def _encrypt_k0(self, order_data: Dict[str, Any]) -> str: """Encrypt k0 parameter (needs reverse engineering)""" return "placeholder_k0_encryption" # This needs proper implementation def _generate_chash(self, order_data: Dict[str, Any]) -> str: """Generate chash parameter (needs reverse engineering)""" return "d6c64d28e362f314071b3f9d78ff7494d9cd7177ae0465e772d1840e9f7905d8" # Placeholder def get_account_info(self) -> Dict[str, Any]: """Get account information including positions and balances""" if not self.is_authenticated: return {'success': False, 'error': 'Not authenticated'} # This would need to be implemented by reverse engineering the account info endpoints logger.info("MEXC: Account info endpoint not yet implemented") return {'success': False, 'error': 'Not implemented'} def get_open_positions(self) -> List[Dict[str, Any]]: """Get list of open futures positions""" if not self.is_authenticated: return [] # This would need to be implemented by reverse engineering the positions endpoint logger.info("MEXC: Open positions endpoint not yet implemented") return []