mcframework.core.MonteCarloSimulation#

class mcframework.core.MonteCarloSimulation[source]#

Bases: ABC

Abstract base class for Monte Carlo simulations.

Subclass this and implement single_simulation(). The framework takes care of reproducible seeding, (optional) parallel execution, statistics, and percentiles.

Notes

Parallel backend. The parallel_backend attribute can be "auto", "thread", or "process". With NumPy RNGs (which release the GIL), threads are usually faster and avoid process-spawn overhead.

Percentiles. If compute_stats=True, the stats engine computes defaults _PCTS = (5, 25, 50, 75, 95) and merges them with user-requested percentiles. The original user request is preserved in result.metadata["requested_percentiles"] and enforced by MonteCarloFramework.compare_results() for percentile metrics.

Examples

>>> from mcframework.simulation import MonteCarloSimulation
>>> class PiSim(MonteCarloSimulation):
...     def single_simulation(self, _rng=None, n_points: int = 10_000):
...         rng = self._rng(_rng, self.rng)
...         x, y = rng.random(n_points), rng.random(n_points)
...         return 4.0 * ((x*x + y*y) <= 1.0).mean()
...
>>> sim = PiSim()
>>> sim.set_seed(42)
>>> res = sim.run(10_000, parallel=True, compute_stats=True)

Methods

`analytic_reference`	Known expected value of `single_simulation()` ("oracle"), if one exists.
`curand_batch`	Optional vectorized cuRAND implementation using CuPy.
`run`	Run the Monte Carlo simulation.
`set_seed`	Set the random seed for reproducible experiments.
`single_simulation`	Perform a single simulation run.
`torch_batch`	Optional vectorized Torch implementation.

analytic_reference(**params: Any) → float | None[source]#

Known expected value of single_simulation() (“oracle”), if one exists.

Override in subclasses that have a closed-form or otherwise-known answer for \(\mathbb{E}[\texttt{single\_simulation}]\) under the given parameters. The value is consumed by mcframework.validation.validate_convergence() to assert that the Monte Carlo estimate converges to ground truth.

Parameters:

**paramsAny: The same keyword parameters passed to single_simulation() / run(), so the reference can depend on the configuration.

Returns:

float or None: The oracle value, or None (default) when no reference exists. Returning None signals that the simulation has no oracle and cannot be convergence -validated (it is research, not a verifiable demo).

Examples

>>> import math
>>> class PiSim(MonteCarloSimulation):
...     reference_source = "Closed form: pi"
...     def single_simulation(self, _rng=None):
...         ...
...     def analytic_reference(self, **params):
...         return math.pi

curand_batch(n: int, device_id: int, rng: cupy.random.RandomState) → cupy.ndarray[source]#

Optional vectorized cuRAND implementation using CuPy.

Override this method in subclasses to enable GPU-accelerated batch execution via cuRAND. When implemented alongside supports_batch = True and use_curand=True, the CUDA backend will call this method instead of torch_batch().

Parameters:

nint: Number of simulation draws.
device_idint: CUDA device index.
rngcupy.random.RandomState: cuRAND generator for reproducible random sampling.

Returns:

cupy.ndarray: A 1D array of length n containing simulation results.

run(n_simulations: int, *, backend: str = 'auto', torch_device: str = 'cpu', cuda_device_id: int = 0, cuda_use_curand: bool = False, cuda_batch_size: int | None = None, cuda_use_streams: bool = True, n_workers: int | None = None, progress_callback: Callable[[int, int], None] | None = None, percentiles: Iterable[int] | None = None, compute_stats: bool = True, stats_engine: StatsEngine | None = None, confidence: float = 0.95, ci_method: str = 'auto', extra_context: Mapping[str, Any] | None = None, **simulation_kwargs: Any) → SimulationResult[source]#

Run the Monte Carlo simulation.

Parameters:

n_simulationsint

Number of simulation draws.

backend{“auto”, “sequential”, “thread”, “process”, “torch”}, default "auto"

Execution backend to use:

"auto" — Sequential for small jobs, parallel (thread/process) for large jobs
"sequential" — Single-threaded execution
"thread" — Thread-based parallelism (best when NumPy releases GIL)
"process" — Process-based parallelism (required on Windows for true parallelism)
"torch" — Torch batch execution (requires supports_batch = True)

torch_device{“cpu”, “mps”, “cuda”}, default "cpu"

Torch device for backend="torch". Ignored for other backends.

"cpu" — Safe default, works everywhere
"mps" — Apple Metal Performance Shaders (M1/M2/M3 Macs)
"cuda" — NVIDIA GPU acceleration

cuda_device_idint, default 0

CUDA device index for multi-GPU systems. Only used when backend="torch" and torch_device="cuda".

cuda_use_curandbool, default False

Use cuRAND (via CuPy) instead of torch.Generator for maximum GPU performance. Requires CuPy and curand_batch() implementation.

cuda_batch_sizeint or None, default None

Fixed batch size for CUDA execution. If None, automatically estimates optimal batch size based on available GPU memory.

cuda_use_streamsbool, default True

Use CUDA streams for overlapped execution. Recommended for performance.

n_workersint, optional

Worker count for parallel backends. Defaults to CPU count.

progress_callbackcallable(), optional

A function f(completed: int, total: int) called periodically.

percentilesiterable of int, optional

Percentiles to compute from raw results. If None and compute_stats=True, the stats engine’s defaults (_PCTS) are used; if compute_stats=False, no percentiles are computed unless explicitly provided.

compute_statsbool, default True

Compute additional metrics via a StatsEngine.

stats_engineStatsEngine, optional

Custom engine (defaults to mcframework.stats_engine.DEFAULT_ENGINE).

confidencefloat, default 0.95

Confidence level for CI-related metrics.

ci_method{“auto”,”z”,”t”}, default "auto"

Which critical values the stats engine should use.

extra_contextmapping, optional

Extra context forwarded to the stats engine.

**simulation_kwargsAny

Keyword arguments forwarded to single_simulation().

Returns:

SimulationResult: See SimulationResult.

See also

run_simulation(): Run a registered simulation by name.

Notes

MPS determinism caveat. When using torch_device="mps", the framework preserves RNG stream structure but does not guarantee bitwise reproducibility due to Metal backend scheduling and float32 arithmetic. Statistical properties (mean, variance, CI coverage) remain correct.

set_seed(seed: int | None) → None[source]#

Set the random seed for reproducible experiments.

Parameters:

seedint or None: Seed for numpy.random.SeedSequence. None chooses entropy from the OS.

Notes

The framework spawns independent child sequences per worker/chunk via numpy.random.SeedSequence.spawn(), ensuring deterministic parallel streams given the same seed and the same block layout.

Warning

Reproducibility is per-(backend, block-layout), not absolute. The number of spawned child streams depends on the resolved backend and the work partition:

The sequential backend spawns one child stream and draws every sample from it.
The thread/process backends spawn one child stream per block, where the block count depends on n_workers, n_simulations, and _CHUNKS_PER_WORKER.

Because backend="auto" switches from sequential to parallel once n_simulations crosses _PARALLEL_THRESHOLD (20_000), the same seed can produce different draws above vs. below that threshold, and sequential vs. parallel results are not bitwise identical. For run-to-run reproducible numbers, pin the backend and n_workers explicitly. Statistical properties (mean, variance, CI coverage) are unaffected.

abstractmethod single_simulation(*args, **kwargs) → float[source]#

Perform a single simulation run.

Returns:

float: The result of the simulation run.

Notes

Subclasses must implement this method.

torch_batch(n: int, *, device: torch.device, generator: torch.Generator) → torch.Tensor[source]#

Optional vectorized Torch implementation.

Override this method in subclasses to enable GPU-accelerated batch execution. When implemented alongside supports_batch = True, the framework will use this method instead of repeated single_simulation calls.

Parameters:

nint: Number of simulation draws.
devicetorch.device: Device to use for the simulation ("cpu", "mps", or "cuda").
generatortorch.Generator: Explicit Torch generator for reproducible random sampling. This generator is seeded from numpy.random.SeedSequence to maintain the same spawning semantics as the NumPy backend.

Returns:

torch.Tensor: A 1D tensor of length n containing simulation results. Use float32 for MPS compatibility; the framework promotes to float64 after moving to CPU.

Raises:

NotImplementedError: If the subclass does not implement this method.

Notes

RNG discipline. All random sampling must use the provided generator explicitly. Never use global Torch RNG (torch.manual_seed).

Dtype policy (device-specific):

MPS (Apple Silicon): Must return float32 (Metal doesn’t support float64). Framework promotes to float64 on CPU.
CUDA (NVIDIA): Can return float32 or float64. Float64 preferred for zero conversion overhead and full precision.
CPU: Can return float32 or float64. Float64 preferred for consistency with framework precision.

This method is optional and must be implemented by subclasses that support the Torch backend. If not implemented, the framework will fall back to the NumPy backend.

Examples

>>> class PiSim(MonteCarloSimulation):
...     supports_batch = True
...     def torch_batch(self, n, *, device, generator):
...         import torch
...         x = torch.rand(n, device=device, generator=generator)
...         y = torch.rand(n, device=device, generator=generator)
...         inside = (x * x + y * y) <= 1.0
...         return 4.0 * inside.float()  # float32 for MPS compatibility

__init__(name: str = 'Simulation')[source]#

classmethod __new__(*args, **kwargs)#