""" Config Synchronization Module This module handles automatic synchronization of trading fees and other parameters between the MEXC API and the local configuration files. """ import logging import os import yaml import json from datetime import datetime, timedelta from typing import Dict, Any, Optional from pathlib import Path import time logger = logging.getLogger(__name__) class ConfigSynchronizer: """Handles automatic synchronization of config parameters with MEXC API""" def __init__(self, config_path: str = "config.yaml", mexc_interface=None): """Initialize the config synchronizer Args: config_path: Path to the main config file mexc_interface: MEXCInterface instance for API calls """ self.config_path = config_path self.mexc_interface = mexc_interface self.last_sync_time = None self.sync_interval = 3600 # Sync every hour by default self.backup_enabled = True # Track sync history self.sync_history_path = "logs/config_sync_history.json" self.sync_history = self._load_sync_history() def _load_sync_history(self) -> list: """Load sync history from file""" try: if os.path.exists(self.sync_history_path): with open(self.sync_history_path, 'r') as f: return json.load(f) except Exception as e: logger.warning(f"Could not load sync history: {e}") return [] def _save_sync_history(self, sync_record: Dict[str, Any]): """Save sync record to history""" try: self.sync_history.append(sync_record) # Keep only last 100 sync records self.sync_history = self.sync_history[-100:] # Ensure logs directory exists os.makedirs(os.path.dirname(self.sync_history_path), exist_ok=True) with open(self.sync_history_path, 'w') as f: json.dump(self.sync_history, f, indent=2, default=str) except Exception as e: logger.error(f"Could not save sync history: {e}") def _load_config(self) -> Dict[str, Any]: """Load current configuration from file""" try: with open(self.config_path, 'r') as f: return yaml.safe_load(f) except Exception as e: logger.error(f"Error loading config from {self.config_path}: {e}") return {} def _save_config(self, config: Dict[str, Any], backup: bool = True) -> bool: """Save configuration to file with optional backup Args: config: Configuration dictionary to save backup: Whether to create a backup of the existing config Returns: bool: True if save successful, False otherwise """ try: # Create backup if requested if backup and self.backup_enabled and os.path.exists(self.config_path): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = f"{self.config_path}.backup_{timestamp}" import shutil shutil.copy2(self.config_path, backup_path) logger.info(f"Created config backup: {backup_path}") # Save new config with open(self.config_path, 'w') as f: yaml.dump(config, f, indent=2, default_flow_style=False) logger.info(f"Config saved successfully to {self.config_path}") return True except Exception as e: logger.error(f"Error saving config to {self.config_path}: {e}") return False def sync_trading_fees(self, force: bool = False) -> Dict[str, Any]: """Sync trading fees from MEXC API to config Args: force: Force sync even if last sync was recent Returns: dict: Sync result with status and details """ sync_record = { 'timestamp': datetime.now().isoformat(), 'type': 'trading_fees', 'status': 'started', 'changes': {}, 'api_response': {}, 'errors': [] } try: # Check if sync is needed if not force and self.last_sync_time: time_since_sync = datetime.now() - self.last_sync_time if time_since_sync.total_seconds() < self.sync_interval: sync_record['status'] = 'skipped' sync_record['reason'] = f'Last sync was {time_since_sync.total_seconds():.0f}s ago' logger.info(f"CONFIG SYNC: Skipping sync, last sync was recent") return sync_record if not self.mexc_interface: sync_record['status'] = 'error' sync_record['errors'].append('No MEXC interface available') logger.error("CONFIG SYNC: No MEXC interface available for fee sync") return sync_record # Get current fees from MEXC API logger.info("CONFIG SYNC: Fetching trading fees from MEXC API") api_fees = self.mexc_interface.get_trading_fees() sync_record['api_response'] = api_fees if api_fees.get('source') == 'fallback': sync_record['status'] = 'warning' sync_record['errors'].append('API returned fallback values') logger.warning("CONFIG SYNC: API returned fallback fee values") # Load current config config = self._load_config() if not config: sync_record['status'] = 'error' sync_record['errors'].append('Could not load current config') return sync_record # Update trading fees in config changes_made = False # Ensure trading fees section exists if 'trading' not in config: config['trading'] = {} if 'trading_fees' not in config['trading']: config['trading']['trading_fees'] = {} # Check and update maker fee current_maker = config['trading']['trading_fees'].get('maker', 0.0) new_maker = api_fees.get('maker_rate', current_maker) if abs(current_maker - new_maker) > 0.000001: # Significant difference config['trading']['trading_fees']['maker'] = new_maker sync_record['changes']['maker_fee'] = { 'old': current_maker, 'new': new_maker, 'change_percent': f"{((new_maker - current_maker) / max(current_maker, 0.000001)) * 100:.2f}%" } changes_made = True logger.info(f"CONFIG SYNC: Updated maker fee: {current_maker:.4f} -> {new_maker:.4f}") # Check and update taker fee current_taker = config['trading']['trading_fees'].get('taker', 0.0005) new_taker = api_fees.get('taker_rate', current_taker) if abs(current_taker - new_taker) > 0.000001: # Significant difference config['trading']['trading_fees']['taker'] = new_taker sync_record['changes']['taker_fee'] = { 'old': current_taker, 'new': new_taker, 'change_percent': f"{((new_taker - current_taker) / max(current_taker, 0.000001)) * 100:.2f}%" } changes_made = True logger.info(f"CONFIG SYNC: Updated taker fee: {current_taker:.4f} -> {new_taker:.4f}") # Update default fee to match taker fee current_default = config['trading']['trading_fees'].get('default', 0.0005) if abs(current_default - new_taker) > 0.000001: config['trading']['trading_fees']['default'] = new_taker sync_record['changes']['default_fee'] = { 'old': current_default, 'new': new_taker } changes_made = True logger.info(f"CONFIG SYNC: Updated default fee: {current_default:.4f} -> {new_taker:.4f}") # Add sync metadata if 'fee_sync_metadata' not in config['trading']: config['trading']['fee_sync_metadata'] = {} config['trading']['fee_sync_metadata'] = { 'last_sync': datetime.now().isoformat(), 'api_source': 'mexc', 'sync_enabled': True, 'api_commission_rates': { 'maker': api_fees.get('maker_commission', 0), 'taker': api_fees.get('taker_commission', 0) } } # Save config if changes were made if changes_made or 'fee_sync_metadata' not in config.get('trading', {}): if self._save_config(config, backup=True): sync_record['status'] = 'success' sync_record['changes_made'] = changes_made self.last_sync_time = datetime.now() logger.info(f"CONFIG SYNC: Successfully synced trading fees") if changes_made: logger.info(f"CONFIG SYNC: Changes made: {list(sync_record['changes'].keys())}") else: logger.info("CONFIG SYNC: No fee changes needed, metadata updated") else: sync_record['status'] = 'error' sync_record['errors'].append('Failed to save updated config') else: sync_record['status'] = 'no_changes' sync_record['changes_made'] = False self.last_sync_time = datetime.now() logger.info("CONFIG SYNC: No changes needed, fees are already current") except Exception as e: sync_record['status'] = 'error' sync_record['errors'].append(str(e)) logger.error(f"CONFIG SYNC: Error during fee sync: {e}") finally: # Save sync record to history self._save_sync_history(sync_record) return sync_record def auto_sync_fees(self) -> bool: """Automatically sync fees if conditions are met Returns: bool: True if sync was performed, False if skipped """ try: # Check if auto-sync is enabled in config config = self._load_config() auto_sync_enabled = config.get('trading', {}).get('fee_sync_metadata', {}).get('sync_enabled', True) if not auto_sync_enabled: logger.debug("CONFIG SYNC: Auto-sync is disabled") return False # Perform sync result = self.sync_trading_fees(force=False) return result.get('status') in ['success', 'no_changes'] except Exception as e: logger.error(f"CONFIG SYNC: Error in auto-sync: {e}") return False def get_sync_status(self) -> Dict[str, Any]: """Get current sync status and history Returns: dict: Sync status information """ try: config = self._load_config() metadata = config.get('trading', {}).get('fee_sync_metadata', {}) # Get latest sync from history latest_sync = self.sync_history[-1] if self.sync_history else None return { 'sync_enabled': metadata.get('sync_enabled', True), 'last_sync': metadata.get('last_sync'), 'api_source': metadata.get('api_source'), 'sync_interval_seconds': self.sync_interval, 'latest_sync_result': latest_sync, 'total_syncs': len(self.sync_history), 'mexc_interface_available': self.mexc_interface is not None } except Exception as e: logger.error(f"Error getting sync status: {e}") return {'error': str(e)} def enable_auto_sync(self, enabled: bool = True): """Enable or disable automatic fee synchronization Args: enabled: Whether to enable auto-sync """ try: config = self._load_config() if 'trading' not in config: config['trading'] = {} if 'fee_sync_metadata' not in config['trading']: config['trading']['fee_sync_metadata'] = {} config['trading']['fee_sync_metadata']['sync_enabled'] = enabled if self._save_config(config, backup=False): logger.info(f"CONFIG SYNC: Auto-sync {'enabled' if enabled else 'disabled'}") else: logger.error("CONFIG SYNC: Failed to update auto-sync setting") except Exception as e: logger.error(f"CONFIG SYNC: Error updating auto-sync setting: {e}")