epydemix.calibration package

Submodules

epydemix.calibration.abc module

class epydemix.calibration.abc.ABCSampler(simulation_function: ~typing.Callable, priors: ~typing.Dict[str, ~typing.Any], parameters: ~typing.Dict[str, ~typing.Any], observed_data: ~typing.Any, distance_function: ~typing.Callable = <function rmse>, rng: ~typing.Any | None = None)[source]

Bases: object

Approximate Bayesian Computation (ABC) class implementing different ABC strategies.

calibrate(strategy: str = 'smc', **kwargs) → CalibrationResults[source]

Run calibration using the specified strategy.

This function allows the user to run Approximate Bayesian Computation (ABC) calibration using one of the available strategies: Sequential Monte Carlo (SMC), rejection sampling, or top fraction selection. The appropriate method is selected based on the strategy argument, and additional keyword arguments (**kwargs) are passed to the corresponding function.

### Available Strategies: - “smc”: Uses Sequential Monte Carlo (ABC-SMC) for calibration. - “rejection”: Uses ABC rejection sampling. - “top_fraction”: Selects the best fraction of simulated results.

### Arguments: - strategy (str, default: “smc”): Specifies the calibration strategy. Must be one of {“smc”, “rejection”, “top_fraction”}. - kwargs: Additional parameters depending on the chosen strategy.

### Strategy-Specific Arguments:

#### “smc” (Sequential Monte Carlo) - num_particles (int, default: 1000): Number of particles (samples) per generation. - num_generations (int, default: 10): Number of generations for the ABC-SMC process. - epsilon_schedule (Optional[List[float]], default: None): Predefined schedule for epsilon values. - epsilon_quantile_level (float, default: 0.5): Quantile level to adapt epsilon if no schedule is provided. - minimum_epsilon (Optional[float], default: None): Minimum allowable epsilon value. - max_time (Optional[timedelta], default: None): Maximum allowed runtime. - total_simulations_budget (Optional[int], default: None): Maximum number of allowed simulations. - perturbations (Optional[Dict[str, Any]], default: None): Perturbation kernels for parameters. - verbose (bool, default: True): Whether to print progress updates.

#### “rejection” (ABC Rejection Sampling) - epsilon (float, default: 0.1): Distance threshold for accepting samples. - num_particles (int, default: 1000): Number of accepted samples. - max_time (Optional[timedelta], default: None): Maximum allowed runtime. - total_simulations_budget (Optional[int], default: None): Maximum number of allowed simulations. - verbose (bool, default: True): Whether to print progress updates. - progress_update_interval (int, default: 1000): Interval at which progress updates are printed.

#### “top_fraction” (ABC Top-Fraction Selection) - top_fraction (float, default: 0.05): Fraction of best-fitting simulations to keep. - Nsim (int, default: 100): Total number of simulations to run. - verbose (bool, default: True): Whether to print progress updates.

### Returns: - CalibrationResults: A deep copy of the results from the chosen calibration strategy.

### Raises: - ValueError: If an unknown strategy is specified.

Example Usage: ```python # Run SMC calibration with custom parameters results = model.calibrate(strategy=”smc”, num_particles=500, num_generations=15)

# Run rejection sampling with a different epsilon threshold results = model.calibrate(strategy=”rejection”, epsilon=0.05, num_particles=2000)

# Run top fraction selection with a different fraction results = model.calibrate(strategy=”top_fraction”, top_fraction=0.1, Nsim=500) ```

run_projections(parameters: Dict[str, Any], iterations: int = 100, generation: int | None = None, scenario_id: str = 'baseline', rng: Any | None = None) → CalibrationResults[source]

Run projections using parameters sampled from the posterior distribution.

Parameters:

parameters – Dictionary of parameters for the projections
iterations – Number of projection iterations to run. Default is 100.
generation – Which generation to use for posterior. If None, the last generation is used.
scenario_id – Identifier for this projection scenario. Default is “baseline”.
rng – Optional seed or np.random.Generator for this call, seeding both the posterior resampling and the simulation of every iteration. Results depend only on the seed, so two scenarios given the same seed are paired: they draw the same posterior samples and diverge only where their scenario-specific parameters differ. If None, the seed is taken from an "rng" key in parameters, otherwise from the sampler’s own rng (so seeding the ABCSampler already makes its projections reproducible). Pass rng explicitly only to override this, e.g. to draw an independent ensemble from the same calibration.

Returns:

A new CalibrationResults object containing the original results plus the new projections

Return type:

CalibrationResults

run_rejection(epsilon: float = 0.1, num_particles: int = 1000, max_time: timedelta | None = None, total_simulations_budget: int | None = None, verbose: bool = True, progress_update_interval: int = 1000) → CalibrationResults[source]: Run ABC rejection sampling.

run_smc(num_particles: int = 1000, num_generations: int = 10, epsilon_schedule: List[float] | None = None, epsilon_quantile_level: float = 0.5, minimum_epsilon: float | None = None, max_time: timedelta | None = None, total_simulations_budget: int | None = None, perturbations: Dict[str, Any] | None = None, verbose: bool = True) → CalibrationResults[source]: Run ABC-SMC calibration.

run_top_fraction(top_fraction: float = 0.05, Nsim: int = 100, verbose: bool = True) → CalibrationResults[source]: Run ABC top fraction selection.

epydemix.calibration.calibration_results module

class epydemix.calibration.calibration_results.CalibrationResults(calibration_strategy: str | None = None, posterior_distributions: Dict[int, ~pandas.core.frame.DataFrame]=<factory>, selected_trajectories: Dict[int, ~typing.List[~typing.Any]]=<factory>, observed_data: Any | None = None, priors: Dict[str, ~typing.Any]=<factory>, calibration_params: Dict[str, ~typing.Any]=<factory>, distances: Dict[int, ~typing.List[~typing.Any]]=<factory>, weights: Dict[int, ~typing.List[~typing.Any]]=<factory>, projections: Dict[str, ~typing.List[~typing.Any]]=<factory>, projection_parameters: Dict[str, ~pandas.core.frame.DataFrame]=<factory>)[source]

Bases: object

Class to store and manage the results of a calibration process.

calibration_strategy

The strategy used for calibration

Type:: str | None

posterior_distributions

Dictionary of posterior distributions per generation

Type:: Dict[int, pandas.core.frame.DataFrame]

selected_trajectories

Dictionary of selected trajectories per generation

Type:: Dict[int, List[Any]]

observed_data

Observed data used for calibration

Type:: Any | None

priors

Dictionary of prior distributions for parameters

Type:: Dict[str, Any]

calibration_params

Dictionary of parameters used in calibration

Type:: Dict[str, Any]

distances

Dictionary of distances per generation

Type:: Dict[int, List[Any]]

weights

Dictionary of weights per generation

Type:: Dict[int, List[Any]]

projections

Dictionary of projections

Type:: Dict[str, List[Any]]

projection_parameters

Dictionary of projection parameters

Type:: Dict[str, pandas.core.frame.DataFrame]

calibration_params: Dict[str, Any]

calibration_strategy: str | None = None

distances: Dict[int, List[Any]]

get_calibration_quantiles(dates: List[date] | None = None, quantiles: List[float] = [0.05, 0.5, 0.95], generation: int | None = None, variables: List[str] | None = None, ignore_nan: bool = False) → DataFrame[source]

Compute quantiles from calibration results.

Parameters:

dates – Optional list of dates for the output
quantiles – List of quantile values to compute (default: [0.05, 0.5, 0.95])
generation – Optional generation number to use (default: latest generation)
variables – Optional list of variables to include
ignore_nan – If True, use np.nanquantile to ignore NaN values. Defaults to False.

get_calibration_trajectories(generation: int | None = None, variables: List[str] | None = None) → Dict[str, ndarray][source]

Get stacked trajectories from calibration results.

Parameters:

generation – The generation to get trajectories from. If None, uses the last generation.
variables – Optional list of variable names to include. If None, all variables are included.

get_distances(generation: int | None = None) → List[Any][source]: Gets the distances for a specific generation.

get_posterior_distribution(generation: int | None = None) → DataFrame[source]: Gets the posterior distribution DataFrame for a specific generation.

get_projection_quantiles(dates: List[date] | None = None, quantiles: List[float] = [0.05, 0.5, 0.95], scenario_id: str = 'baseline', variables: List[str] | None = None, ignore_nan: bool = False) → DataFrame[source]

Compute quantiles from projection results.

Parameters:

dates – Optional list of dates for the output
quantiles – List of quantile values to compute (default: [0.05, 0.5, 0.95])
scenario_id – Scenario identifier (default: “baseline”)
variables – Optional list of variables to include
ignore_nan – If True, use np.nanquantile to ignore NaN values. Defaults to False.

get_projection_trajectories(scenario_id: str = 'baseline', variables: List[str] | None = None) → Dict[str, ndarray][source]

Get stacked trajectories from projection results.

Parameters:

scenario_id – The scenario identifier to get projections for.
variables – Optional list of variable names to include. If None, all variables are included.

get_selected_trajectories(generation: int | None = None) → List[Any][source]: Gets the selected trajectories for a specific generation.

get_weights(generation: int | None = None) → List[Any][source]: Gets the weights for a specific generation.

observed_data: Any | None = None

posterior_distributions: Dict[int, DataFrame]

priors: Dict[str, Any]

projection_parameters: Dict[str, DataFrame]

projections: Dict[str, List[Any]]

selected_trajectories: Dict[int, List[Any]]

weights: Dict[int, List[Any]]

epydemix.calibration.metrics module

epydemix.calibration.metrics.ae(data: Dict, simulation: Dict) → ndarray[source]

Computes the Absolute Error (AE) between the observed data and the simulated data.

Parameters:

data (Dict) – A Dictionary containing the observed data with a key “data” pointing to an array of observations.
simulation (Dict) – A Dictionary containing the simulated data with a key “data” pointing to an array of simulated values.

Returns:

An array of absolute errors between the observed and simulated data.

Return type:

np.ndarray

epydemix.calibration.metrics.mae(data: Dict, simulation: Dict) → float[source]

Computes the Mean Absolute Error (MAE) between the observed data and the simulated data.

Parameters:

data (Dict) – A Dictionary containing the observed data with a key “data” pointing to an array of observations.
simulation (Dict) – A Dictionary containing the simulated data with a key “data” pointing to an array of simulated values.

Returns:

The MAE value indicating the average of the absolute errors between the observed and simulated data.

Return type:

float

epydemix.calibration.metrics.mape(data: Dict, simulation: Dict) → float[source]

Computes the Mean Absolute Percentage Error (MAPE) between the observed data and the simulated data.

Parameters:

data (Dict) – A Dictionary containing the observed data with a key “data” pointing to an array of observations.
simulation (Dict) – A Dictionary containing the simulated data with a key “data” pointing to an array of simulated values.

The percentage error is undefined where the observed value is zero, so those points are excluded from the average and a warning is issued. Use wmape or mae if the zeros matter.

Returns:: The MAPE value indicating the average of the absolute percentage errors between the observed and simulated data.
Return type:: float
Raises:: ValueError – If all observed values are zero.

epydemix.calibration.metrics.rmse(data: Dict, simulation: Dict) → float[source]

Computes the Root Mean Square Error (RMSE) between the observed data and the simulated data.

Parameters:

data (Dict) – A Dictionary containing the observed data with a key “data” pointing to an array of observations.
simulation (Dict) – A Dictionary containing the simulated data with a key “data” pointing to an array of simulated values.

Returns:

The RMSE value indicating the average magnitude of the error between the observed and simulated data.

Return type:

float

epydemix.calibration.metrics.validate_data(data: Dict, simulation: Dict, shape_check: bool = True) → Tuple[ndarray, ndarray][source]

Validates that the input Dictionaries contain the required key ‘data’ and that the shapes of the data arrays match.

Parameters:

data (Dict) – A Dictionary containing the observed data with a key “data” pointing to an array of observations.
simulation (Dict) – A Dictionary containing the simulated data with a key “data” pointing to an array of simulated values.

Returns:

A tuple containing two NumPy arrays: (observed_data, simulated_data).

Return type:

tuple[np.ndarray, np.ndarray]

Raises:

ValueError – If the required key ‘data’ is missing in either Dictionary or if the shapes of the data arrays do not match.

epydemix.calibration.metrics.wmape(data: Dict, simulation: Dict) → float[source]

Computes the Weighted Mean Absolute Percentage Error (wMAPE) between the observed data and the simulated data.

Parameters:

data (Dict) – A Dictionary containing the observed data with a key “data” pointing to an array of observations.
simulation (Dict) – A Dictionary containing the simulated data with a key “data” pointing to an array of simulated values.

Returns:

The wMAPE value indicating the weighted average of the absolute percentage errors between the observed and simulated data.

Return type:

float

Module contents

class epydemix.calibration.ABCSampler(simulation_function: ~typing.Callable, priors: ~typing.Dict[str, ~typing.Any], parameters: ~typing.Dict[str, ~typing.Any], observed_data: ~typing.Any, distance_function: ~typing.Callable = <function rmse>, rng: ~typing.Any | None = None)[source]