Live Transforms

Live transforms build incremental processing pipelines on top of LiveSeries. Events flow through a chain of views and accumulators, each subscribing to its upstream source and emitting results to downstream consumers. Everything is synchronous and driven by data — no background timers, no implicit scheduling.

The same vocabulary as batch TimeSeries: filter, map, select, window, aggregate, rolling, diff, rate, pctChange, fill, cumulative, sample. Each method exists in two flavors — batch over an immutable snapshot, live over an incoming stream — with the same shape of output. If you learned the vocabulary in Transforms, most of this page will read as "the same thing, but incremental."

LiveView: filter, map, select, window

LiveView wraps any live source with a per-event transform. The built-in methods on LiveSeries return pre-configured views:

import { LiveSeries, Sequence } from 'pond-ts';

const live = new LiveSeries({
  name: 'metrics',
  schema: [
    { name: 'time', kind: 'time' },
    { name: 'cpu', kind: 'number' },
    { name: 'mem', kind: 'number' },
    { name: 'host', kind: 'string' },
  ] as const,
});

// Filter: only events from one host
const api1 = live.filter((e) => e.get('host') === 'api-1');

// Map: transform values. `e.get('cpu')` narrows to `number | undefined`
// from the schema; pass the event through unchanged when the value is
// missing, otherwise write the transformed number back.
const pct = live.map((e) => {
  const cpu = e.get('cpu');
  return cpu === undefined ? e : e.set('cpu', cpu * 100);
});

// Select: narrow the schema to specific columns
const cpuOnly = live.select('cpu');

// Window: bounded view over recent data
const last5m = live.window('5m'); // time-based
const last100 = live.window(100); // count-based

Under ordering: 'reorder', window() eviction is not re-applied when a late event is inserted — a late event can temporarily sit "outside" the declared window until the next prune. See Late-event scope for the full picture.

Views maintain their own event buffer and implement the LiveSource interface, so they compose by stacking:

const pipeline = live
  .filter((e) => e.get('host') === 'api-1')
  .select('cpu')
  .window('5m');

// Snapshot the windowed view into a TimeSeries for batch analytics
const ts = pipeline.toTimeSeries();

Subscriptions

Every view exposes on('event', fn) for push-based notification:

const unsub = api1.on('event', (event) => {
  console.log('api-1 cpu:', event.get('cpu'));
});

// Stop listening
unsub();

Cleanup

Call dispose() to unsubscribe a view from its source and stop receiving events.

LiveAggregation: bucketed aggregation

LiveAggregation incrementally aggregates events into fixed-size time buckets. Buckets finalize as the data advances past their boundary.

const agg = live.aggregate(Sequence.every('1m'), {
  cpu: 'avg',
  mem: 'max',
});

// closed() returns only finalized buckets
const closed = agg.closed();

// snapshot() includes the current open bucket's partial value
const snap = agg.snapshot();

Or create it from a view for filtered aggregation:

const agg = live
  .filter((e) => e.get('host') === 'api-1')
  .aggregate(Sequence.every('1m'), { cpu: 'avg' });

Reading the current value

This is a common first stumble: LiveAggregation implements LiveSource<S>, so .length and .at(index) are available — but they expose only closed buckets. The in-progress bucket is not reachable through index-style access. Reach for .at(0) / .at(agg.length - 1) right after a few pushes and you'll get undefined until the first bucket closes.

For the current partial value you have three paths, from lightest to heaviest:

// 1. 'bucket' event — fires on every contributing push with the
//    provisional bucket. Zero allocation on the happy path.
agg.on('bucket', (event) => {
  renderLiveValue(event.get('cpu'));
});

// 2. snapshot() — returns a TimeSeries that includes the open bucket
//    as its last row. Good for "render on a tick" dashboards.
const snap = agg.snapshot();
const currentCpu = snap.last()?.get('cpu');

// 3. Snapshot the source and run batch aggregate. Every bucket in
//    `{ range }` is populated immediately, no waiting for closure.
const batch = live
  .toTimeSeries()
  .aggregate(Sequence.every('1m'), { cpu: 'avg' }, { range });
const currentCpu = batch.last()?.get('cpu');

When to use which:

Need	Use
Alert / export / store finalized minute totals	agg.on('close', ...) or agg.closed()
Live UI that tracks the current bucket as it grows	agg.on('bucket', ...) or agg.snapshot()
Dashboard render with a uniform grid incl. empty buckets	live.toTimeSeries().aggregate(seq, m, { range })
Cheap one-shot read of the current partial value	agg.snapshot().last()

The batch path (live.toTimeSeries().aggregate(...)) re-validates events on each call, so it's not free — but for typical dashboard cadences (seconds to tens of seconds) the cost is negligible, and you get batch semantics for free: every bucket in the range appears, the partial trailing bucket included. If the reducer you want is array-producing (unique, top(n)) and you need a uniform grid of per-minute values with an undefined / empty bucket on quiet minutes, the batch path is the right call — live LiveAggregation only materializes buckets when an event lands in them.

tail + reduce for "current state" without bucketing

Sometimes you don't want a bucketed time series at all — you want a single record summarizing the source's current or recent state. tail gives you a trailing slice; reduce collapses a series to one record:

// "Current" — whole series, one record
const state = live.toTimeSeries().reduce({
  cpu: 'avg',
  host: 'unique',
});
// => { cpu: number | undefined;
//      host: ReadonlyArray<ScalarValue> | undefined }

// "Recent" — trailing 30s, one record
const recent = live.toTimeSeries().tail('30s').reduce({
  cpu: 'p95',
  host: 'top3',
});

Fields narrow per-entry from the reducer name and from the source column's declared kind: numeric reducers ('sum', 'avg', 'count', 'median', 'stdev', 'difference', any p${number}) yield number | undefined; 'unique' and any top${number} yield ReadonlyArray<T> | undefined where T matches the source column (ReadonlyArray<string> for kind: 'string', etc.); 'first' / 'last' / 'keep' preserve the source column kind. No as-casts needed at the call site.

tail(duration) is the temporal counterpart to Array.slice(-n): it keeps events whose begin() is strictly greater than lastEvent.begin() - duration. Called with no argument it's the identity (whole series), and it composes with every other TimeSeries method, not just reduce — plot the last 30s, aggregate just the last hour, etc.

Use this shape for stat cards, alert decisions, and the top-of-dashboard "right now" readout — anywhere a bucketed grid would be overkill.

In React: useCurrent

For React consumers, @pond-ts/react packages the subscribe + tail + reduce dance into a single hook. Full reference: useCurrent. The short version:

import { useCurrent } from '@pond-ts/react';

function StatCards({ live }) {
  const { cpu, host } = useCurrent(live, {
    cpu: 'avg',
    host: 'unique',
  });
  // cpu:  number | undefined
  // host: ReadonlyArray<string | number | boolean> | undefined

  const recent = useCurrent(
    live,
    { cpu: 'p95' },
    { tail: '30s', throttle: 200 },
  );

  return (
    <>
      <Stat label="CPU (all)" value={cpu} />
      <Stat label="CPU p95 (30s)" value={recent.cpu} />
      <Stat label="Hosts seen" value={host?.join(', ') ?? '—'} />
    </>
  );
}

useCurrent returns a stable-shape record even when the source is empty (every mapped field is undefined), so destructuring on first render is safe. Types narrow per-entry: 'avg' yields number, 'unique' yields ReadonlyArray<ScalarValue>, `top${number}` yields an array, other reducers fall back to the source column kind.

Reference stability (v0.5.6+): the returned record and each of its fields are reference-stable across renders as long as the underlying value is structurally unchanged. A new event push that leaves host: 'unique' with the same elements keeps current.host pointing at the same array, so useMemo([current.host], …) and useEffect([current.host], …) only re-run when the set actually changes. Per-field — if cpu shifts but host doesn't, current.cpu is a new number (as it should be) while current.host stays stable.

Subscriptions

// Fires when a bucket finalizes
agg.on('close', (event) => {
  console.log(
    `[${event.begin()} – ${event.end()}] avg cpu: ${event.get('cpu')}`,
  );
});

// Fires on every accumulation with the provisional bucket state
agg.on('bucket', (event) => {
  updateDashboard(event.begin(), event.get('cpu'));
});

// Fires on every source event (no payload)
agg.on('update', () => {
  rerenderSnapshot(agg.snapshot());
});

The bucket event is especially useful for live dashboards — it gives you the evolving partial aggregate as events accumulate, without waiting for the bucket to close.

Grace period for out-of-order events

In messaging systems, events often arrive slightly out of order. The grace option delays bucket closing so late events can still accumulate into their correct bucket:

const live = new LiveSeries({
  name: 'metrics',
  schema,
  ordering: 'reorder',
  graceWindow: '10s',
});

// Grace defaults from the source's graceWindow — no need to repeat it
const agg = live.aggregate(Sequence.every('1m'), { cpu: 'avg' });

// Or override with an explicit value
const agg2 = live.aggregate(
  Sequence.every('1m'),
  { cpu: 'avg' },
  { grace: '5s' },
);

When the source is a LiveSeries with a graceWindow, LiveAggregation inherits that value as its default grace period. You can still override it with an explicit grace option. If the source has no grace window (e.g. strict ordering), the default is zero — buckets close immediately.

With a grace period active, a bucket doesn't finalize until the data's watermark (highest timestamp seen) advances past bucket.end + grace. During that window:

• Late events route to their correct bucket and update the partial aggregate
• The bucket event fires with each update, reflecting the revised state
• snapshot() includes all pending buckets as provisional results

Once the grace window expires, the bucket closes normally and fires the close event with the final value. Events arriving after their bucket's grace window has expired are silently dropped.

Tip: pair ordering: 'reorder' on the source LiveSeries with a graceWindow — the aggregation will automatically inherit the same tolerance for late events.

Late-event scope

graceWindow works at exactly two boundaries in the live pipeline:

• LiveSeries ingest — rejects events older than latest − graceWindow under ordering: 'reorder'. Guarantees the buffer's temporal extent covers any event it accepts.
• LiveAggregation bucket closure — buckets stay open until the watermark passes bucket.end + grace, so late events within the grace window land in the correct bucket.

That's it. Downstream live transforms do not re-flow late events through historical state:

• LiveRollingAggregation treats a late insertion as a new output event at its insertion point. It does not re-scan historical windows to include the late event in their totals.
• LiveView.window() doesn't re-apply eviction when an event is reordered in — a late event can temporarily sit "outside the window" until the next prune.
• Subscribers' 'event' callback fires identically for on-time and reordered insertions; there is no "this was late" payload.
• React hooks (@pond-ts/react) inherit the above.

If you need late-event correctness propagated through every operator — a late event at t=T triggering recomputation of every window and aggregation that covers t=T — pond-ts is not the right tool. Reach for a streaming engine with watermarks, triggers, and retraction. See Akidau's Streaming 102 for what full late-event propagation looks like in that class of system.

For the common "moderate reorder at ingest, bucketed aggregation for dashboards" case, the two supported boundaries are sufficient — and the retention guard (graceWindow ≤ retention.maxAge) is enforced at construction to prevent the accept-then-evict race.

As a LiveSource

LiveAggregation satisfies the LiveSource interface. Its output events are interval-keyed closed buckets:

// Chain further processing on closed bucket stream
import { LiveView } from 'pond-ts';

const alerts = new LiveView(agg, (event) =>
  event.get('cpu') > 0.9 ? event : undefined,
);

alerts.on('event', (bucket) => {
  notify(`High CPU in bucket ${bucket.begin()}`);
});

• agg.length — number of closed buckets (does not include the open bucket)
• agg.at(i) — access the i-th closed bucket event; the in-progress bucket is not reachable this way. Reach for agg.snapshot().last() or the 'bucket' event if you want the partial current value.
• agg.on('event', fn) — fires on bucket close (same as on('close'))

LiveRollingAggregation: sliding-window reduction

LiveRollingAggregation maintains a sliding-window aggregate that updates on every source event. It supports time-based and count-based windows:

Heads up on late events

Under ordering: 'reorder', a late event becomes a new output event at its insertion point — this method does not re-scan historical windows to include it. See Late-event scope for the full story and the set of guarantees that hold at each point in the pipeline.

const rolling = live.rolling('5m', { cpu: 'avg' });

// Current aggregate value
rolling.value(); // { cpu: 0.42 }

// Updates on every source event
rolling.on('update', (val) => {
  console.log('rolling avg:', val.cpu);
});

As a LiveSource

LiveRollingAggregation also satisfies LiveSource. Each source event produces an output event containing the rolling aggregate at that point, preserving the source event's timestamp:

const rolling = live.rolling('5m', { cpu: 'avg' });

// Output buffer grows with each source event
rolling.length; // number of output events so far
rolling.at(0); // first output event (aggregate after first source event)
rolling.at(-1); // most recent output event

// Push-based notification
rolling.on('event', (event) => {
  // event.begin() is the source event's timestamp
  // event.get('cpu') is the rolling average at that point
});

This makes LiveRollingAggregation the live equivalent of the batch rolling() method. The output buffer grows unboundedly — downstream consumers can compose with window() to bound it if needed.

Trigger-based emission

A rolling aggregation's emission cadence is controlled by an optional trigger on the rolling call:

import { Trigger } from 'pond-ts';

const rolling = timings.rolling(
  '1m',
  { latency: 'p95' },
  { trigger: Trigger.every('30s') },
);

Three triggers are available — Trigger.event() (default, per source event), Trigger.every(duration) / Trigger.clock(sequence) (per epoch-aligned boundary), and Trigger.count(n) (per N events). Synchronised partitioned rolling builds on clock triggers — every known partition emits together at each boundary tick.

For the full reference (parameters, emission semantics, trigger × partition matrix, sync-rolling restrictions), see Triggering.

Multi-window rolling: many windows, one ingest

When the use case wants several trailing windows over the same source (e.g. a 1-minute baseline of avg/stdev plus a 200-ms current-tick window of raw samples for anomaly detection), pass a keyed record of mappings to live.rolling:

import { LiveSeries, Trigger } from 'pond-ts';

const live = new LiveSeries({ name: 'cpu', schema });

const fused = live.rolling(
  {
    '1m': {
      cpu_avg: { from: 'cpu', using: 'avg' },
      cpu_sd: { from: 'cpu', using: 'stdev' },
    },
    '200ms': { cpu_samples: { from: 'cpu', using: 'samples' } },
  },
  { trigger: Trigger.every('200ms') },
);

fused.on('event', (event) => {
  // All four columns from both windows on a single event.
  const samples = event.get('cpu_samples') as ReadonlyArray<number>;
  const z = samples.map(
    (s) => (s - event.get('cpu_avg')) / event.get('cpu_sd'),
  );
  // anomaly density, threshold counting, etc.
});

Conceptually equivalent to declaring two live.rolling(...) calls over the same source plus a per-(ts, key) join — but one ingest pass updates all windows' reducer state, one merged event is emitted per trigger fire, and the consumer code is one event handler.

Properties

• Single trigger across all windows by design. Per-window cadence is rare; users who need it fall back to two separate rolling() calls (paying the per-event pipeline cost twice).
• Time-based windows only. Object keys are duration strings ('1m', '200ms', '5s'); count-based windows (live.rolling(100, ...)) stay on the single-window form.
• Duplicate output column names across windows are rejected at construction. Each column name appears once in the merged schema.
• Single-window equivalence. live.rolling('1m', m, opts) and live.rolling({ '1m': m }, opts) produce identical output.

Per-window options

When one window needs different options from the rest (minSamples, e.g.), the value form switches from a bare mapping to an elaborated { mapping, minSamples } wrapper:

live.rolling(
  {
    '1m': { cpu_avg: { from: 'cpu', using: 'avg' } },
    '200ms': {
      mapping: { cpu_samples: { from: 'cpu', using: 'samples' } },
      minSamples: 5, // per-window override
    },
  },
  { trigger, minSamples: 2 }, // top-level default
);

Top-level options apply as defaults across all windows; per-window elaborated minSamples overrides for that window.

Partitioned variant

The same form composes with partitionBy(...) — and through chained pipelines like partitionBy(...).fill(...).rolling({...}):

const fused = live
  .partitionBy('host')
  .fill({ cpu: 'hold' })
  .rolling(
    {
      '1m': { cpu_avg: { from: 'cpu', using: 'avg' } },
      '200ms': { cpu_samples: { from: 'cpu', using: 'samples' } },
    },
    { trigger: Trigger.every('200ms') },
  );
// One merged event per partition per boundary; partition column
// auto-injected at the front of the schema.

On the partitioned form, clock trigger is required — synced cross-partition emission needs a single shared boundary detector.

As a LiveSource

Like LiveRollingAggregation, multi-window rolling implements the LiveSource<Out> contract — length, at(i), on('event', fn), dispose(). Output buffer grows unbounded; compose with .window(duration) for a bounded read-side view.

Performance characteristics

The architectural win is that per-event pipeline overhead (#routeEvent, _pushTrustedEvents, ingest) runs once regardless of N windows; only per-window reducer-state updates remain O(N) — and separate rollings pay that same cost. At N=5 windows over 100k events × 100 hosts, multi-window is 2.4× faster and uses 34% of the heap of N separate rollings. See Rolling → Performance for the full bench table.

For the conceptual framing alongside the other windowing modes, see Windowing.

Per-event transforms: diff, rate, pctChange

These methods compute per-event deltas by comparing consecutive events. Each returns a LiveView that tracks the previous event's values in a closure.

// Absolute difference between consecutive events
const deltas = live.diff('requests');

// Per-second rate of change (delta / time gap in seconds)
const perSec = live.rate('requests');

// Percentage change: (curr - prev) / prev
const pctChg = live.pctChange('requests');

Multiple columns work:

const deltas = live.diff(['cpu', 'mem']);

The first event has no predecessor, so target columns are undefined. Use { drop: true } to omit it, or chain with fill('zero') to replace with zeros:

const clean = live.diff('requests', { drop: true });
const filled = live.diff('requests').fill('zero');

Non-target columns pass through unchanged.

Carry-forward transforms: fill, cumulative

fill

Replaces undefined values as events arrive. Only forward-looking strategies are supported in live mode — bfill and linear require future values and will throw.

// Carry forward the last known value for all columns
const held = live.fill('hold');

// Fill undefined with zero
const zeroed = live.fill('zero');

// Per-column strategies and literal fill values
const mixed = live.fill({ cpu: 'hold', host: 'unknown' });

// Cap consecutive fills
const limited = live.fill('hold', { limit: 3 });

cumulative

Running accumulation over the stream. Supports sum, max, min, count, and custom functions:

const running = live.cumulative({ requests: 'sum' });
const peaks = live.cumulative({ cpu: 'max' });
const custom = live.cumulative({
  requests: (acc, value) => acc + value * 2,
});

Non-accumulated columns pass through unchanged. When a target column's value is not a number, the accumulator holds its previous value.

Sampling: bounded-memory thinning

sample({ stride: N }) thins the event stream going to downstream consumers without affecting this LiveSeries's own length, at(i), listeners, or stats() counters. The headline use case: decouple downstream baseline window length from event rate.

Streaming aggregator memory scales as O(window_seconds × event_rate × per_partition_count). At firehose rates (70k events/s × 80 partitions), a 1-minute rolling baseline holds ≈4M events; a 5-minute baseline at the same rate is the kind of number where you'd hit Node's heap ceiling. Stride-thinning the input stream means a longer baseline for the same memory budget — and sd / sqrt(N) standard error typically still drops far below per-event noise even at stride 10.

Returns a LiveView<S>, so the chainable surface (filter, rolling, reduce, select, map, diff, rate, cumulative, fill, …) is immediately available downstream of the sample.

// Canonical shape: per-host stride before per-host rolling.
const baselines = live
  .partitionBy('host')
  .sample({ stride: 10 })
  .rolling('5m', { cpu_avg: 'avg', cpu_sd: 'stdev' });

baselines.on('event', (e) => {
  // 1-in-10 of each host's stream feeds a 5-minute rolling window.
});

Multi-entity series — see partitionBy

A pre-partition live.sample({ stride: N }) applied to a structured input stream (events arriving in round-robin host order) silently keeps the same subset of partitions and drops the rest. A stride-10 filter on a 80-host round-robin stream keeps 8 of 80 hosts; the other 72 vanish from downstream rollings without warning.

The fix is the canonical shape above: chain sample after partitionBy. Each partition gets its own stride counter, so every host's stream is thinned independently.

This is the same multi-entity consideration that applies to every stateful live operator (rolling, aggregate, fill, diff, rate, cumulative, pctChange, reduce) — see Cleaning data → Multi-entity series for the broader pattern. Sample follows the same convention.

Reducer counts after sampling

Reducers downstream of sample ('count', 'sum', topN) reflect the sampled stream, not the source. Multiply count-shaped outputs by the stride to estimate true counts:

const sampled = live.partitionBy('host').sample({ stride: 10 });
const counts = sampled.rolling('1m', {
  events: { from: 'cpu', using: 'count' },
});
// counts.value().events × 10 ≈ true count over the 1m window

live.stats().ingested and live.on('batch', cb) are upstream of any .sample(...) op — they continue counting true throughput. Only consumers downstream of sample see the thinned stream.

Reservoir sampling on the live side

Not in v0.17.0. Random K-of-N (Algorithm R) replacement on a live source emits non-prefix evictions, which the current eviction-mirroring protocol ('evict' event + cutoff-based prune) doesn't handle. It's queued for a later release once the streaming RFC's LiveChange exact-removal channel lands.

Reservoir sampling on the snapshot side is unconstrained and ships in v0.17.0 — see Sampling → Reservoir (random K-of-N). For the firehose visualization shape (series.sample({ reservoir }).toRows()), materialize the live buffer first via live.toTimeSeries().

Composition patterns

Multi-stage pipeline

const pipeline = live
  .filter((e) => e.get('host') === 'api-1')
  .select('cpu')
  .window('10m')
  .aggregate(Sequence.every('1m'), { cpu: 'avg' });

pipeline.on('close', (bucket) => {
  storeMetric('api-1-cpu-1m', bucket.get('cpu'), bucket.begin());
});

Diff → fill pipeline

A common pattern: compute deltas and fill the first-event gap:

const requestRate = live
  .filter((e) => e.get('host') === 'api-1')
  .diff('requests')
  .fill('zero');

requestRate.on('event', (event) => {
  updateChart(event.begin(), event.get('requests'));
});

Multiple consumers from one source

// One filtered view, two different aggregations
const api1 = live.filter((e) => e.get('host') === 'api-1');

const minuteAgg = api1.aggregate(Sequence.every('1m'), { cpu: 'avg' });
const rollingAvg = api1.rolling('5m', { cpu: 'avg' });

Each consumer maintains independent state but shares the upstream event stream.

EMA smoothing via map

Exponential moving average doesn't need a dedicated class — it's a stateful closure inside map():

function ema(alpha: number) {
  let prev: number | undefined;
  return live.map((event) => {
    const raw = event.get('cpu'); // number | undefined from schema
    if (raw === undefined) return event; // pass through; no new prev
    prev = prev === undefined ? raw : alpha * raw + (1 - alpha) * prev;
    return event.set('cpu', prev);
  });
}

const smoothed = ema(0.3);

Snapshot for batch analytics

When you need the full batch toolkit, snapshot any live source into a TimeSeries:

const ts = live.window('1h').toTimeSeries();

// Full batch API available
const grouped = ts.groupBy('host', (g) => g.reduce({ cpu: 'avg' }));
const filled = ts.fill({ cpu: 'linear' });

Partitioned rolling → one unified series

partitionBy(col) runs a per-partition pipeline; collect() fans the per-partition outputs back into a single LiveSeries you can snapshot, window(), or hand to a React hook:

const baseline = live
  .partitionBy('host')
  .rolling('1m', {
    host: { from: 'host', using: 'last' }, // partition tag drops by default — carry it
    avg: { from: 'cpu', using: 'avg' },
    sd: { from: 'cpu', using: 'stdev' },
  })
  .collect({ retention: { maxAge: '30m' } });
// LiveSeries<{ time, host, avg, sd }> — one series, every host interleaved

Two things to know about collect():

• It is an append-only fan-in — it subscribes to every partition (current and future) and pushes their output events into the unified buffer. Per-partition retention bounds each partition; the unified buffer's retention is independent and does not inherit. Pass { retention: … } to bound it, or it grows unbounded.
• On the per-event (non-clock) partitioned rolling, the partition column drops unless you carry it through with a passthrough reducer (host: { from: 'host', using: 'last' }). The synced Trigger.clock(...) and fused forms auto-inject it instead.

This composition is the basis of an incremental dashboard baseline that costs a gather per render instead of a full re-walk — see Recipe: Incremental ±σ baselines for the end-to-end pattern (≈16× at dashboard scale).

Cleanup

Call dispose() on any view or accumulator to unsubscribe from its source. After disposal, the object stops receiving events but its current state remains readable.

agg.dispose();
rolling.dispose();
view.dispose();