Skip to main content

LiveSeries

LiveSeries is a mutable, append-optimized buffer that shares the same schema type system as TimeSeries. Use it when data arrives incrementally — from WebSockets, message queues, polling loops, or any streaming source — and you need to maintain a rolling window of recent events.

LiveSeries is a bounded in-order buffer with an opt-in grace window for moderate reordering at ingest. It's not a full streaming engine like Apache Beam or Apache Flink: no distributed execution, engine-managed watermarks, processing-time timers, checkpointed state, late panes, retractions, or exactly-once/replay guarantees. pond-ts instead uses a simpler "data is the clock, ordered ingest with bounded reorder tolerance" model. For the specifics of what that does and doesn't cover once data leaves the buffer, see Live Transforms → Late-event scope.

If you're coming from pandas: there's no direct equivalent — pandas works on already-materialized DataFrame / Series objects rather than a live source with subscribers. The closest analogue is a bounded deque you periodically .toTimeSeries() on and run batch methods against: aggregate() for .resample().agg(), align() for .asfreq() / fill / interpolation, rolling() for .rolling().agg(), and groupBy() for .groupby().

If you're coming from PondJS: there is no one-class equivalent to LiveSeries. PondJS split live work across Stream for unbounded input, Pipeline for streaming or batch processing, Aggregator for windowed/grouped aggregation inside a pipeline, and Collection / TimeSeries for event containers. LiveSeries is the pond-ts in-process push target and retained live source.

PondJSpond-ts
Stream + Pipeline().from(stream)LiveSeries.push(...) and live subscriptions
Pipeline().windowBy(...).aggregate(...)live.aggregate(...) → LiveAggregation
pipeline processors and output callbacksmethod chain on LiveSeries / LiveView
Collection / TimeSeries event containersTimeSeries; live.toTimeSeries() snapshots
source- or pipeline-managed historyretention: { maxEvents / maxAge } on source

Creating a LiveSeries​

import { LiveSeries } from 'pond-ts';

const schema = [
  { name: 'time', kind: 'time' },
  { name: 'cpu', kind: 'number' },
  { name: 'host', kind: 'string' },
] as const;

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

The schema is identical to what you'd pass to TimeSeries. All the same column kinds and required: false options work here.

Pushing events​

live.push([Date.now(), 0.42, 'api-1']);
live.push([Date.now(), 0.51, 'api-1'], [Date.now(), 0.38, 'api-2']);

push() accepts one or more rows. Each row is validated against the schema on arrival — invalid rows throw a ValidationError. Events must arrive in timestamp order by default (see ordering modes to change this).

Pushing columns: pushSeries​

When data already arrives as columns (one typed array per column, off a socket or from a worker), don't turn it into rows. Build a batch with TimeSeries.fromColumns and push the batch:

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

const schema = [
  { name: 'time', kind: 'time' },
  { name: 'price', kind: 'number' },
  { name: 'size', kind: 'number' },
] as const;

const live = new LiveSeries({
  name: 'ticks',
  schema,
  retention: { maxEvents: 100_000 },
});

// One message: a Float64Array per column, the same length each.
function onMessage(msg: {
  time: Float64Array;
  price: Float64Array;
  size: Float64Array;
}) {
  live.pushSeries(
    TimeSeries.fromColumns({ name: 'ticks', schema, columns: msg }),
  );
}

pushSeries follows pushMany's rules for the batch's rows: the same ordering, retention and listeners, the same refusals, and a refused batch stores nothing. (One exception, which pushMany shares: on a default series with partitionBy, a partition error, such as a row for a group the partitions don't declare, is thrown after the batch is stored.) The batch must be keyed the same way as the live series and have exactly its value columns, no more, each of the same kind. Columns are matched by name, in any order; in TypeScript the parameter's type already asks for the live series' schema, so that matters for a batch whose schema is only known at run time. A gap in a column the live schema requires is refused, naming the first such row, even if the batch's own schema called the column optional.

It is faster only on a default series (ordering: 'strict', a time key). There the batch's columns are stored as they are: no row tuples, no check per cell, no event object per row. Anything listening for single events or batches brings the event objects back, and with them most of pushMany's cost; that includes a window(), filter() or other view built on the series. Under ordering: 'drop' or 'reorder', or with a timeRange or interval key, the batch goes through pushMany's row path, so the speed is the same.

100,000 rows of time + two numbers, pushed both ways into a default series (ms, the median of six runs on one 4-core Linux machine, Node 22, 2026-11-01; scripts/perf-live-push-series.mjs; the pushSeries times include building each batch with fromColumns; "Faster" is the median of the six runs' own ratios; most runs' ratios sat within a fifth of the median, the widest single run 23× for one batch of 100,000):

CasepushManypushSeriesFaster
Batches of 1,600 rows18.11.810×
One batch of 100,00015.00.916×
Batches of 1,600, keep the last 10,00015.11.213×
Batches of 1657232.4×
One row at a time389601pushMany 1.5× faster
Batches of 1,600 with a window() view69541.3×
Batches of 1,600, ordering: 'drop'81741.1×
Batches of 1,600, plus a text host column21.44.64.6×
The live series keeps your arrays

On a default series the live series holds the batch's Float64Arrays themselves, not copies (fromColumns adopts them, and pushSeries keeps what fromColumns built). Don't write to, reuse, pool or transfer a pushed array while its rows are still in the series. The rows aren't checked again, so a write changes rows already stored and can leave their times out of order: lookups such as atOrBefore then give wrong answers and a later toTimeSeries() throws, while a row you already read with at() keeps its old value. Sending the array's buffer to a worker with postMessage(msg, [buffer]), or shrinking a resizable buffer, breaks those rows: their values read back as missing, or reading them throws. If you reuse buffers, push buf.slice(0, n); that copy is cheap next to pushMany. A push made from inside a listener waits until the current push ends and reads the batch then, so leave its arrays alone until it has run.

Two kinds of batch are copied for you. A batch of 8 rows or fewer is always copied: for arrays you made for that push, copying costs about what deciding whether to copy would, and keeps less memory (arrays you push a second time pay about 1.5 µs a push for the copy). And a batch that would keep much more memory alive than its own rows need, such as history.slice(-100) of a big series, is copied too. A slice shares its parent's buffers, and its text columns share the parent's whole list of distinct strings, so keeping it would keep all of history alive (asTime({ at: 'center' }) on a slice still shares the number columns). pushSeries copies a batch whose arrays sit in buffers more than twice the size its rows need, or whose text column carries more than twice as many distinct strings as it has rows. In that second case the copy keeps only the strings its rows use; a copy made for another reason keeps the list as it is, which then holds at most twice as many strings as rows. A batch whose columns are views into one buffer about its own size, like a single worker message holding every column, is kept as it is.

A non-finite number (NaN, Infinity) is a gap here, as fromColumns treats it; pushMany refuses one. For one row at a time, keep using push. If each batch is only a few rows and you make its arrays for that push, pushMany is faster too: making three small Float64Arrays and a TimeSeries costs more than it saves. Counting that work, 8 rows a push took 50–72 ms per 100,000 rows through pushMany and 67–90 through pushSeries (eight runs each), and 16 rows about the same either way (the table's batches of 16 come from arrays made beforehand).

Reading events​

live.length; // number of events in the buffer
live.first(); // oldest event, or undefined
live.last(); // newest event, or undefined
live.at(0); // event by index
live.at(-1); // negative indices count from the end

Events are the same immutable Event objects used by TimeSeries.

Snapshots​

Convert the current buffer contents into an immutable TimeSeries for batch analytics:

const ts = live.toTimeSeries();

// All TimeSeries operations work on the snapshot
const smoothed = ts.rolling('5m', { cpu: 'avg' });
const groups = ts.groupBy('host');

The snapshot is independent — future pushes to the LiveSeries don't affect it. You can pass a custom name: live.toTimeSeries('snapshot-12:05').

Retention policies​

Control how much data the buffer holds. Retention is checked after every push() — no background timers, no implicit scheduling.

const live = new LiveSeries({
  name: 'bounded',
  schema,
  retention: {
    maxEvents: 1000, // keep at most 1000 events
    maxAge: '30m', // drop events older than 30 minutes
  },
});

Both policies can be combined. When both apply, the one that evicts the most events wins. Eviction always removes the oldest events first.

maxAge is relative to the latest event's timestamp, not wall-clock time. This follows the "data is the clock" principle — if no events arrive, nothing is evicted.

Duration strings support ms, s, m, h, d suffixes, or plain millisecond numbers.

Ordering modes​

By default, LiveSeries expects events in timestamp order and throws on violations. Two alternative modes handle out-of-order data:

strict (default)​

const live = new LiveSeries({ name: 'strict', schema });
live.push([1000, 0.5, 'a']);
live.push([500, 0.3, 'a']); // throws ValidationError

drop​

Silently discard late events:

const live = new LiveSeries({
  name: 'drop-late',
  schema,
  ordering: 'drop',
});
live.push([1000, 0.5, 'a']);
live.push([500, 0.3, 'a']); // silently dropped
live.length; // 1

reorder​

Insert late events in sorted position:

const live = new LiveSeries({
  name: 'reorder',
  schema,
  ordering: 'reorder',
});
live.push([2000, 0.6, 'a']);
live.push([1000, 0.5, 'a']); // inserted before the 2000 event
live.at(0)?.begin(); // 1000

Add a graceWindow to bound how late an event can be:

const live = new LiveSeries({
  name: 'bounded-reorder',
  schema,
  ordering: 'reorder',
  graceWindow: '5s',
});
live.push([10000, 0.5, 'a']);
live.push([8000, 0.4, 'a']); // ok, within 5s
live.push([1000, 0.1, 'a']); // throws, outside grace window

Without graceWindow, reorder accepts any timestamp.

Subscriptions​

Subscribe to lifecycle events. All callbacks fire synchronously, inline with push().

// Per-event: fires once for each event added
const unsub = live.on('event', (event) => {
  console.log('new:', event.get('cpu'));
});

// Batch: fires once per push() call with all added events
live.on('batch', (events) => {
  console.log(`received ${events.length} events`);
});

// Evict: fires when retention removes events
live.on('evict', (events) => {
  console.log(`evicted ${events.length} old events`);
});

Each on() call returns an unsubscribe function. Call it to stop receiving notifications:

const unsub = live.on('event', handler);
// later...
unsub();

Columns: each push as a TimeSeries​

'columns' fires once per push with just the rows that push added, as a normal TimeSeries. Read whole columns from it instead of one Event per row: it suits code that copies values into a chart buffer or writes them to the wire.

live.on('columns', (batch) => {
  const time = batch.keyColumn();
  const cpu = batch.column('cpu');
  for (let i = 0; i < batch.length; i += 1) {
    send(time.beginAt(i), cpu.read(i));
  }
});

On a default (strict, time-keyed) series this builds no Event and copies no row: the batch shares the arrays the series stores, and handing it over costs well under a microsecond a push. (Partition sub-series gather their rows into fresh arrays, and a pushSeries batch of 8 rows or fewer is copied on the way in.) With ordering: 'reorder' or 'drop' the batch is built from the added events, sorted by key. A batch never changes once handed out, even after retention drops its rows from the series (holding one keeps those rows in memory), unless you write into arrays you handed to pushSeries. Partition sub-series fire it too.

1M rows pushed 1,000 at a time into a 10,000-row buffer (median of 5, packages/core/scripts/perf-live-columns.mjs):

the listener'batch''columns'
(none attached)130 ms130 ms
sums one column320 ms141 ms
copies one column into a buffer356 ms142 ms
writes every row as JSON446 ms251 ms

pond-rs has no 'columns' output yet.

Callback ordering​

Within a single push() call:

  1. event — fires per event, inline with insertion
  2. Retention runs (may evict old events)
  3. batch — fires once with all events that were added
  4. columns — fires once with the added rows as a TimeSeries
  5. evict — fires if retention removed events

clear() fires only the evict subscription.

Clearing​

live.clear();

Removes all events and fires the evict subscription with the removed events. After clearing, the buffer is empty and ready for new pushes.

Typical patterns​

WebSocket ingestion with rolling window​

const live = new LiveSeries({
  name: 'ws-metrics',
  schema,
  retention: { maxAge: '1h' },
});

ws.on('message', (msg) => {
  const { timestamp, cpu, host } = JSON.parse(msg);
  live.push([timestamp, cpu, host]);
});

// Periodic snapshot for rendering
setInterval(() => {
  const ts = live.toTimeSeries();
  renderChart(ts.rolling('5m', { cpu: 'avg' }));
}, 1000);

Monitoring with eviction alerts​

const live = new LiveSeries({
  name: 'monitor',
  schema,
  retention: { maxEvents: 500 },
});

live.on('evict', (evicted) => {
  metrics.increment('events.evicted', evicted.length);
});

live.on('batch', (events) => {
  for (const e of events) {
    const cpu = e.get('cpu'); // number | undefined, narrowed from schema
    if (cpu !== undefined && cpu > 0.9) {
      alert(`High CPU on ${e.get('host')}`);
    }
  }
});

Tolerating out-of-order delivery​

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

// Events from multiple sources may arrive slightly out of order
sourceA.on('data', (row) => live.push(row));
sourceB.on('data', (row) => live.push(row));