Calibration#
modelly.calibrate fits parameters to observed data. How to use it is in
Calibration.
Methods#
What every calibration method shares: model, priors, targets, seeds, runs, checkpoints.
A method subclasses Calibration and implements run. The base class
turns parameter samples into batches of runs on the experiment executors,
measures each point’s distance to the targets, and saves the whole calibration,
state included, to a checkpoint that Calibration.load() resumes.
- exception CalibrationError[source]#
A calibration cannot go on (no particle accepted, a checkpoint is unusable…).
- class Calibration(model: type | str, priors: Mapping[str, Prior], targets: Sequence[Target], steps: int = 100, replicates: int = 1, seed: int = 0, store: str | Path | None = None, fixed: Mapping[str, Any] | None = None)[source]#
Base class: fit the priors’ parameters of
modeltotargets.- Parameters:
model – the Model class, or
"package.module:Class"/"path/model.py:Class".priors –
{parameter: Prior}for every parameter to fit.targets – the observed quantities (
Target).steps – steps per run.
replicates – runs per parameter point, averaged before the distance.
seed – the root of every random stream: same seed, same result.
store – the folder the runs are written to (a temporary one by default); a calibration rerun into the same folder reads the runs already there.
fixed – parameters every run takes, not fitted.
Check the definition; nothing runs before
run.- points(samples: Mapping[str, Sequence[Any]]) list[dict[str, Any]][source]#
One parameter dict per sample, the fixed parameters included.
- evaluate(batch: Batch, execution: Execution) tuple[ndarray, dict[str, ndarray]][source]#
Run
batch; the total distance and every target’s summary, per point.
- measure(results: Sequence[Mapping[str, ndarray]]) tuple[ndarray, dict[str, ndarray]][source]#
The total distance and every target’s summary of simulated
results.
- posterior(samples: Mapping[str, Sequence[Any]], weights: ndarray, distances: ndarray, summaries: Mapping[str, ndarray]) Posterior[source]#
The Posterior of weighted samples, with what it needs to predict and plot.
ABC-SMC: approximate Bayesian computation by sequential Monte Carlo.
Generation 0 samples the priors. Each later generation sets its tolerance to a
quantile of the previous generation’s distances, proposes particles by drawing
an ancestor by weight and moving it with the perturbation kernel
(modelly.calibrate.kernels), keeps those whose distance is within the
tolerance, and weights them by prior over proposal density (Toni et al., 2009;
Beaumont et al., 2009).
Proposals are made in batches of particles; a generation keeps the first
particles accepted in proposal order. Batch b of generation g draws
from the stream (PROPOSAL, g, b) and runs with the seeds of (g, b), so
the posterior is the same on every executor and after a resume.
Example
>>> calibration = ABCSMC(
... model=Epidemic,
... priors={"beta": Prior.uniform(0.05, 0.5), "gamma": Prior.log_normal(-2, 0.5)},
... targets=[Target("infected", observed=cases)],
... steps=60,
... seed=2027,
... )
>>> posterior = calibration.run(budget=5000, executor="processes")
- class Accepted(samples: dict[str, list[Any]], distances: ndarray, summaries: dict[str, ndarray])[source]#
The particles of one proposal batch that met the tolerance.
- class Generation(samples: dict[str, list[Any]], weights: ndarray, distances: ndarray, summaries: dict[str, ndarray], tolerance: float, runs: int, acceptance: float)[source]#
One accepted population: samples, weights, distances, summaries and costs.
- class ABCSMC(*args: Any, particles: int = 100, quantile: float = 0.5, min_tolerance: float = 0.0, min_acceptance: float = 0.01, **kwargs: Any)[source]#
ABC-SMC with an adaptive tolerance schedule.
Takes the
Calibrationarguments and:- Parameters:
particles – the population size (and the size of a proposal batch).
quantile – the next tolerance is this quantile of the last distances.
min_tolerance – stop once the tolerance reaches this value.
min_acceptance – stop when a generation accepts a smaller share of its proposals; also bounds the cost of one generation.
Check the settings; nothing runs before
run.- run(budget: int | None = None, generations: int = 5, executor: Any = 'serial', max_parallel: int | None = None, **options: Any) Posterior[source]#
Add generations until one of the stopping rules holds; the last posterior.
Stops after
generationsgenerations in all, before a generation that would take the model runs pastbudget, when the tolerance reachesmin_toleranceor the acceptance rate falls belowmin_acceptance.checkpoint=path(inoptions) saves after every generation. Callingrunagain, afterloador with a largergenerations, goes on from the last generation.
- posterior_of(generation: Generation) Posterior[source]#
The Posterior of
generation, with the run’s diagnostics ininfo.
ABC rejection: sample the priors, run every sample, keep the closest.
The simplest approximate Bayesian computation: budget parameter points are
drawn from the priors (stream (PROPOSAL, REJECTION_KEY) of the seed), run
as one batch on the chosen executor, and kept when their distance is within
tolerance, or, without a tolerance, when they are among the quantile
closest. The accepted samples have equal weights.
Example
>>> calibration = ABCRejection(Epidemic, priors, targets, steps=60, seed=1)
>>> posterior = calibration.run(budget=2000, quantile=0.05)
- class ABCRejection(model: type | str, priors: Mapping[str, Prior], targets: Sequence[Target], steps: int = 100, replicates: int = 1, seed: int = 0, store: str | Path | None = None, fixed: Mapping[str, Any] | None = None)[source]#
ABC rejection sampling over the priors (see the module docstring).
Check the definition; nothing runs before
run.- run(budget: int = 1000, quantile: float | None = 0.1, tolerance: float | None = None, executor: Any = 'serial', **options: Any) Posterior[source]#
Run
budgetprior samples and keep the closest (max_parallelin options).tolerancekeeps every sample within it; otherwisequantilekeeps that share of the samples, the closest first.
The method of simulated moments: match simulated statistics to observed ones.
Each target contributes its observed moments (a series, or the statistics its
summary returns). At a parameter point the replicate mean of the simulated
moments gives the gap g = m(theta) - m_observed (scaled by the square root
of each target’s weight), and the objective is g' W g. W is the
weighting matrix: the identity, the inverse of the moments’ variances
("diagonal") or of their covariance ("optimal"), both estimated from
the replicates at the starting point, or a matrix you give.
The objective is minimised in the priors’ unit box with scipy.optimize
(Nelder-Mead or Powell); every evaluation uses the same replicate seeds, so
the search sees a deterministic function.
Example
>>> fit = SimulatedMoments(Market, priors, targets, steps=200, replicates=8)
>>> fit.run(evaluations=150).best
- class SimulatedMoments(*args: Any, weighting: str | ndarray = 'identity', optimizer: str = 'Nelder-Mead', start: dict[str, float] | None = None, **kwargs: Any)[source]#
Simulated method of moments (see the module docstring).
Takes the
Calibrationarguments and:- Parameters:
weighting –
"identity","diagonal","optimal"or a square matrix.optimizer –
"Nelder-Mead"or"Powell".start – the starting parameters (default: the prior medians).
Check the settings; nothing runs before
run.- moments(result: dict[str, ndarray]) ndarray[source]#
The scaled gaps of every replicate:
array[replicate, moment].
Bayesian optimisation of the targets’ distance: a basic Gaussian-process search.
A built-in, deliberately small optimiser for models too expensive for ABC:
initial points of a Latin hypercube over the priors’ unit box, then, one
point at a time, the candidate with the largest expected improvement under a
Gaussian process (modelly.calibrate.gp) fitted to every evaluation so
far. Candidates are random points of the box from a seeded stream, so a run
is reproducible. It returns the best point found, not a posterior
distribution, and suits a handful of parameters and up to a few hundred
evaluations; no external optimisation library is needed.
Example
>>> search = BayesOpt(Epidemic, priors, targets, steps=60, replicates=4, seed=3)
>>> search.run(evaluations=40).best
- class BayesOpt(*args: Any, **kwargs: Any)[source]#
Gaussian-process Bayesian optimisation of the total distance (see the module docstring).
Check the priors are continuous; nothing runs before
run.
What the optimising methods share: a unit box over the priors and common random numbers.
SimulatedMoments and
BayesOpt search the box spanned by the
priors (their support, or the central 99.8 % of an unbounded one; log-scaled
priors on a log scale). Every evaluation runs its point with the same
replicate seeds (common random numbers), so the objective is a deterministic,
smoother function of the parameters, and each evaluation is a batch of its own
in the store.
- class UnitBox(priors: Mapping[str, Prior])[source]#
The map between the unit cube and the priors’ search intervals.
One interval per prior; choices are refused.
- class PointEstimate(*args: Any, **kwargs: Any)[source]#
Base of the methods that return the best point found (a one-sample Posterior).
Check the priors are continuous; nothing runs before
run.- simulate(units: Sequence[Sequence[float]], execution: Execution) list[dict[str, ndarray]][source]#
Run each unit-cube point as its own batch, with common random numbers.
Perturbation kernels of ABC-SMC: move a particle, and the density of that move.
Continuous parameters move together under a multivariate normal whose
covariance is twice the weighted covariance of the previous population
(Beaumont et al., 2009). A choice keeps its value with probability stay
and otherwise takes one of the other values, each equally likely.
- joint_logprior(priors: Mapping[str, Prior], samples: Mapping[str, Sequence[Any]]) ndarray[source]#
The log prior density of every sample (independent priors).
- class Kernel(priors: Mapping[str, Prior], population: Mapping[str, Sequence[Any]], weights: ndarray, covariance: ndarray, stay: float = 0.7)[source]#
A perturbation kernel fitted to a weighted population of particles.
- classmethod fit(priors: Mapping[str, Prior], samples: Mapping[str, Sequence[Any]], weights: ndarray) Kernel[source]#
The kernel for moving the weighted population
samples.
- perturb(samples: Mapping[str, Sequence[Any]], rng: Generator) dict[str, list[Any]][source]#
Each of
samplesmoved once (they may leave the prior’s support).
- logpdf(new: Mapping[str, Sequence[Any]]) ndarray[source]#
log K(new[i] | population[j])as an array[i, j].
Priors, targets and distances#
Priors: what is believed about each parameter before the data are seen.
Example
>>> import numpy as np
>>> prior = Prior.uniform(0.05, 0.5)
>>> values = prior.sample(np.random.default_rng(1), 3)
>>> bool(((values >= 0.05) & (values <= 0.5)).all())
True
>>> float(prior.logpdf(0.6))
-inf
Continuous priors sample and evaluate through scipy.stats; a choice draws
one of its values. Every draw comes from the generator passed in, so the same
seed gives the same samples.
- class Prior(kind: str, arguments: dict[str, ~typing.Any]=<factory>)[source]#
A prior distribution over one parameter; build it with a class method.
- Variables:
- classmethod uniform(low: float, high: float) Prior[source]#
Every value in
[low, high]equally likely.
- classmethod log_uniform(low: float, high: float) Prior[source]#
Uniform on a log scale over
[low, high](low > 0).
- classmethod log_normal(mu: float, sigma: float) Prior[source]#
A value whose logarithm is normal with mean
muand sdsigma.
- classmethod truncated_normal(mean: float, sd: float, low: float = -inf, high: float = inf) Prior[source]#
A normal distribution cut to
[low, high].
- classmethod choice(values: Sequence[Any], probabilities: Sequence[float] | None = None) Prior[source]#
One of
values, equally likely unlessprobabilitiesare given.
- logpdf(value: Any) float | ndarray[source]#
The log density (log probability for a choice);
-infoff the support.
Targets: what a calibration compares between the model and the data.
A Target names an output the model records every step (an attribute,
a method or a module-level function of the model, the same things an
Experiment output or a DataCollector model reporter reads) and the
observed values it should match:
Target("infected", observed=cases) # the whole series
Target("infected", observed=[412.0], summary=peak) # one statistic
Target("prey", observed=counts, distance=distance.mae, weight=0.5)
Without summary the observed series is matched to the end of the simulated
one: observed[-1] against the value after the last step. With summary,
summary(series) of every replicate is compared to observed. Replicates
are averaged before the distance is taken.
- class Target(name: str, observed: ~typing.Any, distance: ~collections.abc.Callable[[~numpy.ndarray, ~numpy.ndarray], float] | str = <function rmse>, weight: float = 1.0, summary: ~collections.abc.Callable[[~numpy.ndarray], ~typing.Any] | None = None, output: str | ~collections.abc.Callable[[~typing.Any], ~typing.Any] | None = None)[source]#
One observed quantity and how to measure the model’s distance to it.
- Parameters:
name – the column name; also the output read unless
outputis given.observed – the observed series, or the observed summary statistics.
distance – a function
(simulated, observed) -> floator the name of a built-in ("rmse","mae","euclidean","wasserstein").weight – the factor of this target’s distance in the total.
summary – a function of one simulated series (a 1-d array, step 0 first) returning the statistics compared to
observed.output – what to record, when it is not
name: an attribute path or a module-level function of the model.
- check_length(steps: int) None[source]#
Refuse an observed series longer than a run (
steps + 1values).
- replicate_values(series: ndarray) ndarray[source]#
What is compared, per replicate:
array[replicate, value].
- total_distance(targets: Sequence[Target], simulated: dict[str, ndarray]) float[source]#
The sum of the targets’ weighted distances.
Distances between a simulated summary and the observed one.
A distance is any function distance(simulated, observed) -> float of two
arrays of the same shape; smaller is closer. Example:
>>> import numpy as np
>>> rmse(np.array([1.0, 2.0]), np.array([1.0, 4.0]))
1.4142135623730951
>>> weighted([1.0, 0.0])(np.array([1.0, 2.0]), np.array([1.0, 4.0]))
0.0
Every built-in is a module-level function or a picklable object, so a calibration that uses them can be saved and resumed.
- euclidean(simulated: ndarray, observed: ndarray) float[source]#
The Euclidean norm of the difference (for vectors of summary statistics).
- wasserstein(simulated: ndarray, observed: ndarray) float[source]#
The Wasserstein-1 distance between the two sets of values as distributions.
The values are compared as samples, whatever their order or count.
Posterior#
The result of a calibration: weighted parameter samples and how well they fit.
Example
>>> posterior = calibration.run(budget=2000)
>>> posterior.summary() # mean, median, sd and HDI per parameter
>>> posterior.to_dataframe() # one row per sample, with weight and distance
>>> posterior.predict(draws=20) # posterior predictive runs
>>> posterior.plot() # marginals, pairs, fit against the data
- weighted_quantile(values: ndarray, weights: ndarray, q: float) float[source]#
The
qquantile ofvaluesunderweights(the inverse of the weighted CDF).
- hdi(values: ndarray, weights: ndarray, probability: float) tuple[float, float][source]#
The shortest interval holding
probabilityof the weighted samples.
- class Posterior(samples: Mapping[str, list[Any]], weights: np.ndarray, distances: np.ndarray, summaries: Mapping[str, np.ndarray], calibration: Calibration)[source]#
Weighted samples of the fitted parameters.
- Variables:
samples –
{parameter: values}, one value per sample.weights – the normalised importance weights (all equal for rejection).
distances – each sample’s total distance to the targets.
summaries –
{target: array[sample, value]}, what each sample simulated.calibration – the calibration that produced it (used by
predict).
Keep the samples; the weights are normalised to sum to one.
- summary(probability: float = 0.95) DataFrame[source]#
Weighted mean, median, sd and highest-density interval per numeric parameter.
- resample(draws: int) list[int][source]#
drawssample indices drawn with the weights, from the calibration’s seed.
- predict(draws: int = 20, executor: Any = 'serial', max_parallel: int | None = None, **options: Any) DataFrame[source]#
Run the model at
drawsposterior samples; one row per draw, replicate and step.The samples are drawn with the weights from the calibration’s seed, and the runs go to
executor(withoptions) like the calibration’s own.
Matplotlib figures of a Posterior: marginals, pairs, and the fit to the data.
The figure is built with matplotlib.figure.Figure (no pyplot state), so it
works the same in scripts, notebooks and without a display; save it with
figure.savefig(path).
Runs#
Run batches of parameter points through the experiment executors and read them back.
Calibration and sensitivity analysis both ask the same question many times:
“what does the model produce at these points?”. A Simulator answers
it with the machinery of modelly.experiments: one
RunTask per (point, replicate), run on any
executor, each writing its Parquet file to a store folder per batch.
Seeds follow the experiment discipline. A batch is named by a key of integers;
its runs use SeedSequence(entropy, spawn_key=(point, replicate)) with the
entropy drawn from SeedSequence(seed, spawn_key=(SIMULATION, *key)), or the
same entropy for every batch when the batch asks for common random numbers.
The numbers therefore never depend on the executor, the worker count or the
order of the runs, and a batch already in its folder is read, not rerun.
- class Execution(executor: Any = 'serial', max_parallel: int | None = None, options: Mapping[str, ~typing.Any]=<factory>)[source]#
Where runs execute: an executor name (or instance), its parallelism and options.
Used as a context manager,
"processes"keeps one pool of workers for every batch until the block ends (seemodelly.calibrate.pool).
- class Batch(key: tuple[int, ...], points: Sequence[Mapping[str, Any]], common: bool = False)[source]#
Points to simulate under one key;
commonreuses the same seeds for every batch.
- class Simulator(model: str, outputs: dict[str, str], steps: int, replicates: int, seed: int, root: Path)[source]#
A model, its outputs, steps and replicates, and the folder batches go to.
runreturns, for every point,{output: array[replicate, step]}withsteps + 1columns (a run that stopped early repeats its last value).
A process executor that keeps its workers between batches.
ABC-SMC, simulated moments and Bayesian optimisation submit many small
batches. The experiments’ processes executor starts a fresh pool for each
run call (workers start with forkserver or spawn and import the
model: about a second each time); this one starts it once and reuses it until
close. Its outcomes are the same, task for task.
- class PooledProcessExecutor(**options)[source]#
processes, with one pool for everyruncall untilclose.No pool until the first batch.
A small Gaussian-process regressor on numpy and scipy (basic, by design).
Used by BayesOpt and the sensitivity
emulator. Inputs are points of the unit cube; the kernel is a squared
exponential with one length scale per input plus a noise term, fitted by
maximising the log marginal likelihood with L-BFGS-B from a fixed start. It is
meant for a few hundred points in a handful of dimensions; for more, use a
dedicated library.
Example
>>> import numpy as np
>>> x = np.linspace(0, 1, 8)[:, None]
>>> gp = GaussianProcess.fit(x, np.sin(6 * x[:, 0]))
>>> mean, sd = gp.predict(np.array([[0.5]]))
>>> bool(abs(mean[0] - np.sin(3.0)) < 0.05)
True
- class GaussianProcess(x: ndarray, lengths: ndarray, noise: float, alpha: ndarray, factor: ndarray, mean: float, scale: float)[source]#
A fitted Gaussian process: training data, hyperparameters, and the solved system.
- classmethod fit(x: ndarray, y: ndarray) GaussianProcess[source]#
Fit to inputs
x[point, dimension]in the unit cube and outputsy[point].
Calibration files#
Read a calibration from a TOML file (what modelly calibrate spec.toml runs).
The file reuses the experiment format ([experiment], [space],
[run]) and adds the priors, the targets and the method:
[experiment]
model = "model.py:SIR" # relative to this file, or package.module:Class
steps = 40
replicates = 1
seed = 7
store = "runs/calibration" # optional; a temporary folder otherwise
[space] # a range is a uniform prior (log = true: log-uniform),
beta = { min = 0.05, max = 1.0 } # an integer range or a list a choice,
population = 1000 # anything else a fixed parameter
[priors] # optional: other priors, by kind
gamma = { kind = "log_normal", mu = -2.3, sigma = 0.5 }
[[targets]]
name = "infected"
file = "cases.csv" # or observed = [..]; column defaults to name
column = "cases"
distance = "rmse" # or mae, euclidean, wasserstein, module:function
weight = 1.0
[calibration]
method = "abc_smc" # abc_rejection, simulated_moments, bayes_opt
particles = 100 # the method's settings
budget = 3000 # and its run() arguments
[run]
executor = "processes"
max_parallel = 4
- priors_and_fixed(document: Mapping[str, Any]) tuple[dict[str, Prior], dict[str, Any]][source]#
The priors (from
[space]and[priors]) and the fixed parameters.
- load_calibration(path: str | Path) tuple[Calibration, dict[str, Any], RunSettings][source]#
The calibration a TOML file describes, its
run()arguments and[run]settings.