141 lines
4.6 KiB
Python
141 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from random import Random
|
|
from typing import Dict, List, Literal, Optional, Sequence
|
|
|
|
|
|
DEFAULT_STD_DEV_RATIO = 0.1
|
|
DEFAULT_UNIFORM_SPAN_RATIO = 0.15
|
|
DistributionType = Literal["normal", "uniform", "triangular"]
|
|
|
|
|
|
@dataclass
|
|
class SimulationParameter:
|
|
name: str
|
|
base_value: float
|
|
distribution: DistributionType
|
|
std_dev: Optional[float] = None
|
|
minimum: Optional[float] = None
|
|
maximum: Optional[float] = None
|
|
mode: Optional[float] = None
|
|
|
|
|
|
def _ensure_positive_span(span: float, fallback: float) -> float:
|
|
return span if span and span > 0 else fallback
|
|
|
|
|
|
def _compile_parameters(parameters: Sequence[Dict[str, float]]) -> List[SimulationParameter]:
|
|
compiled: List[SimulationParameter] = []
|
|
for index, item in enumerate(parameters):
|
|
if "value" not in item:
|
|
raise ValueError(
|
|
f"Parameter at index {index} must include 'value'")
|
|
name = str(item.get("name", f"param_{index}"))
|
|
base_value = float(item["value"])
|
|
distribution = str(item.get("distribution", "normal")).lower()
|
|
if distribution not in {"normal", "uniform", "triangular"}:
|
|
raise ValueError(
|
|
f"Parameter '{name}' has unsupported distribution '{distribution}'"
|
|
)
|
|
|
|
span_default = abs(base_value) * DEFAULT_UNIFORM_SPAN_RATIO or 1.0
|
|
|
|
if distribution == "normal":
|
|
std_dev = item.get("std_dev")
|
|
std_dev_value = float(std_dev) if std_dev is not None else abs(
|
|
base_value) * DEFAULT_STD_DEV_RATIO or 1.0
|
|
compiled.append(
|
|
SimulationParameter(
|
|
name=name,
|
|
base_value=base_value,
|
|
distribution="normal",
|
|
std_dev=_ensure_positive_span(std_dev_value, 1.0),
|
|
)
|
|
)
|
|
continue
|
|
|
|
minimum = item.get("min")
|
|
maximum = item.get("max")
|
|
if minimum is None or maximum is None:
|
|
minimum = base_value - span_default
|
|
maximum = base_value + span_default
|
|
minimum = float(minimum)
|
|
maximum = float(maximum)
|
|
if minimum >= maximum:
|
|
raise ValueError(
|
|
f"Parameter '{name}' requires 'min' < 'max' for {distribution} distribution"
|
|
)
|
|
|
|
if distribution == "uniform":
|
|
compiled.append(
|
|
SimulationParameter(
|
|
name=name,
|
|
base_value=base_value,
|
|
distribution="uniform",
|
|
minimum=minimum,
|
|
maximum=maximum,
|
|
)
|
|
)
|
|
else: # triangular
|
|
mode = item.get("mode")
|
|
if mode is None:
|
|
mode = base_value
|
|
mode_value = float(mode)
|
|
if not (minimum <= mode_value <= maximum):
|
|
raise ValueError(
|
|
f"Parameter '{name}' mode must be within min/max bounds for triangular distribution"
|
|
)
|
|
compiled.append(
|
|
SimulationParameter(
|
|
name=name,
|
|
base_value=base_value,
|
|
distribution="triangular",
|
|
minimum=minimum,
|
|
maximum=maximum,
|
|
mode=mode_value,
|
|
)
|
|
)
|
|
return compiled
|
|
|
|
|
|
def _sample_parameter(rng: Random, param: SimulationParameter) -> float:
|
|
if param.distribution == "normal":
|
|
assert param.std_dev is not None
|
|
return rng.normalvariate(param.base_value, param.std_dev)
|
|
if param.distribution == "uniform":
|
|
assert param.minimum is not None and param.maximum is not None
|
|
return rng.uniform(param.minimum, param.maximum)
|
|
# triangular
|
|
assert (
|
|
param.minimum is not None
|
|
and param.maximum is not None
|
|
and param.mode is not None
|
|
)
|
|
return rng.triangular(param.minimum, param.maximum, param.mode)
|
|
|
|
|
|
def run_simulation(
|
|
parameters: Sequence[Dict[str, float]],
|
|
iterations: int = 1000,
|
|
seed: Optional[int] = None,
|
|
) -> List[Dict[str, float]]:
|
|
"""Run a lightweight Monte Carlo simulation using configurable distributions."""
|
|
|
|
if iterations <= 0:
|
|
return []
|
|
|
|
compiled_params = _compile_parameters(parameters)
|
|
if not compiled_params:
|
|
return []
|
|
|
|
rng = Random(seed)
|
|
results: List[Dict[str, float]] = []
|
|
for iteration in range(1, iterations + 1):
|
|
total = 0.0
|
|
for param in compiled_params:
|
|
sample = _sample_parameter(rng, param)
|
|
total += sample
|
|
results.append({"iteration": iteration, "result": total})
|
|
return results
|