mcframework.sims.BlackScholesSimulation#

class mcframework.sims.BlackScholesSimulation[source]#

Bases: MonteCarloSimulation

Monte Carlo simulation for Black-Scholes option pricing.

Uses Geometric Brownian Motion for stock price dynamics and supports European and American options (calls and puts) with Greeks calculation.

Parameters:

namestr, optional: Simulation name. Defaults to “Black-Scholes Option Pricing”.

Methods

`analytic_reference`	Closed-form Black-Scholes-Merton price (the oracle for European exercise).
`calculate_greeks`	Estimate primary Greeks via finite differences.
`curand_batch`	Optional vectorized cuRAND implementation using CuPy.
`price_american`	Price an American option with the Longstaff-Schwartz (LSM) algorithm.
`run`	Run the Monte Carlo simulation.
`set_seed`	Set the random seed for reproducible experiments.
`single_simulation`	Price a single option instance under Black–Scholes dynamics.
`torch_batch`	Optional vectorized Torch implementation.

analytic_reference(*, S0: float = 100.0, K: float = 100.0, T: float = 1.0, r: float = 0.05, sigma: float = 0.2, option_type: str = 'call', exercise_type: str = 'european', **params) → float | None[source]#

Closed-form Black-Scholes-Merton price (the oracle for European exercise).

\[C = S_0\,\Phi(d_1) - K e^{-rT}\,\Phi(d_2), \qquad P = K e^{-rT}\,\Phi(-d_2) - S_0\,\Phi(-d_1),\]

with \(d_1 = \frac{\ln(S_0/K) + (r + \tfrac12\sigma^2)T}{\sigma\sqrt T}\) and \(d_2 = d_1 - \sigma\sqrt T\).

Returns None for exercise_type="american": an American option has no closed-form price, so it is deliberately not convergence-validated. That absence is itself the governance signal.

calculate_greeks(n_simulations: int, S0: float = 100.0, K: float = 100.0, T: float = 1.0, r: float = 0.05, sigma: float = 0.2, option_type: str = 'call', exercise_type: str = 'european', n_steps: int = 252, backend: str = 'sequential', bump_pct: float = 0.01, time_bump_days: float = 1.0) → dict[str, float][source]#: Estimate primary Greeks via finite differences.

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

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.

price_american(n_paths: int, *, S0: float = 100.0, K: float = 100.0, T: float = 1.0, r: float = 0.05, sigma: float = 0.2, option_type: str = 'call', n_steps: int = 50, _rng: Generator | None = None) → float[source]#

Price an American option with the Longstaff-Schwartz (LSM) algorithm.

Unlike single_simulation() with exercise_type="american" (which is a high-biased perfect-foresight bound), this simulates a batch of GBM paths jointly and regresses continuation values across paths, so the exercise decision uses no future information. This is the estimator to use for actual American option valuation.

Parameters:

n_pathsint: Number of Monte Carlo paths to simulate. LSM regression quality improves with more paths.
S0, K, T, r, sigmafloat: Standard Black-Scholes parameters (spot, strike, maturity in years, risk-free rate, volatility).
option_type{“call”, “put”}, default "call": Payoff family. Early exercise matters mainly for puts (and dividend -paying calls, not modeled here).
n_stepsint, default 50: Number of exercise opportunities (time steps) along each path.
_rngnumpy.random.Generator, optional: Explicit generator. Falls back to self.rng (set via set_seed()).

Returns:

float: Estimated arbitrage-free American option price.

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#

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#

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.

single_simulation(*, S0: float = 100.0, K: float = 100.0, T: float = 1.0, r: float = 0.05, sigma: float = 0.2, option_type: str = 'call', exercise_type: str = 'european', n_steps: int = 252, _rng: Generator | None = None, **kwargs) → float[source]#

Price a single option instance under Black–Scholes dynamics.

Notes

For exercise_type="american" this returns the perfect-foresight upper bound (the maximum discounted intrinsic value along the path), which is a high-biased estimator, not a fair price. Use price_american() for a look-ahead-free Longstaff-Schwartz price.

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

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 = 'Black-Scholes Option Pricing')[source]#

classmethod __new__(*args, **kwargs)#