Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,21 @@ Release tags match the version committed in `pyproject.toml`.

## [Unreleased]

### Changed

- `StreamingIsotonicCalibrator` fits with SciPy's `isotonic_regression`
instead of scikit-learn's `IsotonicRegression`, which spent over 90% of each
update validating inputs. Updates and half-life calibration are 11-27x
faster; outputs are bit-for-bit unchanged. Requires SciPy 1.12.
- Batches of 1,024+ labels find their bins, and 2,048+ probabilities
interpolate, by direct calculation on the evenly spaced grid rather than a
binary search, with identical results.
- `import streamcal` no longer loads scikit-learn (1.1 s to 0.08 s). Names that
need it load on first use. A process that only runs the streaming calibrator
peaks at 75 MB rather than 148 MB.
- `BatchCalibrator` stores labels as one byte, so history takes 9 rather than
16 bytes per observation, with identical fits.

### Fixed

- `compare_prequential` no longer times batches with no observed labels as
Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ fixed-size per-bin sufficient statistics, can forget stale history, and makes
the prediction-before-observation order explicit so a forecast never uses its
own outcome.

The numerical fit is scikit-learn's weighted isotonic regression. Streamcal adds
The numerical fit is SciPy's weighted isotonic regression. Streamcal adds
the compact streaming state, prediction-age forgetting, and evaluation needed
to choose a quality/resource operating point.

Expand Down
147 changes: 147 additions & 0 deletions benchmarks/speed.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
"""Time and memory receipts for the hot paths, comparable across commits.

Run from the repository root, optionally pointing at another checkout's
``src`` to compare commits on the same machine::

python -m benchmarks.speed # this checkout
python -m benchmarks.speed --src ../other/src # another commit

Timings are medians of repeated calls with BLAS pinned to one thread. They
describe this machine; compare two commits by running both here.
"""

import argparse
import json
import os
import statistics
import subprocess
import sys
import time
from functools import partial
from pathlib import Path


def median_us(call, repeats):
"""Median wall time of ``call`` in microseconds."""
samples = []
for _ in range(repeats):
start = time.perf_counter_ns()
call()
samples.append(time.perf_counter_ns() - start)
return statistics.median(samples) / 1e3


def probe_import(src):
"""Time ``import streamcal`` and its first update in this (fresh) process.

Import time only means something in a process that has not imported
anything yet, so :func:`import_cost` runs this in a new interpreter each
time, through ``--probe-import``. That is also why this module imports
NumPy inside :func:`main` rather than at the top.
"""
import resource

sys.path.insert(0, src)
start = time.perf_counter()
from streamcal import StreamingIsotonicCalibrator

imported = time.perf_counter()
StreamingIsotonicCalibrator().update([0.2, 0.8], [0, 1])
updated = time.perf_counter()
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
return {
"import_ms": (imported - start) * 1e3,
"first_update_ms": (updated - imported) * 1e3,
# ru_maxrss is bytes on macOS and kilobytes on Linux.
"peak_rss_mb": peak * (1 if sys.platform == "darwin" else 1024) / 2**20,
"imports_sklearn": "sklearn" in sys.modules,
}


def import_cost(src, runs=7):
"""Median of :func:`probe_import` over fresh interpreters."""
env = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}
command = [sys.executable, "-m", "benchmarks.speed", "--probe-import"]
rows = [
json.loads(
subprocess.run( # noqa: S603 - this script, run by the same interpreter
[*command, "--src", src],
capture_output=True,
text=True,
check=True,
env=env,
).stdout
)
for _ in range(runs)
]
return {
key: statistics.median(row[key] for row in rows)
for key in ("import_ms", "first_update_ms", "peak_rss_mb")
} | {"imports_sklearn": rows[0]["imports_sklearn"]}


def main():
"""Measure and print the receipts as JSON."""
parser = argparse.ArgumentParser()
parser.add_argument("--src", default=str(Path("src").resolve()))
parser.add_argument("--output")
parser.add_argument("--probe-import", action="store_true", help="internal")
args = parser.parse_args()
if args.probe_import:
print(json.dumps(probe_import(args.src)))
return
# Probe imports before this process loads anything heavy: on Linux a child's
# peak-memory counter starts from the parent's size when it is forked.
results = {"import": import_cost(args.src)}
import numpy as np

sys.path.insert(0, args.src)
from streamcal import (
BatchCalibrator,
StreamingIsotonicCalibrator,
compare_prequential,
)

rng = np.random.default_rng(0)
for bins in (20, 100, 1000):
for size in (1, 100, 10_000):
p = rng.uniform(0, 1, size)
y = (rng.random(size) < p).astype(float)
repeats = 2000 if size < 10_000 else 300
decayed = StreamingIsotonicCalibrator(n_bins=bins, decay=0.9).update(p, y)
timed = StreamingIsotonicCalibrator(n_bins=bins, half_life_seconds=100.0)
timed.update(p, y, prediction_times=np.zeros(size), observed_at=1.0)
key = f"B={bins} m={size}"
results[f"update {key}"] = median_us(partial(decayed.update, p, y), repeats)
results[f"calibrate {key}"] = median_us(
partial(decayed.calibrate, p), repeats
)
results[f"half-life calibrate {key}"] = median_us(
partial(timed.calibrate, p, current_time=2.0), repeats
)
for history in (10_000, 100_000, 1_000_000):
p = rng.uniform(0, 1, history)
y = (rng.random(history) < p).astype(float)
batch = BatchCalibrator(refit_every=10**9).update(p, y)
results[f"batch update, no refit, H={history:,}"] = median_us(
partial(batch.update, p[:10], y[:10]), 50
)
results[f"batch history bytes, H={history:,}"] = batch.history_bytes
size = 3000
p = rng.uniform(0, 1, size)
y = (rng.random(size) < p).astype(int)
start = time.perf_counter()
compare_prequential(
p, y, {"stream": lambda: StreamingIsotonicCalibrator(n_bins=20)}, batch_size=1
)
results["compare_prequential us/observation, batch 1"] = (
(time.perf_counter() - start) / size * 1e6
)
text = json.dumps(results, indent=1)
if args.output:
Path(args.output).write_text(text)
print(text)


if __name__ == "__main__":
main()
21 changes: 13 additions & 8 deletions docs/formalization.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,8 +50,9 @@ $$
$$

Each target is the bin's observed positive rate, shrunk toward the bin center.
Streamcal passes these weights and targets to scikit-learn's
`IsotonicRegression`; it does not implement the isotonic solver. A bin with
Streamcal passes these weights and targets to SciPy's
`scipy.optimize.isotonic_regression`; it does not implement the isotonic
solver. A bin with
zero weight (possible only when $\lambda = 0$) is left out of the fit, and its
value is interpolated from its neighbours. A raw-history implementation has to
use the same convention for the two to agree, because the objective leaves
Expand Down Expand Up @@ -106,20 +107,24 @@ $$
$$

or 808 bytes at $B = 20$, however many observations $H$ it has seen. Keeping
the raw history instead takes $16H$ bytes (a float64 probability and label per
observation), before counting the fitted model, and a rolling window of $W$
observations takes $16W$ bytes. These figures cover the numerical arrays only,
the raw history instead, as `BatchCalibrator` does, takes $9H$ bytes (a float64
probability and a one-byte label per observation), before counting the fitted
model, and a rolling window of $W$ observations takes $9W$ bytes. These figures cover the numerical arrays only,
not Python objects or the interpreter.

An update with a batch of $m$ labels costs

| Step | Work |
|---|---|
| Find each label's bin | $O(m \log B)$ |
| Find each label's bin | $O(m)$ |
| Decay and accumulate $N_b$, $S_b$ | $O(m + B)$ |
| Refit on $B$ aggregated bins | $O(B \log B)$ |
| Refit on $B$ aggregated bins | $O(B)$ |

so $O(m \log B + B \log B)$ time and $O(m + B)$ temporary memory in total.
so $O(m + B)$ time and $O(m + B)$ temporary memory in total. Because the bins
are evenly spaced, a label's bin is computed directly rather than searched
for; batches under 1,024 labels use a binary search instead, $O(m \log B)$,
which is faster at that size and gives the same bins. The bins are already in
order, so the pool-adjacent-violators refit runs in linear time.
Neither depends on $H$. Calibrating interpolates against $B$ centers; in
half-life mode it also refits the $B$ bins for the elapsed time.

Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ classifiers = [
dependencies = [
"numpy>=2.5.3",
"scikit-learn>=1.8",
"scipy>=1.11",
"scipy>=1.12",
]

[project.urls]
Expand Down
58 changes: 50 additions & 8 deletions src/streamcal/__init__.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,22 @@
"""Bounded-memory calibration for binary probability streams."""

from __future__ import annotations

from importlib import import_module
from importlib.metadata import version
from typing import TYPE_CHECKING, Any

from streamcal.batch import BatchCalibrator
from streamcal.calibrators import CalibratorDiagnostics, StreamingIsotonicCalibrator
from streamcal.evaluation import (
ConfigurationResult,
StreamingIsotonicConfig,
TradeoffReport,
compare_prequential,
)
from streamcal.metrics import binned_calibration_error, brier_score

if TYPE_CHECKING:
from streamcal.batch import BatchCalibrator
from streamcal.evaluation import (
ConfigurationResult,
StreamingIsotonicConfig,
TradeoffReport,
compare_prequential,
)
from streamcal.metrics import binned_calibration_error, brier_score

__version__ = version("streamcal")

Expand All @@ -25,3 +31,39 @@
"brier_score",
"compare_prequential",
]

# Names whose modules import scikit-learn. They load on first access, so a
# service that only runs StreamingIsotonicCalibrator never imports it.
_LAZY = {
"BatchCalibrator": "streamcal.batch",
"ConfigurationResult": "streamcal.evaluation",
"StreamingIsotonicConfig": "streamcal.evaluation",
"TradeoffReport": "streamcal.evaluation",
"compare_prequential": "streamcal.evaluation",
"binned_calibration_error": "streamcal.metrics",
"brier_score": "streamcal.metrics",
}


def __getattr__(name: str) -> Any:
"""Import a public name from its module on first access.

Args:
name: Attribute requested from the package.

Returns:
The requested public object.

Raises:
AttributeError: If ``name`` is not a public name of the package.
"""
if name not in _LAZY:
raise AttributeError(f"module 'streamcal' has no attribute {name!r}")
value = getattr(import_module(_LAZY[name]), name)
globals()[name] = value
return value


def __dir__() -> list[str]:
"""List public names, including those not yet imported."""
return sorted(set(globals()) | set(__all__))
12 changes: 7 additions & 5 deletions src/streamcal/batch.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,9 @@ def __init__(
def reset(self) -> Self:
"""Discard history and the fitted map."""
self._probabilities = np.empty(0)
self._outcomes = np.empty(0)
# Labels are 0 or 1, so one byte each is enough: 9 rather than 16
# bytes of history per observation. Fits convert them back to float.
self._outcomes = np.empty(0, dtype=np.uint8)
self._model: CalibratedClassifierCV | IsotonicRegression | None = None
self.n_updates = 0
self.n_observations = 0
Expand All @@ -110,7 +112,7 @@ def freeze(self) -> Self:
if not self.is_ready:
raise ValueError("cannot freeze before a successful fit")
self._probabilities = np.empty(0)
self._outcomes = np.empty(0)
self._outcomes = np.empty(0, dtype=np.uint8)
self.frozen = True
return self

Expand Down Expand Up @@ -144,7 +146,7 @@ def update(self, probabilities: ArrayLike, outcomes: ArrayLike) -> Self:
if self.frozen:
return self
retained_p = np.concatenate((self._probabilities, p))
retained_y = np.concatenate((self._outcomes, y))
retained_y = np.concatenate((self._outcomes, y.astype(np.uint8)))
if self.window_size is not None:
retained_p = retained_p[-self.window_size :].copy()
retained_y = retained_y[-self.window_size :].copy()
Expand All @@ -155,13 +157,13 @@ def update(self, probabilities: ArrayLike, outcomes: ArrayLike) -> Self:
if self.method == "isotonic":
model = IsotonicRegression(
y_min=0.0, y_max=1.0, out_of_bounds="clip"
).fit(retained_p, retained_y)
).fit(retained_p, retained_y.astype(np.float64))
else:
model = CalibratedClassifierCV(
FrozenEstimator(ProbabilityClassifier()),
method=self.method,
cv=_CalibrationRows(),
).fit(retained_p[:, None], retained_y)
).fit(retained_p[:, None], retained_y.astype(np.float64))
self._probabilities, self._outcomes = retained_p, retained_y
self._model = model
self.n_updates += 1
Expand Down
Loading
Loading