`Pipeline`#

Define a pipeline once, then run it in batch or live with identical results. Input names a source stream; calling functors and stream operators on those handles records the pipeline. Pipeline(inputs=[...], outputs=[...]) compiles it into a callable: pipe(arrays) runs it in batch, pipe(generators) runs it event by event (lazy pull path), and pipe.live() opens an incremental session you drive yourself. All three produce byte-identical output on the same events.

For the conceptual model (how the pipeline is built, when to reach for it, and the define-once-run-anywhere guarantee), see Pipelines. This page is the reference contract.

class Input(name)[source]#

Create a source Node - a named placeholder for a timed stream.

__call__(**kwargs)#: Call self as a function.

__init__(**kwargs)#

class Node(op, inputs=())[source]#

An immutable handle for a stream inside a pipeline.

You do not usually construct a Node directly: Input(name) returns one, and applying a functor or a stream operator to a Node returns another. op records the operation (an input, a functor instance, or an operator) and inputs is a tuple of upstream Nodes.

__init__(op, inputs=())[source]#

class Pipeline(inputs, outputs, align_outputs=True)[source]#

A reusable N-in / M-out function you define once and call on stored or live data.

Arguments:

inputs: ordered list of Input(...) nodes that define the call signature. Feeds are bound positionally, or by name via a keyword call.
outputs: ordered list of output nodes to evaluate.
align_outputs (default True): when True, co-index all M outputs onto a shared, sorted index axis, so each output carries its as-of value at every unique union index (combine_latest's per-event intermediate rows are collapsed to one row per index) and the returned (values, index) pairs have equal length. When False, return independent per-output streams whose lengths may differ.

Call pipe(*feeds) (positional) or pipe(**named_feeds) (by Input name) to run the pipeline. Each feed may be a bare value array (positional, index = row-number), or a (values, index) pair (values-first). Pass generators of (value, index) pairs to run the pipeline lazily, event by event, with byte-identical results (the lazy pull path). Returns a single (values, index) pair when M == 1, or a tuple of pairs when M > 1.

__call__(*args, **kwargs)[source]#: Call self as a function.

__init__(inputs, outputs, align_outputs=True)[source]#

live()[source]#

Open a live streaming session: push events and drive a clock yourself.

Bind data AFTER definition, event by event, on the same Pipeline that runs batch. Push events with .push(input, index, value); close windows whose boundary has passed with .advance(now) (e.g. on a clock tick, finalizing empty bars); force the current partial window with .flush(); collect aligned outputs with .result().

The session drives this Pipeline's single compiled engine (shared with __call__), resetting it on open, so use one session at a time: do not interleave a live session with a batch call or run two live sessions concurrently.

The three names#

Input(name): a function that returns a Node, a named source handle. The inputs list of a Pipeline is made of these, and their order and names define the pipeline's call signature.
Node: the handle type for a stream inside the pipeline. It is user-facing as a type, but you rarely construct Node(...) by hand; you obtain nodes from Input(...) and by applying functors and stream operators to existing nodes.
Pipeline: the compiled, runnable object built from inputs and outputs.

Constructor#

Pipeline(inputs, outputs, align_outputs=True)

inputs: a list of Input(...) nodes. Defines the call signature; feeds are bound positionally in this order, or by the input names.
outputs: a list of nodes to evaluate.
align_outputs: when True (default), all outputs are co-indexed onto a single shared index, so every output is an equal-length (values, index) pair. When False, each output is returned as an independent stream whose length may differ from the others.

The constructor validates the pipeline and raises a clear ValueError if: an inputs entry is not an Input(...) node; an outputs entry is not a Node; an output references an undeclared input; a declared input is never used; or a single functor instance backs more than one node.

Calling the pipeline#

Feeds are passed positionally in input order, pipe(*feeds), or by input name, pipe(**named_feeds). Each feed may be:

a bare array (positional, with the index taken as the row number), or
a (values, index) pair.

The return is always (values, index) tuples:

one output, a single (values, index) pair,
multiple outputs, a tuple of such pairs, one per output.

Pass generators of (value, index) pairs instead of arrays to run the pipeline lazily, event by event: pipe(gen_a, gen_b) returns an iterator that yields output events byte-identical to the batch result.

Live, incremental sessions: `pipe.live()`#

pipe(arrays) and pipe(generators) both consume complete feeds. When you drive the pipeline yourself, one event at a time, pipe.live() opens a session object. It shares the pipeline's single engine and resets it on open, so use one session at a time (do not interleave it with a pipe(...) call). Each method returns the session, so calls chain.

.push(input, index, value) feeds one event. input is an Input name (str) or its position (int) in the inputs list; index is the event's integer index; value is the float value.
.advance(now) moves logical time to now (an integer index), closing every windowing node whose bucket boundary has passed by now. This finalizes time-based bars even when no event fell in them, for example an empty minute bar closed by a clock tick. It is a no-op for event-count (count=) windows and before the first event; call it with non-decreasing now.
.flush() finalizes the current partial window(s) on demand, for example at the end of a processing loop. The end-of-input flush that pipe(...) performs implicitly is the special case.
.result() returns the output accumulated so far, in the same shape pipe(...) returns, and drains the internal buffers.

Feeding the same events in index order and then calling .flush() reproduces the batch result exactly. .advance() (and a clock input wired into the pipeline) additionally let a windowing node emit bars a purely event-driven pass would not, the empty leading and trailing bars in Resample, for instance.

import numpy as np
from screamer import Input, Pipeline, First, Last, Resample, CombineLatest

price = Input("price")
open_b  = Resample(freq=10, fill="nan", agg=First())(price)
close_b = Resample(freq=10, fill="nan", agg=Last())(price)
bars = CombineLatest()(open_b, close_b)
pipe = Pipeline(inputs=[price], outputs=[bars])

live = pipe.live()
live.push("price", 0, 100.0)   # one trade in bar [0, 10)
live.advance(30)               # clock passes bars [10,20) and [20,30): empty -> NaN
live.flush()
values, index = live.result()
print(index)
print(values)

[ 0 10 20]
[[100.  nan]
 [ nan  nan]
 [ nan  nan]]

Example#

Align two streams and take their difference. The same pipeline runs in batch and live.

import numpy as np
from screamer import Input, Pipeline, Sub, CombineLatest

a, b = Input("a"), Input("b")
pipe = Pipeline(inputs=[a, b], outputs=[Sub()(CombineLatest()(a, b))])

# feeds are (values, index) - values-first
fa = (np.array([10.0, 20.0, 30.0]), np.array([1, 2, 3]))
fb = (np.array([1.0, 2.0, 3.0]),   np.array([1, 2, 3]))

spread, idx = pipe(fa, fb)
print(spread.reshape(-1))

# pipe(generators) returns the identical result, event by event (lazy path)
events = list(pipe(
    ((v, k) for v, k in zip(fa[0], fa[1])),
    ((v, k) for v, k in zip(fb[0], fb[1])),
))
sv = np.array([e[0] for e in events])
print(np.array_equal(spread.reshape(-1), sv, equal_nan=True))

[ 9. 18. 27.]
True

Inspecting a pipeline#

print(pipe) (or pipe.to_text()) shows the pipeline as an indented tree, rooted at each output and descending to the inputs. A node shared by several consumers is printed once and then referenced by id, so a diamond reads as a diamond. Node labels carry the functor and operator parameters (RollingMean(window_size=20), Resample(every=5, ...)).

For a diagram, pipe.to_dot() returns a Graphviz DOT string with no dependencies, and pipe.to_graphviz() returns a rendered graphviz.Source when the optional graphviz package is installed (pip install screamer[viz], plus the system Graphviz dot binary). In a Jupyter notebook, displaying a Pipeline shows the diagram inline, falling back to the text tree when graphviz is not available.

from screamer import Input, Pipeline, RollingMean, Sub, CombineLatest

a, b = Input("a"), Input("b")
pipe = Pipeline(inputs=[a, b], outputs=[RollingMean(20)(Sub()(CombineLatest()(a, b)))])
print(pipe.to_text())

Pipeline(2 input(s), 1 output(s), align_outputs=True)
out[0] = RollingMean(window_size=20)  #4
   └─ Sub()  #3
      └─ CombineLatest(emit='when_all')  #2
         ├─ a  #0 (input)
         └─ b  #1 (input)

Saving and loading a pipeline#

pipe.to_json() serializes the pipeline to JSON (its inputs, nodes with their parameters, outputs, and align_outputs), and Pipeline.from_json(text) rebuilds a runnable Pipeline from it. This round-trips exactly, so a pipeline can be saved as a config file and reloaded. to_dict / from_dict give the same round-trip with a plain dict.

import numpy as np
from screamer import Input, Pipeline, RollingMean, Sub, CombineLatest

a, b = Input("a"), Input("b")
pipe = Pipeline(inputs=[a, b], outputs=[RollingMean(20)(Sub()(CombineLatest()(a, b)))])

config = pipe.to_json()          # save this string to a file
restored = Pipeline.from_json(config)

fa = (np.arange(1.0, 61), np.arange(1, 61))
fb = (np.arange(0.0, 60), np.arange(1, 61))
print(np.array_equal(pipe(fa, fb)[0], restored(fa, fb)[0], equal_nan=True))

True

Pipeline#