streaming

js/ml/metrics/streaming.ts

Streaming metric accumulators for batched and sharded evaluation.

Every accumulator here holds constant memory regardless of how many observations it sees, and every one supports merge, so a DataLoader fan-out can score its shards in parallel realms and reduce the partial results at the end. merge returns a new accumulator and leaves both operands untouched.

Interfaces

interface StreamingMetric<Value> {

A metric that can be fed incrementally and combined across shards.

Readonly Properties

readonly count: number

Number of observations accumulated.

Methods

value(): Value

Current value of the metric.

reset(): void

Discard all accumulated state.

interface StreamingRegressionValue {

Regression error summary accumulated in one pass.

Properties

meanSquaredError: number

Mean squared error.

rootMeanSquaredError: number

Root mean squared error.

meanAbsoluteError: number

Mean absolute error.

r2: number

Coefficient of determination.

Classes

class StreamingMean implements StreamingMetric<number> {

Running mean over a stream of values.

Uses compensated summation, so a long run of small values added to a large total does not quietly lose precision.

import { StreamingMean } from 'fino:ml/metrics';

const mean = new StreamingMean();
mean.updateAll([1, 2, 3]);
mean.update(4);
console.log(mean.value()); // 2.5

Getters

get count(): number

Methods

update(value: number): void

Accumulate one value.

updateAll(values: ArrayLike<number>): void

Accumulate a batch of values.

value(): number

Mean of everything accumulated, or 0 before the first observation.

merge(other: StreamingMean): StreamingMean

Combine with another accumulator.

reset(): void

class StreamingVariance implements StreamingMetric<number> {

Running mean and variance in one pass.

Uses Welford's algorithm and the pairwise merge that goes with it, so the variance stays stable for large counts and for shards combined out of order — a naive sum-of-squares would cancel catastrophically here.

import { StreamingVariance } from 'fino:ml/metrics';

const stats = new StreamingVariance();
stats.updateAll([2, 4, 4, 4, 5, 5, 7, 9]);
console.log(stats.mean()); // 5
console.log(stats.variance()); // 4

Getters

get count(): number

Methods

update(value: number): void

Accumulate one value.

updateAll(values: ArrayLike<number>): void

Accumulate a batch of values.

mean(): number

Mean of everything accumulated.

variance(): number

Population variance.

sampleVariance(): number

Sample variance, with Bessel's correction.

standardDeviation(): number

Population standard deviation.

value(): number

Population variance, matching variance().

merge(other: StreamingVariance): StreamingVariance

Combine with another accumulator.

reset(): void

class StreamingAccuracy implements StreamingMetric<number> {

Running classification accuracy.

import { StreamingAccuracy } from 'fino:ml/metrics';

const acc = new StreamingAccuracy();
acc.updateAll([1, 0, 1], [1, 0, 0]);
console.log(acc.value().toFixed(4)); // 0.6667

Getters

get count(): number

Methods

update(trueLabel: Label, predictedLabel: Label): void

Accumulate one prediction.

updateAll(yTrue: ArrayLike<Label>, yPred: ArrayLike<Label>): void

Accumulate a batch of paired labels.

value(): number

Fraction correct so far, or 0 before the first observation.

merge(other: StreamingAccuracy): StreamingAccuracy

Combine with another accumulator.

reset(): void

class StreamingConfusionMatrix implements StreamingMetric<ConfusionMatrix> {

A confusion matrix built incrementally from batches.

Wraps ConfusionMatrix so that the full per-class report — precision, recall, F1, kappa, MCC — is available from a stream without holding the labels themselves. Label universes grow as new classes appear, and merge unions them.

import { StreamingConfusionMatrix } from 'fino:ml/metrics';

const running = new StreamingConfusionMatrix();
running.updateAll(['a', 'b'], ['a', 'a']);
running.updateAll(['b'], ['b']);
console.log(running.value().accuracy().toFixed(4)); // 0.6667

Constructors

constructor(labels?: readonly Label[])

Create an accumulator, optionally over a known label universe.

Fixing the labels up front makes an unexpected class an error instead of a silent schema change, and keeps zero-support classes in the report.

Getters

get count(): number

Methods

update(trueLabel: Label, predictedLabel: Label): void

Accumulate one prediction.

updateAll(yTrue: ArrayLike<Label>, yPred: ArrayLike<Label>): void

Accumulate a batch of paired labels.

value(): ConfusionMatrix

The matrix accumulated so far.

Throws before the first observation, when there is no label universe to report over.

merge(other: StreamingConfusionMatrix): StreamingConfusionMatrix

Combine with another accumulator.

reset(): void

class StreamingRegression implements StreamingMetric<StreamingRegressionValue> {

Running regression errors over a stream of predictions.

Reports MSE, RMSE, MAE, and R² together from a single pass, tracking the target variance with Welford's algorithm so R² needs no second pass over the data.

import { StreamingRegression } from 'fino:ml/metrics';

const running = new StreamingRegression();
running.updateAll([3, -0.5, 2, 7], [2.5, 0, 2, 8]);
console.log(running.value().meanSquaredError); // 0.375

Getters

get count(): number

Methods

update(trueValue: number, predicted: number): void

Accumulate one prediction.

updateAll(yTrue: ArrayLike<number>, yPred: ArrayLike<number>): void

Accumulate a batch of predictions.

value(): StreamingRegressionValue

Error summary so far. Every field is 0 before the first observation.

merge(other: StreamingRegression): StreamingRegression

Combine with another accumulator.

reset(): void