520 lines
24 KiB
Python
520 lines
24 KiB
Python
import logging
|
|
import time
|
|
from typing import Dict, Any, List, Optional
|
|
import requests
|
|
import hmac
|
|
import hashlib
|
|
from urllib.parse import urlencode, quote_plus
|
|
import json # Added for json.dumps
|
|
|
|
from .exchange_interface import ExchangeInterface
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# https://github.com/mexcdevelop/mexc-api-postman/blob/main/MEXC%20V3.postman_collection.json
|
|
# MEXC V3.postman_collection.json
|
|
|
|
class MEXCInterface(ExchangeInterface):
|
|
"""MEXC Exchange API Interface"""
|
|
|
|
def __init__(self, api_key: str = "", api_secret: str = "", test_mode: bool = True, trading_mode: str = 'simulation'):
|
|
"""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)
|
|
trading_mode: 'simulation', 'testnet', or 'live'. Determines API endpoints used.
|
|
"""
|
|
super().__init__(api_key, api_secret, test_mode)
|
|
|
|
self.trading_mode = trading_mode # Store the trading mode
|
|
|
|
# MEXC API Base URLs
|
|
self.base_url = "https://api.mexc.com" # Live API URL
|
|
if self.trading_mode == 'testnet':
|
|
# Note: MEXC does not have a separate testnet for spot trading.
|
|
# We use the live API for 'testnet' mode and rely on 'simulation' for true dry-runs.
|
|
logger.warning("MEXC does not have a separate testnet for spot trading. Using live API for 'testnet' mode.")
|
|
|
|
self.api_version = "api/v3"
|
|
self.recv_window = 5000 # 5 seconds window for request validity
|
|
|
|
# Session for HTTP requests
|
|
self.session = requests.Session()
|
|
|
|
logger.info(f"MEXCInterface initialized in {self.trading_mode} mode. Ensure correct API endpoints are being used.")
|
|
|
|
def connect(self) -> bool:
|
|
"""Test connection to MEXC API by fetching account info."""
|
|
if not self.api_key or not self.api_secret:
|
|
logger.error("MEXC API key or secret not set. Cannot connect.")
|
|
return False
|
|
|
|
# Test connection by making a small, authenticated request
|
|
try:
|
|
account_info = self.get_account_info()
|
|
if account_info:
|
|
logger.info("Successfully connected to MEXC API and retrieved account info.")
|
|
return True
|
|
else:
|
|
logger.error("Failed to connect to MEXC API: Could not retrieve account info.")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Exception during MEXC API connection test: {e}")
|
|
return False
|
|
|
|
def _format_spot_symbol(self, symbol: str) -> str:
|
|
"""Formats a symbol to MEXC spot API standard (e.g., 'ETH/USDT' -> 'ETHUSDC')."""
|
|
if '/' in symbol:
|
|
base, quote = symbol.split('/')
|
|
# Convert USDT to USDC for MEXC spot trading
|
|
if quote.upper() == 'USDT':
|
|
quote = 'USDC'
|
|
return f"{base.upper()}{quote.upper()}"
|
|
else:
|
|
# Convert USDT to USDC for symbols like ETHUSDT
|
|
symbol = symbol.upper()
|
|
if symbol.endswith('USDT'):
|
|
symbol = symbol.replace('USDT', 'USDC')
|
|
return symbol
|
|
|
|
def _format_futures_symbol(self, symbol: str) -> str:
|
|
"""Formats a symbol to MEXC futures API standard (e.g., 'ETH/USDT' -> 'ETH_USDT')."""
|
|
# This method is included for completeness but should not be used for spot trading
|
|
return symbol.replace('/', '_').upper()
|
|
|
|
def _generate_signature(self, timestamp: str, method: str, endpoint: str, params: Dict[str, Any]) -> str:
|
|
"""Generate signature for private API calls using MEXC's official method"""
|
|
# MEXC signature format varies by method:
|
|
# For GET/DELETE: URL-encoded query string of alphabetically sorted parameters.
|
|
# For POST: JSON string of parameters (no sorting needed).
|
|
# The API-Secret is used as the HMAC SHA256 key.
|
|
|
|
# Remove signature from params to avoid circular inclusion
|
|
clean_params = {k: v for k, v in params.items() if k != 'signature'}
|
|
|
|
parameter_string: str
|
|
|
|
if method.upper() == "POST":
|
|
# For POST requests, the signature parameter is a JSON string
|
|
# Ensure sorting keys for consistent JSON string generation across runs
|
|
# even though MEXC says sorting is not required for POST params, it's good practice.
|
|
parameter_string = json.dumps(clean_params, sort_keys=True, separators=(',', ':'))
|
|
else:
|
|
# For GET/DELETE requests, parameters are spliced in dictionary order with & interval
|
|
sorted_params = sorted(clean_params.items())
|
|
parameter_string = '&'.join(f"{key}={str(value)}" for key, value in sorted_params)
|
|
|
|
# The string to be signed is: accessKey + timestamp + obtained parameter string.
|
|
string_to_sign = f"{self.api_key}{timestamp}{parameter_string}"
|
|
|
|
logger.debug(f"MEXC string to sign (method {method}): {string_to_sign}")
|
|
|
|
# Generate HMAC SHA256 signature
|
|
signature = hmac.new(
|
|
self.api_secret.encode('utf-8'),
|
|
string_to_sign.encode('utf-8'),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
logger.debug(f"MEXC generated signature: {signature}")
|
|
return signature
|
|
|
|
def _send_public_request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Send a public API request to MEXC."""
|
|
if params is None:
|
|
params = {}
|
|
|
|
url = f"{self.base_url}/{self.api_version}/{endpoint}"
|
|
|
|
headers = {'Accept': 'application/json'}
|
|
|
|
try:
|
|
response = requests.request(method, url, params=params, headers=headers, timeout=10)
|
|
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as http_err:
|
|
logger.error(f"HTTP error in public request to {endpoint}: {response.status_code} {response.reason}")
|
|
logger.error(f"Response content: {response.text}")
|
|
return {}
|
|
except requests.exceptions.ConnectionError as conn_err:
|
|
logger.error(f"Connection error in public request to {endpoint}: {conn_err}")
|
|
return {}
|
|
except requests.exceptions.Timeout as timeout_err:
|
|
logger.error(f"Timeout error in public request to {endpoint}: {timeout_err}")
|
|
return {}
|
|
except Exception as e:
|
|
logger.error(f"Error in public request to {endpoint}: {e}")
|
|
return {}
|
|
|
|
def _send_private_request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
|
|
"""Send a private request to the exchange with proper signature"""
|
|
if params is None:
|
|
params = {}
|
|
|
|
timestamp = str(int(time.time() * 1000))
|
|
# Add timestamp and recvWindow to params for signature and request
|
|
params['timestamp'] = timestamp
|
|
params['recvWindow'] = self.recv_window
|
|
signature = self._generate_signature(timestamp, method, endpoint, params)
|
|
params['signature'] = signature
|
|
|
|
headers = {
|
|
"X-MEXC-APIKEY": self.api_key,
|
|
"Request-Time": timestamp
|
|
}
|
|
|
|
# For spot API, use the correct endpoint format
|
|
if not endpoint.startswith('api/v3/'):
|
|
endpoint = f"api/v3/{endpoint}"
|
|
url = f"{self.base_url}/{endpoint}"
|
|
try:
|
|
if method.upper() == "GET":
|
|
response = self.session.get(url, headers=headers, params=params, timeout=10)
|
|
elif method.upper() == "POST":
|
|
# MEXC expects POST parameters as JSON in the request body, not as query string
|
|
# The signature is generated from the JSON string of parameters.
|
|
# We need to exclude 'signature' from the JSON body sent, as it's for the header.
|
|
params_for_body = {k: v for k, v in params.items() if k != 'signature'}
|
|
response = self.session.post(url, headers=headers, json=params_for_body, timeout=10)
|
|
else:
|
|
logger.error(f"Unsupported method: {method}")
|
|
return None
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
# For successful responses, return the data directly
|
|
# MEXC doesn't always use 'success' field for successful operations
|
|
if response.status_code == 200:
|
|
return data
|
|
else:
|
|
logger.error(f"API error: Status Code: {response.status_code}, Response: {response.text}")
|
|
return None
|
|
except requests.exceptions.HTTPError as http_err:
|
|
logger.error(f"HTTP error for {endpoint}: Status Code: {response.status_code}, Response: {response.text}")
|
|
logger.error(f"HTTP error details: {http_err}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Request error for {endpoint}: {e}")
|
|
return None
|
|
|
|
def get_account_info(self) -> Dict[str, Any]:
|
|
"""Get account information"""
|
|
endpoint = "account"
|
|
result = self._send_private_request("GET", endpoint, {})
|
|
return result if result is not None else {}
|
|
|
|
def get_balance(self, asset: str) -> float:
|
|
"""Get available balance for a specific asset."""
|
|
account_info = self.get_account_info()
|
|
if account_info and 'balances' in account_info:
|
|
for balance in account_info['balances']:
|
|
if balance.get('asset') == asset.upper():
|
|
return float(balance.get('free', 0.0))
|
|
logger.warning(f"Could not retrieve free balance for {asset}")
|
|
return 0.0
|
|
|
|
def get_ticker(self, symbol: str) -> Optional[Dict[str, Any]]:
|
|
"""Get ticker information for a symbol."""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
endpoint = "ticker/24hr"
|
|
params = {'symbol': formatted_symbol}
|
|
|
|
response = self._send_public_request('GET', endpoint, params)
|
|
|
|
if isinstance(response, dict):
|
|
ticker_data: Dict[str, Any] = response
|
|
elif isinstance(response, list) and len(response) > 0:
|
|
found_ticker = next((item for item in response if item.get('symbol') == formatted_symbol), None)
|
|
if found_ticker:
|
|
ticker_data = found_ticker
|
|
else:
|
|
logger.error(f"Ticker data for {formatted_symbol} not found in response list.")
|
|
return None
|
|
else:
|
|
logger.error(f"Unexpected ticker response format: {response}")
|
|
return None
|
|
|
|
# At this point, ticker_data is guaranteed to be a Dict[str, Any] due to the above logic
|
|
# If it was None, we would have returned early.
|
|
|
|
# Extract relevant info and format for universal use
|
|
last_price = float(ticker_data.get('lastPrice', 0))
|
|
bid_price = float(ticker_data.get('bidPrice', 0))
|
|
ask_price = float(ticker_data.get('askPrice', 0))
|
|
volume = float(ticker_data.get('volume', 0)) # Base asset volume
|
|
|
|
# Determine price change and percent change
|
|
price_change = float(ticker_data.get('priceChange', 0))
|
|
price_change_percent = float(ticker_data.get('priceChangePercent', 0))
|
|
|
|
logger.info(f"MEXC: Got ticker from {endpoint} for {symbol}: ${last_price:.2f}")
|
|
|
|
return {
|
|
'symbol': formatted_symbol,
|
|
'last': last_price,
|
|
'bid': bid_price,
|
|
'ask': ask_price,
|
|
'volume': volume,
|
|
'high': float(ticker_data.get('highPrice', 0)),
|
|
'low': float(ticker_data.get('lowPrice', 0)),
|
|
'change': price_change_percent, # This is usually priceChangePercent
|
|
'exchange': 'MEXC',
|
|
'raw_data': ticker_data
|
|
}
|
|
|
|
def get_api_symbols(self) -> List[str]:
|
|
"""Get list of symbols supported for API trading"""
|
|
try:
|
|
endpoint = "selfSymbols"
|
|
result = self._send_private_request("GET", endpoint, {})
|
|
if result and 'data' in result:
|
|
return result['data']
|
|
elif isinstance(result, list):
|
|
return result
|
|
else:
|
|
logger.warning(f"Unexpected response format for API symbols: {result}")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"Error getting API symbols: {e}")
|
|
return []
|
|
|
|
def is_symbol_supported(self, symbol: str) -> bool:
|
|
"""Check if a symbol is supported for API trading"""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
supported_symbols = self.get_api_symbols()
|
|
return formatted_symbol in supported_symbols
|
|
|
|
def place_order(self, symbol: str, side: str, order_type: str, quantity: float, price: Optional[float] = None) -> Dict[str, Any]:
|
|
"""Place a new order on MEXC."""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
|
|
# Check if symbol is supported for API trading
|
|
if not self.is_symbol_supported(symbol):
|
|
supported_symbols = self.get_api_symbols()
|
|
logger.error(f"Symbol {formatted_symbol} is not supported for API trading")
|
|
logger.info(f"Supported symbols include: {supported_symbols[:10]}...") # Show first 10
|
|
return {}
|
|
|
|
# Format quantity according to symbol precision requirements
|
|
formatted_quantity = self._format_quantity_for_symbol(formatted_symbol, quantity)
|
|
if formatted_quantity is None:
|
|
logger.error(f"MEXC: Failed to format quantity {quantity} for {formatted_symbol}")
|
|
return {}
|
|
|
|
# Handle order type restrictions for specific symbols
|
|
final_order_type = self._adjust_order_type_for_symbol(formatted_symbol, order_type.upper())
|
|
|
|
# Get price for limit orders
|
|
final_price = price
|
|
if final_order_type == 'LIMIT' and price is None:
|
|
# Get current market price
|
|
ticker = self.get_ticker(symbol)
|
|
if ticker and 'last' in ticker:
|
|
final_price = ticker['last']
|
|
logger.info(f"MEXC: Using market price ${final_price:.2f} for LIMIT order")
|
|
else:
|
|
logger.error(f"MEXC: Could not get market price for LIMIT order on {formatted_symbol}")
|
|
return {}
|
|
|
|
endpoint = "order"
|
|
|
|
params: Dict[str, Any] = {
|
|
'symbol': formatted_symbol,
|
|
'side': side.upper(),
|
|
'type': final_order_type,
|
|
'quantity': str(formatted_quantity) # Quantity must be a string
|
|
}
|
|
if final_price is not None:
|
|
params['price'] = str(final_price) # Price must be a string for limit orders
|
|
|
|
logger.info(f"MEXC: Placing {side.upper()} {final_order_type} order for {formatted_quantity} {formatted_symbol} at price {final_price}")
|
|
|
|
try:
|
|
# MEXC API endpoint for placing orders is /api/v3/order (POST)
|
|
order_result = self._send_private_request('POST', endpoint, params)
|
|
if order_result is not None:
|
|
logger.info(f"MEXC: Order placed successfully: {order_result}")
|
|
return order_result
|
|
else:
|
|
logger.error(f"MEXC: Error placing order: request returned None")
|
|
return {}
|
|
except Exception as e:
|
|
logger.error(f"MEXC: Exception placing order: {e}")
|
|
return {}
|
|
|
|
def _format_quantity_for_symbol(self, formatted_symbol: str, quantity: float) -> Optional[float]:
|
|
"""Format quantity according to symbol precision requirements"""
|
|
try:
|
|
# Symbol-specific precision rules
|
|
if formatted_symbol == 'ETHUSDC':
|
|
# ETHUSDC requires max 5 decimal places, step size 0.000001
|
|
formatted_qty = round(quantity, 5)
|
|
# Ensure it meets minimum step size
|
|
step_size = 0.000001
|
|
formatted_qty = round(formatted_qty / step_size) * step_size
|
|
# Round again to remove floating point errors
|
|
formatted_qty = round(formatted_qty, 6)
|
|
logger.info(f"MEXC: Formatted ETHUSDC quantity {quantity} -> {formatted_qty}")
|
|
return formatted_qty
|
|
elif formatted_symbol == 'BTCUSDC':
|
|
# Assume similar precision for BTC
|
|
formatted_qty = round(quantity, 6)
|
|
step_size = 0.000001
|
|
formatted_qty = round(formatted_qty / step_size) * step_size
|
|
formatted_qty = round(formatted_qty, 6)
|
|
return formatted_qty
|
|
else:
|
|
# Default formatting - 6 decimal places
|
|
return round(quantity, 6)
|
|
except Exception as e:
|
|
logger.error(f"Error formatting quantity for {formatted_symbol}: {e}")
|
|
return None
|
|
|
|
def _adjust_order_type_for_symbol(self, formatted_symbol: str, order_type: str) -> str:
|
|
"""Adjust order type based on symbol restrictions"""
|
|
if formatted_symbol == 'ETHUSDC':
|
|
# ETHUSDC only supports LIMIT and LIMIT_MAKER orders
|
|
if order_type == 'MARKET':
|
|
logger.info(f"MEXC: Converting MARKET order to LIMIT for {formatted_symbol} (MARKET not supported)")
|
|
return 'LIMIT'
|
|
return order_type
|
|
|
|
def cancel_order(self, symbol: str, order_id: str) -> Dict[str, Any]:
|
|
"""Cancel an existing order on MEXC."""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
endpoint = "order"
|
|
params = {
|
|
'symbol': formatted_symbol,
|
|
'orderId': order_id
|
|
}
|
|
logger.info(f"MEXC: Cancelling order {order_id} for {formatted_symbol}")
|
|
try:
|
|
# MEXC API endpoint for cancelling orders is /api/v3/order (DELETE)
|
|
cancel_result = self._send_private_request('DELETE', endpoint, params)
|
|
if cancel_result:
|
|
logger.info(f"MEXC: Order cancelled successfully: {cancel_result}")
|
|
return cancel_result
|
|
else:
|
|
logger.error(f"MEXC: Error cancelling order: {cancel_result}")
|
|
return {}
|
|
except Exception as e:
|
|
logger.error(f"MEXC: Exception cancelling order: {e}")
|
|
return {}
|
|
|
|
def get_order_status(self, symbol: str, order_id: str) -> Dict[str, Any]:
|
|
"""Get the status of an order on MEXC."""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
endpoint = "order"
|
|
params = {
|
|
'symbol': formatted_symbol,
|
|
'orderId': order_id
|
|
}
|
|
logger.info(f"MEXC: Getting status for order {order_id} for {formatted_symbol}")
|
|
try:
|
|
# MEXC API endpoint for order status is /api/v3/order (GET)
|
|
status_result = self._send_private_request('GET', endpoint, params)
|
|
if status_result:
|
|
logger.info(f"MEXC: Order status retrieved: {status_result}")
|
|
return status_result
|
|
else:
|
|
logger.error(f"MEXC: Error getting order status: {status_result}")
|
|
return {}
|
|
except Exception as e:
|
|
logger.error(f"MEXC: Exception getting order status: {e}")
|
|
return {}
|
|
|
|
def get_open_orders(self, symbol: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Get all open orders on MEXC for a symbol or all symbols."""
|
|
endpoint = "openOrders"
|
|
params = {}
|
|
if symbol:
|
|
params['symbol'] = self._format_spot_symbol(symbol)
|
|
|
|
logger.info(f"MEXC: Getting open orders for {symbol if symbol else 'all symbols'}")
|
|
try:
|
|
# MEXC API endpoint for open orders is /api/v3/openOrders (GET)
|
|
open_orders = self._send_private_request('GET', endpoint, params)
|
|
if open_orders and isinstance(open_orders, list):
|
|
logger.info(f"MEXC: Retrieved {len(open_orders)} open orders.")
|
|
return open_orders
|
|
else:
|
|
logger.error(f"MEXC: Error getting open orders: {open_orders}")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"MEXC: Exception getting open orders: {e}")
|
|
return []
|
|
|
|
def get_my_trades(self, symbol: str, limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""Get trade history for a specific symbol."""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
endpoint = "myTrades"
|
|
params = {'symbol': formatted_symbol, 'limit': limit}
|
|
|
|
logger.info(f"MEXC: Getting trade history for {formatted_symbol} (limit: {limit})")
|
|
try:
|
|
# MEXC API endpoint for trade history is /api/v3/myTrades (GET)
|
|
trade_history = self._send_private_request('GET', endpoint, params)
|
|
if trade_history and isinstance(trade_history, list):
|
|
logger.info(f"MEXC: Retrieved {len(trade_history)} trade records.")
|
|
return trade_history
|
|
else:
|
|
logger.error(f"MEXC: Error getting trade history: {trade_history}")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"MEXC: Exception getting trade history: {e}")
|
|
return []
|
|
|
|
def get_server_time(self) -> int:
|
|
"""Get current MEXC server time in milliseconds."""
|
|
endpoint = "time"
|
|
response = self._send_public_request('GET', endpoint)
|
|
if response and 'serverTime' in response:
|
|
return int(response['serverTime'])
|
|
logger.error("Failed to get MEXC server time.")
|
|
return int(time.time() * 1000) # Fallback to local time
|
|
|
|
def get_all_balances(self) -> Dict[str, Dict[str, float]]:
|
|
"""Get all asset balances from MEXC account."""
|
|
account_info = self.get_account_info()
|
|
balances = {}
|
|
if account_info and 'balances' in account_info:
|
|
for balance in account_info['balances']:
|
|
asset = balance.get('asset')
|
|
free = float(balance.get('free'))
|
|
locked = float(balance.get('locked'))
|
|
if asset:
|
|
balances[asset.upper()] = {'free': free, 'locked': locked, 'total': free + locked}
|
|
return balances
|
|
|
|
def get_trading_fees(self) -> Dict[str, Any]:
|
|
"""Get current trading fee rates from MEXC API"""
|
|
endpoint = "account/commission"
|
|
response = self._send_private_request('GET', endpoint)
|
|
if response and 'data' in response:
|
|
fees_data = response['data']
|
|
return {
|
|
'maker': float(fees_data.get('makerCommission', 0.0)),
|
|
'taker': float(fees_data.get('takerCommission', 0.0)),
|
|
'default': float(fees_data.get('defaultCommission', 0.0))
|
|
}
|
|
logger.error("Failed to get trading fees from MEXC API.")
|
|
return {}
|
|
|
|
def get_symbol_trading_fees(self, symbol: str) -> Dict[str, Any]:
|
|
"""Get trading fee rates for a specific symbol from MEXC API"""
|
|
formatted_symbol = self._format_spot_symbol(symbol)
|
|
endpoint = "account/commission"
|
|
params = {'symbol': formatted_symbol}
|
|
response = self._send_private_request('GET', endpoint, params)
|
|
if response and 'data' in response:
|
|
fees_data = response['data']
|
|
return {
|
|
'maker': float(fees_data.get('makerCommission', 0.0)),
|
|
'taker': float(fees_data.get('takerCommission', 0.0)),
|
|
'default': float(fees_data.get('defaultCommission', 0.0))
|
|
}
|
|
logger.error(f"Failed to get trading fees for {symbol} from MEXC API.")
|
|
return {} |