Transformations

The simplest shape: events in, events out. One event per source event, same temporal key, sometimes a different schema. These are the operators that preserve the event grid — they don't bucket, window, or aggregate.

For operations that change the event grid (rolling windows, fixed-bucket aggregation, alignment), see Concepts → Windowing and the Aggregation, Alignment, and Rolling Windows operator pages.

Two flavors:

• Per-event-only — operate on one event at a time, no memory. map, select, rename, collapse, filter, plus set / merge on individual events.
• Per-event-with-tiny-state — look at the previous event (or a small accumulator) to compute each output. diff, rate, pctChange, cumulative, scan, shift.

Coming from pandas: this section covers .assign(), .rename(), [cols], .where() (≈ filter), .diff(), .pct_change(), .cumsum(), .shift(). Coming from JavaScript arrays: map, filter do exactly what you'd expect; the rest are time-series-aware.

Per-event-only

map(nextSchema, fn) — transform each event

The most general. Pass the output schema (so the result type is narrow) and a function that takes a source event and returns the new one. The schema argument is mandatory even when the shape doesn't change — pond-ts always wants the next schema explicit.

// Same schema; just transform the cpu value.
const scaled = cpu.map(cpu.schema, (e) => {
  const value = e.get('cpu');
  return value === undefined ? e : e.set('cpu', value * 100);
});

// New schema — adds a "flagged" column.
const flaggedSchema = [
  ...cpu.schema,
  { name: 'flagged', kind: 'boolean' },
] as const;
const flagged = cpu.map(flaggedSchema, (e) =>
  e.merge({ flagged: (e.get('cpu') ?? 0) > 0.9 }),
);

Event instances are immutable; set returns a new Event with the field updated. merge(patch) works the same way for multi-field updates and can introduce new columns.

filter(predicate) — drop events

Predicate gets an event and an index, returns true to keep:

const active = cpu.filter((e) => e.get('cpu') !== undefined);
const apiOnly = events.filter((e) => e.get('host')?.startsWith('api'));

Same schema in, same schema out; just fewer rows. See Cleaning data for filter as a "remove bad rows" pattern.

select(...keys) — keep only some columns

Schema-narrowing — type-level too:

const just = full.select('cpu', 'host');
// just.schema === [time, cpu, host]; the other columns are gone.
just.at(0)?.get('cpu'); // still narrows to number | undefined

rename(mapping) — rename payload columns

const renamed = cpu.rename({ cpu: 'usage' });
renamed.at(0)?.get('usage'); // number | undefined
// renamed.schema has 'usage' where 'cpu' was.

The temporal key column can't be renamed.

collapse(keys, output, reducer) — combine columns into one

Reducer receives a typed record of just the named columns, returns a single value:

// Average of cpu and requests into a "load" column.
const summary = metrics.collapse(
  ['cpu', 'requests'],
  'load',
  ({ cpu, requests }) => ((cpu ?? 0) + (requests ?? 0)) / 2,
);
// schema: [time, load, ...other source columns]

The named columns are removed by default; { append: true } keeps them and adds the new one alongside.

Event.set(field, value) and Event.merge(patch)

Work directly on an Event. You'll use them inside map callbacks:

events.map((e) => e.set('cpuPct', (e.get('cpu') ?? 0) * 100));

events.map((e) =>
  e.merge({
    cpu: (e.get('cpu') ?? 0) * 100,
    flagged: (e.get('cpu') ?? 0) > 0.9,
  }),
);

set returns the same key; merge does too. Both are immutable.

Per-event-with-tiny-state

These look at one event plus the previous (or an accumulator) to compute each output. Length is preserved; the first event has no predecessor, so target columns are undefined (or you can drop the first event with { drop: true }).

diff(cols) — per-event delta

const deltas = series.diff('requests');
// requests = curr - prev. First event: undefined.
const both = series.diff(['cpu', 'mem']);
const dropped = series.diff('requests', { drop: true });

rate(cols) — delta per second

diff divided by the time gap (in seconds). When two consecutive events share a timestamp (dt = 0), rate returns undefined rather than dividing by zero — filter per-producer first if you have concurrent emissions.

const rps = series.rate('requests');

pctChange(cols) — relative change

(curr - prev) / prev. Time-gap-independent. Returns undefined when the previous value is 0.

const rets = prices.pctChange('price');

cumulative(spec) — running total / max / min / count

const running = series.cumulative({ requests: 'sum', cpu: 'max' });
const errors = series.cumulative({ errors: 'count' });
const product = series.cumulative({ factor: (acc, v) => acc * v });

When a source value is undefined, the accumulator carries forward its previous value — no NaN propagation.

scan(source, step, init, options?) — typed-accumulator running fold

scan is the general form of cumulative — the classic mapAccumL. cumulative locks three things together (the accumulator is the output is a number, written in place); scan decouples them: the accumulator A can be any value, the numeric output is separate, and you choose whether to replace the source or append a new column.

// running sum — the cumulative special case, replacing in place:
series.scan('work', (acc, v) => [acc + v, acc + v], 0);
//   ≡ series.cumulative({ work: 'sum' })

// typed accumulator into a NEW column: hysteresis elevation gain
// carries (ref, gain) but emits only the gain.
const T = 3; // ±3 m deadband
const withGain = track.scan<'cumGain', { ref: number | null; gain: number }>(
  'ele',
  (acc, ele) => {
    if (acc.ref === null) return [{ ref: ele, gain: 0 }, 0];
    const d = ele - acc.ref;
    if (d >= T) return [{ ref: ele, gain: acc.gain + d }, acc.gain + d];
    if (d <= -T) return [{ ref: ele, gain: acc.gain }, acc.gain];
    return [acc, acc.gain]; // within deadband — carry
  },
  { ref: null, gain: 0 },
  { output: 'cumGain' },
);

step(acc, value, i) returns [nextAcc, output]. The accumulator A is inferred from init. With no options.output the source column is replaced (widened to optional number, as cumulative does); with options.output a new column of that name is appended and the source is left intact (the name must not already exist).

Missing cells inherit cumulative's behavior exactly: a missing / undefined source cell does not call step — the accumulator is held and the row re-emits the last output (so it holds flat across a gap), undefined only until the first defined value. A stored NaN is a defined number and is passed to step. On a multi-entity series the accumulator interleaves across entities; scope it per entity with partitionBy, run scan inside each partition, and fan back in with collect:

// One running total per host, not a single interleaved accumulator.
const perHost = series
  .partitionBy('host')
  .scan('requests', (acc, v) => [acc + v, acc + v], 0)
  .collect();

split = scan + byColumn. The reason scan exists: it isolates the order-dependent state into a column so value-axis aggregation (byColumn) can stay pure and order-free. Materialize the carried state with scan, then segment statelessly — e.g. per-kilometre split gain is last − first of the scanned cumGain in each distance bin. See Value axis for the mental model behind this composition.

tail(duration) — keep the trailing window

tail is technically a temporal selection rather than a per-event op, but it's tiny and lives next to these in practice — keep events whose begin() is greater than lastEvent.begin() - duration. Same schema, fewer rows.

const recent = series.tail('30s');
const all = series.tail(); // identity (whole series) when no duration

Often paired with reduce for "current state" readouts — see Rolling windows → tail for the composition.

shift(cols, n) — lag / lead

Move column values forward (positive n = lag) or backward (negative n = lead). Vacated slots become undefined. Keys are unchanged.

const lagged = series.shift('value', 1);
const lead = series.shift('value', -1);
const multi = series.shift(['cpu', 'mem'], 2);

For time-based shifts on irregular data, align to a regular grid first (Alignment), then shift by the corresponding number of events.

Method comparison

Method	Output
map(nextSchema, fn)	New schema possible; one event in, one out
filter(pred)	Same schema; subset of rows
select(...keys)	Schema narrowed to chosen columns
rename(mapping)	Schema with renamed columns
collapse(cols, ...)	Schema replaces several columns with one
diff(cols)	Same schema; per-event delta column
rate(cols)	Same schema; delta-per-second column
pctChange(cols)	Same schema; relative-change column
cumulative(spec)	Same schema; running aggregate column
scan(src, …)	Replace source, or append output column
tail(duration)	Same schema; trailing temporal slice
shift(cols, n)	Same schema; column values shifted
Event.set(field, v)	Single field on one event
Event.merge(patch)	Multi-field patch on one event