popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

i think we fixed mexc interface at the end!!!

Browse Source
This commit is contained in:
Dobromir Popov
2025-07-04 02:14:29 +03:00
parent 978cecf0c5
commit cf91e090c8
7 changed files with 445 additions and 613 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
NN/exchanges/mexc_interface.py
View File

@ -65,11 +65,19 @@ class MEXCInterface(ExchangeInterface):
return False
def _format_spot_symbol(self, symbol: str) -> str:
"""Formats a symbol to MEXC spot API standard (e.g., 'ETH/USDT' -> 'ETHUSDT')."""
"""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()}"
return symbol.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')."""
@ -77,22 +85,37 @@ class MEXCInterface(ExchangeInterface):
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"""
# Build the string to sign
sign_str = self.api_key + timestamp
if params:
# Append all parameters sorted by key, without URL encoding for signature
query_str = "&".join([f"{k}={v}" for k, v in sorted(params.items()) if k != 'signature'])
if query_str:
sign_str += query_str
"""Generate signature for private API calls using MEXC's expected parameter order"""
# MEXC requires specific parameter ordering, not alphabetical
# Based on successful test: symbol, side, type, quantity, timestamp, then other params
mexc_param_order = ['symbol', 'side', 'type', 'quantity', 'timestamp', 'recvWindow']
# Build ordered parameter list
ordered_params = []
# Add parameters in MEXC's expected order
for param_name in mexc_param_order:
if param_name in params and param_name != 'signature':
ordered_params.append(f"{param_name}={params[param_name]}")
# Add any remaining parameters not in the standard order (alphabetically)
remaining_params = {k: v for k, v in params.items() if k not in mexc_param_order and k != 'signature'}
for key in sorted(remaining_params.keys()):
ordered_params.append(f"{key}={remaining_params[key]}")
# Create query string (MEXC doesn't use the api_key + timestamp prefix)
query_string = '&'.join(ordered_params)
logger.debug(f"MEXC signature query string: {query_string}")
logger.debug(f"Signature string: {sign_str}")
# Generate HMAC SHA256 signature
signature = hmac.new(
self.api_secret.encode('utf-8'),
sign_str.encode('utf-8'),
query_string.encode('utf-8'),
hashlib.sha256
).hexdigest()
logger.debug(f"MEXC signature: {signature}")
return signature
def _send_public_request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
@ -139,24 +162,26 @@ class MEXCInterface(ExchangeInterface):
"Request-Time": timestamp
}
# Ensure endpoint does not start with a slash to avoid double slashes
if endpoint.startswith('/'):
endpoint = endpoint.lstrip('/')
# 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":
headers["Content-Type"] = "application/x-www-form-urlencoded"
response = self.session.post(url, headers=headers, data=params, timeout=10)
# MEXC expects POST parameters as query string, not in body
response = self.session.post(url, headers=headers, params=params, timeout=10)
else:
logger.error(f"Unsupported method: {method}")
return None
response.raise_for_status()
data = response.json()
if data.get('success', False):
return data.get('data', data)
# 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
@ -170,7 +195,7 @@ class MEXCInterface(ExchangeInterface):
def get_account_info(self) -> Dict[str, Any]:
"""Get account information"""
endpoint = "/api/v3/account"
endpoint = "account"
result = self._send_private_request("GET", endpoint, {})
return result if result is not None else {}
@ -235,9 +260,39 @@ class MEXCInterface(ExchangeInterface):
logger.error(f"Failed to get ticker for {symbol}")
return None
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 {}
endpoint = "order"
params: Dict[str, Any] = {