"""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)