210 lines
7.6 KiB
Python
210 lines
7.6 KiB
Python
"""
|
|
Test script for automatic fee synchronization with MEXC API
|
|
|
|
This script demonstrates how the system can automatically sync trading fees
|
|
from the MEXC API to the local configuration file.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from dotenv import load_dotenv
|
|
|
|
# Add NN directory to path
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), 'NN'))
|
|
|
|
from NN.exchanges.mexc_interface import MEXCInterface
|
|
from core.config_sync import ConfigSynchronizer
|
|
from core.trading_executor import TradingExecutor
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def test_mexc_fee_retrieval():
|
|
"""Test retrieving fees directly from MEXC API"""
|
|
logger.info("=== Testing MEXC Fee Retrieval ===")
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Initialize MEXC interface
|
|
api_key = os.getenv('MEXC_API_KEY')
|
|
api_secret = os.getenv('MEXC_SECRET_KEY')
|
|
|
|
if not api_key or not api_secret:
|
|
logger.error("MEXC API credentials not found in environment variables")
|
|
return None
|
|
|
|
try:
|
|
mexc = MEXCInterface(api_key=api_key, api_secret=api_secret, test_mode=False)
|
|
|
|
# Test connection
|
|
if mexc.connect():
|
|
logger.info("MEXC: Connection successful")
|
|
else:
|
|
logger.error("MEXC: Connection failed")
|
|
return None
|
|
|
|
# Get trading fees
|
|
logger.info("MEXC: Fetching trading fees...")
|
|
fees = mexc.get_trading_fees()
|
|
|
|
if fees:
|
|
logger.info(f"MEXC Trading Fees Retrieved:")
|
|
logger.info(f" Maker Rate: {fees.get('maker_rate', 0)*100:.3f}%")
|
|
logger.info(f" Taker Rate: {fees.get('taker_rate', 0)*100:.3f}%")
|
|
logger.info(f" Source: {fees.get('source', 'unknown')}")
|
|
|
|
if fees.get('source') == 'mexc_api':
|
|
logger.info(f" Raw Commission Rates:")
|
|
logger.info(f" Maker: {fees.get('maker_commission', 0)} basis points")
|
|
logger.info(f" Taker: {fees.get('taker_commission', 0)} basis points")
|
|
else:
|
|
logger.warning("Using fallback fee values - API may not be working")
|
|
else:
|
|
logger.error("Failed to retrieve trading fees")
|
|
|
|
return fees
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error testing MEXC fee retrieval: {e}")
|
|
return None
|
|
|
|
def test_config_synchronization():
|
|
"""Test automatic config synchronization"""
|
|
logger.info("\n=== Testing Config Synchronization ===")
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
try:
|
|
# Initialize MEXC interface
|
|
api_key = os.getenv('MEXC_API_KEY')
|
|
api_secret = os.getenv('MEXC_SECRET_KEY')
|
|
|
|
if not api_key or not api_secret:
|
|
logger.error("MEXC API credentials not found")
|
|
return False
|
|
|
|
mexc = MEXCInterface(api_key=api_key, api_secret=api_secret, test_mode=False)
|
|
|
|
# Initialize config synchronizer
|
|
config_sync = ConfigSynchronizer(
|
|
config_path="config.yaml",
|
|
mexc_interface=mexc
|
|
)
|
|
|
|
# Get current sync status
|
|
logger.info("Current sync status:")
|
|
status = config_sync.get_sync_status()
|
|
for key, value in status.items():
|
|
if key != 'latest_sync_result':
|
|
logger.info(f" {key}: {value}")
|
|
|
|
# Perform manual sync
|
|
logger.info("\nPerforming manual fee synchronization...")
|
|
sync_result = config_sync.sync_trading_fees(force=True)
|
|
|
|
logger.info(f"Sync Result:")
|
|
logger.info(f" Status: {sync_result.get('status')}")
|
|
logger.info(f" Changes Made: {sync_result.get('changes_made', False)}")
|
|
|
|
if sync_result.get('changes'):
|
|
logger.info(" Fee Changes:")
|
|
for fee_type, change in sync_result['changes'].items():
|
|
logger.info(f" {fee_type}: {change['old']:.6f} -> {change['new']:.6f}")
|
|
|
|
if sync_result.get('errors'):
|
|
logger.warning(f" Errors: {sync_result['errors']}")
|
|
|
|
# Test auto-sync
|
|
logger.info("\nTesting auto-sync...")
|
|
auto_sync_success = config_sync.auto_sync_fees()
|
|
logger.info(f"Auto-sync result: {'Success' if auto_sync_success else 'Failed/Skipped'}")
|
|
|
|
return sync_result.get('status') in ['success', 'no_changes']
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error testing config synchronization: {e}")
|
|
return False
|
|
|
|
def test_trading_executor_integration():
|
|
"""Test fee sync integration in TradingExecutor"""
|
|
logger.info("\n=== Testing TradingExecutor Integration ===")
|
|
|
|
try:
|
|
# Initialize trading executor (this should trigger automatic fee sync)
|
|
logger.info("Initializing TradingExecutor with auto fee sync...")
|
|
executor = TradingExecutor("config.yaml")
|
|
|
|
# Check if config synchronizer was initialized
|
|
if hasattr(executor, 'config_synchronizer') and executor.config_synchronizer:
|
|
logger.info("Config synchronizer successfully initialized")
|
|
|
|
# Get sync status
|
|
sync_status = executor.get_fee_sync_status()
|
|
logger.info("Fee sync status:")
|
|
for key, value in sync_status.items():
|
|
if key not in ['latest_sync_result']:
|
|
logger.info(f" {key}: {value}")
|
|
|
|
# Test manual sync through executor
|
|
logger.info("\nTesting manual sync through TradingExecutor...")
|
|
manual_sync = executor.sync_fees_with_api(force=True)
|
|
logger.info(f"Manual sync result: {manual_sync.get('status')}")
|
|
|
|
# Test auto sync
|
|
logger.info("Testing auto sync...")
|
|
auto_sync = executor.auto_sync_fees_if_needed()
|
|
logger.info(f"Auto sync result: {'Success' if auto_sync else 'Skipped/Failed'}")
|
|
|
|
return True
|
|
else:
|
|
logger.error("Config synchronizer not initialized in TradingExecutor")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error testing TradingExecutor integration: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""Run all tests"""
|
|
logger.info("Starting Fee Synchronization Tests")
|
|
logger.info("=" * 50)
|
|
|
|
# Test 1: Direct API fee retrieval
|
|
fees = test_mexc_fee_retrieval()
|
|
|
|
# Test 2: Config synchronization
|
|
if fees:
|
|
sync_success = test_config_synchronization()
|
|
else:
|
|
logger.warning("Skipping config sync test due to API failure")
|
|
sync_success = False
|
|
|
|
# Test 3: TradingExecutor integration
|
|
if sync_success:
|
|
integration_success = test_trading_executor_integration()
|
|
else:
|
|
logger.warning("Skipping TradingExecutor test due to sync failure")
|
|
integration_success = False
|
|
|
|
# Summary
|
|
logger.info("\n" + "=" * 50)
|
|
logger.info("TEST SUMMARY:")
|
|
logger.info(f" MEXC API Fee Retrieval: {'PASS' if fees else 'FAIL'}")
|
|
logger.info(f" Config Synchronization: {'PASS' if sync_success else 'FAIL'}")
|
|
logger.info(f" TradingExecutor Integration: {'PASS' if integration_success else 'FAIL'}")
|
|
|
|
if fees and sync_success and integration_success:
|
|
logger.info("\nALL TESTS PASSED! Fee synchronization is working correctly.")
|
|
logger.info("Your system will now automatically sync trading fees from MEXC API.")
|
|
else:
|
|
logger.warning("\nSome tests failed. Check the logs above for details.")
|
|
|
|
if __name__ == "__main__":
|
|
main() |