479 lines
19 KiB
Python
479 lines
19 KiB
Python
"""
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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, session_cookies: Dict[str, str] = None):
|
|
"""
|
|
Initialize the MEXC Futures Web Client
|
|
|
|
Args:
|
|
session_cookies: Dictionary of cookies from an authenticated browser session
|
|
"""
|
|
self.session = requests.Session()
|
|
|
|
# Base URLs for different endpoints
|
|
self.base_url = "https://www.mexc.com"
|
|
self.futures_api_url = "https://futures.mexc.com/api/v1"
|
|
self.captcha_url = f"{self.base_url}/ucgateway/captcha_api/captcha/robot"
|
|
|
|
# Session state
|
|
self.is_authenticated = False
|
|
self.user_id = None
|
|
self.auth_token = None
|
|
self.fingerprint = None
|
|
self.visitor_id = None
|
|
|
|
# Load session cookies if provided
|
|
if session_cookies:
|
|
self.load_session_cookies(session_cookies)
|
|
|
|
# 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}"
|
|
|
|
# Setup headers for captcha request
|
|
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}"
|
|
}
|
|
|
|
# Add captcha token if available (this would need to be extracted from browser)
|
|
# For now, we'll make the request without it and see what happens
|
|
|
|
try:
|
|
response = self.session.get(url, headers=headers, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('success') and data.get('code') == 0:
|
|
logger.info(f"MEXC: Captcha verification successful for {side} {symbol}")
|
|
return True
|
|
else:
|
|
logger.warning(f"MEXC: Captcha verification failed: {data}")
|
|
return False
|
|
else:
|
|
logger.error(f"MEXC: Captcha request failed with status {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"MEXC: Captcha verification error: {e}")
|
|
return False
|
|
|
|
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 [] |