Reducer Reference

pond-ts ships 15 built-in reducers, plus percentile patterns (p50, p95, …) and top-N-by-frequency patterns (top3, top10, …). Every built-in works in aggregate(), reduce(), rolling(), and collapse(). The two array-producing reducers (unique and top(n)) also slot into arrayAggregate().

Choosing a reducer

Quick guide — skip ahead to the detailed sections when you know what you want:

You want…	Reducer
Running total or counter	`sum`, `count`
Central tendency	`avg`, `median`
Spread / volatility	`stdev`, `difference`
Extremes	`min`, `max`
Tail behavior	`p95`, `p99`, `p99.9`
First or last seen	`first`, `last`
Constant metadata column (e.g. `host` after bucketing)	`keep`
Distinct values in a bucket	`unique`
Most-frequent values	`top(n)`
All values (with duplicates, for downstream computation)	`samples`
None of the above	Custom reducer function

Pandas analogues in order: .sum(), .count(), .mean(), .median(), .std(), .max() − .min(), .min(), .max(), .quantile(0.95), .first(), .last(), a value that matches .nunique() == 1, .unique(), .value_counts().head(n), and any lambda you want.

Type narrowing on `reduce(mapping)`

TimeSeries.reduce(mapping) narrows each output field to the reducer's actual value type — no as casts at the call site:

const result = series.reduce({
  cpu: 'avg', //   -> number | undefined
  latency: 'p95', //   -> number | undefined
  host: 'unique', //   -> ReadonlyArray<string> | undefined   (source: kind 'string')
  values: 'unique', //   -> ReadonlyArray<number> | undefined   (source: kind 'number')
  paths: 'top3', //   -> ReadonlyArray<string> | undefined
  status: 'last', //   -> string | undefined   (source column kind)
});

const avgCpu: number | undefined = result.cpu;
const hosts: ReadonlyArray<string> | undefined = result.host;

Array-output reducers thread source column kind through. 'unique' and any top${number} on a kind: 'string' column narrow to ReadonlyArray<string>, on a kind: 'number' column to ReadonlyArray<number>, on a kind: 'boolean' column to ReadonlyArray<boolean>. Array-kind source columns fall back to the wide ReadonlyArray<ScalarValue> union because tracking element kind on an array column is out of scope for the schema.

Custom reducer functions and AggregateOutputSpec entries ({ from, using, kind? }) fall back to ColumnValue | undefined because their output kind is set at runtime and the type system can't see through it.

Numeric reducers

These operate on numeric values only. Non-numeric and undefined values are filtered out before the reducer runs.

`sum`

Total of all numeric values in the bucket.

series.aggregate(Sequence.every('5m'), { requests: 'sum' });

Returns 0 for an empty bucket (there are zero values to sum).

`avg`

Arithmetic mean of numeric values.

series.reduce('cpu', 'avg');

Returns undefined for an empty bucket.

`min`

Smallest numeric value.

series.aggregate(Sequence.every('1h'), { temperature: 'min' });

Returns undefined for an empty bucket.

`max`

Largest numeric value.

series.aggregate(Sequence.every('1h'), { temperature: 'max' });

Returns undefined for an empty bucket.

`median`

Middle value of the sorted numeric values. For even-length sets, linearly interpolates between the two middle values (equivalent to p50).

series.reduce('latency', 'median');

Returns undefined for an empty bucket.

`stdev`

Population standard deviation. Measures how spread out the values are around the mean.

series.aggregate(Sequence.every('10m'), { latency: 'stdev' });

Returns 0 when all values are identical. Returns undefined for an empty bucket.

`difference`

Range within the bucket: max - min. Useful for measuring volatility or stability within a time window.

series.aggregate(Sequence.every('10m'), { temperature: 'difference' });

Returns 0 for a single value. Returns undefined for an empty bucket.

`count`

Number of defined (non-undefined) values. Works on any column type, not just numeric.

series.aggregate(Sequence.every('1h'), { errors: 'count' });

Returns 0 for an empty bucket.

Percentiles: `p0` through `p100`

The p prefix followed by a number (0--100) computes that percentile with linear interpolation between adjacent ranks.

series.aggregate(Sequence.every('5m'), {
  p50: { from: 'latency', using: 'p50', kind: 'number' },
  p95: { from: 'latency', using: 'p95', kind: 'number' },
  p99: { from: 'latency', using: 'p99', kind: 'number' },
});

Shorthand when aggregating a single column:

series.reduce('latency', 'p95');

Fractional percentiles are supported: 'p99.9'.

p0 returns the minimum, p50 is equivalent to median, p100 returns the maximum.

Returns undefined for an empty bucket.

Any-type reducers

These work on values of any column kind (number, string, boolean).

`first`

First non-undefined value in the bucket (by event order).

series.aggregate(Sequence.every('1h'), { status: 'first' });

Returns undefined for an empty bucket.

`last`

Last non-undefined value in the bucket.

series.aggregate(Sequence.every('1h'), { status: 'last' });

Returns undefined for an empty bucket.

`keep`

Returns the value if every value in the bucket is identical; undefined otherwise. Useful for preserving constant metadata columns (like host) through aggregation.

const hourly = series.aggregate(Sequence.every('1h'), {
  cpu: 'avg',
  host: 'keep',
});

If the series was already partitioned by groupBy('host'), keep will always return the host name since all events in each group share the same value.

Returns undefined for an empty bucket or mixed values.

Array-producing reducers

These two reducers output an array column (kind: 'array') instead of a scalar. They drive most of the tagging / distinct-value workflows — see the Array Columns guide for the full treatment.

`unique`

Distinct non-undefined values, sorted. Works on any column kind.

On a scalar source column. Produces one array per bucket containing every distinct value seen, sorted deterministically (numbers < strings < booleans, JS default within a kind).

const hosts = new TimeSeries({
  name: 'hosts',
  schema: [
    { name: 'time', kind: 'time' },
    { name: 'host', kind: 'string' },
  ] as const,
  rows: [
    // bucket [0s, 3s) ─────────────────
    [0, 'api-1'],
    [1000, 'api-1'],
    [2000, 'api-2'],
    // bucket [3s, 6s) ─────────────────
    [3500, 'api-3'],
    [4000, 'api-1'],
    [5500, 'api-1'],
  ],
});

const agg = hosts.aggregate(Sequence.every('3s'), {
  host: 'unique', // 'string' -> 'array'
});

// agg.at(0)!.get('host')  =>  ['api-1', 'api-2']
// agg.at(1)!.get('host')  =>  ['api-1', 'api-3']

On an array-kind source column. unique flattens one level and takes the set union of elements across every array in the bucket.

const tagged = new TimeSeries({
  name: 'tagged',
  schema: [
    { name: 'time', kind: 'time' },
    { name: 'tags', kind: 'array' },
  ] as const,
  rows: [
    // bucket [0s, 3s) ─────────────────
    [0, ['5xx', 'timeout']],
    [500, ['5xx']],
    [1000, ['retry']],
    // bucket [3s, 6s) ─────────────────
    [3500, ['timeout']],
    [4000, []], // empty arrays contribute nothing
  ],
});

const agg = tagged.aggregate(Sequence.every('3s'), {
  tags: 'unique',
});

// agg.at(0)!.get('tags')  =>  ['5xx', 'retry', 'timeout']
//   union of ['5xx','timeout'] + ['5xx'] + ['retry'] = {5xx, timeout, retry}
// agg.at(1)!.get('tags')  =>  ['timeout']

Returns [] for an empty bucket.

`top(n)` / `top${number}`

Top N values by frequency, sorted by count descending. Ties break by scalar order (numbers < strings < booleans, JS default within a kind). If N exceeds the unique count, all unique values are returned. Parsed from the string pattern `top${number}` (like p${number}); the top(n) helper returns the typed string literal — either form is accepted.

On a scalar source column. Counts occurrences in the bucket and keeps the most frequent N.

import { top } from 'pond-ts';

const hosts = new TimeSeries({
  name: 'hosts',
  schema: [
    { name: 'time', kind: 'time' },
    { name: 'host', kind: 'string' },
  ] as const,
  rows: [
    // bucket [0s, 3s) ─────────────────
    [0, 'api-1'],
    [500, 'api-2'],
    [1000, 'api-1'],
    [1500, 'api-3'],
    [2500, 'api-1'],
    // bucket [3s, 6s) ─────────────────
    [3500, 'api-4'],
    [4000, 'api-4'],
    [4500, 'api-2'],
  ],
});

const agg = hosts.aggregate(Sequence.every('3s'), {
  host: top(2), // 'string' -> 'array', top 2 most frequent
});

// agg.at(0)!.get('host')  =>  ['api-1', 'api-2']
//   counts: api-1=3, api-2=1, api-3=1. api-1 wins outright;
//   api-2 beats api-3 on the tie-break (scalar order).
// agg.at(1)!.get('host')  =>  ['api-4', 'api-2']
//   counts: api-4=2, api-2=1.

The string form is interchangeable: { host: 'top2' } does the same thing.

On an array-kind source column. Counts elements across every array in the bucket, not the arrays themselves.

const tagged = new TimeSeries({
  name: 'tagged',
  schema: [
    { name: 'time', kind: 'time' },
    { name: 'tags', kind: 'array' },
  ] as const,
  rows: [
    [0, ['5xx', 'timeout']],
    [500, ['5xx']],
    [1000, ['retry', '5xx']],
    [1500, ['timeout']],
  ],
});

const agg = tagged.aggregate(Sequence.every('3s'), {
  tags: top(3),
});

// agg.at(0)!.get('tags')  =>  ['5xx', 'timeout', 'retry']
//   flattened elements: 5xx, timeout, 5xx, retry, 5xx, timeout
//   counts: 5xx=3, timeout=2, retry=1 — already three unique, all kept.

Invalid names ('top0', 'top-1') throw at resolve time.

Rolling-window complexity is O(1) per add/remove (count map); final sort on snapshot is O(k log k) where k is the distinct value count.

Returns [] for an empty bucket.

`samples`

All defined values in the bucket, returned as an array in arrival order. Duplicates are preserved (this is what distinguishes samples from unique and top${N}). Reach for samples when a downstream computation needs the raw value list — anomaly density against an externally-computed baseline, custom thresholding, histogramming, or feeding into a non-pond pipeline.

const cpu = new TimeSeries({
  name: 'cpu',
  schema: [
    { name: 'time', kind: 'time' },
    { name: 'cpu', kind: 'number' },
  ] as const,
  rows: [
    // bucket [0s, 3s) ─────────────────
    [0, 0.4],
    [1000, 0.5],
    [2000, 0.4],
    // bucket [3s, 6s) ─────────────────
    [3500, 0.6],
    [4000, 0.7],
  ],
});

const agg = cpu.aggregate(Sequence.every('3s'), {
  vals: { from: 'cpu', using: 'samples' },
});

// agg.at(0)!.get('vals')  =>  [0.4, 0.5, 0.4]   — duplicates preserved
// agg.at(1)!.get('vals')  =>  [0.6, 0.7]

Like unique, samples flattens one level on array-kind source columns:

tagged.aggregate(Sequence.every('3s'), {
  flat: { from: 'tags', using: 'samples' },
});
// bucket [0s, 3s) — sources [['5xx','timeout'], ['5xx'], ['retry']]
//   .get('flat')  =>  ['5xx', 'timeout', '5xx', 'retry']   — flat, ordered, with duplicates

Use on bounded windows. Memory is O(window size); per-event cost is O(1) add and O(1) remove (Map-keyed by event index); snapshot is O(N) for the array copy. For unbounded reduce(mapping) on a large series, expect to allocate one array of every defined value — count or a custom function is usually cheaper if you don't actually need the values.

Returns [] for an empty bucket.

Custom reducers

Supply a function wherever a built-in name is accepted. The function receives all values in the bucket (including undefined entries) and returns a scalar.

series.aggregate(Sequence.every('10m'), {
  value: (values) => {
    const nums = values.filter((v): v is number => typeof v === 'number');
    return nums.length === 0 ? undefined : nums.reduce((a, b) => a + b, 0);
  },
});

Custom reducers work in aggregate(), reduce(), rolling(), collapse(), and (since v0.14.1) every live counterpart — live.aggregate(), live.rolling(), and partitionBy(c).rolling(...).

Performance characteristic on live

Built-in reducers maintain incremental state — add/remove/snapshot are each O(1). Custom functions don't have that machinery, so the live adapter buffers values and re-runs the function over the entire current window at each snapshot() call. Per-event cost is still O(1) for state maintenance, but snapshot() is O(window size) — the function re-runs every time the accumulator emits.

For high-throughput live use, prefer:

a built-in reducer (avg, p95, stdev, ...) when one fits, or
'samples' to collapse the window to a value list once, then run your custom logic on the consumer side after the snapshot has fired.

Custom-function reducers shine on low-rate streams where convenience matters more than per-snapshot cost — debug aggregations, prototype pipelines, dashboards under ~1k events/sec.

Empty-bucket behavior

Reducer	Empty bucket
`count`	`0`
`sum`	`0`
`avg`	`undefined`
`min`	`undefined`
`max`	`undefined`
`median`	`undefined`
`stdev`	`undefined`
`difference`	`undefined`
`first`	`undefined`
`last`	`undefined`
`keep`	`undefined`
`p50`...	`undefined`
`unique`	`[]`
`top3`...	`[]`
`samples`	`[]`

Rolling window behavior

All built-in reducers work with rolling(). Most use O(1) incremental add/remove:

Reducer	Rolling complexity
`sum`, `avg`, `count`, `stdev`	O(1) per event
`min`, `max`	O(1) amortized (monotone deque)
`first`, `last`	O(1) amortized
`median`, percentiles, `difference`	O(N) per event (sorted array insert/remove)
`keep`, `unique`	O(1) per event (value count map)
`top(n)`	O(1) add/remove; O(k log k) snapshot sort
`samples`	O(1) add/remove; O(N) snapshot
Custom function	O(1) add/remove; O(N) snapshot

For typical rolling windows (tens to hundreds of events), the sorted array approach is fast enough. For very large windows with median/percentile, consider a custom reducer — but note that custom functions on live also pay O(N) per snapshot (the function re-runs over the current window each emit). See Custom reducers for the trade-off.

Multi-window rolling: merged output schema

Multi-window rolling (live.rolling({ '1m': m1, '200ms': m2 }, opts)) emits one merged event per trigger fire with all windows' columns concatenated into one record. Two rules govern the merged output schema:

Output column names must be unique across all windows. { '1m': { cpu_avg: 'avg' }, '5m': { cpu_avg: 'avg' } } is rejected at construction. Reuse implies ambiguity at the merged-record level — rename one of the columns ({ '5m': { cpu_avg_5m: { from: 'cpu', using: 'avg' } } }) or drop the duplicating window.
Per-window kind narrowing follows the same dispatch as single-window rolling. Each entry's from / using (or the AggregateMap shorthand cpu: 'avg') resolves to a column kind using the same OutputSpecKind rules as aggregate and rolling. The merged schema is the union of all windows' per-column ColumnDefs.

Per-event reducer cost is unchanged from the single-window case — each window's reducers update incrementally per source event. The shared work is the per-event ingest pipeline (one routing pass instead of N), which compounds at higher window counts. See Rolling → Multi-window rolling → Performance for the bench table.

Choosing a reducer​

Type narrowing on reduce(mapping)​

Numeric reducers​

sum​

avg​

min​

max​

median​

stdev​

difference​

count​

Percentiles: p0 through p100​

Any-type reducers​

first​

last​

keep​

Array-producing reducers​

unique​

top(n) / `top${number}`​

samples​

Custom reducers​

Empty-bucket behavior​

Rolling window behavior​

Multi-window rolling: merged output schema​

Choosing a reducer

Type narrowing on `reduce(mapping)`

Numeric reducers

`sum`

`avg`

`min`

`max`

`median`

`stdev`

`difference`

`count`

Percentiles: `p0` through `p100`

Any-type reducers

`first`

`last`

`keep`

Array-producing reducers

`unique`

`top(n)` / `top${number}`

`samples`

Custom reducers

Empty-bucket behavior

Rolling window behavior

Multi-window rolling: merged output schema