Multi-stream operators

A stream is values paired with an index that places each event on a shared timeline (a timestamp or sequence number). When streams tick on different clocks, the operators use that index rather than row position. The full model is in Streams, values, and alignment.

screamer has a small set of operators that reshape streams. CombineLatest aligns them on their index; Merge and split interleave and separate them; Dropna removes NaN events and Filter gates events by a mask. The last three change the number of events in the output. All are causal and composable, and each returns results as a (values, index) pair.

import numpy as np
import pandas as pd
from screamer import CombineLatest, Merge, split, Dropna, Filter, GreaterThan, And

# two feeds on one timeline, ticking at different times
a_idx = np.array([1, 3, 5, 7, 9], dtype=np.int64)
a_val = np.array([10., 11., 12., 11., 13.])
b_idx = np.array([2, 4, 6, 8], dtype=np.int64)
b_val = np.array([5., 6., 5.5, 6.5])

Combining streams on different clocks

Stream a ticks at odd indices, b at even, so they never share a timestamp. CombineLatest emits a row on every distinct index and forward-fills each stream's latest value across the gaps (an as-of join). Watch a value repeat down the rows where only the other stream ticked. With the default emit="when_all", output begins only once every stream has ticked at least once, so there are no cold-start NaNs.

The aligned (T, 2) values feed straight into any function, for example RollingCorr(10)(aligned); inside a Pipeline, the shorter RollingCorr(10)(CombineLatest()(a, b)) form works too.

aligned, idx = CombineLatest()(a_val, b_val, index=[a_idx, b_idx])
spread = aligned[:, 0] - aligned[:, 1]

pd.DataFrame({
    "a (carried)": aligned[:, 0],
    "b (carried)": aligned[:, 1],
    "spread": spread,
}, index=idx).rename_axis("index")

	a (carried)	b (carried)	spread
index
2	10.0	5.0	5.0
3	11.0	5.0	6.0
4	11.0	6.0	5.0
5	12.0	6.0	6.0
6	12.0	5.5	6.5
7	11.0	5.5	5.5
8	11.0	6.5	4.5
9	13.0	6.5	6.5

Emit modes: when_all vs on_any

emit="on_any" starts at the very first event, filling streams not yet seen with NaN until their first value arrives. when_all (used above) is the same result without those leading NaN rows. Use on_any when you want the earliest possible output and can drop the warm-up NaNs downstream, an idiom shown at the end.

aligned_any, idx_any = CombineLatest(emit="on_any")(a_val, b_val, index=[a_idx, b_idx])
print(f"on_any: {len(idx_any)} rows, when_all: {len(idx)} rows")

pd.DataFrame({
    "a (carried)": aligned_any[:, 0],
    "b (carried)": aligned_any[:, 1],
}, index=idx_any).rename_axis("index")

on_any: 9 rows, when_all: 8 rows

	a (carried)	b (carried)
index
1	10.0	NaN
2	10.0	5.0
3	11.0	5.0
4	11.0	6.0
5	12.0	6.0
6	12.0	5.5
7	11.0	5.5
8	11.0	6.5
9	13.0	6.5

Interleaving and separating: Merge and split

Merge interleaves the streams into one index-sorted sequence, tagging each event with the stream it came from. It does not align or forward-fill; it just orders every event on one timeline. split is its inverse: given the tagged stream, it routes the events back into per-source (values, index) pairs.

values, sources, m_idx = Merge()(a_val, b_val, index=[a_idx, b_idx])

pd.DataFrame({
    "value": values,
    "source": np.where(sources == 0, "a", "b"),
}, index=m_idx).rename_axis("index")

	value	source
index
1	10.0	a
2	5.0	b
3	11.0	a
4	6.0	b
5	12.0	a
6	5.5	b
7	11.0	a
8	6.5	b
9	13.0	a

(a_back, a_back_idx), (b_back, b_back_idx) = split(values, sources, index=m_idx)

print("recovered a:", a_back, "at", a_back_idx)
print("recovered b:", b_back, "at", b_back_idx)
print("round trip matches the originals:",
      np.array_equal(a_back, a_val) and np.array_equal(b_back, b_val))

recovered a: [10. 11. 12. 11. 13.] at [1 3 5 7 9]
recovered b: [5.  6.  5.5 6.5] at [2 4 6 8]
round trip matches the originals: True

Dropping events: Dropna

Dropna removes events whose value is NaN, giving a shorter, gap-free stream. For a 2-D aligned stream, how="any" (the default) drops a row if any column is NaN, while how="all" drops it only if every column is NaN.

readings_idx = np.array([1, 2, 3, 4, 5], dtype=np.int64)
readings = np.array([1.0, np.nan, 3.0, -4.0, 5.0])

clean, clean_idx = Dropna()(readings, index=readings_idx)
print("input  :", readings, "at", readings_idx)
print("cleaned:", clean, "at", clean_idx, "(index 2 removed)")

# 2-D: how='any' vs how='all'
grid_idx = np.array([1, 2, 3], dtype=np.int64)
grid = np.array([[1.0, np.nan],
                 [np.nan, np.nan],
                 [3.0, 4.0]])
_, keep_any = Dropna(how="any")(grid, index=grid_idx)
_, keep_all = Dropna(how="all")(grid, index=grid_idx)

pd.DataFrame({
    "col 0": grid[:, 0],
    "col 1": grid[:, 1],
    "kept (how=any)": np.isin(grid_idx, keep_any),
    "kept (how=all)": np.isin(grid_idx, keep_all),
}, index=grid_idx).rename_axis("index")

input  : [ 1. nan  3. -4.  5.] at [1 2 3 4 5]
cleaned: [ 1.  3. -4.  5.] at [1 3 4 5] (index 2 removed)

	col 0	col 1	kept (how=any)	kept (how=all)
index
1	1.0	NaN	False	True
2	NaN	NaN	False	False
3	3.0	4.0	True	True

Comparison and logic operators

A mask is a stream of 1.0 (keep) and 0.0/NaN (drop). Build one from the comparison and logic operators, which apply element-by-element to any stream.

Below, a range mask keeps values that lie strictly inside (-2.0, 2.0). Two comparison streams are combined with And:

from screamer import (
    LessThan, GreaterEqual, LessEqual, Equal, NotEqual,
    Or, Not, Where, IsNan, IsFinite,
)

x_demo = np.array([0.5, 1.8, -0.3, 2.5, -1.1])
demo_idx = np.arange(len(x_demo), dtype=np.int64)
lo_bound = np.full_like(x_demo, -2.0)
hi_bound = np.full_like(x_demo, 2.0)

mask_demo = And()(GreaterThan()(x_demo, lo_bound), LessThan()(x_demo, hi_bound))
in_range, in_range_idx = Filter()((x_demo, demo_idx), (mask_demo, demo_idx))
print("mask:", mask_demo)
print("kept:", in_range, "at index", in_range_idx)

# Where selects between two streams element-by-element based on a mask.
zero = np.zeros_like(x_demo)
print("Where(mask, x, 0):", Where()(mask_demo, x_demo, zero))

mask: [1. 1. 1. 0. 1.]
kept: [ 0.5  1.8 -0.3 -1.1] at index [0 1 2 4]
Where(mask, x, 0): [ 0.5  1.8 -0.3  0.  -1.1]

Comparison and logic operator reference:

• GreaterThan, LessThan, GreaterEqual, LessEqual

• Equal, NotEqual

• And, Or, Not

• Where - selects between two streams by a mask

• IsNan, IsFinite

Keeping events: Filter

Filter takes two streams: a data stream and a mask stream. Each event in the data stream is kept when its aligned mask value is nonzero; a zero or NaN mask drops the event. Build the mask with the comparison and logic operators shown above. For plain NaN removal, Dropna is simpler.

mask = GreaterThan()(readings, np.zeros_like(readings))
kept, kept_idx = Filter()((readings, readings_idx), (mask, readings_idx))
print("kept:", kept, "at index", kept_idx)

pd.DataFrame({
    "value": readings,
    "mask (v > 0)": mask,
    "kept": np.isin(readings_idx, kept_idx),
}, index=readings_idx).rename_axis("index")

kept: [1. 3. 5.] at index [1 3 5]

	value	mask (v > 0)	kept
index
1	1.0	1.0	True
2	NaN	NaN	False
3	3.0	1.0	True
4	-4.0	0.0	False
5	5.0	1.0	True

Composing them

The operators chain. A common idiom removes the cold-start NaN that emit="on_any" produces: align with on_any to start as early as possible, then Dropna to drop the warm-up rows, giving a clean aligned stream ready for any function.

raw, raw_idx = CombineLatest(emit="on_any")(a_val, b_val, index=[a_idx, b_idx])
clean, clean_idx = Dropna()(raw, index=raw_idx)

print("on_any rows :", len(raw_idx), "(the first has a NaN)")
print("after dropna:", len(clean_idx), "clean aligned rows")

pd.DataFrame({
    "a": raw[:, 0],
    "b": raw[:, 1],
    "kept (dropna)": ~np.isnan(raw).any(axis=1),
}, index=raw_idx).rename_axis("index")

on_any rows : 9 (the first has a NaN)
after dropna: 8 clean aligned rows

	a	b	kept (dropna)
index
1	10.0	NaN	False
2	10.0	5.0	True
3	11.0	5.0	True
4	11.0	6.0	True
5	12.0	6.0	True
6	12.0	5.5	True
7	11.0	5.5	True
8	11.0	6.5	True
9	13.0	6.5	True