""" Utility functions for loading and managing strategy configurations. """ import json import os import logging from typing import List, Dict, Any, Optional from dash import Output, Input, State logger = logging.getLogger(__name__) def load_strategy_templates() -> Dict[str, Dict[str, Any]]: """Load all strategy templates from the templates directory. Returns: Dict[str, Dict[str, Any]]: Dictionary mapping strategy type to template configuration """ templates = {} try: # Get the templates directory path templates_dir = os.path.join(os.path.dirname(__file__), 'templates') if not os.path.exists(templates_dir): logger.error(f"Templates directory not found at {templates_dir}") return {} # Load all JSON files from templates directory for filename in os.listdir(templates_dir): if filename.endswith('_template.json'): file_path = os.path.join(templates_dir, filename) try: with open(file_path, 'r', encoding='utf-8') as f: template = json.load(f) strategy_type = template.get('type') if strategy_type: templates[strategy_type] = template else: logger.warning(f"Template {filename} missing 'type' field") except json.JSONDecodeError as e: logger.error(f"Error decoding JSON from {filename}: {e}") except Exception as e: logger.error(f"Error loading template {filename}: {e}") except Exception as e: logger.error(f"Error loading strategy templates: {e}") return templates def get_strategy_dropdown_options() -> List[Dict[str, str]]: """Generate dropdown options for strategy types from templates. Returns: List[Dict[str, str]]: List of dropdown options with label and value """ templates = load_strategy_templates() options = [] for strategy_type, template in templates.items(): option = { 'label': template.get('name', strategy_type.upper()), 'value': strategy_type } options.append(option) # Sort by label for consistent UI options.sort(key=lambda x: x['label']) return options def get_strategy_parameter_schema(strategy_type: str) -> Optional[Dict[str, Any]]: """Get parameter schema for a specific strategy type. Args: strategy_type (str): The strategy type (e.g., 'ema_crossover', 'rsi') Returns: Optional[Dict[str, Any]]: Parameter schema or None if not found """ templates = load_strategy_templates() template = templates.get(strategy_type) if template: return template.get('parameter_schema', {}) return None def get_strategy_default_parameters(strategy_type: str) -> Optional[Dict[str, Any]]: """Get default parameters for a specific strategy type. Args: strategy_type (str): The strategy type (e.g., 'ema_crossover', 'rsi') Returns: Optional[Dict[str, Any]]: Default parameters or None if not found """ templates = load_strategy_templates() template = templates.get(strategy_type) if template: return template.get('default_parameters', {}) return None def get_strategy_metadata(strategy_type: str) -> Optional[Dict[str, Any]]: """Get metadata for a specific strategy type. Args: strategy_type (str): The strategy type (e.g., 'ema_crossover', 'rsi') Returns: Optional[Dict[str, Any]]: Strategy metadata or None if not found """ templates = load_strategy_templates() template = templates.get(strategy_type) if template: return template.get('metadata', {}) return None def get_strategy_required_indicators(strategy_type: str) -> List[str]: """Get required indicators for a specific strategy type. Args: strategy_type (str): The strategy type (e.g., 'ema_crossover', 'rsi') Returns: List[str]: List of required indicator types """ metadata = get_strategy_metadata(strategy_type) if metadata: return metadata.get('required_indicators', []) return [] def generate_parameter_fields_config(strategy_type: str) -> Optional[Dict[str, Any]]: """Generate parameter field configuration for dynamic UI generation. Args: strategy_type (str): The strategy type (e.g., 'ema_crossover', 'rsi') Returns: Optional[Dict[str, Any]]: Configuration for generating parameter input fields """ schema = get_strategy_parameter_schema(strategy_type) defaults = get_strategy_default_parameters(strategy_type) if not schema or not defaults: return None fields_config = {} for param_name, param_schema in schema.items(): field_config = { 'type': param_schema.get('type', 'int'), 'label': param_name.replace('_', ' ').title(), 'default': defaults.get(param_name, param_schema.get('default')), 'description': param_schema.get('description', ''), 'input_id': f'{strategy_type}-{param_name.replace("_", "-")}-input' } # Add validation constraints if present if 'min' in param_schema: field_config['min'] = param_schema['min'] if 'max' in param_schema: field_config['max'] = param_schema['max'] if 'step' in param_schema: field_config['step'] = param_schema['step'] if 'options' in param_schema: field_config['options'] = param_schema['options'] fields_config[param_name] = field_config return fields_config def validate_strategy_parameters(strategy_type: str, parameters: Dict[str, Any]) -> tuple[bool, List[str]]: """Validate strategy parameters against schema. Args: strategy_type (str): The strategy type parameters (Dict[str, Any]): Parameters to validate Returns: tuple[bool, List[str]]: (is_valid, list_of_errors) """ schema = get_strategy_parameter_schema(strategy_type) if not schema: return False, [f"No schema found for strategy type: {strategy_type}"] errors = [] # Check required parameters for param_name, param_schema in schema.items(): if param_schema.get('required', True) and param_name not in parameters: errors.append(f"Missing required parameter: {param_name}") continue if param_name not in parameters: continue value = parameters[param_name] param_type = param_schema.get('type', 'int') # Type validation if param_type == 'int' and not isinstance(value, int): errors.append(f"Parameter {param_name} must be an integer") elif param_type == 'float' and not isinstance(value, (int, float)): errors.append(f"Parameter {param_name} must be a number") elif param_type == 'str' and not isinstance(value, str): errors.append(f"Parameter {param_name} must be a string") elif param_type == 'bool' and not isinstance(value, bool): errors.append(f"Parameter {param_name} must be a boolean") # Range validation if 'min' in param_schema and value < param_schema['min']: errors.append(f"Parameter {param_name} must be >= {param_schema['min']}") if 'max' in param_schema and value > param_schema['max']: errors.append(f"Parameter {param_name} must be <= {param_schema['max']}") # Options validation if 'options' in param_schema and value not in param_schema['options']: errors.append(f"Parameter {param_name} must be one of: {param_schema['options']}") return len(errors) == 0, errors def save_user_strategy(strategy_name: str, config: Dict[str, Any]) -> bool: """Save a user-defined strategy configuration. Args: strategy_name (str): Name of the strategy configuration config (Dict[str, Any]): Strategy configuration Returns: bool: True if saved successfully, False otherwise """ try: user_strategies_dir = os.path.join(os.path.dirname(__file__), 'user_strategies') os.makedirs(user_strategies_dir, exist_ok=True) filename = f"{strategy_name.lower().replace(' ', '_')}.json" file_path = os.path.join(user_strategies_dir, filename) with open(file_path, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) logger.info(f"Saved user strategy configuration: {strategy_name}") return True except Exception as e: logger.error(f"Error saving user strategy {strategy_name}: {e}") return False def load_user_strategies() -> Dict[str, Dict[str, Any]]: """Load all user-defined strategy configurations. Returns: Dict[str, Dict[str, Any]]: Dictionary mapping strategy name to configuration """ strategies = {} try: user_strategies_dir = os.path.join(os.path.dirname(__file__), 'user_strategies') if not os.path.exists(user_strategies_dir): return {} for filename in os.listdir(user_strategies_dir): if filename.endswith('.json'): file_path = os.path.join(user_strategies_dir, filename) try: with open(file_path, 'r', encoding='utf-8') as f: config = json.load(f) strategy_name = config.get('name', filename.replace('.json', '')) strategies[strategy_name] = config except Exception as e: logger.error(f"Error loading user strategy {filename}: {e}") except Exception as e: logger.error(f"Error loading user strategies: {e}") return strategies def delete_user_strategy(strategy_name: str) -> bool: """Delete a user-defined strategy configuration. Args: strategy_name (str): Name of the strategy to delete Returns: bool: True if deleted successfully, False otherwise """ try: user_strategies_dir = os.path.join(os.path.dirname(__file__), 'user_strategies') filename = f"{strategy_name.lower().replace(' ', '_')}.json" file_path = os.path.join(user_strategies_dir, filename) if os.path.exists(file_path): os.remove(file_path) logger.info(f"Deleted user strategy configuration: {strategy_name}") return True else: logger.warning(f"User strategy file not found: {file_path}") return False except Exception as e: logger.error(f"Error deleting user strategy {strategy_name}: {e}") return False def export_strategy_config(strategy_name: str, config: Dict[str, Any]) -> str: """Export strategy configuration as JSON string. Args: strategy_name (str): Name of the strategy config (Dict[str, Any]): Strategy configuration Returns: str: JSON string representation of the configuration """ export_data = { 'name': strategy_name, 'config': config, 'exported_at': str(os.times()), 'version': '1.0' } return json.dumps(export_data, indent=2, ensure_ascii=False) def import_strategy_config(json_string: str) -> tuple[bool, Optional[Dict[str, Any]], List[str]]: """Import strategy configuration from JSON string. Args: json_string (str): JSON string containing strategy configuration Returns: tuple[bool, Optional[Dict[str, Any]], List[str]]: (success, config, errors) """ try: data = json.loads(json_string) if 'name' not in data or 'config' not in data: return False, None, ['Invalid format: missing name or config fields'] # Validate the configuration if it has a strategy type config = data['config'] if 'strategy' in config: is_valid, errors = validate_strategy_parameters(config['strategy'], config) if not is_valid: return False, None, errors return True, data, [] except json.JSONDecodeError as e: return False, None, [f'Invalid JSON format: {e}'] except Exception as e: return False, None, [f'Error importing configuration: {e}']