Skip to content

Analytics

Unify the datasets of an experiment (one run or many) into analysis-ready tables.

From the CLI, TUI and GUI

  • lb runs analyze --experiment ID (or --folder for every run of the output folder, written to data_exports/_folders/<folder>/) writes the tables to data_exports/<experiment>/, after printing a summary; --host and --workload filter them. Without either option, an interactive terminal opens a picker with a preview, then the summary with Unify, Filters and Cancel. The result ends with the equivalent command.
  • lb runs list → Analyze opens the same flow on the run's experiment.
  • lb run --analyze unifies the run's experiment when the run ends; an interactive lb run asks instead (default No), and --no-analyze never asks.
  • The GUI Analytics view lists the experiments, prepares the same summary and unifies in the background.

All of them use UnificationService from lb_app.api.

UnificationService

UnificationService(catalog, export_root, config_path=None)

List experiments, preview their unification, write it.

Source code in lb_app/services/unification_service.py
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    catalog: RunCatalogService,
    export_root: Path,
    config_path: Path | None = None,
) -> None:
    self._catalog = catalog
    # Absolute, so the printed output folder means the same from anywhere.
    self._export_root = export_root.resolve()
    self._config_path = config_path

Attributes

output_dir property

output_dir

Methods:

find

find(experiment_id)

The named experiment, or the whole folder for None.

Source code in lb_app/services/unification_service.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def find(self, experiment_id: str | None) -> ExperimentInfo:
    """The named experiment, or the whole folder for ``None``."""
    if experiment_id is None:
        folder = self._catalog.folder_experiment()
        if not folder.runs:
            raise UnificationError(f"No runs in {self.output_dir}")
        return folder
    found = self._catalog.get_experiment(experiment_id)
    if found is None:
        recent = [e.id for e in self._catalog.list_experiments()[:5]]
        hint = f"; recent experiments: {', '.join(recent)}" if recent else ""
        raise UnificationError(
            f"Experiment {experiment_id!r} not found in {self.output_dir}{hint}"
        )
    return found

list_targets

list_targets()

Experiments newest first, then the whole folder; [] without runs.

Source code in lb_app/services/unification_service.py
93
94
95
96
97
98
def list_targets(self) -> list[ExperimentInfo]:
    """Experiments newest first, then the whole folder; [] without runs."""
    experiments = self._catalog.list_experiments()
    if not experiments:
        return []
    return [*experiments, self._catalog.folder_experiment()]

prepare

prepare(experiment, hosts=(), workloads=())

Load and filter the experiment; nothing is written yet.

Source code in lb_app/services/unification_service.py
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
def prepare(
    self,
    experiment: ExperimentInfo,
    hosts: Sequence[str] = (),
    workloads: Sequence[str] = (),
) -> UnificationPreview:
    """Load and filter the experiment; nothing is written yet."""
    if not _has_pyarrow():
        raise UnificationError(
            "Writing Parquet needs pyarrow: "
            "pip install 'linux-benchmark-lib[controller]'"
        )
    _check_names("host", hosts, experiment.hosts)
    _check_names("workload", workloads, experiment.workloads)
    host_filter = _effective(hosts, experiment.hosts)
    workload_filter = _effective(workloads, experiment.workloads)
    data = load_experiment(experiment.runs).filter(host_filter, workload_filter)
    return UnificationPreview(
        experiment=experiment,
        data=data,
        hosts=host_filter,
        workloads=workload_filter,
        out_dir=self._out_dir(experiment),
        command=self._command(experiment, host_filter, workload_filter),
    )

write

write(preview)

Write the previewed tables, replacing an earlier unification.

Source code in lb_app/services/unification_service.py
142
143
144
145
146
147
148
149
150
151
def write(self, preview: UnificationPreview) -> list[Path]:
    """Write the previewed tables, replacing an earlier unification."""
    if preview.is_empty:
        raise UnificationError(
            f"Nothing to unify for {preview.experiment.id}: {preview.empty_reason}"
        )
    try:
        return preview.data.to_parquet(preview.out_dir)
    except Exception as exc:  # disk, permissions, pyarrow: say why, no traceback
        raise UnificationError(f"Could not write {preview.out_dir}: {exc}") from exc

UnificationPreview dataclass

UnificationPreview(experiment, data, hosts, workloads, out_dir, command)

What a unification will write, loaded once and shown before writing.

Attributes

command instance-attribute

command

data instance-attribute

data

empty_reason property

empty_reason

experiment instance-attribute

experiment

hosts instance-attribute

hosts

is_empty property

is_empty

out_dir instance-attribute

out_dir

row_counts property

row_counts

workloads instance-attribute

workloads

Methods:

summary_rows

summary_rows()

Label/value pairs every UI shows before the user confirms.

Source code in lb_app/services/unification_service.py
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
def summary_rows(self) -> list[tuple[str, str]]:
    """Label/value pairs every UI shows before the user confirms."""
    report = self.data.report
    hosts = self.hosts or self.experiment.hosts
    workloads = self.workloads or self.experiment.workloads
    rows = [
        ("Experiment", self.experiment.id),
        ("Kind", self.experiment.kind),
        ("Runs", str(len(self.experiment.runs))),
        ("Hosts", ", ".join(hosts) or "-"),
        ("Workloads", ", ".join(workloads) or "-"),
    ]
    rows += [(f"Rows: {name}", str(n)) for name, n in self.row_counts.items()]
    rows += [
        ("Failed repetitions", str(len(report.failed_repetitions))),
        ("Load errors", str(len(report.errors))),
        ("Load warnings", str(len(report.warnings))),
    ]
    shown = report.errors[:MAX_SUMMARY_ERRORS]
    rows += [(f"Error: {e['source']}", e["message"]) for e in shown]
    hidden = len(report.errors) - len(shown)
    if hidden:
        rows.append(("More errors", f"{hidden} more in load_report.json"))
    if self.is_empty:
        rows.append(("Nothing to unify", self.empty_reason))
    rows.append(("Output", str(self.out_dir)))
    return rows

Unified datasets

load_experiment(runs) reads every datasets.json manifest of the given runs and returns runs, host_info, repetitions, results and samples as pandas DataFrames plus a LoadReport; ExperimentData.to_parquet(dir) writes them.

load_experiment

load_experiment(runs)

Load every declared dataset of runs into unified tables.

Source code in lb_analytics/unify/loader.py
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
def load_experiment(runs: Sequence[RunInfo]) -> ExperimentData:
    """Load every declared dataset of ``runs`` into unified tables."""
    report = LoadReport()
    frames: dict[str, list[pd.DataFrame]] = {
        "results": [],
        "samples": [],
        "host_info": [],
    }
    run_rows: list[dict[str, Any]] = []
    repetition_rows: list[dict[str, Any]] = []
    for run in runs:
        root = run.output_root
        run_rows.append(_run_row(run, report))
        for manifest_dir in _manifest_dirs(root):
            try:
                manifest = read_manifest(manifest_dir)
            except Exception as exc:
                report.error(str(manifest_dir), f"invalid {MANIFEST_FILENAME}: {exc}")
                continue
            ctx = SourceContext(
                run.run_id,
                _host_of(root, manifest_dir, manifest),
                manifest.workload,
                manifest.plugin,
            )
            _load_manifest(manifest, manifest_dir, ctx, frames, report)
            repetition_rows.extend(_repetitions(manifest, manifest_dir, ctx, report))
        _report_unmanifested(root, report)
    return ExperimentData(
        runs=_runs_frame(run_rows),
        host_info=_concat(frames["host_info"], HOST_INFO_COLUMNS),
        repetitions=_repetitions_frame(repetition_rows),
        results=_concat(frames["results"], RESULTS_COLUMNS),
        samples=_concat(frames["samples"], SAMPLES_COLUMNS),
        report=report,
    )

ExperimentData dataclass

ExperimentData(runs, host_info, repetitions, results, samples, report)

Attributes

host_info instance-attribute

host_info

is_empty property

is_empty

repetitions instance-attribute

repetitions

report instance-attribute

report

results instance-attribute

results

row_counts property

row_counts

runs instance-attribute

runs

samples instance-attribute

samples

Methods:

filter

filter(hosts=(), workloads=())

Keep the given hosts and workloads; an empty filter keeps everything.

A filter applies to every table that has its column, so runs is never filtered and host_info only by host.

Source code in lb_analytics/unify/experiment.py
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def filter(
    self, hosts: Sequence[str] = (), workloads: Sequence[str] = ()
) -> ExperimentData:
    """Keep the given hosts and workloads; an empty filter keeps everything.

    A filter applies to every table that has its column, so ``runs`` is
    never filtered and ``host_info`` only by host.
    """
    tables: dict[str, pd.DataFrame] = {}
    for name in TABLES:
        frame = getattr(self, name)
        if hosts and "host" in frame.columns:
            frame = frame[frame["host"].isin(hosts)]
        if workloads and "workload" in frame.columns:
            frame = frame[frame["workload"].isin(workloads)]
        tables[name] = frame.reset_index(drop=True)
    return ExperimentData(**tables, report=self.report)

to_parquet

to_parquet(out_dir)

Write one Parquet file per table plus load_report.json.

Source code in lb_analytics/unify/experiment.py
52
53
54
55
56
57
58
59
60
61
62
63
def to_parquet(self, out_dir: Path) -> list[Path]:
    """Write one Parquet file per table plus ``load_report.json``."""
    out_dir.mkdir(parents=True, exist_ok=True)
    paths = []
    for name in TABLES:
        path = out_dir / f"{name}.parquet"
        getattr(self, name).to_parquet(path, index=False)
        paths.append(path)
    report_path = out_dir / "load_report.json"
    report_path.write_text(json.dumps(self.report.to_dict(), indent=2))
    paths.append(report_path)
    return paths

Runs are grouped into experiments by RunCatalogService: list_experiments(), get_experiment(id) and folder_experiment() return ExperimentInfo objects whose runs feed load_experiment directly:

experiment = RunCatalogService(output_dir).get_experiment("exp-20261009-151200")
data = load_experiment(experiment.runs)

Prediction

lb_analytics.api can answer, per benchmark metric, how much of a machine's performance its specifications explain, and predict a machine that was not benchmarked. It is a library for notebooks; there is no command.

from lb_analytics.api import evaluate, fit, load_experiment, machines, predict, targets
from lb_app.api import RunCatalogService

catalog = RunCatalogService(output_dir)
data = [load_experiment(catalog.get_experiment(e).runs) for e in experiment_ids]
m, t = machines(*data), targets(*data)
report = evaluate(m, t, ["physical_cpus"], targets_filter="stress_ng")
target = "stress_ng/stress_ng/stress_ng_plugin/bogo_ops_per_s_real[stressor=cpu]"
model = fit(m, t, target, ["physical_cpus"])

new = machines(load_experiment(catalog.get_experiment("exp-new-host").runs))
predict(model, new.iloc[0])  # Prediction(value, low, high, unit, extrapolated)

Pass the same experiments to machines and targets: the two tables are joined on the machine name.

  • machines has one row per machine: a host with one set of features (cores, MHz, caches, memory, disk). The same host re-created with another size is another machine (host#<hash>); small drifts (a few KiB of MemTotal, bogomips, swap) do not split a host.
  • targets has one row per machine and target, with the median over the successful repetitions, the IQR and the count. A run whose host_info is missing counts for its host's machine when the host has only one; otherwise its results are left out with a warning.
  • fit is a log-linear regression: a coefficient of 0.9 on physical_cpus means doubling the cores multiplies the metric by about 1.87.
  • evaluate leaves each machine out in turn and compares the model with the mean of the other machines and with the nearest one on the chosen features. A model is worth using only where beats_baselines is true.
  • predict gives a range: the worst error seen leaving machines out, plus the run-to-run noise. With one repetition per machine the noise is unknown, and predict warns that the range covers only the validation error. A feature outside the training range, or a combination of features unlike any training machine (16 cores with the smallest memory), warns that the prediction is an extrapolation.

Limits: the machines must differ in the chosen features (identical VMs explain nothing), and features that move together (cores and memory on VM sizes that scale both) are refused as collinear; use one to three features with fewer than 15 machines; the kernel reports virtio disks as rotational, so disk_rotational is unreliable on VMs. The features of a machine come from any run on it, so a short baseline run is enough to predict it.

machines

machines(*data)

One row per machine: a host with one set of identifying features.

A host seen with one size is named after the host. A host seen with several sizes (a VM re-created) gives one machine per size, named host#<hash of its features> so the name does not depend on which experiments are passed. Feature values come from the machine's first run.

Source code in lb_analytics/predict/features.py
 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
def machines(*data: ExperimentData) -> pd.DataFrame:
    """One row per machine: a host with one set of identifying features.

    A host seen with one size is named after the host. A host seen with
    several sizes (a VM re-created) gives one machine per size, named
    ``host#<hash of its features>`` so the name does not depend on which
    experiments are passed. Feature values come from the machine's first run.
    """
    info = pd.concat([d.host_info for d in data], ignore_index=True)
    if info.empty:
        return pd.DataFrame(columns=MACHINE_COLUMNS)
    runs = pd.concat([d.runs for d in data], ignore_index=True)
    started = runs.drop_duplicates("run_id").set_index("run_id")["created_at"]
    rows = []
    for (run_id, host), group in info.groupby(["run_id", "host"], sort=False):
        values = {
            (category, name): value
            for category, name, value in group[
                ["category", "name", "value"]
            ].itertuples(index=False)
            if not pd.isna(value)
        }
        rows.append(
            {
                "host": host,
                "run_id": run_id,
                "started": started.get(run_id),
                **_features(values),
            }
        )
    frame = pd.DataFrame(rows).sort_values(
        ["started", "run_id"], na_position="last", kind="stable"
    )
    frame["mem_gib"] = (frame["mem_bytes"] / _GIB).round()
    frame["disk_gib"] = (frame["disk_bytes"] / _GIB).round()
    keys = ["host", *_IDENTITY, "mem_gib", "disk_gib"]
    groups = list(frame.groupby(keys, dropna=False, sort=False))
    sizes = Counter(key[0] for key, _ in groups)
    out = []
    for key, group in groups:
        host = str(key[0])
        name = host if sizes[host] == 1 else f"{host}#{_digest(key[1:])}"
        first = group.iloc[0].to_dict()
        out.append({**first, "machine": name, "runs": list(group["run_id"])})
    return pd.DataFrame(out, columns=MACHINE_COLUMNS)

targets

targets(*data)

Median, IQR and count of every target on every machine.

Failed repetitions are left out. A run whose host_info is missing counts for its host's machine when the host has only one; otherwise (a host never described, or one with several sizes) its results are left out with a warning, since they have no features to model.

Source code in lb_analytics/predict/features.py
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
def targets(*data: ExperimentData) -> pd.DataFrame:
    """Median, IQR and count of every target on every machine.

    Failed repetitions are left out. A run whose ``host_info`` is missing
    counts for its host's machine when the host has only one; otherwise (a
    host never described, or one with several sizes) its results are left out
    with a warning, since they have no features to model.
    """
    found = machines(*data)
    owner = {
        (run_id, host): machine
        for machine, host, run_ids in zip(
            found["machine"], found["host"], found["runs"], strict=True
        )
        for run_id in run_ids
    }
    results = pd.concat([d.results for d in data], ignore_index=True)
    reps = pd.concat([d.repetitions for d in data], ignore_index=True)
    keys = ["run_id", "host", "workload", "repetition"]
    failed_rows = reps.loc[reps["success"].eq(False), keys]
    failed = set(failed_rows.itertuples(index=False, name=None))
    keep = [
        key not in failed for key in results[keys].itertuples(index=False, name=None)
    ]
    results = results[keep].copy()
    by_host = found.groupby("host")["machine"].unique()
    results["machine"] = [
        owner.get((run_id, host))
        or (by_host[host][0] if host in by_host and len(by_host[host]) == 1 else None)
        for run_id, host in zip(results["run_id"], results["host"], strict=True)
    ]
    unowned = results.loc[results["machine"].isna(), "host"].unique().tolist()
    if unowned:
        warnings.warn(
            f"Results of {', '.join(map(str, unowned))} are left out: no "
            "host_info tells which machine produced them.",
            stacklevel=2,
        )
    results["value"] = pd.to_numeric(results["value"], errors="coerce")
    results = results.dropna(subset=["machine", "value"])
    if results.empty:
        return pd.DataFrame(columns=TARGET_COLUMNS)
    results["target"] = _target_names(results)
    grouped = results.groupby(["machine", "target"], sort=True)
    out = grouped.agg(
        workload=("workload", "first"),
        metric=("metric", "first"),
        unit=("unit", "first"),
        median=("value", "median"),
        n=("value", "size"),
    )
    out["iqr"] = grouped["value"].quantile(0.75) - grouped["value"].quantile(0.25)
    return out.reset_index()[TARGET_COLUMNS]

fit

fit(machines, targets, target, features)

Fit log(median) = b0 + Σ bᵢ·log(featureᵢ) for one target.

disk_rotational enters as 0/1 instead of its log. The returned model carries its leave-one-machine-out error, which sets the range of predict, and the run-to-run noise of machines measured at least twice (None when none was).

Source code in lb_analytics/predict/model.py
 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
def fit(
    machines: pd.DataFrame,
    targets: pd.DataFrame,
    target: str,
    features: Sequence[str],
) -> Model:
    """Fit ``log(median) = b0 + Σ bᵢ·log(featureᵢ)`` for one target.

    ``disk_rotational`` enters as 0/1 instead of its log. The returned model
    carries its leave-one-machine-out error, which sets the range of
    ``predict``, and the run-to-run noise of machines measured at least twice
    (``None`` when none was).
    """
    features = list(features)
    table = _table(machines, targets, target, features)
    design = _design(table, features)
    coef = _solve(design, np.log(table["median"].to_numpy(float)))
    # _table refused collinear features, so the inverse exists.
    xtx_inv = np.linalg.inv(design.T @ design)
    held_out = _loo(table, features)
    held_out = held_out[held_out["method"] == "model"]
    ratios = held_out["predicted"].to_numpy(float) / held_out["actual"].to_numpy(float)
    log_errors = np.abs(np.log(ratios))
    repeated = table["n"] >= 2
    rel_iqr = (table["iqr"] / table["median"])[repeated]
    return Model(
        target=target,
        unit=_unit(table),
        features=tuple(features),
        intercept=float(coef[0]),
        coefficients={f: float(c) for f, c in zip(features, coef[1:], strict=True)},
        machines=tuple(table["machine"]),
        feature_ranges={
            f: (float(table[f].min()), float(table[f].max())) for f in features
        },
        max_log_error=float(log_errors.max()),
        rel_iqr=float(rel_iqr.median()) if repeated.any() else None,
        xtx_inv=tuple(tuple(float(v) for v in row) for row in xtx_inv),
        max_leverage=float(_leverage(design, xtx_inv).max()),
    )

evaluate

evaluate(machines, targets, features, targets_filter=None)

Leave-one-machine-out errors of the model and both baselines.

One row per target and method. Targets that cannot be fitted get one skipped row with the reason in note.

Source code in lb_analytics/predict/model.py
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
226
227
def evaluate(
    machines: pd.DataFrame,
    targets: pd.DataFrame,
    features: Sequence[str],
    targets_filter: str | None = None,
) -> pd.DataFrame:
    """Leave-one-machine-out errors of the model and both baselines.

    One row per target and method. Targets that cannot be fitted get one
    ``skipped`` row with the reason in ``note``.
    """
    features = list(features)
    _check_features(features)
    names = sorted(targets["target"].unique())
    if targets_filter:
        names = [name for name in names if targets_filter in name]
    rows: list[dict[str, Any]] = []
    for name in names:
        count = int((targets["target"] == name).sum())
        try:
            table = _table(machines, targets, name, features)
        except PredictionError as exc:
            rows.append(
                {
                    "target": name,
                    "method": "skipped",
                    "machines": count,
                    "note": str(exc),
                }
            )
            continue
        errors = _loo(table, features).groupby("method")["pct_error"]
        medians = errors.apply(lambda e: e.abs().median())
        maxima = errors.apply(lambda e: e.abs().max())
        wins = bool(medians["model"] < min(medians["mean"], medians["nearest"]))
        rows.extend(
            {
                "target": name,
                "method": method,
                "machines": count,
                "median_abs_pct_error": float(medians[method]),
                "max_abs_pct_error": float(maxima[method]),
                "beats_baselines": wins if method == "model" else None,
                "note": "",
            }
            for method in METHODS
        )
    return pd.DataFrame(rows, columns=EVALUATE_COLUMNS)

loo_predictions

loo_predictions(machines, targets, target, features)

Every held-out prediction behind evaluate for one target.

One row per machine and method: actual, predicted, pct_error.

Source code in lb_analytics/predict/model.py
166
167
168
169
170
171
172
173
174
175
176
177
def loo_predictions(
    machines: pd.DataFrame,
    targets: pd.DataFrame,
    target: str,
    features: Sequence[str],
) -> pd.DataFrame:
    """Every held-out prediction behind ``evaluate`` for one target.

    One row per machine and method: ``actual``, ``predicted``, ``pct_error``.
    """
    features = list(features)
    return _loo(_table(machines, targets, target, features), features)

predict

predict(model, machine)

Predict model.target for one machine, with the model's range.

The range is the worst leave-one-machine-out error plus the target's run-to-run noise. A feature outside the training range, or a combination of features unlike any training machine, warns and sets extrapolated.

Source code in lb_analytics/predict/model.py
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
def predict(model: Model, machine: Mapping[str, Any] | pd.Series) -> Prediction:
    """Predict ``model.target`` for one machine, with the model's range.

    The range is the worst leave-one-machine-out error plus the target's
    run-to-run noise. A feature outside the training range, or a combination
    of features unlike any training machine, warns and sets ``extrapolated``.
    """
    log_value = model.intercept
    extrapolated = False
    point = [1.0]
    for feature in model.features:
        x = to_number(machine.get(feature))
        if math.isnan(x) or (x <= 0 and feature not in BINARY_FEATURES):
            raise PredictionError(
                f"Feature '{feature}' is missing or not positive "
                f"({machine.get(feature)!r})."
            )
        low, high = model.feature_ranges[feature]
        if not low <= x <= high:
            extrapolated = True
            warnings.warn(
                f"{feature}={x:g} is outside the training range "
                f"[{low:g}, {high:g}]: the prediction is an extrapolation.",
                stacklevel=2,
            )
        term = x if feature in BINARY_FEATURES else math.log(x)
        point.append(term)
        log_value += model.coefficients[feature] * term
    row = np.array([point])
    leverage = float(_leverage(row, np.array(model.xtx_inv))[0])
    if not extrapolated and leverage > model.max_leverage * (1 + 1e-9):
        extrapolated = True
        warnings.warn(
            "This combination of features is unlike any training machine, "
            "although each one is in range: the prediction is an extrapolation.",
            stacklevel=2,
        )
    noise = model.rel_iqr
    if noise is None:
        warnings.warn(
            "The run-to-run noise is unknown (one repetition per machine): "
            "the range covers only the validation error.",
            stacklevel=2,
        )
        noise = 0.0
    value = math.exp(log_value)
    spread = math.exp(model.max_log_error + math.log1p(noise))
    return Prediction(value, value / spread, value * spread, model.unit, extrapolated)