import json import datetime import logging from typing import Dict, List, Optional, Any from pathlib import Path class ConfigManager: """Manages configuration loading, validation, and default values for backtest operations""" DEFAULT_CONFIG = { "start_date": "2025-05-01", "stop_date": datetime.datetime.today().strftime('%Y-%m-%d'), "initial_usd": 10000, "timeframes": ["1D", "6h", "3h", "1h", "30m", "15m", "5m", "1m"], "stop_loss_pcts": [0.01, 0.02, 0.03, 0.05], "data_dir": "../data", "results_dir": "results" } def __init__(self, logging_instance: Optional[logging.Logger] = None): """ Initialize configuration manager Args: logging_instance: Optional logging instance for output """ self.logging = logging_instance self.config = {} def load_config(self, config_path: Optional[str] = None) -> Dict[str, Any]: """ Load configuration from file or interactive input Args: config_path: Path to JSON config file, if None prompts for interactive input Returns: Dictionary containing validated configuration Raises: FileNotFoundError: If config file doesn't exist json.JSONDecodeError: If config file has invalid JSON ValueError: If configuration values are invalid """ if config_path: self.config = self._load_from_file(config_path) else: self.config = self._load_interactive() self._validate_config() return self.config def _load_from_file(self, config_path: str) -> Dict[str, Any]: """Load configuration from JSON file""" try: config_file = Path(config_path) if not config_file.exists(): raise FileNotFoundError(f"Configuration file not found: {config_path}") with open(config_file, 'r') as f: config = json.load(f) if self.logging: self.logging.info(f"Configuration loaded from {config_path}") return config except json.JSONDecodeError as e: error_msg = f"Invalid JSON in configuration file {config_path}: {e}" if self.logging: self.logging.error(error_msg) raise json.JSONDecodeError(error_msg, e.doc, e.pos) def _load_interactive(self) -> Dict[str, Any]: """Load configuration through interactive prompts""" print("No config file provided. Please enter the following values (press Enter to use default):") config = {} # Start date start_date = input(f"Start date [{self.DEFAULT_CONFIG['start_date']}]: ") or self.DEFAULT_CONFIG['start_date'] config['start_date'] = start_date # Stop date stop_date = input(f"Stop date [{self.DEFAULT_CONFIG['stop_date']}]: ") or self.DEFAULT_CONFIG['stop_date'] config['stop_date'] = stop_date # Initial USD initial_usd_str = input(f"Initial USD [{self.DEFAULT_CONFIG['initial_usd']}]: ") or str(self.DEFAULT_CONFIG['initial_usd']) try: config['initial_usd'] = float(initial_usd_str) except ValueError: raise ValueError(f"Invalid initial USD value: {initial_usd_str}") # Timeframes timeframes_str = input(f"Timeframes (comma separated) [{', '.join(self.DEFAULT_CONFIG['timeframes'])}]: ") or ','.join(self.DEFAULT_CONFIG['timeframes']) config['timeframes'] = [tf.strip() for tf in timeframes_str.split(',') if tf.strip()] # Stop loss percentages stop_loss_pcts_str = input(f"Stop loss pcts (comma separated) [{', '.join(str(x) for x in self.DEFAULT_CONFIG['stop_loss_pcts'])}]: ") or ','.join(str(x) for x in self.DEFAULT_CONFIG['stop_loss_pcts']) try: config['stop_loss_pcts'] = [float(x.strip()) for x in stop_loss_pcts_str.split(',') if x.strip()] except ValueError: raise ValueError(f"Invalid stop loss percentages: {stop_loss_pcts_str}") # Add default directories config['data_dir'] = self.DEFAULT_CONFIG['data_dir'] config['results_dir'] = self.DEFAULT_CONFIG['results_dir'] return config def _validate_config(self) -> None: """ Validate configuration values Raises: ValueError: If any configuration value is invalid """ # Validate initial USD if self.config.get('initial_usd', 0) <= 0: raise ValueError("Initial USD must be positive") # Validate stop loss percentages stop_loss_pcts = self.config.get('stop_loss_pcts', []) for pct in stop_loss_pcts: if not 0 < pct < 1: raise ValueError(f"Stop loss percentage must be between 0 and 1, got: {pct}") # Validate dates try: datetime.datetime.strptime(self.config['start_date'], '%Y-%m-%d') datetime.datetime.strptime(self.config['stop_date'], '%Y-%m-%d') except ValueError as e: raise ValueError(f"Invalid date format (should be YYYY-MM-DD): {e}") # Validate timeframes timeframes = self.config.get('timeframes', []) if not timeframes: raise ValueError("At least one timeframe must be specified") # Validate directories exist or can be created for dir_key in ['data_dir', 'results_dir']: dir_path = Path(self.config.get(dir_key, '')) try: dir_path.mkdir(parents=True, exist_ok=True) except Exception as e: raise ValueError(f"Cannot create directory {dir_path}: {e}") if self.logging: self.logging.info("Configuration validation completed successfully") def get_config(self) -> Dict[str, Any]: """Return the current configuration""" return self.config.copy() def save_config(self, output_path: str) -> None: """ Save current configuration to file Args: output_path: Path where to save the configuration """ try: with open(output_path, 'w') as f: json.dump(self.config, f, indent=2) if self.logging: self.logging.info(f"Configuration saved to {output_path}") except Exception as e: error_msg = f"Failed to save configuration to {output_path}: {e}" if self.logging: self.logging.error(error_msg) raise