Experiments and Monte Carlo¶
experiment ¶
Experiments: scenarios x replications, with provenance, checkpointing and parallelism.
Experiment ¶
Experiment(model: Model, scenarios: Scenario | Sequence[Scenario] | None = None, *, replications: int = 10, seed: int = 0, workers: int = 1, name: str | None = None, common_random_numbers: bool = True, on_error: Literal['raise', 'record'] = 'raise', antithetic: bool = False)
Run a model across scenarios and independent replications.
Seeds: replication r uses derive_seed(seed, "replication", r).
With common_random_numbers=True (default) that seed is shared by all
scenarios, so scenario comparisons are paired and much less noisy.
Each replication is independent and fully determined by its seed, so
results are identical whether run serially or with workers > 1.
grid ¶
grid(axes: Mapping[str, Sequence[Any]], base: Scenario | None = None) -> list[Scenario]
Replace the scenarios with a full-factorial grid and return them.
run ¶
run(scenario: Scenario | Sequence[Scenario] | None = None, *, replications: int | None = None, seed: int | None = None, workers: int | None = None, checkpoint: str | Path | None = None, progress: Callable[[int, int], None] | None = None, executor: Executor | None = None, _reuse: dict[tuple[str, int], ReplicationRecord] | None = None) -> ExperimentResult
Run every (scenario, replication) pair and collect the results.
checkpoint names a JSON-lines file: finished replications are
appended as they complete, and a re-run with the same file skips them
(after checking that the experiment definition matches).
executor runs the replications on any :class:concurrent.futures.Executor
instead of local worker processes - for example a Dask cluster
(distributed.Client(...).get_executor()), or a thread pool for
models that release the GIL. Results are identical, because every
replication's seed is fixed in advance.
run_until ¶
run_until(relative_precision: float, metrics: Sequence[str], *, confidence: float = 0.95, max_replications: int = 1000, checkpoint: str | Path | None = None, progress: Callable[[int, int], None] | None = None) -> ExperimentResult
Add replications until every metric in every scenario reaches relative_precision.
Starts with self.replications (at least 3), then repeatedly asks
:func:~simulsi.statistics.required_replications how many are needed
and runs only the missing ones (replication seeds do not depend on the
total, so earlier runs are reused). Stops at max_replications.
The stopping decision itself uses the data, so the final interval's
coverage is approximate - the standard caveat of sequential procedures.
ExperimentResult ¶
ExperimentResult(metadata: ExperimentMetadata, records: Sequence[ReplicationRecord])
All replication records plus provenance, with analysis and export helpers.
derive ¶
derive(fn: Callable[[Mapping[str, float]], Mapping[str, float]]) -> ExperimentResult
Add derived metrics to every successful record in place, e.g. result.derive(costs.metrics).
values ¶
values(metric: str, scenario: str | None = None) -> np.ndarray[Any, np.dtype[np.float64]]
Independent observations of metric, ordered by replication (NaN if missing).
For antithetic experiments each value is the mean of one antithetic
pair (the pair means are the independent units); use :meth:raw_values
for one value per run.
raw_values ¶
raw_values(metric: str, scenario: str | None = None) -> np.ndarray[Any, np.dtype[np.float64]]
One value per successful run, ordered by replication (no antithetic pairing).
summary ¶
summary(metrics: Iterable[str] | None = None, *, confidence: float = 0.95, scenarios: Iterable[str] | None = None) -> list[dict[str, Any]]
One row per (scenario, metric) with mean, std and a t confidence interval.
control_variate ¶
control_variate(metric: str, controls: Mapping[str, float], scenario: str | None = None, *, confidence: float = 0.95) -> Any
Variance-reduced estimate of metric using metrics with known means.
controls maps control metrics to their true expectations, e.g. the
sampled mean service time and 1 / service_rate.
report ¶
report(path: str | Path | None = None, *, metrics: Iterable[str] | None = None, baseline: str | None = None, confidence: float = 0.95, adjust: str = 'holm', title: str | None = None) -> str
A self-contained HTML report; written to path if given. Returns the HTML.
rows ¶
rows() -> list[dict[str, Any]]
Long-form-friendly wide rows: one per replication with metric columns.
save ¶
save(directory: str | Path, *, formats: Sequence[str] = ('json', 'csv')) -> Path
Write experiment.json (+ replications.csv / .parquet) into directory.
load
classmethod
¶
load(path: str | Path) -> ExperimentResult
Load from an experiment.json file or a directory containing one.
shutdown_workers ¶
shutdown_workers() -> None
Stop the reusable worker processes (e.g. after editing a model file in a notebook).
montecarlo ¶
Monte Carlo simulation: propagate input uncertainty through a model.
MonteCarloResult
dataclass
¶
MonteCarloResult(inputs: dict[str, NDArray[Any]], outputs: dict[str, FloatArray], iterations: int, seed: int, runtime: float, sampling: str = 'random', metadata: dict[str, Any] = dict())
Sampled inputs and outputs, one entry per iteration.
quantile_ci ¶
quantile_ci(q: float, output: str | None = None, *, confidence: float = 0.95, n_resamples: int = 2000) -> tuple[float, float]
Bootstrap interval for a quantile (e.g. the 95th percentile of cost).
probability ¶
probability(output: str | None, op: Comparator, threshold: float, *, confidence: float = 0.95) -> tuple[float, tuple[float, float]]
P(output op threshold) with a Wilson interval, e.g. probability("cost", ">", 1e6).
sample_inputs ¶
sample_inputs(parameters: Mapping[str, DistributionLike | Mapping[str, Any]], iterations: int, seed: int, sampling: Literal['random', 'lhs'] = 'random') -> dict[str, npt.NDArray[Any]]
Draw iterations values for every uncertain parameter (independent streams).
monte_carlo ¶
monte_carlo(model: Callable[..., Any] | Model, parameters: Mapping[str, DistributionLike | Mapping[str, Any]], iterations: int = 10000, *, seed: int = 0, fixed: Mapping[str, Any] | None = None, vectorized: bool = False, objective: Callable[[Mapping[str, Any]], float] | None = None, sampling: Literal['random', 'lhs'] = 'random') -> MonteCarloResult
Sample uncertain parameters and evaluate model for each draw.
model may be:
- a function
f(**params) -> float | dict[str, float]; if it has anrngargument it also receives a per-iteration :class:RandomStreamfor internal randomness; - with
vectorized=True, a function receiving whole arrays (one per parameter) and returning an array or a dict of arrays - much faster; - a :class:
~simulsi.core.model.Model, in which case every iteration is one simulation run (outputs = the run's flat metrics).
objective derives an extra output named "objective" from each
iteration's outputs (and fixed/sampled inputs).