Type System and Error Handling

# Nevergrad-specific type aliases
ArgsKwargs = Tuple[Tuple[Any, ...], Dict[str, Any]]
"""Type for function arguments and keyword arguments."""

ArrayLike = Union[Tuple[float, ...], List[float], np.ndarray]
"""Array-like structures for numerical data."""

PathLike = Union[str, Path]
"""Path-like objects for file system operations."""

FloatLoss = float
"""Single-objective loss value."""

Loss = Union[float, ArrayLike]
"""Single or multi-objective loss values."""

BoundValue = Optional[Union[float, int, np.int_, np.float64, ArrayLike]]
"""Bound values for parameter constraints."""

# Standard typing imports (re-exported)
Any: Type
Generic: Type
Type: Type
TypeVar: Type
Optional: Type
Union: Type
Dict: Type
Tuple: Type
List: Type
Set: Type
Deque: Type
Sequence: Type
NamedTuple: Type
MutableMapping: Type
Iterator: Type
Iterable: Type
Generator: Type
KeysView: Type
ValuesView: Type
ItemsView: Type
Callable: Type
Hashable: Type
Match: Type
Path: Type
Protocol: Type

def cast(typ: Type, val: Any) -> Any:
    """Type casting function."""

class JobLike(Protocol[X]):
    """
    Protocol for job-like objects with asynchronous execution capabilities.
    
    Defines interface for objects that can be executed asynchronously
    and provide completion status and result retrieval.
    """
    
    def done(self) -> bool:
        """
        Check if job execution is complete.
        
        Returns:
            True if job has finished execution
        """
    
    def result(self) -> X:
        """
        Get job execution result.
        
        Returns:
            Result of job execution
            
        Raises:
            Exception if job failed or is not yet complete
        """

class ExecutorLike(Protocol):
    """
    Protocol for executor-like objects that can submit tasks for execution.
    
    Defines interface for parallel execution frameworks compatible
    with nevergrad's optimization workflows.
    """
    
    def submit(self, fn: Callable, *args, **kwargs) -> JobLike:
        """
        Submit function for execution.
        
        Args:
            fn: Function to execute
            *args: Positional arguments for function
            **kwargs: Keyword arguments for function
            
        Returns:
            Job-like object for tracking execution
        """

class NevergradError(Exception):
    """
    Base exception class for all Nevergrad errors.
    
    All custom exceptions in nevergrad inherit from this class,
    enabling catch-all error handling for nevergrad-specific issues.
    """

class NevergradWarning(UserWarning):
    """
    Base warning class for all Nevergrad warnings.
    
    All custom warnings in nevergrad inherit from this class,
    enabling structured warning handling and filtering.
    """

class NevergradEarlyStopping(StopIteration, NevergradError):
    """
    Stops the minimization loop when raised.
    
    Used by early stopping mechanisms and convergence criteria
    to cleanly terminate optimization runs.
    """

class NevergradRuntimeError(RuntimeError, NevergradError):
    """
    Runtime errors raised by Nevergrad during execution.
    
    Covers general runtime issues that occur during optimization
    or parameter manipulation.
    """

class NevergradTypeError(TypeError, NevergradError):
    """
    Type errors raised by Nevergrad for incorrect types.
    
    Raised when function arguments or parameter types don't match
    expected interfaces or protocols.
    """

class NevergradValueError(ValueError, NevergradError):
    """
    Value errors raised by Nevergrad for incorrect values.
    
    Raised when parameter values are outside valid ranges or
    configuration settings are invalid.
    """

class NevergradNotImplementedError(NotImplementedError, NevergradError):
    """
    Raised when functionality is not yet implemented.
    
    Used for features that are planned but not yet available,
    or abstract methods that must be implemented by subclasses.
    """

class TellNotAskedNotSupportedError(NevergradNotImplementedError):
    """
    Raised by optimizers which do not support tell_not_asked interface.
    
    Some optimizers cannot handle evaluation results for candidates
    they did not generate through the ask() method.
    """

class ExperimentFunctionCopyError(NevergradNotImplementedError):
    """
    Raised when experiment function fails to copy itself.
    
    Occurs when benchmark functions cannot be properly serialized
    or copied for parallel execution.
    """

class UnsupportedExperiment(unittest.SkipTest, NevergradRuntimeError):
    """
    Raised if experiment is not compatible with current settings.
    
    Used in benchmarking when experiment requirements don't match
    the current execution environment or configuration.
    """

class NevergradDeprecationError(NevergradRuntimeError):
    """
    Raised when deprecated functionality is used.
    
    Indicates that code uses deprecated APIs that have been removed
    or are no longer supported.
    """

class UnsupportedParameterOperationError(NevergradRuntimeError):
    """
    Raised when operation is not supported by parameter type.
    
    Occurs when attempting operations that are not valid for
    specific parameter types (e.g., bounds on discrete parameters).
    """

class NevergradDeprecationWarning(DeprecationWarning, NevergradWarning):
    """
    Deprecated function/class warnings.
    
    Issued when deprecated APIs are used but still functional,
    providing migration guidance to users.
    """

class NevergradRuntimeWarning(RuntimeWarning, NevergradWarning):
    """
    Runtime warnings raised by nevergrad.
    
    General warnings about potentially problematic runtime conditions
    or suboptimal configurations.
    """

class InefficientSettingsWarning(NevergradRuntimeWarning):
    """
    Optimization settings are not optimal.
    
    Warns when algorithm or parameter settings may lead to
    poor optimization performance.
    """

class BadLossWarning(NevergradRuntimeWarning):
    """
    Provided loss is unhelpful for optimization.
    
    Issued when loss values are constant, infinite, or otherwise
    don't provide optimization guidance.
    """

class LossTooLargeWarning(BadLossWarning):
    """
    Loss is clipped because it is too large.
    
    Warns when loss values exceed internal limits and are
    automatically clipped to prevent numerical issues.
    """

class NevergradBehaviorChangesWarning(NevergradRuntimeWarning):
    """
    Notifies about changes in nevergrad behavior.
    
    Informs users about behavioral changes in algorithms or
    interfaces that may affect existing code.
    """

class FinishedUnderlyingOptimizerWarning(NevergradRuntimeWarning):
    """
    Underlying scipy optimizer finished early.
    
    Warns when wrapped scipy optimizers terminate before
    the allocated budget is exhausted.
    """

class FailedConstraintWarning(NevergradRuntimeWarning):
    """
    Constraint could not be applied successfully.
    
    Issued when constraint satisfaction fails or constraint
    functions cannot be properly evaluated.
    """

import nevergrad as ng
from nevergrad.typing import ArrayLike, Loss, ExecutorLike

def optimization_function(x: ArrayLike) -> Loss:
    """Type-safe optimization function."""
    import numpy as np
    return float(np.sum(np.array(x) ** 2))

def parallel_optimize(
    function: Callable[[ArrayLike], Loss],
    parametrization: ng.p.Parameter,
    executor: ExecutorLike,
    budget: int
) -> ng.p.Parameter:
    """Type-safe parallel optimization."""
    optimizer = ng.optimizers.CMA(parametrization=parametrization, budget=budget)
    
    jobs = []
    for _ in range(budget):
        x = optimizer.ask()
        job = executor.submit(function, x.value)
        jobs.append((x, job))
    
    for x, job in jobs:
        loss = job.result()
        optimizer.tell(x, loss)
    
    return optimizer.provide_recommendation()

import nevergrad as ng
from nevergrad import errors

def robust_optimization(function, parametrization, budget=100):
    """Optimization with comprehensive error handling."""
    try:
        optimizer = ng.optimizers.CMA(parametrization=parametrization, budget=budget)
        
        for i in range(budget):
            try:
                x = optimizer.ask()
                loss = function(x.value)
                
                if not isinstance(loss, (int, float)):
                    raise errors.NevergradTypeError(f"Loss must be numeric, got {type(loss)}")
                
                if loss != loss:  # Check for NaN
                    raise errors.NevergradValueError("Loss is NaN")
                
                optimizer.tell(x, loss)
                
            except errors.NevergradEarlyStopping:
                print(f"Early stopping at iteration {i}")
                break
            
            except errors.TellNotAskedNotSupportedError:
                print("Optimizer doesn't support tell_not_asked, skipping...")
                continue
                
            except errors.UnsupportedParameterOperationError as e:
                print(f"Parameter operation not supported: {e}")
                continue
        
        return optimizer.provide_recommendation()
        
    except errors.NevergradRuntimeError as e:
        print(f"Runtime error during optimization: {e}")
        return None
        
    except errors.NevergradError as e:
        print(f"Nevergrad error: {e}")
        return None

import warnings
import nevergrad as ng
from nevergrad import errors

# Configure warning filters
warnings.filterwarnings("always", category=errors.InefficientSettingsWarning)
warnings.filterwarnings("ignore", category=errors.NevergradDeprecationWarning)

def optimization_with_warnings(function, parametrization):
    """Handle optimization warnings appropriately."""
    
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        
        optimizer = ng.optimizers.CMA(parametrization=parametrization, budget=10)
        result = optimizer.minimize(function)
        
        # Process warnings
        for warning in w:
            if issubclass(warning.category, errors.InefficientSettingsWarning):
                print(f"Efficiency warning: {warning.message}")
            elif issubclass(warning.category, errors.BadLossWarning):
                print(f"Loss warning: {warning.message}")
            elif issubclass(warning.category, errors.NevergradBehaviorChangesWarning):
                print(f"Behavior change: {warning.message}")
    
    return result

from nevergrad import errors

class CustomOptimizationError(errors.NevergradRuntimeError):
    """Custom error for specific optimization scenarios."""
    pass

class ConvergenceFailedError(CustomOptimizationError):
    """Raised when optimization fails to converge."""
    
    def __init__(self, iterations, tolerance):
        super().__init__(f"Failed to converge after {iterations} iterations with tolerance {tolerance}")
        self.iterations = iterations
        self.tolerance = tolerance

def strict_optimization(function, parametrization, tolerance=1e-6):
    """Optimization with strict convergence requirements."""
    optimizer = ng.optimizers.CMA(parametrization=parametrization, budget=1000)
    
    prev_loss = float('inf')
    patience = 0
    max_patience = 50
    
    for i in range(optimizer.budget):
        x = optimizer.ask()
        loss = function(x.value)
        optimizer.tell(x, loss)
        
        if abs(prev_loss - loss) < tolerance:
            patience += 1
        else:
            patience = 0
        
        if patience >= max_patience:
            raise ConvergenceFailedError(i, tolerance)
        
        prev_loss = loss
    
    return optimizer.provide_recommendation()

from nevergrad.typing import ExecutorLike, JobLike
from concurrent.futures import ThreadPoolExecutor, Future

class NevergradExecutor(ExecutorLike):
    """Custom executor implementation for nevergrad."""
    
    def __init__(self, max_workers=4):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    def submit(self, fn, *args, **kwargs) -> JobLike:
        """Submit function for execution."""
        future = self.executor.submit(fn, *args, **kwargs)
        return NevergradJob(future)

class NevergradJob(JobLike):
    """Job wrapper for Future objects."""
    
    def __init__(self, future: Future):
        self.future = future
    
    def done(self) -> bool:
        """Check if job is complete."""
        return self.future.done()
    
    def result(self):
        """Get job result."""
        return self.future.result()

# Usage with custom executor
executor = NevergradExecutor(max_workers=8)
result = parallel_optimize(sphere_function, parametrization, executor, budget=100)