Source code for src.driver.exceptions

"""Custom exception classes for the AMMM driver module.

This module defines domain-specific exceptions for better error handling
and more informative error messages throughout the driver workflow.
"""

from typing import Optional, Dict, Any, List


[docs] class AMMBaseException(Exception): """Base exception class for all AMMM-specific exceptions.""" def __init__(self, message: str, details: Optional[Dict[str, Any]] = None, suggestions: Optional[List[str]] = None): """Initialize the exception with a message and optional details. Args: message: The error message details: Optional dictionary with additional error details suggestions: Optional list of suggestions for fixing the error """ super().__init__(message) self.message = message self.details = details or {} self.suggestions = suggestions or [] def __str__(self) -> str: """Format the exception message with details and suggestions.""" parts = [self.message] if self.details: parts.append("\nDetails:") for key, value in self.details.items(): parts.append(f" - {key}: {value}") if self.suggestions: parts.append("\nSuggestions:") for suggestion in self.suggestions: parts.append(f" • {suggestion}") return "\n".join(parts)
[docs] class ConfigurationError(AMMBaseException): """Raised when there are issues with the configuration file or settings. This includes: - Invalid configuration format - Missing required configuration fields - Invalid parameter values - Incompatible configuration combinations """ pass
[docs] class DataValidationError(AMMBaseException): """Raised when input data fails validation checks. This includes: - Missing required columns - Invalid data types - Data quality issues (NaN, infinite values) - Insufficient data for modeling """ pass
[docs] class FileNotFoundError(AMMBaseException): """Raised when required files are missing or inaccessible. This includes: - Missing configuration file - Missing input data file - Missing holidays file - Unreadable files due to permissions """ def __init__(self, filepath: str, file_type: str = "file"): """Initialize with specific file information. Args: filepath: Path to the missing file file_type: Type of file (config, data, holidays, etc.) """ message = f"Required {file_type} not found: {filepath}" details = { "filepath": filepath, "file_type": file_type } suggestions = [ f"Check that the {file_type} exists at the specified path", "Verify the file path is correct", "Ensure you have read permissions for the file" ] super().__init__(message, details, suggestions)
[docs] class InsufficientDataError(DataValidationError): """Raised when there's not enough data for model fitting. This includes: - Too few observations - Not enough data points after train/test split - Insufficient variation in features """ def __init__(self, required: int, actual: int, context: str = "observations"): """Initialize with specific data size information. Args: required: Minimum required number of data points actual: Actual number of data points available context: Context for the data (observations, training samples, etc.) """ message = f"Insufficient data: {actual} {context} available, but {required} required" details = { "required": required, "actual": actual, "context": context } suggestions = [ "Provide more historical data", "Reduce the train/test split ratio", "Consider using a simpler model with fewer parameters" ] super().__init__(message, details, suggestions)
[docs] class ModelFittingError(AMMBaseException): """Raised when model fitting fails. This includes: - Convergence failures - Numerical issues during fitting - Memory errors - Invalid model configuration """ pass
[docs] class OptimizationError(AMMBaseException): """Raised when budget optimization fails. This includes: - Invalid constraint configurations - Infeasible optimization problems - Solver convergence failures """ pass
[docs] class ConstraintValidationError(OptimizationError): """Raised when optimization constraints are invalid. This includes: - Negative ramp constraints - Invalid percentage values (not in 0-100 range) - Unknown channel names in constraints - Contradictory constraint combinations """ def __init__(self, message: str, constraint_type: str = "constraint", channel: Optional[str] = None, value: Optional[Any] = None): """Initialize with constraint validation details. Args: message: The error message constraint_type: Type of constraint (ramp, budget, etc.) channel: Optional channel name involved value: Optional invalid value """ details = {"constraint_type": constraint_type} if channel: details["channel"] = channel if value is not None: details["invalid_value"] = value suggestions = [ "Check constraint values are non-negative", "Ensure percentage values are between 0 and 100", "Verify channel names match those in your data" ] super().__init__(message, details, suggestions)
[docs] class InfeasibleOptimizationError(OptimizationError): """Raised when the optimization problem has no feasible solution. This includes: - Budget constraints that cannot be satisfied - Ramp constraints too restrictive given budget - Conflicting upper/lower bounds """ def __init__(self, message: str, budget: Optional[float] = None, constraints: Optional[Dict[str, Any]] = None): """Initialize with optimization context. Args: message: The error message budget: Total budget that couldn't be allocated constraints: Dictionary of active constraints """ details = {} if budget is not None: details["budget"] = budget if constraints: details["constraints"] = str(constraints) suggestions = [ "Relax ramp constraints to allow larger budget changes", "Increase the number of optimization periods", "Check that budget bounds allow the total budget to be allocated" ] super().__init__(message, details, suggestions)
[docs] class ModelNotFittedError(AMMBaseException): """Raised when operations requiring a fitted model are called before fitting. This includes: - Attempting predictions before fitting - Trying to visualize results before fitting - Accessing model parameters before fitting """ def __init__(self, operation: str): """Initialize with the operation that requires a fitted model. Args: operation: The operation that was attempted """ message = f"Cannot perform '{operation}': Model has not been fitted yet" details = { "operation": operation } suggestions = [ "Call fit_model() before attempting this operation", "Load a pre-fitted model using the model_filename parameter" ] super().__init__(message, details, suggestions)
[docs] class DataQualityWarning(UserWarning): """Warning for data quality issues that don't prevent execution but may affect results. This includes: - High correlation between features - Low variance in some features - Potential outliers detected - Missing values that were imputed """ pass
[docs] class ColumnMismatchError(DataValidationError): """Raised when expected columns are missing or there's a mismatch in column structure. This includes: - Missing media columns specified in config - Missing control columns - Column name mismatches between config and data """ def __init__(self, expected_columns: List[str], actual_columns: List[str], missing_columns: Optional[List[str]] = None): """Initialize with column information. Args: expected_columns: List of expected column names actual_columns: List of actual column names in the data missing_columns: Optional list of specifically missing columns """ if missing_columns is None: missing_columns = list(set(expected_columns) - set(actual_columns)) message = f"Column mismatch: {len(missing_columns)} required columns missing from data" details = { "missing_columns": missing_columns, "expected_count": len(expected_columns), "actual_count": len(actual_columns) } # Try to find close matches for suggestions from difflib import get_close_matches suggestions = ["Check column names in your data file match the configuration"] for missing_col in missing_columns[:3]: # Show suggestions for first 3 missing columns close_matches = get_close_matches(missing_col, actual_columns, n=2, cutoff=0.6) if close_matches: suggestions.append(f"Did you mean '{close_matches[0]}' instead of '{missing_col}'?") super().__init__(message, details, suggestions)
[docs] class DateContinuityError(DataValidationError): """Raised when there are gaps or issues in the date sequence. This includes: - Missing dates in time series - Duplicate dates - Non-sequential dates - Invalid date formats """ def __init__(self, issue_type: str, problem_dates: Optional[List[str]] = None): """Initialize with date continuity information. Args: issue_type: Type of date issue (gaps, duplicates, etc.) problem_dates: Optional list of problematic dates """ message = f"Date continuity issue: {issue_type}" details = { "issue_type": issue_type } if problem_dates: details["problem_dates"] = str(problem_dates[:10]) # Show first 10 problematic dates if len(problem_dates) > 10: details["total_issues"] = str(len(problem_dates)) suggestions = [] if issue_type == "gaps": suggestions.append("Fill missing dates with interpolated or zero values") suggestions.append("Check if data collection was paused during certain periods") elif issue_type == "duplicates": suggestions.append("Remove or aggregate duplicate date entries") suggestions.append("Check for data processing errors causing duplicates") super().__init__(message, details, suggestions)
[docs] class TensorShapeError(ModelFittingError): """Raised when there's a tensor shape mismatch, often due to configuration changes. This is a common issue when the PyMC session isn't properly cleared between runs with different configurations. """ def __init__(self, expected_shape: Optional[tuple] = None, actual_shape: Optional[tuple] = None): """Initialize with shape information. Args: expected_shape: Expected tensor shape actual_shape: Actual tensor shape encountered """ message = "Tensor shape mismatch detected - likely due to cached model state" details = {} if expected_shape: details["expected_shape"] = expected_shape if actual_shape: details["actual_shape"] = actual_shape suggestions = [ "Clear the PyMC session using clear_pymc_session()", "Restart the Python kernel for a complete reset", "Check if the number of media channels changed between runs", "Ensure configuration matches the data structure" ] super().__init__(message, details, suggestions)