Calibrator
black_it.calibrator.Calibrator
The class used to perform a calibration.
Source code in black_it/calibrator.py
class Calibrator: # pylint: disable=too-many-instance-attributes
"""The class used to perform a calibration."""
STATE_VERSION = 0
def __init__( # pylint: disable=too-many-arguments
self,
samplers: List[BaseSampler],
loss_function: BaseLoss,
real_data: NDArray[np.float64],
model: Callable,
parameters_bounds: Union[NDArray[np.float64], List[List[float]]],
parameters_precision: Union[NDArray[np.float64], List[float]],
ensemble_size: int,
sim_length: Optional[int] = None,
convergence_precision: Optional[int] = None,
verbose: bool = True,
saving_folder: Optional[str] = None,
random_state: Optional[int] = None,
n_jobs: Optional[int] = None,
):
"""
Initialize the Calibrator object.
It must be initialized with details on the parameters to explore,
on the model to calibrate, on the samplers and on the loss function to use.
Args:
samplers: list of methods to be used in the calibration procedure
loss_function: a loss function which evaluates the similarity between simulated and real datasets
real_data: an array containing the real time series
model: a model with free parameters to be calibrated
parameters_bounds: the bounds of the parameter space
parameters_precision: the precisions to be used for the discretization of the parameters
ensemble_size: number of repetitions to be run for each set of parameters to decrease statistical
fluctuations. For deterministic models this should be set to 1.
sim_length: number of periods to simulate the model for, by default this is equal to the length of the
real time series.
convergence_precision: number of significant digits to consider in the convergence check. The check is
not performed if this is set to 'None'.
verbose: whether to print calibration updates
saving_folder: the name of the folder where data should be saved and/or retrieved
random_state: random state of the calibrator, used for model simulations and to initialise the samplers
n_jobs: the maximum number of concurrently running jobs. For more details, see the
[joblib.Parallel documentation](https://joblib.readthedocs.io/en/latest/generated/joblib.Parallel.html).
"""
self.samplers = samplers
self.loss_function = loss_function
self.model = model
self.random_state = random_state
self.real_data = real_data
self.ensemble_size = ensemble_size
if sim_length is None:
self.N = self.real_data.shape[0]
else:
if sim_length != self.real_data.shape[0]:
warnings.warn(
"The length of real time series is different from the simulation length, "
f"got {self.real_data.shape[0]} and {sim_length}. This may or may not be a problem depending "
"on the loss function used.",
RuntimeWarning,
)
self.N = sim_length
self.D = self.real_data.shape[1]
self.verbose = verbose
self.convergence_precision = (
self._validate_convergence_precision(convergence_precision)
if convergence_precision is not None
else None
)
self.saving_folder = saving_folder
# Initialize search grid
self.param_grid = SearchSpace(parameters_bounds, parameters_precision, verbose)
# initialize arrays
self.params_samp = np.zeros((0, self.param_grid.dims))
self.losses_samp = np.zeros(0)
self.batch_num_samp = np.zeros(0, dtype=int)
self.method_samp = np.zeros(0, dtype=int)
self.series_samp = np.zeros((0, self.ensemble_size, self.N, self.D))
# initialize variables before calibration
self.n_sampled_params = 0
self.current_batch_index = 0
# set number of processes for parallel evaluation of model
self.n_jobs = n_jobs if n_jobs is not None else multiprocessing.cpu_count()
print(
f"Selecting {self.n_jobs} processes for the parallel evaluation of the model"
)
self.samplers_id_table = self._construct_samplers_id_table(samplers)
@property
def random_state(self) -> Optional[int]:
"""Get the random state."""
return self._random_state
@random_state.setter
def random_state(self, random_state: Optional[int]) -> None:
"""Set the random state."""
self._random_state = random_state
self._random_generator = default_rng(self.random_state)
@property
def random_generator(self) -> np.random.Generator:
"""Get the random generator."""
return self._random_generator
def _get_random_seed(self) -> int:
"""Get new random seed from the current random generator."""
return get_random_seed(self._random_generator)
def _set_samplers_seeds(self) -> None:
"""Set the calibration seed."""
for sampler in self.samplers:
sampler.random_state = self._get_random_seed()
@staticmethod
def _construct_samplers_id_table(samplers: List[BaseSampler]) -> Dict[str, int]:
"""
Construct the samplers-by-id table.
Given the list (built-in or user-defined) of samplers a calibration
session is going to use, return a map from the sampler human-readable
name to a numeric id (starting from 0).
Different calibration sessions may result in different conversion
tables.
Args:
samplers: the list of samplers of the calibrator
Returns:
A dict that maps from the given sampler names to unique ids.
"""
samplers_id_table = {}
sampler_id = 0
for sampler in samplers:
sampler_name = type(sampler).__name__
if sampler_name in samplers_id_table:
continue
samplers_id_table[sampler_name] = sampler_id
sampler_id = sampler_id + 1
return samplers_id_table
def set_samplers(self, samplers: List[BaseSampler]) -> None:
"""Set the samplers list of the calibrator.
This method overwrites the samplers of a calibrator object with a custom list of samplers.
Args:
samplers: a list of samplers
"""
# overwrite the list of samplers
self.samplers = samplers
# update the samplers_id_table with the new samplers, only if necessary
sampler_id = max(self.samplers_id_table.values()) + 1
for sampler in samplers:
sampler_name = type(sampler).__name__
if sampler_name in self.samplers_id_table:
continue
self.samplers_id_table[sampler_name] = sampler_id
sampler_id = sampler_id + 1
@classmethod
def restore_from_checkpoint( # pylint: disable=too-many-locals
cls, checkpoint_path: str, model: Callable
) -> "Calibrator":
"""
Return an instantiated class from a database file and a model simulator.
Args:
checkpoint_path: the name of the database file to read from
model: the model to calibrate. It must be equal to the one already calibrated
Returns:
An initialised Calibrator object.
"""
(
parameters_bounds,
parameters_precision,
real_data,
ensemble_size,
N,
_D,
convergence_precision,
verbose,
saving_file,
random_state,
random_generator_state,
model_name,
samplers,
loss_function,
current_batch_index,
n_sampled_params,
n_jobs,
params_samp,
losses_samp,
series_samp,
batch_num_samp,
method_samp,
) = load_calibrator_state(checkpoint_path, cls.STATE_VERSION)
_assert(
model_name == model.__name__,
(
"Error: the model provided appears to be different from the one present "
"in the database"
),
)
calibrator = cls(
samplers,
loss_function,
real_data,
model,
parameters_bounds,
parameters_precision,
ensemble_size,
N,
convergence_precision,
verbose,
saving_file,
random_state,
n_jobs,
)
calibrator.current_batch_index = current_batch_index
calibrator.n_sampled_params = n_sampled_params
calibrator.params_samp = params_samp
calibrator.losses_samp = losses_samp
calibrator.series_samp = series_samp
calibrator.batch_num_samp = batch_num_samp
calibrator.method_samp = method_samp
# reset the random number generator state
calibrator.random_generator.bit_generator.state = random_generator_state
return calibrator
def simulate_model(self, params: NDArray) -> NDArray:
"""
Simulate the model.
This method calls the model simulator in parallel on a given set of parameter values, a number of repeated
evaluations are performed for each parameter to average out random fluctuations.
Args:
params: the array of parameters for which the model should be evaluated
# noqa
Returns:
simulated_data: an array of dimensions (batch_size, ensemble_size, N, D) containing all
simulated time series
"""
rep_params = np.repeat(params, self.ensemble_size, axis=0)
simulated_data_list = Parallel(n_jobs=self.n_jobs)(
delayed(self.model)(param, self.N, self._get_random_seed())
for i, param in enumerate(rep_params)
)
simulated_data = np.array(simulated_data_list)
simulated_data = np.reshape(
simulated_data, (params.shape[0], self.ensemble_size, self.N, self.D)
)
return simulated_data
def calibrate(self, n_batches: int) -> Tuple[NDArray, NDArray]:
"""
Run calibration for n batches.
Args:
n_batches (int): number of 'batches' to be executed. Each batch runs over all methods
Returns:
The sampled parameters and the corresponding sampled losses.
Both arrays are sorted by increasing loss values
"""
if self.current_batch_index == 0:
# we only set the samplers' random state at the start of a calibration
self._set_samplers_seeds()
for _ in range(n_batches):
print()
print(f"BATCH NUMBER: {self.current_batch_index + 1}")
print(f"PARAMS SAMPLED: {self.n_sampled_params}")
for method in self.samplers:
t_start = time.time()
print()
print(f"METHOD: {type(method).__name__}")
# get new params from a specific sampler
new_params = method.sample(
self.param_grid,
self.params_samp,
self.losses_samp,
)
t_eval = time.time()
# simulate an ensemble of models for different parameters
new_simulated_data = self.simulate_model(new_params)
new_losses = []
for sim_data_ensemble in new_simulated_data:
new_loss = self.loss_function.compute_loss(
sim_data_ensemble, self.real_data
)
new_losses.append(new_loss)
# update arrays
self.params_samp = np.vstack((self.params_samp, new_params))
self.losses_samp = np.hstack((self.losses_samp, new_losses))
self.series_samp = np.vstack((self.series_samp, new_simulated_data))
self.batch_num_samp = np.hstack(
(
self.batch_num_samp,
[self.current_batch_index] * method.batch_size,
)
)
self.method_samp = np.hstack(
(
self.method_samp,
[self.samplers_id_table[type(method).__name__]]
* method.batch_size,
)
)
# logging
t_end = time.time()
if self.verbose:
min_dist_new_points = np.round(np.min(new_losses), 2)
avg_dist_new_points = np.round(np.average(new_losses), 2)
avg_dist_existing_points = np.round(np.average(self.losses_samp), 2)
elapsed_tot = np.round(t_end - t_start, 1)
elapsed_eval = np.round(t_end - t_eval, 1)
print(
textwrap.dedent(
f"""\
----> sim exec elapsed time: {elapsed_eval}s
----> min loss new params: {min_dist_new_points}
----> avg loss new params: {avg_dist_new_points}
----> avg loss exist params: {avg_dist_existing_points}
----> curr min loss: {np.min(self.losses_samp)}
====> total elapsed time: {elapsed_tot}s
"""
),
end="",
)
# update count of number of params sampled
self.n_sampled_params = self.n_sampled_params + len(new_params)
self.current_batch_index += 1
# check convergence for early termination
if self.convergence_precision is not None:
converged = self.check_convergence(
self.losses_samp, self.n_sampled_params, self.convergence_precision
)
if converged and self.verbose:
print("\nCONVERGENCE CHECK:")
print("Achieved convergence loss, stopping search.")
break
if self.saving_folder is not None:
self.create_checkpoint(self.saving_folder)
idx = np.argsort(self.losses_samp)
return self.params_samp[idx], self.losses_samp[idx]
@staticmethod
def check_convergence(
losses_samp: NDArray, n_sampled_params: int, convergence_precision: int
) -> bool:
"""
Check convergence of the calibration.
Args:
losses_samp: the sampled losses
n_sampled_params: the number of sampled params
convergence_precision: the required convergence precision.
Returns:
True if the calibration converged, False otherwise.
"""
converged = (
np.round(np.min(losses_samp[:n_sampled_params]), convergence_precision)
== 0.0
)
return converged
def create_checkpoint(self, file_name: Union[str, os.PathLike]) -> None:
"""
Save the current state of the object.
Args:
file_name: the name of the folder where the data will be saved
"""
checkpoint_path: str = os.path.join(os.path.realpath(file_name))
t_start = time.time()
model_name = self.model.__name__
save_calibrator_state(
checkpoint_path,
self.param_grid.parameters_bounds,
self.param_grid.parameters_precision,
self.real_data,
self.ensemble_size,
self.N,
self.D,
self.convergence_precision,
self.verbose,
self.saving_folder,
self.random_state,
self.random_generator.bit_generator.state,
model_name,
self.samplers,
self.loss_function,
self.current_batch_index,
self.n_sampled_params,
self.n_jobs,
self.params_samp,
self.losses_samp,
self.series_samp,
self.batch_num_samp,
self.method_samp,
)
t_end = time.time()
elapsed = np.round(t_end - t_start, 1)
print(f"Checkpoint saved in {elapsed}s")
@staticmethod
def _validate_convergence_precision(convergence_precision: int) -> int:
"""Validate convergence precision input."""
_assert(
convergence_precision >= 0,
f"convergence precision must be an integer greater than 0, got {convergence_precision}",
exception_class=ValueError,
)
return convergence_precision
random_generator: Generator
property
readonly
Get the random generator.
random_state: Optional[int]
property
writable
Get the random state.
__init__(self, samplers, loss_function, real_data, model, parameters_bounds, parameters_precision, ensemble_size, sim_length=None, convergence_precision=None, verbose=True, saving_folder=None, random_state=None, n_jobs=None)
special
Initialize the Calibrator object.
It must be initialized with details on the parameters to explore, on the model to calibrate, on the samplers and on the loss function to use.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
samplers |
List[black_it.samplers.base.BaseSampler] |
list of methods to be used in the calibration procedure |
required |
loss_function |
BaseLoss |
a loss function which evaluates the similarity between simulated and real datasets |
required |
real_data |
ndarray |
an array containing the real time series |
required |
model |
Callable |
a model with free parameters to be calibrated |
required |
parameters_bounds |
Union[numpy.ndarray[Any, numpy.dtype[numpy.float64]], List[List[float]]] |
the bounds of the parameter space |
required |
parameters_precision |
Union[numpy.ndarray[Any, numpy.dtype[numpy.float64]], List[float]] |
the precisions to be used for the discretization of the parameters |
required |
ensemble_size |
int |
number of repetitions to be run for each set of parameters to decrease statistical fluctuations. For deterministic models this should be set to 1. |
required |
sim_length |
Optional[int] |
number of periods to simulate the model for, by default this is equal to the length of the real time series. |
None |
convergence_precision |
Optional[int] |
number of significant digits to consider in the convergence check. The check is not performed if this is set to 'None'. |
None |
verbose |
bool |
whether to print calibration updates |
True |
saving_folder |
Optional[str] |
the name of the folder where data should be saved and/or retrieved |
None |
random_state |
Optional[int] |
random state of the calibrator, used for model simulations and to initialise the samplers |
None |
n_jobs |
Optional[int] |
the maximum number of concurrently running jobs. For more details, see the joblib.Parallel documentation. |
None |
Source code in black_it/calibrator.py
def __init__( # pylint: disable=too-many-arguments
self,
samplers: List[BaseSampler],
loss_function: BaseLoss,
real_data: NDArray[np.float64],
model: Callable,
parameters_bounds: Union[NDArray[np.float64], List[List[float]]],
parameters_precision: Union[NDArray[np.float64], List[float]],
ensemble_size: int,
sim_length: Optional[int] = None,
convergence_precision: Optional[int] = None,
verbose: bool = True,
saving_folder: Optional[str] = None,
random_state: Optional[int] = None,
n_jobs: Optional[int] = None,
):
"""
Initialize the Calibrator object.
It must be initialized with details on the parameters to explore,
on the model to calibrate, on the samplers and on the loss function to use.
Args:
samplers: list of methods to be used in the calibration procedure
loss_function: a loss function which evaluates the similarity between simulated and real datasets
real_data: an array containing the real time series
model: a model with free parameters to be calibrated
parameters_bounds: the bounds of the parameter space
parameters_precision: the precisions to be used for the discretization of the parameters
ensemble_size: number of repetitions to be run for each set of parameters to decrease statistical
fluctuations. For deterministic models this should be set to 1.
sim_length: number of periods to simulate the model for, by default this is equal to the length of the
real time series.
convergence_precision: number of significant digits to consider in the convergence check. The check is
not performed if this is set to 'None'.
verbose: whether to print calibration updates
saving_folder: the name of the folder where data should be saved and/or retrieved
random_state: random state of the calibrator, used for model simulations and to initialise the samplers
n_jobs: the maximum number of concurrently running jobs. For more details, see the
[joblib.Parallel documentation](https://joblib.readthedocs.io/en/latest/generated/joblib.Parallel.html).
"""
self.samplers = samplers
self.loss_function = loss_function
self.model = model
self.random_state = random_state
self.real_data = real_data
self.ensemble_size = ensemble_size
if sim_length is None:
self.N = self.real_data.shape[0]
else:
if sim_length != self.real_data.shape[0]:
warnings.warn(
"The length of real time series is different from the simulation length, "
f"got {self.real_data.shape[0]} and {sim_length}. This may or may not be a problem depending "
"on the loss function used.",
RuntimeWarning,
)
self.N = sim_length
self.D = self.real_data.shape[1]
self.verbose = verbose
self.convergence_precision = (
self._validate_convergence_precision(convergence_precision)
if convergence_precision is not None
else None
)
self.saving_folder = saving_folder
# Initialize search grid
self.param_grid = SearchSpace(parameters_bounds, parameters_precision, verbose)
# initialize arrays
self.params_samp = np.zeros((0, self.param_grid.dims))
self.losses_samp = np.zeros(0)
self.batch_num_samp = np.zeros(0, dtype=int)
self.method_samp = np.zeros(0, dtype=int)
self.series_samp = np.zeros((0, self.ensemble_size, self.N, self.D))
# initialize variables before calibration
self.n_sampled_params = 0
self.current_batch_index = 0
# set number of processes for parallel evaluation of model
self.n_jobs = n_jobs if n_jobs is not None else multiprocessing.cpu_count()
print(
f"Selecting {self.n_jobs} processes for the parallel evaluation of the model"
)
self.samplers_id_table = self._construct_samplers_id_table(samplers)
calibrate(self, n_batches)
Run calibration for n batches.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n_batches |
int |
number of 'batches' to be executed. Each batch runs over all methods |
required |
Returns:
| Type | Description |
|---|---|
Tuple[numpy.ndarray[Any, numpy.dtype[+ScalarType]], numpy.ndarray[Any, numpy.dtype[+ScalarType]]] |
The sampled parameters and the corresponding sampled losses. Both arrays are sorted by increasing loss values |
Source code in black_it/calibrator.py
def calibrate(self, n_batches: int) -> Tuple[NDArray, NDArray]:
"""
Run calibration for n batches.
Args:
n_batches (int): number of 'batches' to be executed. Each batch runs over all methods
Returns:
The sampled parameters and the corresponding sampled losses.
Both arrays are sorted by increasing loss values
"""
if self.current_batch_index == 0:
# we only set the samplers' random state at the start of a calibration
self._set_samplers_seeds()
for _ in range(n_batches):
print()
print(f"BATCH NUMBER: {self.current_batch_index + 1}")
print(f"PARAMS SAMPLED: {self.n_sampled_params}")
for method in self.samplers:
t_start = time.time()
print()
print(f"METHOD: {type(method).__name__}")
# get new params from a specific sampler
new_params = method.sample(
self.param_grid,
self.params_samp,
self.losses_samp,
)
t_eval = time.time()
# simulate an ensemble of models for different parameters
new_simulated_data = self.simulate_model(new_params)
new_losses = []
for sim_data_ensemble in new_simulated_data:
new_loss = self.loss_function.compute_loss(
sim_data_ensemble, self.real_data
)
new_losses.append(new_loss)
# update arrays
self.params_samp = np.vstack((self.params_samp, new_params))
self.losses_samp = np.hstack((self.losses_samp, new_losses))
self.series_samp = np.vstack((self.series_samp, new_simulated_data))
self.batch_num_samp = np.hstack(
(
self.batch_num_samp,
[self.current_batch_index] * method.batch_size,
)
)
self.method_samp = np.hstack(
(
self.method_samp,
[self.samplers_id_table[type(method).__name__]]
* method.batch_size,
)
)
# logging
t_end = time.time()
if self.verbose:
min_dist_new_points = np.round(np.min(new_losses), 2)
avg_dist_new_points = np.round(np.average(new_losses), 2)
avg_dist_existing_points = np.round(np.average(self.losses_samp), 2)
elapsed_tot = np.round(t_end - t_start, 1)
elapsed_eval = np.round(t_end - t_eval, 1)
print(
textwrap.dedent(
f"""\
----> sim exec elapsed time: {elapsed_eval}s
----> min loss new params: {min_dist_new_points}
----> avg loss new params: {avg_dist_new_points}
----> avg loss exist params: {avg_dist_existing_points}
----> curr min loss: {np.min(self.losses_samp)}
====> total elapsed time: {elapsed_tot}s
"""
),
end="",
)
# update count of number of params sampled
self.n_sampled_params = self.n_sampled_params + len(new_params)
self.current_batch_index += 1
# check convergence for early termination
if self.convergence_precision is not None:
converged = self.check_convergence(
self.losses_samp, self.n_sampled_params, self.convergence_precision
)
if converged and self.verbose:
print("\nCONVERGENCE CHECK:")
print("Achieved convergence loss, stopping search.")
break
if self.saving_folder is not None:
self.create_checkpoint(self.saving_folder)
idx = np.argsort(self.losses_samp)
return self.params_samp[idx], self.losses_samp[idx]
check_convergence(losses_samp, n_sampled_params, convergence_precision)
staticmethod
Check convergence of the calibration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
losses_samp |
ndarray |
the sampled losses |
required |
n_sampled_params |
int |
the number of sampled params |
required |
convergence_precision |
int |
the required convergence precision. |
required |
Returns:
| Type | Description |
|---|---|
bool |
True if the calibration converged, False otherwise. |
Source code in black_it/calibrator.py
@staticmethod
def check_convergence(
losses_samp: NDArray, n_sampled_params: int, convergence_precision: int
) -> bool:
"""
Check convergence of the calibration.
Args:
losses_samp: the sampled losses
n_sampled_params: the number of sampled params
convergence_precision: the required convergence precision.
Returns:
True if the calibration converged, False otherwise.
"""
converged = (
np.round(np.min(losses_samp[:n_sampled_params]), convergence_precision)
== 0.0
)
return converged
create_checkpoint(self, file_name)
Save the current state of the object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_name |
Union[str, os.PathLike] |
the name of the folder where the data will be saved |
required |
Source code in black_it/calibrator.py
def create_checkpoint(self, file_name: Union[str, os.PathLike]) -> None:
"""
Save the current state of the object.
Args:
file_name: the name of the folder where the data will be saved
"""
checkpoint_path: str = os.path.join(os.path.realpath(file_name))
t_start = time.time()
model_name = self.model.__name__
save_calibrator_state(
checkpoint_path,
self.param_grid.parameters_bounds,
self.param_grid.parameters_precision,
self.real_data,
self.ensemble_size,
self.N,
self.D,
self.convergence_precision,
self.verbose,
self.saving_folder,
self.random_state,
self.random_generator.bit_generator.state,
model_name,
self.samplers,
self.loss_function,
self.current_batch_index,
self.n_sampled_params,
self.n_jobs,
self.params_samp,
self.losses_samp,
self.series_samp,
self.batch_num_samp,
self.method_samp,
)
t_end = time.time()
elapsed = np.round(t_end - t_start, 1)
print(f"Checkpoint saved in {elapsed}s")
restore_from_checkpoint(checkpoint_path, model)
classmethod
Return an instantiated class from a database file and a model simulator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
checkpoint_path |
str |
the name of the database file to read from |
required |
model |
Callable |
the model to calibrate. It must be equal to the one already calibrated |
required |
Returns:
| Type | Description |
|---|---|
Calibrator |
An initialised Calibrator object. |
Source code in black_it/calibrator.py
@classmethod
def restore_from_checkpoint( # pylint: disable=too-many-locals
cls, checkpoint_path: str, model: Callable
) -> "Calibrator":
"""
Return an instantiated class from a database file and a model simulator.
Args:
checkpoint_path: the name of the database file to read from
model: the model to calibrate. It must be equal to the one already calibrated
Returns:
An initialised Calibrator object.
"""
(
parameters_bounds,
parameters_precision,
real_data,
ensemble_size,
N,
_D,
convergence_precision,
verbose,
saving_file,
random_state,
random_generator_state,
model_name,
samplers,
loss_function,
current_batch_index,
n_sampled_params,
n_jobs,
params_samp,
losses_samp,
series_samp,
batch_num_samp,
method_samp,
) = load_calibrator_state(checkpoint_path, cls.STATE_VERSION)
_assert(
model_name == model.__name__,
(
"Error: the model provided appears to be different from the one present "
"in the database"
),
)
calibrator = cls(
samplers,
loss_function,
real_data,
model,
parameters_bounds,
parameters_precision,
ensemble_size,
N,
convergence_precision,
verbose,
saving_file,
random_state,
n_jobs,
)
calibrator.current_batch_index = current_batch_index
calibrator.n_sampled_params = n_sampled_params
calibrator.params_samp = params_samp
calibrator.losses_samp = losses_samp
calibrator.series_samp = series_samp
calibrator.batch_num_samp = batch_num_samp
calibrator.method_samp = method_samp
# reset the random number generator state
calibrator.random_generator.bit_generator.state = random_generator_state
return calibrator
set_samplers(self, samplers)
Set the samplers list of the calibrator.
This method overwrites the samplers of a calibrator object with a custom list of samplers.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
samplers |
List[black_it.samplers.base.BaseSampler] |
a list of samplers |
required |
Source code in black_it/calibrator.py
def set_samplers(self, samplers: List[BaseSampler]) -> None:
"""Set the samplers list of the calibrator.
This method overwrites the samplers of a calibrator object with a custom list of samplers.
Args:
samplers: a list of samplers
"""
# overwrite the list of samplers
self.samplers = samplers
# update the samplers_id_table with the new samplers, only if necessary
sampler_id = max(self.samplers_id_table.values()) + 1
for sampler in samplers:
sampler_name = type(sampler).__name__
if sampler_name in self.samplers_id_table:
continue
self.samplers_id_table[sampler_name] = sampler_id
sampler_id = sampler_id + 1
simulate_model(self, params)
Simulate the model.
This method calls the model simulator in parallel on a given set of parameter values, a number of repeated evaluations are performed for each parameter to average out random fluctuations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
params |
ndarray |
the array of parameters for which the model should be evaluated |
required |
noqa
Returns:
| Type | Description |
|---|---|
simulated_data |
an array of dimensions (batch_size, ensemble_size, N, D) containing all simulated time series |
Source code in black_it/calibrator.py
def simulate_model(self, params: NDArray) -> NDArray:
"""
Simulate the model.
This method calls the model simulator in parallel on a given set of parameter values, a number of repeated
evaluations are performed for each parameter to average out random fluctuations.
Args:
params: the array of parameters for which the model should be evaluated
# noqa
Returns:
simulated_data: an array of dimensions (batch_size, ensemble_size, N, D) containing all
simulated time series
"""
rep_params = np.repeat(params, self.ensemble_size, axis=0)
simulated_data_list = Parallel(n_jobs=self.n_jobs)(
delayed(self.model)(param, self.N, self._get_random_seed())
for i, param in enumerate(rep_params)
)
simulated_data = np.array(simulated_data_list)
simulated_data = np.reshape(
simulated_data, (params.shape[0], self.ensemble_size, self.N, self.D)
)
return simulated_data