Skip to content

Aggregated Results

AggregatedExecutionResults dataclass

All three violation metrics plus cost, aggregated across scenarios.

Optionally stores the per-scenario :class:ExecutionResult objects that were aggregated, so downstream code can inspect individual scenario values (e.g. for min–max range display).

Source code in src/autoslo/workload_execution/aggregated_execution_results.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@dataclass(frozen=True)
class AggregatedExecutionResults:
    """All three violation metrics plus cost, aggregated across scenarios.

    Optionally stores the per-scenario :class:`ExecutionResult` objects
    that were aggregated, so downstream code can inspect individual
    scenario values (e.g. for min–max range display).
    """

    violation_rate: float
    violation_amount_s: float
    violation_relative_mean: float
    cost: float
    scenario_results: list[ExecutionResult]

    def primary_violation(self, slo_metric: str | SloMetric) -> float:
        """Extract the primary violation value for the given SLO metric."""
        slo_metric_obj = (
            slo_metric
            if isinstance(slo_metric, SloMetric)
            else SloMetric(slo_metric)
        )

        if slo_metric_obj == SloMetric.BINARY:
            return self.violation_rate
        elif slo_metric_obj == SloMetric.ABSOLUTE_S:
            return self.violation_amount_s
        elif slo_metric_obj == SloMetric.RELATIVE:
            return self.violation_relative_mean
        else:
            raise ValueError(f"Unknown SLO metric: {slo_metric}")

    @staticmethod
    def _fmt_cell(
        agg_val: float,
        scenario_vals: list[float],
    ) -> str:
        """Format a metric cell: aggregated value with dim min–max range."""
        main = f"{agg_val:.4f}"
        if len(scenario_vals) >= 2:
            lo, hi = min(scenario_vals), max(scenario_vals)
            main += f"\n[dim]{lo:.4f} … {hi:.4f}[/dim]"
        return main

    @staticmethod
    def print_comparison(
        *entries: tuple[str, AggregatedExecutionResults],
        console: Any,
        agg_method: str = "p90",
        slo_metric: str | SloMetric = "binary",
        highlight_best: bool = True,
    ) -> None:
        """Print a table comparing multiple AggregatedExecutionResults.

        Parameters
        ----------
        *entries :
            ``(label, agg)`` pairs.  Each gets one row.  Use labels like
            ``"Initial (train)"`` / ``"Initial (val)"`` to distinguish splits.
        agg_method :
            Aggregation metric shown in the title.
        slo_metric :
            The SLO metric that was actually optimised (``"binary"``,
            ``"absolute_s"``, or ``"relative"``).  The best cell in this
            column and in the Cost column is highlighted green; the best
            cell in the other two violation columns is highlighted yellow.
        highlight_best :
            Whether to highlight the best values in the table.
        """
        # Map slo_metric → column index (0-2 are the three violation metrics).
        slo_metric_str = (
            slo_metric.value
            if isinstance(slo_metric, SloMetric)
            else slo_metric
        )
        _SLO_TO_COL = {"binary": 0, "absolute_s": 1, "relative": 2}
        targeted_col = _SLO_TO_COL.get(slo_metric_str, 0)
        cost_col = 3  # always

        table = Table(
            title=f"Comparison  [dim](agg: {agg_method})[/dim]",
            show_lines=True,
        )
        table.add_column("Config", justify="left")

        metric_labels = [
            "Viol. Rate",
            "Viol. Amt (s)",
            "Viol. Rel.",
            "Cost ($)",
        ]
        for ml in metric_labels:
            table.add_column(ml, justify="right")

        fmt = AggregatedExecutionResults._fmt_cell

        def _extract(
            agg: AggregatedExecutionResults,
        ) -> tuple[list[float], list[list[float]]]:
            aggs = [
                agg.violation_rate,
                agg.violation_amount_s,
                agg.violation_relative_mean,
                agg.cost,
            ]
            if agg.scenario_results:
                per = [
                    [r.violation_rate for r in agg.scenario_results],
                    [r.violation_amount_s for r in agg.scenario_results],
                    [r.violation_relative_mean for r in agg.scenario_results],
                    [r.total_cost for r in agg.scenario_results],
                ]
            else:
                per = [[], [], [], []]
            return aggs, per

        # Build row data.
        row_data: list[tuple[str, list[float], list[list[float]]]] = []
        for label, agg in entries:
            a, p = _extract(agg)
            row_data.append((label, a, p))

        # Find the best (lowest) aggregated value per column.
        n_metric = 4
        best_per_col: list[int] = []
        if row_data:
            for c in range(n_metric):
                best_per_col.append(
                    min(
                        range(len(row_data)),
                        key=lambda i, _c=c: row_data[i][1][_c],  # type: ignore
                    )
                )

        for row_idx, (label, aggs, per) in enumerate(row_data):
            cells: list[str] = []
            for c in range(n_metric):
                cell = fmt(aggs[c], per[c])
                if (
                    len(row_data) > 1
                    and (row_idx == best_per_col[c])
                    and highlight_best
                ):
                    if c == targeted_col or c == cost_col:
                        cell = f"[green]{cell}[/green]"
                    else:
                        cell = f"[yellow]{cell}[/yellow]"
                cells.append(cell)
            table.add_row(label, *cells)

        console.print(table)

    @staticmethod
    def aggregate_from(
        results: list[ExecutionResult], metric: str = "p90"
    ) -> AggregatedExecutionResults:
        """Compute a summary statistic over scenario results.

        Parameters
        ----------
        results :
            Per-scenario results to aggregate.
        metric :
            ``"mean"``, ``"max"``, or any ``"pNN"`` quantile.

        Returns
        -------
        AggregatedExecutionResults with all three violation metrics and cost.
        """
        if not results:
            return AggregatedExecutionResults(
                violation_rate=0.0,
                violation_amount_s=0.0,
                violation_relative_mean=0.0,
                cost=0.0,
                scenario_results=[],
            )

        rates = [r.violation_rate for r in results]
        amounts = [r.violation_amount_s for r in results]
        relatives = [r.violation_relative_mean for r in results]
        costs = [r.total_cost for r in results]

        if metric == "mean":
            return AggregatedExecutionResults(
                violation_rate=statistics.mean(rates),
                violation_amount_s=statistics.mean(amounts),
                violation_relative_mean=statistics.mean(relatives),
                cost=statistics.mean(costs),
                scenario_results=results,
            )
        if metric == "max":
            return AggregatedExecutionResults(
                violation_rate=max(rates),
                violation_amount_s=max(amounts),
                violation_relative_mean=max(relatives),
                cost=max(costs),
                scenario_results=results,
            )

        # pNN quantile
        if metric.startswith("p") and metric[1:].isdigit():
            q = int(metric[1:]) / 100.0
            return AggregatedExecutionResults(
                violation_rate=float(np.quantile(rates, q)),
                violation_amount_s=float(np.quantile(amounts, q)),
                violation_relative_mean=float(np.quantile(relatives, q)),
                cost=float(np.quantile(costs, q)),
                scenario_results=results,
            )

        raise ValueError(f"Unknown aggregation metric: {metric!r}")

primary_violation(slo_metric)

Extract the primary violation value for the given SLO metric.

Source code in src/autoslo/workload_execution/aggregated_execution_results.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def primary_violation(self, slo_metric: str | SloMetric) -> float:
    """Extract the primary violation value for the given SLO metric."""
    slo_metric_obj = (
        slo_metric
        if isinstance(slo_metric, SloMetric)
        else SloMetric(slo_metric)
    )

    if slo_metric_obj == SloMetric.BINARY:
        return self.violation_rate
    elif slo_metric_obj == SloMetric.ABSOLUTE_S:
        return self.violation_amount_s
    elif slo_metric_obj == SloMetric.RELATIVE:
        return self.violation_relative_mean
    else:
        raise ValueError(f"Unknown SLO metric: {slo_metric}")

_fmt_cell(agg_val, scenario_vals) staticmethod

Format a metric cell: aggregated value with dim min–max range.

Source code in src/autoslo/workload_execution/aggregated_execution_results.py 
46
47
48
49
50
51
52
53
54
55
56
@staticmethod
def _fmt_cell(
    agg_val: float,
    scenario_vals: list[float],
) -> str:
    """Format a metric cell: aggregated value with dim min–max range."""
    main = f"{agg_val:.4f}"
    if len(scenario_vals) >= 2:
        lo, hi = min(scenario_vals), max(scenario_vals)
        main += f"\n[dim]{lo:.4f} … {hi:.4f}[/dim]"
    return main

print_comparison(*entries, console, agg_method='p90', slo_metric='binary', highlight_best=True) staticmethod

Print a table comparing multiple AggregatedExecutionResults.

Parameters

*entries : (label, agg) pairs. Each gets one row. Use labels like "Initial (train)" / "Initial (val)" to distinguish splits. agg_method : Aggregation metric shown in the title. slo_metric : The SLO metric that was actually optimised ("binary", "absolute_s", or "relative"). The best cell in this column and in the Cost column is highlighted green; the best cell in the other two violation columns is highlighted yellow. highlight_best : Whether to highlight the best values in the table.

Source code in src/autoslo/workload_execution/aggregated_execution_results.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@staticmethod
def print_comparison(
    *entries: tuple[str, AggregatedExecutionResults],
    console: Any,
    agg_method: str = "p90",
    slo_metric: str | SloMetric = "binary",
    highlight_best: bool = True,
) -> None:
    """Print a table comparing multiple AggregatedExecutionResults.

    Parameters
    ----------
    *entries :
        ``(label, agg)`` pairs.  Each gets one row.  Use labels like
        ``"Initial (train)"`` / ``"Initial (val)"`` to distinguish splits.
    agg_method :
        Aggregation metric shown in the title.
    slo_metric :
        The SLO metric that was actually optimised (``"binary"``,
        ``"absolute_s"``, or ``"relative"``).  The best cell in this
        column and in the Cost column is highlighted green; the best
        cell in the other two violation columns is highlighted yellow.
    highlight_best :
        Whether to highlight the best values in the table.
    """
    # Map slo_metric → column index (0-2 are the three violation metrics).
    slo_metric_str = (
        slo_metric.value
        if isinstance(slo_metric, SloMetric)
        else slo_metric
    )
    _SLO_TO_COL = {"binary": 0, "absolute_s": 1, "relative": 2}
    targeted_col = _SLO_TO_COL.get(slo_metric_str, 0)
    cost_col = 3  # always

    table = Table(
        title=f"Comparison  [dim](agg: {agg_method})[/dim]",
        show_lines=True,
    )
    table.add_column("Config", justify="left")

    metric_labels = [
        "Viol. Rate",
        "Viol. Amt (s)",
        "Viol. Rel.",
        "Cost ($)",
    ]
    for ml in metric_labels:
        table.add_column(ml, justify="right")

    fmt = AggregatedExecutionResults._fmt_cell

    def _extract(
        agg: AggregatedExecutionResults,
    ) -> tuple[list[float], list[list[float]]]:
        aggs = [
            agg.violation_rate,
            agg.violation_amount_s,
            agg.violation_relative_mean,
            agg.cost,
        ]
        if agg.scenario_results:
            per = [
                [r.violation_rate for r in agg.scenario_results],
                [r.violation_amount_s for r in agg.scenario_results],
                [r.violation_relative_mean for r in agg.scenario_results],
                [r.total_cost for r in agg.scenario_results],
            ]
        else:
            per = [[], [], [], []]
        return aggs, per

    # Build row data.
    row_data: list[tuple[str, list[float], list[list[float]]]] = []
    for label, agg in entries:
        a, p = _extract(agg)
        row_data.append((label, a, p))

    # Find the best (lowest) aggregated value per column.
    n_metric = 4
    best_per_col: list[int] = []
    if row_data:
        for c in range(n_metric):
            best_per_col.append(
                min(
                    range(len(row_data)),
                    key=lambda i, _c=c: row_data[i][1][_c],  # type: ignore
                )
            )

    for row_idx, (label, aggs, per) in enumerate(row_data):
        cells: list[str] = []
        for c in range(n_metric):
            cell = fmt(aggs[c], per[c])
            if (
                len(row_data) > 1
                and (row_idx == best_per_col[c])
                and highlight_best
            ):
                if c == targeted_col or c == cost_col:
                    cell = f"[green]{cell}[/green]"
                else:
                    cell = f"[yellow]{cell}[/yellow]"
            cells.append(cell)
        table.add_row(label, *cells)

    console.print(table)

aggregate_from(results, metric='p90') staticmethod

Compute a summary statistic over scenario results.

Parameters

results : Per-scenario results to aggregate. metric : "mean", "max", or any "pNN" quantile.

Returns

AggregatedExecutionResults with all three violation metrics and cost.

Source code in src/autoslo/workload_execution/aggregated_execution_results.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@staticmethod
def aggregate_from(
    results: list[ExecutionResult], metric: str = "p90"
) -> AggregatedExecutionResults:
    """Compute a summary statistic over scenario results.

    Parameters
    ----------
    results :
        Per-scenario results to aggregate.
    metric :
        ``"mean"``, ``"max"``, or any ``"pNN"`` quantile.

    Returns
    -------
    AggregatedExecutionResults with all three violation metrics and cost.
    """
    if not results:
        return AggregatedExecutionResults(
            violation_rate=0.0,
            violation_amount_s=0.0,
            violation_relative_mean=0.0,
            cost=0.0,
            scenario_results=[],
        )

    rates = [r.violation_rate for r in results]
    amounts = [r.violation_amount_s for r in results]
    relatives = [r.violation_relative_mean for r in results]
    costs = [r.total_cost for r in results]

    if metric == "mean":
        return AggregatedExecutionResults(
            violation_rate=statistics.mean(rates),
            violation_amount_s=statistics.mean(amounts),
            violation_relative_mean=statistics.mean(relatives),
            cost=statistics.mean(costs),
            scenario_results=results,
        )
    if metric == "max":
        return AggregatedExecutionResults(
            violation_rate=max(rates),
            violation_amount_s=max(amounts),
            violation_relative_mean=max(relatives),
            cost=max(costs),
            scenario_results=results,
        )

    # pNN quantile
    if metric.startswith("p") and metric[1:].isdigit():
        q = int(metric[1:]) / 100.0
        return AggregatedExecutionResults(
            violation_rate=float(np.quantile(rates, q)),
            violation_amount_s=float(np.quantile(amounts, q)),
            violation_relative_mean=float(np.quantile(relatives, q)),
            cost=float(np.quantile(costs, q)),
            scenario_results=results,
        )

    raise ValueError(f"Unknown aggregation metric: {metric!r}")