gds_psuu.metric

Metric and Aggregation primitives for composable KPI construction.

Metric and Aggregation primitives for composable KPI construction.

Metric

Bases: BaseModel

Per-run scalar extracted from simulation output.

Source code in packages/gds-psuu/gds_psuu/metric.py

class Metric(BaseModel):
    """Per-run scalar extracted from simulation output."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    name: str
    fn: MetricFn

Aggregation

Bases: BaseModel

Combines per-run metric values across Monte Carlo runs.

Source code in packages/gds-psuu/gds_psuu/metric.py

class Aggregation(BaseModel):
    """Combines per-run metric values across Monte Carlo runs."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    name: str
    fn: AggregationFn

final_value(key)

Metric: value of a state variable at the final timestep of a run.

Source code in packages/gds-psuu/gds_psuu/metric.py

def final_value(key: str) -> Metric:
    """Metric: value of a state variable at the final timestep of a run."""
    return Metric(
        name=f"final_{key}",
        fn=lambda results, run, _key=key: _final_value_fn(results, run, _key),
    )

trajectory_mean(key)

Metric: mean of a state variable over time for a single run.

Source code in packages/gds-psuu/gds_psuu/metric.py

def trajectory_mean(key: str) -> Metric:
    """Metric: mean of a state variable over time for a single run."""
    return Metric(
        name=f"mean_{key}",
        fn=lambda results, run, _key=key: _trajectory_mean_fn(results, run, _key),
    )

max_value(key)

Metric: maximum value of a state variable within a single run.

Source code in packages/gds-psuu/gds_psuu/metric.py

def max_value(key: str) -> Metric:
    """Metric: maximum value of a state variable within a single run."""
    return Metric(
        name=f"max_{key}",
        fn=lambda results, run, _key=key: _max_value_fn(results, run, _key),
    )

min_value(key)

Metric: minimum value of a state variable within a single run.

Source code in packages/gds-psuu/gds_psuu/metric.py

def min_value(key: str) -> Metric:
    """Metric: minimum value of a state variable within a single run."""
    return Metric(
        name=f"min_{key}",
        fn=lambda results, run, _key=key: _min_value_fn(results, run, _key),
    )

percentile_agg(p)

Aggregation: p-th percentile across runs.

Source code in packages/gds-psuu/gds_psuu/metric.py

def percentile_agg(p: float) -> Aggregation:
    """Aggregation: p-th percentile across runs."""

    def _fn(vals: list[float]) -> float:
        if not vals:
            return 0.0
        s = sorted(vals)
        k = (p / 100.0) * (len(s) - 1)
        lo = int(k)
        hi = min(lo + 1, len(s) - 1)
        frac = k - lo
        return s[lo] + frac * (s[hi] - s[lo])

    return Aggregation(name=f"p{p}", fn=_fn)

probability_above(threshold)

Aggregation: fraction of runs where metric exceeds threshold.

Source code in packages/gds-psuu/gds_psuu/metric.py

def probability_above(threshold: float) -> Aggregation:
    """Aggregation: fraction of runs where metric exceeds threshold."""
    return Aggregation(
        name=f"P(>{threshold})",
        fn=lambda vals: (
            sum(1 for v in vals if v > threshold) / len(vals) if vals else 0.0
        ),
    )

probability_below(threshold)

Aggregation: fraction of runs where metric is below threshold.

Source code in packages/gds-psuu/gds_psuu/metric.py

def probability_below(threshold: float) -> Aggregation:
    """Aggregation: fraction of runs where metric is below threshold."""
    return Aggregation(
        name=f"P(<{threshold})",
        fn=lambda vals: (
            sum(1 for v in vals if v < threshold) / len(vals) if vals else 0.0
        ),
    )