mexc futures web client wip

This commit is contained in:
Dobromir Popov
2025-06-24 12:59:28 +03:00
parent d902e01197
commit 1f3166e1e5
11 changed files with 2139 additions and 0 deletions

View File

@ -0,0 +1,9 @@
# Run the automation
python run_mexc_browser.py
# Browser opens with MEXC futures page
# Log in manually → Choose option 1 to verify login
# Choose option 5 for guided test trading
# Perform small trade → All requests captured
# Choose option 4 to save data
# Use captured cookies with MEXCFuturesWebClient

View File

@ -0,0 +1,8 @@
# MEXC Web Client Module
#
# This module provides web-based trading capabilities for MEXC futures trading
# which is not supported by their official API.
from .mexc_futures_client import MEXCFuturesWebClient
__all__ = ['MEXCFuturesWebClient']

View File

@ -0,0 +1,502 @@
#!/usr/bin/env python3
"""
MEXC Auto Browser with Request Interception
This script automatically spawns a ChromeDriver instance and captures
all MEXC futures trading requests in real-time, including full request
and response data needed for reverse engineering.
"""
import logging
import time
import json
import sys
import os
from typing import Dict, List, Optional, Any
from datetime import datetime
import threading
import queue
# Selenium imports
try:
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, WebDriverException
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
except ImportError:
print("Please install selenium and webdriver-manager:")
print("pip install selenium webdriver-manager")
sys.exit(1)
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class MEXCRequestInterceptor:
"""
Automatically spawns ChromeDriver and intercepts all MEXC API requests
"""
def __init__(self, headless: bool = False, save_to_file: bool = True):
"""
Initialize the request interceptor
Args:
headless: Run browser in headless mode
save_to_file: Save captured requests to JSON file
"""
self.driver = None
self.headless = headless
self.save_to_file = save_to_file
self.captured_requests = []
self.captured_responses = []
self.session_cookies = {}
self.monitoring = False
self.request_queue = queue.Queue()
# File paths for saving data
self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
self.requests_file = f"mexc_requests_{self.timestamp}.json"
self.cookies_file = f"mexc_cookies_{self.timestamp}.json"
def setup_chrome_with_logging(self) -> webdriver.Chrome:
"""Setup Chrome with performance logging enabled"""
logger.info("Setting up ChromeDriver with request interception...")
# Chrome options
chrome_options = Options()
if self.headless:
chrome_options.add_argument("--headless")
logger.info("Running in headless mode")
# Essential options for automation
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_argument("--disable-web-security")
chrome_options.add_argument("--allow-running-insecure-content")
chrome_options.add_argument("--disable-features=VizDisplayCompositor")
# User agent to avoid detection
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"
chrome_options.add_argument(f"--user-agent={user_agent}")
# Disable automation flags
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)
# Enable performance logging for network requests
chrome_options.add_argument("--enable-logging")
chrome_options.add_argument("--log-level=0")
chrome_options.add_argument("--v=1")
# Set capabilities for performance logging
caps = DesiredCapabilities.CHROME
caps['goog:loggingPrefs'] = {
'performance': 'ALL',
'browser': 'ALL'
}
try:
# Automatically download and install ChromeDriver
logger.info("Downloading/updating ChromeDriver...")
service = Service(ChromeDriverManager().install())
# Create driver
driver = webdriver.Chrome(
service=service,
options=chrome_options,
desired_capabilities=caps
)
# Hide automation indicators
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
driver.execute_cdp_cmd('Network.setUserAgentOverride', {
"userAgent": user_agent
})
# Enable network domain for CDP
driver.execute_cdp_cmd('Network.enable', {})
driver.execute_cdp_cmd('Runtime.enable', {})
logger.info("ChromeDriver setup complete!")
return driver
except Exception as e:
logger.error(f"Failed to setup ChromeDriver: {e}")
raise
def start_monitoring(self):
"""Start the browser and begin monitoring"""
logger.info("Starting MEXC Request Interceptor...")
try:
# Setup ChromeDriver
self.driver = self.setup_chrome_with_logging()
# Navigate to MEXC futures
mexc_url = "https://www.mexc.com/en-GB/futures/ETH_USDT?type=linear_swap"
logger.info(f"Navigating to: {mexc_url}")
self.driver.get(mexc_url)
# Wait for page load
WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.TAG_NAME, "body"))
)
logger.info("✅ MEXC page loaded successfully!")
logger.info("📝 Please log in manually in the browser window")
logger.info("🔍 Request monitoring is now active...")
# Start monitoring in background thread
self.monitoring = True
monitor_thread = threading.Thread(target=self._monitor_requests, daemon=True)
monitor_thread.start()
# Wait for manual login
self._wait_for_login()
return True
except Exception as e:
logger.error(f"Failed to start monitoring: {e}")
return False
def _wait_for_login(self):
"""Wait for user to log in and show interactive menu"""
logger.info("\n" + "="*60)
logger.info("MEXC REQUEST INTERCEPTOR - INTERACTIVE MODE")
logger.info("="*60)
while True:
print("\nOptions:")
print("1. Check login status")
print("2. Extract current cookies")
print("3. Show captured requests summary")
print("4. Save captured data to files")
print("5. Perform test trade (manual)")
print("6. Monitor for 60 seconds")
print("0. Stop and exit")
choice = input("\nEnter choice (0-6): ").strip()
if choice == "1":
self._check_login_status()
elif choice == "2":
self._extract_cookies()
elif choice == "3":
self._show_requests_summary()
elif choice == "4":
self._save_all_data()
elif choice == "5":
self._guide_test_trade()
elif choice == "6":
self._monitor_for_duration(60)
elif choice == "0":
break
else:
print("Invalid choice. Please try again.")
self.stop_monitoring()
def _check_login_status(self):
"""Check if user is logged into MEXC"""
try:
cookies = self.driver.get_cookies()
auth_cookies = ['uc_token', 'u_id', 'x-mxc-fingerprint']
found_auth = []
for cookie in cookies:
if cookie['name'] in auth_cookies and cookie['value']:
found_auth.append(cookie['name'])
if len(found_auth) >= 2:
print("✅ LOGIN DETECTED - You appear to be logged in!")
print(f" Found auth cookies: {', '.join(found_auth)}")
return True
else:
print("❌ NOT LOGGED IN - Please log in to MEXC in the browser")
print(" Missing required authentication cookies")
return False
except Exception as e:
print(f"❌ Error checking login: {e}")
return False
def _extract_cookies(self):
"""Extract and display current session cookies"""
try:
cookies = self.driver.get_cookies()
cookie_dict = {}
for cookie in cookies:
cookie_dict[cookie['name']] = cookie['value']
self.session_cookies = cookie_dict
print(f"\n📊 Extracted {len(cookie_dict)} cookies:")
# Show important cookies
important = ['uc_token', 'u_id', 'x-mxc-fingerprint', 'mexc_fingerprint_visitorId']
for name in important:
if name in cookie_dict:
value = cookie_dict[name]
display_value = value[:20] + "..." if len(value) > 20 else value
print(f"{name}: {display_value}")
else:
print(f"{name}: Missing")
# Save cookies to file
if self.save_to_file:
with open(self.cookies_file, 'w') as f:
json.dump(cookie_dict, f, indent=2)
print(f"\n💾 Cookies saved to: {self.cookies_file}")
except Exception as e:
print(f"❌ Error extracting cookies: {e}")
def _monitor_requests(self):
"""Background thread to monitor network requests"""
last_log_count = 0
while self.monitoring:
try:
# Get performance logs
logs = self.driver.get_log('performance')
for log in logs:
try:
message = json.loads(log['message'])
method = message.get('message', {}).get('method', '')
# Capture network requests
if method == 'Network.requestWillBeSent':
self._process_request(message['message']['params'])
elif method == 'Network.responseReceived':
self._process_response(message['message']['params'])
except (json.JSONDecodeError, KeyError) as e:
continue
# Show progress every 10 new requests
if len(self.captured_requests) >= last_log_count + 10:
last_log_count = len(self.captured_requests)
logger.info(f"📈 Captured {len(self.captured_requests)} requests, {len(self.captured_responses)} responses")
except Exception as e:
if self.monitoring: # Only log if we're still supposed to be monitoring
logger.debug(f"Monitor error: {e}")
time.sleep(0.5) # Check every 500ms
def _process_request(self, request_data):
"""Process a captured network request"""
try:
url = request_data.get('request', {}).get('url', '')
# Filter for MEXC API requests
if self._is_mexc_request(url):
request_info = {
'type': 'request',
'timestamp': datetime.now().isoformat(),
'url': url,
'method': request_data.get('request', {}).get('method', ''),
'headers': request_data.get('request', {}).get('headers', {}),
'postData': request_data.get('request', {}).get('postData', ''),
'requestId': request_data.get('requestId', '')
}
self.captured_requests.append(request_info)
# Show important requests immediately
if ('futures.mexc.com' in url or 'captcha' in url):
print(f"\n🚀 CAPTURED REQUEST: {request_info['method']} {url}")
if request_info['postData']:
print(f" 📄 POST Data: {request_info['postData'][:100]}...")
except Exception as e:
logger.debug(f"Error processing request: {e}")
def _process_response(self, response_data):
"""Process a captured network response"""
try:
url = response_data.get('response', {}).get('url', '')
# Filter for MEXC API responses
if self._is_mexc_request(url):
response_info = {
'type': 'response',
'timestamp': datetime.now().isoformat(),
'url': url,
'status': response_data.get('response', {}).get('status', 0),
'headers': response_data.get('response', {}).get('headers', {}),
'requestId': response_data.get('requestId', '')
}
self.captured_responses.append(response_info)
# Show important responses immediately
if ('futures.mexc.com' in url or 'captcha' in url):
status = response_info['status']
status_emoji = "" if status == 200 else ""
print(f" {status_emoji} RESPONSE: {status} for {url}")
except Exception as e:
logger.debug(f"Error processing response: {e}")
def _is_mexc_request(self, url: str) -> bool:
"""Check if URL is a relevant MEXC API request"""
mexc_indicators = [
'futures.mexc.com',
'ucgateway/captcha_api',
'api/v1/private',
'api/v3/order',
'mexc.com/api'
]
return any(indicator in url for indicator in mexc_indicators)
def _show_requests_summary(self):
"""Show summary of captured requests"""
print(f"\n📊 CAPTURE SUMMARY:")
print(f" Total Requests: {len(self.captured_requests)}")
print(f" Total Responses: {len(self.captured_responses)}")
# Group by URL pattern
url_counts = {}
for req in self.captured_requests:
base_url = req['url'].split('?')[0] # Remove query params
url_counts[base_url] = url_counts.get(base_url, 0) + 1
print("\n🔗 Top URLs:")
for url, count in sorted(url_counts.items(), key=lambda x: x[1], reverse=True)[:5]:
print(f" {count}x {url}")
# Show recent futures API calls
futures_requests = [r for r in self.captured_requests if 'futures.mexc.com' in r['url']]
if futures_requests:
print(f"\n🚀 Futures API Calls: {len(futures_requests)}")
for req in futures_requests[-3:]: # Show last 3
print(f" {req['method']} {req['url']}")
def _save_all_data(self):
"""Save all captured data to files"""
if not self.save_to_file:
print("File saving is disabled")
return
try:
# Save requests
with open(self.requests_file, 'w') as f:
json.dump({
'requests': self.captured_requests,
'responses': self.captured_responses,
'summary': {
'total_requests': len(self.captured_requests),
'total_responses': len(self.captured_responses),
'capture_session': self.timestamp
}
}, f, indent=2)
# Save cookies if we have them
if self.session_cookies:
with open(self.cookies_file, 'w') as f:
json.dump(self.session_cookies, f, indent=2)
print(f"\n💾 Data saved to:")
print(f" 📋 Requests: {self.requests_file}")
if self.session_cookies:
print(f" 🍪 Cookies: {self.cookies_file}")
except Exception as e:
print(f"❌ Error saving data: {e}")
def _guide_test_trade(self):
"""Guide user through performing a test trade"""
print("\n🧪 TEST TRADE GUIDE:")
print("1. Make sure you're logged into MEXC")
print("2. Go to the trading interface")
print("3. Try to place a SMALL test trade (it may fail, but we'll capture the requests)")
print("4. Watch the console for captured API calls")
print("\n⚠️ IMPORTANT: Use very small amounts for testing!")
input("\nPress Enter when you're ready to start monitoring...")
self._monitor_for_duration(120) # Monitor for 2 minutes
def _monitor_for_duration(self, seconds: int):
"""Monitor requests for a specific duration"""
print(f"\n🔍 Monitoring requests for {seconds} seconds...")
print("Perform your trading actions now!")
start_time = time.time()
initial_count = len(self.captured_requests)
while time.time() - start_time < seconds:
current_count = len(self.captured_requests)
new_requests = current_count - initial_count
remaining = seconds - int(time.time() - start_time)
print(f"\r⏱️ Time remaining: {remaining}s | New requests: {new_requests}", end="", flush=True)
time.sleep(1)
final_count = len(self.captured_requests)
new_total = final_count - initial_count
print(f"\n✅ Monitoring complete! Captured {new_total} new requests")
def stop_monitoring(self):
"""Stop monitoring and close browser"""
logger.info("Stopping request monitoring...")
self.monitoring = False
if self.driver:
self.driver.quit()
logger.info("Browser closed")
# Final save
if self.save_to_file and (self.captured_requests or self.captured_responses):
self._save_all_data()
logger.info("Final data save complete")
def main():
"""Main function to run the interceptor"""
print("🚀 MEXC Request Interceptor with ChromeDriver")
print("=" * 50)
print("This will automatically:")
print("✅ Download/setup ChromeDriver")
print("✅ Open MEXC futures page")
print("✅ Capture all API requests/responses")
print("✅ Extract session cookies")
print("✅ Save data to JSON files")
print("\nPress Ctrl+C to stop at any time")
# Ask for preferences
headless = input("\nRun in headless mode? (y/n): ").lower().strip() == 'y'
interceptor = MEXCRequestInterceptor(headless=headless, save_to_file=True)
try:
success = interceptor.start_monitoring()
if not success:
print("❌ Failed to start monitoring")
return
except KeyboardInterrupt:
print("\n\n⏹️ Stopping interceptor...")
except Exception as e:
print(f"\n❌ Error: {e}")
finally:
interceptor.stop_monitoring()
print("\n👋 Goodbye!")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,358 @@
"""
MEXC Browser Automation for Cookie Extraction and Request Monitoring
This module uses Selenium to automate browser interactions and extract
session cookies and request data for MEXC futures trading.
"""
import logging
import time
import json
from typing import Dict, List, Optional, Any
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, WebDriverException
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
logger = logging.getLogger(__name__)
class MEXCBrowserAutomation:
"""
Browser automation for MEXC futures trading session management
"""
def __init__(self, headless: bool = False, proxy: Optional[str] = None):
"""
Initialize browser automation
Args:
headless: Run browser in headless mode
proxy: HTTP proxy to use (format: host:port)
"""
self.driver = None
self.headless = headless
self.proxy = proxy
self.logged_in = False
def setup_chrome_driver(self) -> webdriver.Chrome:
"""Setup Chrome driver with appropriate options"""
chrome_options = Options()
if self.headless:
chrome_options.add_argument("--headless")
# Basic Chrome options for automation
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)
# Set user agent to avoid detection
chrome_options.add_argument("--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")
# Proxy setup if provided
if self.proxy:
chrome_options.add_argument(f"--proxy-server=http://{self.proxy}")
# Enable network logging
chrome_options.add_argument("--enable-logging")
chrome_options.add_argument("--log-level=0")
chrome_options.set_capability("goog:loggingPrefs", {"performance": "ALL"})
# Automatically download and setup ChromeDriver
service = Service(ChromeDriverManager().install())
try:
driver = webdriver.Chrome(service=service, options=chrome_options)
# Execute script to avoid detection
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
return driver
except WebDriverException as e:
logger.error(f"Failed to setup Chrome driver: {e}")
raise
def start_browser(self):
"""Start the browser session"""
if self.driver is None:
logger.info("Starting Chrome browser for MEXC automation")
self.driver = self.setup_chrome_driver()
logger.info("Browser started successfully")
def stop_browser(self):
"""Stop the browser session"""
if self.driver:
logger.info("Stopping browser")
self.driver.quit()
self.driver = None
def navigate_to_mexc_futures(self, symbol: str = "ETH_USDT"):
"""
Navigate to MEXC futures trading page
Args:
symbol: Trading symbol to navigate to
"""
if not self.driver:
self.start_browser()
url = f"https://www.mexc.com/en-GB/futures/{symbol}?type=linear_swap"
logger.info(f"Navigating to MEXC futures: {url}")
self.driver.get(url)
# Wait for page to load
try:
WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.TAG_NAME, "body"))
)
logger.info("MEXC futures page loaded")
except TimeoutException:
logger.error("Timeout waiting for MEXC page to load")
def wait_for_login(self, timeout: int = 300) -> bool:
"""
Wait for user to manually log in to MEXC
Args:
timeout: Maximum time to wait for login (seconds)
Returns:
bool: True if login detected, False if timeout
"""
logger.info("Please log in to MEXC manually in the browser window")
logger.info("Waiting for login completion...")
start_time = time.time()
while time.time() - start_time < timeout:
# Check if we can find elements that indicate logged in state
try:
# Look for user-specific elements that appear after login
cookies = self.driver.get_cookies()
# Check for authentication cookies
auth_cookies = ['uc_token', 'u_id']
logged_in_indicators = 0
for cookie in cookies:
if cookie['name'] in auth_cookies and cookie['value']:
logged_in_indicators += 1
if logged_in_indicators >= 2:
logger.info("Login detected!")
self.logged_in = True
return True
except Exception as e:
logger.debug(f"Error checking login status: {e}")
time.sleep(2) # Check every 2 seconds
logger.error(f"Login timeout after {timeout} seconds")
return False
def extract_session_cookies(self) -> Dict[str, str]:
"""
Extract all cookies from current browser session
Returns:
Dictionary of cookie name-value pairs
"""
if not self.driver:
logger.error("Browser not started")
return {}
cookies = {}
try:
browser_cookies = self.driver.get_cookies()
for cookie in browser_cookies:
cookies[cookie['name']] = cookie['value']
logger.info(f"Extracted {len(cookies)} cookies from browser session")
# Log important cookies (without values for security)
important_cookies = ['uc_token', 'u_id', 'x-mxc-fingerprint', 'mexc_fingerprint_visitorId']
for cookie_name in important_cookies:
if cookie_name in cookies:
logger.info(f"Found important cookie: {cookie_name}")
else:
logger.warning(f"Missing important cookie: {cookie_name}")
return cookies
except Exception as e:
logger.error(f"Failed to extract cookies: {e}")
return {}
def monitor_network_requests(self, duration: int = 60) -> List[Dict[str, Any]]:
"""
Monitor network requests for the specified duration
Args:
duration: How long to monitor requests (seconds)
Returns:
List of captured network requests
"""
if not self.driver:
logger.error("Browser not started")
return []
logger.info(f"Starting network monitoring for {duration} seconds")
logger.info("Please perform trading actions in the browser (open/close positions)")
start_time = time.time()
captured_requests = []
while time.time() - start_time < duration:
try:
# Get performance logs (network requests)
logs = self.driver.get_log('performance')
for log in logs:
message = json.loads(log['message'])
# Filter for relevant MEXC API requests
if (message.get('message', {}).get('method') == 'Network.responseReceived'):
response = message['message']['params']['response']
url = response.get('url', '')
# Look for futures API calls
if ('futures.mexc.com' in url or
'ucgateway/captcha_api' in url or
'api/v1/private' in url):
request_data = {
'url': url,
'method': response.get('mimeType', ''),
'status': response.get('status'),
'headers': response.get('headers', {}),
'timestamp': log['timestamp']
}
captured_requests.append(request_data)
logger.info(f"Captured request: {url}")
except Exception as e:
logger.debug(f"Error in network monitoring: {e}")
time.sleep(1)
logger.info(f"Network monitoring complete. Captured {len(captured_requests)} requests")
return captured_requests
def perform_test_trade(self, symbol: str = "ETH_USDT", volume: float = 1.0, leverage: int = 200):
"""
Attempt to perform a test trade to capture the complete request flow
Args:
symbol: Trading symbol
volume: Position size
leverage: Leverage multiplier
"""
if not self.logged_in:
logger.error("Not logged in - cannot perform test trade")
return
logger.info(f"Attempting test trade: {symbol}, Volume: {volume}, Leverage: {leverage}x")
logger.info("This will attempt to click trading interface elements")
try:
# This would need to be implemented based on MEXC's specific UI elements
# For now, just wait and let user perform manual actions
logger.info("Please manually place a small test trade while monitoring is active")
time.sleep(30)
except Exception as e:
logger.error(f"Error during test trade: {e}")
def full_session_capture(self, symbol: str = "ETH_USDT") -> Dict[str, Any]:
"""
Complete session capture workflow
Args:
symbol: Trading symbol to use
Returns:
Dictionary containing cookies and captured requests
"""
logger.info("Starting full MEXC session capture")
try:
# Start browser and navigate to MEXC
self.navigate_to_mexc_futures(symbol)
# Wait for manual login
if not self.wait_for_login():
return {'success': False, 'error': 'Login timeout'}
# Extract session cookies
cookies = self.extract_session_cookies()
if not cookies:
return {'success': False, 'error': 'Failed to extract cookies'}
# Monitor network requests while user performs actions
logger.info("Starting network monitoring - please perform trading actions now")
requests = self.monitor_network_requests(duration=120) # 2 minutes
return {
'success': True,
'cookies': cookies,
'network_requests': requests,
'timestamp': int(time.time())
}
except Exception as e:
logger.error(f"Error in session capture: {e}")
return {'success': False, 'error': str(e)}
finally:
self.stop_browser()
def main():
"""Main function for standalone execution"""
logging.basicConfig(level=logging.INFO)
print("MEXC Browser Automation - Session Capture")
print("This will open a browser window for you to log into MEXC")
print("Make sure you have Chrome browser installed")
automation = MEXCBrowserAutomation(headless=False)
try:
result = automation.full_session_capture()
if result['success']:
print(f"\nSession capture successful!")
print(f"Extracted {len(result['cookies'])} cookies")
print(f"Captured {len(result['network_requests'])} network requests")
# Save results to file
output_file = f"mexc_session_capture_{int(time.time())}.json"
with open(output_file, 'w') as f:
json.dump(result, f, indent=2)
print(f"Results saved to: {output_file}")
else:
print(f"Session capture failed: {result['error']}")
except KeyboardInterrupt:
print("\nSession capture interrupted by user")
except Exception as e:
print(f"Error: {e}")
finally:
automation.stop_browser()
if __name__ == "__main__":
main()

View File

@ -0,0 +1,474 @@
"""
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'
})
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,
'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 []

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@ -0,0 +1,259 @@
"""
MEXC Session Manager
Helper utilities for managing MEXC web sessions and extracting cookies from browser.
"""
import logging
import json
import re
from typing import Dict, Optional, Any
from pathlib import Path
logger = logging.getLogger(__name__)
class MEXCSessionManager:
"""
Helper class for managing MEXC web sessions and extracting browser cookies
"""
def __init__(self):
self.session_file = Path("mexc_session.json")
def extract_cookies_from_network_tab(self, cookie_header: str) -> Dict[str, str]:
"""
Extract cookies from browser Network tab cookie header
Args:
cookie_header: Raw cookie string from browser (copy from Request Headers)
Returns:
Dictionary of parsed cookies
"""
cookies = {}
# Remove 'Cookie: ' prefix if present
if cookie_header.startswith('Cookie: '):
cookie_header = cookie_header[8:]
elif cookie_header.startswith('cookie: '):
cookie_header = cookie_header[8:]
# Split by semicolon and parse each cookie
cookie_pairs = cookie_header.split(';')
for pair in cookie_pairs:
pair = pair.strip()
if '=' in pair:
name, value = pair.split('=', 1)
cookies[name.strip()] = value.strip()
logger.info(f"Extracted {len(cookies)} cookies from browser")
return cookies
def validate_session_cookies(self, cookies: Dict[str, str]) -> bool:
"""
Validate that essential cookies are present for authentication
Args:
cookies: Dictionary of cookie name-value pairs
Returns:
bool: True if cookies appear valid for authentication
"""
required_cookies = [
'uc_token', # User authentication token
'u_id', # User ID
'x-mxc-fingerprint', # Browser fingerprint
'mexc_fingerprint_visitorId' # Visitor ID
]
missing_cookies = []
for cookie_name in required_cookies:
if cookie_name not in cookies or not cookies[cookie_name]:
missing_cookies.append(cookie_name)
if missing_cookies:
logger.warning(f"Missing required cookies: {missing_cookies}")
return False
logger.info("All required cookies are present")
return True
def save_session(self, cookies: Dict[str, str], metadata: Optional[Dict[str, Any]] = None):
"""
Save session cookies to file for reuse
Args:
cookies: Dictionary of cookies to save
metadata: Optional metadata about the session
"""
session_data = {
'cookies': cookies,
'metadata': metadata or {},
'timestamp': int(time.time())
}
try:
with open(self.session_file, 'w') as f:
json.dump(session_data, f, indent=2)
logger.info(f"Session saved to {self.session_file}")
except Exception as e:
logger.error(f"Failed to save session: {e}")
def load_session(self) -> Optional[Dict[str, str]]:
"""
Load session cookies from file
Returns:
Dictionary of cookies if successful, None otherwise
"""
if not self.session_file.exists():
logger.info("No saved session found")
return None
try:
with open(self.session_file, 'r') as f:
session_data = json.load(f)
cookies = session_data.get('cookies', {})
timestamp = session_data.get('timestamp', 0)
# Check if session is too old (24 hours)
import time
if time.time() - timestamp > 24 * 3600:
logger.warning("Saved session is too old (>24h), may be expired")
if self.validate_session_cookies(cookies):
logger.info("Loaded valid session from file")
return cookies
else:
logger.warning("Loaded session has invalid cookies")
return None
except Exception as e:
logger.error(f"Failed to load session: {e}")
return None
def extract_from_curl_command(self, curl_command: str) -> Dict[str, str]:
"""
Extract cookies from a curl command copied from browser
Args:
curl_command: Complete curl command from browser "Copy as cURL"
Returns:
Dictionary of extracted cookies
"""
cookies = {}
# Find cookie header in curl command
cookie_match = re.search(r'-H [\'"]cookie: ([^\'"]+)[\'"]', curl_command, re.IGNORECASE)
if not cookie_match:
cookie_match = re.search(r'--header [\'"]cookie: ([^\'"]+)[\'"]', curl_command, re.IGNORECASE)
if cookie_match:
cookie_header = cookie_match.group(1)
cookies = self.extract_cookies_from_network_tab(cookie_header)
logger.info(f"Extracted {len(cookies)} cookies from curl command")
else:
logger.warning("No cookie header found in curl command")
return cookies
def print_cookie_extraction_guide(self):
"""Print instructions for extracting cookies from browser"""
print("\n" + "="*80)
print("MEXC COOKIE EXTRACTION GUIDE")
print("="*80)
print("""
To extract cookies from your browser for MEXC futures trading:
METHOD 1: Browser Network Tab
1. Open MEXC futures page and log in: https://www.mexc.com/en-GB/futures/ETH_USDT
2. Open browser Developer Tools (F12)
3. Go to Network tab
4. Try to place a small futures trade (it will fail, but we need the request)
5. Find the request to 'futures.mexc.com' in the Network tab
6. Right-click on the request -> Copy -> Copy request headers
7. Find the 'Cookie:' line and copy everything after 'Cookie: '
METHOD 2: Copy as cURL
1. Follow steps 1-5 above
2. Right-click on the futures API request -> Copy -> Copy as cURL
3. Paste the entire cURL command
METHOD 3: Manual Cookie Extraction
1. While logged into MEXC, press F12 -> Application/Storage tab
2. On the left, expand 'Cookies' -> click on 'https://www.mexc.com'
3. Copy the values for these important cookies:
- uc_token
- u_id
- x-mxc-fingerprint
- mexc_fingerprint_visitorId
IMPORTANT NOTES:
- Cookies expire after some time (usually 24 hours)
- You must be logged into MEXC futures (not just spot trading)
- Keep your cookies secure - they provide access to your account
- Test with small amounts first
Example usage:
session_manager = MEXCSessionManager()
# Method 1: From cookie header
cookie_header = "uc_token=ABC123; u_id=DEF456; ..."
cookies = session_manager.extract_cookies_from_network_tab(cookie_header)
# Method 2: From cURL command
curl_cmd = "curl 'https://futures.mexc.com/...' -H 'cookie: uc_token=ABC123...'"
cookies = session_manager.extract_from_curl_command(curl_cmd)
# Save session for reuse
session_manager.save_session(cookies)
""")
print("="*80)
if __name__ == "__main__":
# When run directly, show the extraction guide
import time
manager = MEXCSessionManager()
manager.print_cookie_extraction_guide()
print("\nWould you like to:")
print("1. Load saved session")
print("2. Extract cookies from clipboard")
print("3. Exit")
choice = input("\nEnter choice (1-3): ").strip()
if choice == "1":
cookies = manager.load_session()
if cookies:
print(f"\nLoaded {len(cookies)} cookies from saved session")
if manager.validate_session_cookies(cookies):
print("Session appears valid for trading")
else:
print("Warning: Session may be incomplete or expired")
else:
print("No valid saved session found")
elif choice == "2":
print("\nPaste your cookie header or cURL command:")
user_input = input().strip()
if user_input.startswith('curl'):
cookies = manager.extract_from_curl_command(user_input)
else:
cookies = manager.extract_cookies_from_network_tab(user_input)
if cookies and manager.validate_session_cookies(cookies):
print(f"\nSuccessfully extracted {len(cookies)} valid cookies")
save = input("Save session for reuse? (y/n): ").strip().lower()
if save == 'y':
manager.save_session(cookies)
else:
print("Failed to extract valid cookies")
else:
print("Goodbye!")

52
run_mexc_browser.py Normal file
View File

@ -0,0 +1,52 @@
#!/usr/bin/env python3
"""
One-Click MEXC Browser Launcher
Simply run this script to start capturing MEXC futures trading requests.
"""
import sys
import os
# Add project root to path
project_root = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, project_root)
def main():
"""Launch MEXC browser automation"""
print("🚀 MEXC Futures Request Interceptor")
print("=" * 50)
print("This will automatically:")
print("✅ Install ChromeDriver")
print("✅ Open MEXC futures page")
print("✅ Capture all API requests")
print("✅ Extract session cookies")
print("✅ Save data to JSON files")
print("\nRequirements will be installed automatically if missing.")
try:
# First try to run the auto browser directly
from core.mexc_webclient.auto_browser import main as run_auto_browser
run_auto_browser()
except ImportError as e:
print(f"\n⚠️ Import error: {e}")
print("Installing requirements first...")
# Try to install requirements and run setup
try:
from setup_mexc_browser import main as setup_main
setup_main()
except ImportError:
print("❌ Could not find setup script")
print("Please run: pip install selenium webdriver-manager")
except Exception as e:
print(f"❌ Error: {e}")
print("\nTroubleshooting:")
print("1. Make sure you have Chrome browser installed")
print("2. Check your internet connection")
print("3. Try running: pip install selenium webdriver-manager")
if __name__ == "__main__":
main()

88
setup_mexc_browser.py Normal file
View File

@ -0,0 +1,88 @@
#!/usr/bin/env python3
"""
MEXC Browser Setup & Runner
This script automatically installs dependencies and runs the MEXC browser automation.
"""
import subprocess
import sys
import os
import importlib
def check_and_install_requirements():
"""Check and install required packages"""
required_packages = [
'selenium',
'webdriver-manager',
'requests'
]
print("🔍 Checking required packages...")
missing_packages = []
for package in required_packages:
try:
importlib.import_module(package.replace('-', '_'))
print(f"{package} - already installed")
except ImportError:
missing_packages.append(package)
print(f"{package} - missing")
if missing_packages:
print(f"\n📦 Installing missing packages: {', '.join(missing_packages)}")
for package in missing_packages:
try:
subprocess.check_call([sys.executable, '-m', 'pip', 'install', package])
print(f"✅ Successfully installed {package}")
except subprocess.CalledProcessError as e:
print(f"❌ Failed to install {package}: {e}")
return False
print("✅ All requirements satisfied!")
return True
def run_browser_automation():
"""Run the MEXC browser automation"""
try:
# Import and run the auto browser
from core.mexc_webclient.auto_browser import main as auto_browser_main
auto_browser_main()
except ImportError:
print("❌ Could not import auto browser module")
print("Make sure core/mexc_webclient/auto_browser.py exists")
except Exception as e:
print(f"❌ Error running browser automation: {e}")
def main():
"""Main setup and run function"""
print("🚀 MEXC Browser Automation Setup")
print("=" * 40)
# Check Python version
if sys.version_info < (3, 7):
print("❌ Python 3.7+ required")
return
print(f"✅ Python {sys.version.split()[0]} detected")
# Install requirements
if not check_and_install_requirements():
print("❌ Failed to install requirements")
return
print("\n🌐 Starting browser automation...")
print("This will:")
print("• Download ChromeDriver automatically")
print("• Open MEXC futures page")
print("• Capture all trading requests")
print("• Extract session cookies")
input("\nPress Enter to continue...")
# Run the automation
run_browser_automation()
if __name__ == "__main__":
main()

View File

@ -0,0 +1,208 @@
#!/usr/bin/env python3
"""
Test MEXC Futures Web Client
This script demonstrates how to use the MEXC Futures Web Client
for futures trading that isn't supported by their official API.
IMPORTANT: This requires extracting cookies from your browser session.
"""
import logging
import sys
import os
import time
# Add the project root to path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from core.mexc_webclient import MEXCFuturesWebClient
from core.mexc_webclient.session_manager import MEXCSessionManager
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_basic_connection():
"""Test basic connection and authentication"""
logger.info("Testing MEXC Futures Web Client")
# Initialize session manager
session_manager = MEXCSessionManager()
# Try to load saved session first
cookies = session_manager.load_session()
if not cookies:
print("\nNo saved session found. You need to extract cookies from your browser.")
session_manager.print_cookie_extraction_guide()
print("\nPaste your cookie header or cURL command (or press Enter to exit):")
user_input = input().strip()
if not user_input:
print("No input provided. Exiting.")
return False
# Extract cookies from user input
if user_input.startswith('curl'):
cookies = session_manager.extract_from_curl_command(user_input)
else:
cookies = session_manager.extract_cookies_from_network_tab(user_input)
if not cookies:
logger.error("Failed to extract cookies from input")
return False
# Validate and save session
if session_manager.validate_session_cookies(cookies):
session_manager.save_session(cookies)
logger.info("Session saved for future use")
else:
logger.warning("Extracted cookies may be incomplete")
# Initialize the web client
client = MEXCFuturesWebClient(cookies)
if not client.is_authenticated:
logger.error("Failed to authenticate with extracted cookies")
return False
logger.info("Successfully authenticated with MEXC")
logger.info(f"User ID: {client.user_id}")
logger.info(f"Auth Token: {client.auth_token[:20]}..." if client.auth_token else "No auth token")
return True
def test_captcha_verification(client: MEXCFuturesWebClient):
"""Test captcha verification system"""
logger.info("Testing captcha verification...")
# Test captcha for ETH_USDT long position with 200x leverage
success = client.verify_captcha('ETH_USDT', 'openlong', '200X')
if success:
logger.info("Captcha verification successful")
else:
logger.warning("Captcha verification failed - this may be normal if no position is being opened")
return success
def test_position_opening(client: MEXCFuturesWebClient, dry_run: bool = True):
"""Test opening a position (dry run by default)"""
if dry_run:
logger.info("DRY RUN: Testing position opening (no actual trade)")
else:
logger.warning("LIVE TRADING: Opening actual position!")
symbol = 'ETH_USDT'
volume = 1 # Small test position
leverage = 200
logger.info(f"Attempting to open long position: {symbol}, Volume: {volume}, Leverage: {leverage}x")
if not dry_run:
result = client.open_long_position(symbol, volume, leverage)
if result['success']:
logger.info(f"Position opened successfully!")
logger.info(f"Order ID: {result['order_id']}")
logger.info(f"Timestamp: {result['timestamp']}")
return True
else:
logger.error(f"Failed to open position: {result['error']}")
return False
else:
logger.info("DRY RUN: Would attempt to open position here")
# Test just the captcha verification part
return client.verify_captcha(symbol, 'openlong', f'{leverage}X')
def interactive_menu(client: MEXCFuturesWebClient):
"""Interactive menu for testing different functions"""
while True:
print("\n" + "="*50)
print("MEXC Futures Web Client Test Menu")
print("="*50)
print("1. Test captcha verification")
print("2. Test position opening (DRY RUN)")
print("3. Test position opening (LIVE - BE CAREFUL!)")
print("4. Test position closing (DRY RUN)")
print("5. Show session info")
print("6. Refresh session")
print("0. Exit")
choice = input("\nEnter choice (0-6): ").strip()
if choice == "1":
test_captcha_verification(client)
elif choice == "2":
test_position_opening(client, dry_run=True)
elif choice == "3":
confirm = input("Are you sure you want to open a LIVE position? (type 'YES' to confirm): ")
if confirm == "YES":
test_position_opening(client, dry_run=False)
else:
print("Cancelled live trading")
elif choice == "4":
logger.info("DRY RUN: Position closing test")
success = client.verify_captcha('ETH_USDT', 'closelong', '200X')
if success:
logger.info("DRY RUN: Would close position here")
else:
logger.warning("Captcha verification failed for position closing")
elif choice == "5":
print(f"\nSession Information:")
print(f"Authenticated: {client.is_authenticated}")
print(f"User ID: {client.user_id}")
print(f"Auth Token: {client.auth_token[:20]}..." if client.auth_token else "None")
print(f"Fingerprint: {client.fingerprint}")
print(f"Visitor ID: {client.visitor_id}")
elif choice == "6":
session_manager = MEXCSessionManager()
session_manager.print_cookie_extraction_guide()
elif choice == "0":
print("Goodbye!")
break
else:
print("Invalid choice. Please try again.")
def main():
"""Main test function"""
print("MEXC Futures Web Client Test")
print("WARNING: This is experimental software for futures trading")
print("Use at your own risk and test with small amounts first!")
# Test basic connection
if not test_basic_connection():
logger.error("Failed to establish connection. Exiting.")
return
# Create client with loaded session
session_manager = MEXCSessionManager()
cookies = session_manager.load_session()
if not cookies:
logger.error("No valid session available")
return
client = MEXCFuturesWebClient(cookies)
if not client.is_authenticated:
logger.error("Authentication failed")
return
# Show interactive menu
interactive_menu(client)
if __name__ == "__main__":
main()