Files
calminer/services/simulation.py
zwitschi 97b1c0360b
Some checks failed
Run Tests / e2e tests (push) Failing after 1m27s
Run Tests / lint tests (push) Failing after 6s
Run Tests / unit tests (push) Failing after 7s
Refactor test cases for improved readability and consistency
- Updated test functions in various test files to enhance code clarity by formatting long lines and improving indentation.
- Adjusted assertions to use multi-line formatting for better readability.
- Added new test cases for theme settings API to ensure proper functionality.
- Ensured consistent use of line breaks and spacing across test files for uniformity.
2025-10-27 10:32:55 +01:00

145 lines
4.7 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