estim8.estimator

This module implements the Estimator class for parameter estimation and uncertainty quantification.

class estim8.estimator.Estimator(model: ~estim8.models.Estim8Model, bounds: dict[str, ~typing.List[float]], data: ~estim8.datatypes.Experiment | ~typing.Dict[str, ~estim8.datatypes.Experiment] | ~pandas.core.frame.DataFrame | ~typing.Dict[str, ~pandas.core.frame.DataFrame], t: ~typing.List[float] | None = None, metric: ~typing.Literal['SS', 'WSS', 'negLL'] = 'SS', error_model: ~estim8.error_models.BaseErrorModel = <estim8.error_models.LinearErrorModel object>, parameter_mapping: list[~estim8.utils.ModelHelpers.ParameterMapper] | ~estim8.utils.ModelHelpers.ParameterMapping | None = None)

Bases: object

Parameter estimation and uncertainty quantification manager.

This class provides a comprehensive interface for: - Parameter estimation using various optimization algorithms - Monte Carlo sampling for uncertainty quantification - Profile likelihood analysis - Handling multiple experimental replicates - Distributed computation through federated workers

Parameters:

modelmodels.Estim8Model

Model to fit to experimental data

boundsdict[str, List[float]]

Parameter bounds formatted as {parameter_name: [lower_bound, upper_bound]}

dataUnion[Experiment, Dict[str, Experiment], pd.DataFrame, Dict[str, pd.DataFrame]]

Experimental measurements to fit model against. Can be: - Single experiment as Experiment or DataFrame - Multiple replicates as Dict[str, Experiment] or Dict[str, DataFrame]

tList[float], optional

Simulation time settings as [t_start, t_end, stepsize]. If None, derived from data

metricLiteral[“SS”, “WSS”, “negLL”], optional

Loss function for optimization: - “SS”: Sum of squares - “WSS”: Weighted sum of squares - “negLL”: Negative log-likelihood Default is “SS”

error_modelerror_models.BaseErrorModel, optional

Error model for experimental data. Only used when data is DataFrame. Default is LinearErrorModel()

parameter_mappingUnion[List[utils.ModelHelpers.ParameterMapper], utils.ModelHelpers.ParameterMapping], optional

Maps parameters between replicates. Either: - List of ParameterMapper objects - Complete ParameterMapping object Default is None (no mapping)

Attributes:

replicate_IDsList[str]

IDs of experimental replicates

modelmodels.Estim8Model

Reference to model being fit

boundsdict

Parameter bounds

metricstr

Current loss function

error_modelerror_models.BaseErrorModel

Current error model

parameter_mappingutils.ModelHelpers.ParameterMapping

Current parameter mapping

Methods

estimate(method, max_iter, optimizer_kwargs={}, n_jobs=1, federated_workers=False)	Main parameter estimation method
mc_sampling(method, n_jobs, max_iter, n_samples=100, mcs_at_once=1)	Monte Carlo sampling for uncertainty quantification
profile_likelihood(p_opt, method, max_iter, n_points=3, dp_rel=0.1)	Profile likelihood analysis for parameter confidence intervals

Notes

• Supports both single-core and parallel computation

• Can use federated worker architecture for distributed computing, enabling parallel simulation of model replicates within a single objective function call

• Handles replicate-specific parameters through parameter mapping

• Provides multiple optimization algorithms through scipy, scikit-optimize and pygmo

Examples

Basic parameter estimation:

>>> estimator = Estimator(model=model, bounds=bounds, data=data)
>>> estimates, info = estimator.estimate(method='de', max_iter=1000)

Parallelized parameter estimation: >>> estimates, info = estimator.estimate(method=’de’, n_jobs=2, max_iter=1000)

With federated workers:

>>> estimates, info = estimator.estimate(method='de', max_iter=1000, federated_workers=4)

Monte Carlo sampling:

>>> results = estimator.mc_sampling(method='de', n_jobs=4, max_iter=100, n_samples=10)

check_problem_input() → None

Validate estimation problem setup.

Checks: 1. Model observables match data observation mappings 2. Data completeness and consistency

Raises:

KeyError

If observation mappings don’t match model observables

property data

estimate(method: str | List[str], max_iter: int, optimizer_kwargs: dict = {}, n_jobs: int = 1, federated_workers: int | bool = False, worker_kwargs: dict = {})

Estimate the parameters of the model.

Parameters:

methodstr | List[str]

The optimization method(s).

max_iterint

The maximum number of iterations.

optimizer_kwargsdict, optional

Additional keyword arguments for the optimizer, by default {}.

n_jobs: int, optional

The number of parallel jobs, by default 1.

federated_workersint | bool, optional

The number of federated workers, by default False.

worker_kwargsdict, optional

Additional keyword arguments for federated workers. These are passed to the launch_workers() method (see launch_workers() for details), by default {}.

Returns:

tuple

The optimization result and additional information.

property hosts_and_ports

launch_workers(n_workers: int | None = None, host: str = 'localhost', ports: List[int] | None = None, start_at_port: int = 9500, mc_sampling=False) → None

Launch federated worker processes for distributed computation.

Parameters:

n_workersint, optional

Number of worker processes, by default None

hoststr, optional

Host address for workers, by default “localhost”

portsList[int], optional

Specific ports for workers, by default None

start_at_portint, optional

Starting port number when auto-assigning, by default 9500

mc_samplingbool, optional

Whether workers will be used for Monte Carlo sampling, by default False

Raises:

ValueError

If neither n_workers nor ports are specified

mc_sampling(method: str | List[str], n_jobs: int, max_iter: int, federated_workers: int = 0, optimizer_kwargs: dict = {}, n_samples: int = 100, mcs_at_once: int = 1, worker_kwargs: dict = {})

Perform Monte Carlo sampling for parameter estimation problem.

Parameters:

methodstr | List[str]

The optimization method(s).

n_jobsint

The number of parallel jobs.

max_iterint

The maximum number of iterations.

federated_workersint, optional

The number of federated workers, by default 0.

optimizer_kwargsdict, optional

Additional keyword arguments for the optimizer, by default {}.

n_samplesint, optional

The number of Monte Carlo samples, by default 100.

mcs_at_onceint, optional

The number of Monte Carlo samples to process at once, by default 1.

worker_kwargsdict, optional

Additional keyword arguments for federated workers. These are passed to the launch_workers() method (see launch_workers() for details), by default {}.

Returns:

list

The results of the Monte Carlo sampling.

property metric

objective(theta: array) → float

Calculate total loss function value across all replicates.

Parameters:

thetanp.array

Parameter vector to evaluate

Returns:

float

Total loss function value

objective_for_replicate(parameters: dict, data: Experiment, metric: Literal['SS', 'WSS', 'negLL'] | None = None) → float

Calculate loss function value for a single experimental replicate.

Parameters:

parametersdict

Model parameters as {parameter_name: value}

dataExperiment

Experimental data for this replicate

metricLiteral[“SS”, “WSS”, “negLL”], optional

Loss function type, by default None (uses instance metric)

Returns:

float

Loss function value. Returns np.inf if simulator crashes with RuntimeError, ValueError, ZeroDivisionError, OverflowError or fmpy.fmi1.FMICallException,.

property parameter_mapping

profile_likelihood(p_opt: dict, method: str | List[str], max_iter: int, n_jobs: int = 1, federated_workers: int = 0, optimizer_kwargs: dict = {}, p_at_once=1, max_steps: int | None = None, stepsize: float = 0.02, p_inv: list | None = None, alpha: float = 0.05, worker_kwargs: dict = {})

Calculate the profile likelihood for the estimated parameters.

Parameters:

p_optdict

The optimal parameter values.

methodstr | List[str]

The optimization method(s).

max_iterint

The maximum number of iterations.

n_jobsint, optional

The number of parallel jobs, by default 1.

federated_workersint, optional

The number of federated workers, by default 0.

optimizer_kwargsdict, optional

Additional keyword arguments for the optimizer, by default {}.

p_at_onceint, optional

The number of parameters to profile at once, by default 1.

n_pointsint, optional

The number of points to evaluate, by default 3.

dp_relfloat, optional

The relative parameter variation width, by default 0.1.

p_invlist, optional

The parameters to investigate, by default None.

worker_kwargsdict, optional

Additional keyword arguments for federated workers. These are passed to the launch_workers() method (see launch_workers() for details), by default {}.

Returns:

dict

The profile likelihood results.

Raises:

KeyError

If the parameter names in p_inv or p_opt are not specified in the bounds.

ValueError

If dp_rel is not in the range (0,1).

shutdown_workers()

Shut down the federated workers.

property t

test_workers()

Test the responsiveness of the federated workers.

Raises:

utils.WorkerNotUpError

If any worker is not responsive.

static use_federated_workers(func: Callable)

Decorator for managing federated worker lifecycle.

This decorator handles: 1. Worker initialization 2. Worker testing 3. Function execution 4. Worker cleanup 5. Error handling

Parameters:

funccallable

Function to wrap

Returns:

callable

Wrapped function with worker lifecycle managementy