""" Central Configuration Management This module handles all configuration for the trading system. It loads settings from config.yaml and provides easy access to all components. """ import os import yaml import logging from pathlib import Path from typing import Dict, List, Any, Optional logger = logging.getLogger(__name__) class Config: """Central configuration management for the trading system""" def __init__(self, config_path: str = "config.yaml"): """Initialize configuration from YAML file""" self.config_path = Path(config_path) self._config = self._load_config() self._setup_directories() def _load_config(self) -> Dict[str, Any]: """Load configuration from YAML file""" try: if not self.config_path.exists(): logger.warning(f"Config file {self.config_path} not found, using defaults") return self._get_default_config() with open(self.config_path, 'r') as f: config = yaml.safe_load(f) logger.info(f"Loaded configuration from {self.config_path}") return config except Exception as e: logger.error(f"Error loading config: {e}") logger.info("Using default configuration") return self._get_default_config() def _get_default_config(self) -> Dict[str, Any]: """Get default configuration if file is missing""" return { 'symbols': ['ETH/USDT', 'BTC/USDT'], 'timeframes': ['1m', '5m', '15m', '1h', '4h', '1d'], 'data': { 'provider': 'binance', 'cache_enabled': True, 'cache_dir': 'cache', 'historical_limit': 1000, 'real_time_enabled': True, 'websocket_reconnect': True }, 'cnn': { 'window_size': 20, 'features': ['open', 'high', 'low', 'close', 'volume'], 'hidden_layers': [64, 32, 16], 'dropout': 0.2, 'learning_rate': 0.001, 'batch_size': 32, 'epochs': 100, 'confidence_threshold': 0.6 }, 'rl': { 'state_size': 100, 'action_space': 3, 'epsilon': 1.0, 'epsilon_decay': 0.995, 'epsilon_min': 0.01, 'learning_rate': 0.0001, 'gamma': 0.99, 'memory_size': 10000, 'batch_size': 64, 'target_update_freq': 1000 }, 'orchestrator': { 'cnn_weight': 0.7, 'rl_weight': 0.3, 'confidence_threshold': 0.5, 'decision_frequency': 60 }, 'trading': { 'max_position_size': 0.1, 'stop_loss': 0.02, 'take_profit': 0.05, 'trading_fee': 0.0002, 'min_trade_interval': 60 }, 'web': { 'host': '127.0.0.1', 'port': 8050, 'debug': False, 'update_interval': 1000, 'chart_history': 100 }, 'logging': { 'level': 'INFO', 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s', 'file': 'logs/trading.log', 'max_size': 10485760, 'backup_count': 5 }, 'performance': { 'use_gpu': True, 'mixed_precision': True, 'num_workers': 4, 'batch_size_multiplier': 1.0 }, 'paths': { 'models': 'models', 'data': 'data', 'logs': 'logs', 'cache': 'cache', 'plots': 'plots' }, 'training': { 'use_only_real_data': True, 'batch_size': 32, 'learning_rate': 0.001, 'epochs': 100, 'validation_split': 0.2, 'early_stopping_patience': 10 } } def _setup_directories(self): """Create necessary directories""" try: paths = self._config.get('paths', {}) for path_name, path_value in paths.items(): Path(path_value).mkdir(parents=True, exist_ok=True) # Also create specific model subdirectories models_dir = Path(paths.get('models', 'models')) (models_dir / 'cnn' / 'saved').mkdir(parents=True, exist_ok=True) (models_dir / 'rl' / 'saved').mkdir(parents=True, exist_ok=True) except Exception as e: logger.error(f"Error creating directories: {e}") # Property accessors for easy access @property def symbols(self) -> List[str]: """Get list of trading symbols""" return self._config.get('symbols', ['ETH/USDT']) @property def timeframes(self) -> List[str]: """Get list of timeframes""" return self._config.get('timeframes', ['1m', '5m', '1h']) @property def data(self) -> Dict[str, Any]: """Get data provider settings""" return self._config.get('data', {}) @property def cnn(self) -> Dict[str, Any]: """Get CNN model settings""" return self._config.get('cnn', {}) @property def rl(self) -> Dict[str, Any]: """Get RL agent settings""" return self._config.get('rl', {}) @property def orchestrator(self) -> Dict[str, Any]: """Get orchestrator settings""" return self._config.get('orchestrator', {}) @property def trading(self) -> Dict[str, Any]: """Get trading execution settings""" return self._config.get('trading', {}) @property def web(self) -> Dict[str, Any]: """Get web dashboard settings""" return self._config.get('web', {}) @property def logging(self) -> Dict[str, Any]: """Get logging settings""" return self._config.get('logging', {}) @property def performance(self) -> Dict[str, Any]: """Get performance settings""" return self._config.get('performance', {}) @property def paths(self) -> Dict[str, str]: """Get file paths""" return self._config.get('paths', {}) @property def training(self) -> Dict[str, Any]: """Training configuration""" return { 'use_only_real_data': True, 'batch_size': self._config.get('training', {}).get('batch_size', 32), 'learning_rate': self._config.get('training', {}).get('learning_rate', 0.001), 'epochs': self._config.get('training', {}).get('epochs', 100), 'validation_split': self._config.get('training', {}).get('validation_split', 0.2), 'early_stopping_patience': self._config.get('training', {}).get('early_stopping_patience', 10) } def get(self, key: str, default: Any = None) -> Any: """Get configuration value by key with optional default""" return self._config.get(key, default) def update(self, key: str, value: Any): """Update configuration value""" self._config[key] = value def save(self): """Save current configuration back to file""" try: with open(self.config_path, 'w') as f: yaml.dump(self._config, f, default_flow_style=False, indent=2) logger.info(f"Configuration saved to {self.config_path}") except Exception as e: logger.error(f"Error saving configuration: {e}") # Global configuration instance _config_instance = None def get_config(config_path: str = "config.yaml") -> Config: """Get global configuration instance (singleton pattern)""" global _config_instance if _config_instance is None: _config_instance = Config(config_path) return _config_instance def setup_logging(config: Optional[Config] = None): """Setup logging based on configuration""" if config is None: config = get_config() log_config = config.logging # Create logs directory log_file = Path(log_config.get('file', 'logs/trading.log')) log_file.parent.mkdir(parents=True, exist_ok=True) # Setup logging logging.basicConfig( level=getattr(logging, log_config.get('level', 'INFO')), format=log_config.get('format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s'), handlers=[ logging.FileHandler(log_file), logging.StreamHandler() ] ) logger.info("Logging configured successfully")