Skip to main content

Analytics library

Project description

Python Data Streams

Build Status Coverage PyPI PyPI Docs Binder

Tributary is a library for constructing dataflow graphs in python. Unlike many other DAG libraries in python (airflow, luigi, prefect, dagster, dask, kedro, etc), tributary is not designed with data/etl pipelines or scheduling in mind. Instead, tributary is more similar to libraries like mdf, pyungo, streamz, or pyfunctional, in that it is designed to be used as the implementation for a data model. One such example is the greeks library, which leverages tributary to build data models for options pricing.


Install with pip:

pip install tributary

or with conda:

conda install -c conda-forge tributary

or from source:

python install

Note: If installing from source or with pip, you'll also need Graphviz itself if you want to visualize the graph using the .graphviz() method.

Stream Types

Tributary offers several kinds of streams:


These are synchronous, reactive data streams, built using asynchronous python generators. They are designed to mimic complex event processors in terms of event ordering.


These are functional streams, built by currying python functions (callbacks).


These are lazily-evaluated python streams, where outputs are propogated only as inputs change. They are implemented as directed acyclic graphs.


  • Streaming: In this example, we construct a variety of forward propogating reactive graphs.
  • Lazy: In this example, we construct a variety of lazily-evaluated directed acyclic computation graphs.
  • Automatic Differentiation: In this example, we use tributary to perform automatic differentiation on both lazy and streaming graphs.

Graph Visualization

You can visualize the graph with Graphviz. All streaming and lazy nodes support a graphviz method.

Streaming and lazy nodes also support ipydagred3 for live update monitoring.


Here green indicates executing, yellow indicates stalled for backpressure, and red indicates that StreamEnd has been propogated (e.g. stream has ended).


Here green indicates executing, and red indicates that the node is dirty. Note the the determination if a node is dirty is also done lazily (we can check with isDirty whcih will update the node's graph state.

Sources and Sinks


  • Python Function/Generator/Async Function/Async Generator
  • Curve - yield through an iterable
  • Const - yield a constant
  • Timer - yield on an interval
  • Random - generates a random dictionary of values
  • File - streams data from a file, optionally loading each line as a json
  • HTTP - polls a url with GET requests, streams data out
  • HTTPServer - runs an http server and streams data sent by clients
  • Websocket - strams data from a websocket
  • WebsocketServer - runs a websocket server and streams data sent by clients
  • SocketIO - streams data from a socketIO connection
  • SocketIOServer - streams data from a socketIO connection
  • SSE - streams data from an SSE connection
  • Kafka - streams data from kafka
  • Postgres - streams data from postgres


  • Foo - data to a python function
  • File - data to a file
  • HTTP - POSTs data to an url
  • HTTPServer - runs an http server and streams data to connections
  • Websocket - streams data to a websocket
  • WebsocketServer - runs a websocket server and streams data to connections
  • SocketIO - streams data to a socketIO connection
  • SocketIOServer - runs a socketio server and streams data to connections
  • SSE - runs an SSE server and streams data to connections
  • Kafka - streams data to kafka
  • Postgres - streams data to postgres
  • Email - streams data and sends it in emails
  • TextMessage - streams data and sends it via text message



  • Delay - Streaming wrapper to delay a stream
  • Throttle - Streaming wrapper to only tick at most every interval
  • Debounce - Streaming wrapper to only tick on new values
  • Apply - Streaming wrapper to apply a function to an input stream
  • Window - Streaming wrapper to collect a window of values
  • Unroll - Streaming wrapper to unroll an iterable stream
  • UnrollDataFrame - Streaming wrapper to unroll a dataframe into a stream
  • Merge - Streaming wrapper to merge 2 inputs into a single output
  • ListMerge - Streaming wrapper to merge 2 input lists into a single output list
  • DictMerge - Streaming wrapper to merge 2 input dicts into a single output dict. Preference is given to the second input (e.g. if keys overlap)
  • Reduce - Streaming wrapper to merge any number of inputs
  • FixedMap - Map input stream to fixed number of outputs
  • Subprocess - Open a subprocess and yield results as they come. Can also stream data to subprocess (either instantaneous or long-running subprocess)


Note that tributary can also be configured to operate on dual numbers for things like lazy or streaming autodifferentiation.

Arithmetic Operators

  • Noop (unary) - Pass input to output
  • Negate (unary) - -1 * input
  • Invert (unary) - 1/input
  • Add (binary) - add 2 inputs
  • Sub (binary) - subtract second input from first
  • Mult (binary) - multiple inputs
  • Div (binary) - divide first input by second
  • RDiv (binary) - divide second input by first
  • Mod (binary) - first input % second input
  • Pow (binary) - first input^second input
  • Sum (n-ary) - sum all inputs
  • Average (n-ary) - average of all inputs
  • Round (unary)
  • Floor (unary)
  • Ceil (unary)

Boolean Operators

  • Not (unary) - Not input
  • And (binary) - And inputs
  • Or (binary) - Or inputs


  • Equal (binary) - inputs are equal
  • NotEqual (binary) - inputs are not equal
  • Less (binary) - first input is less than second input
  • LessOrEqual (binary) - first input is less than or equal to second input
  • Greater (binary) - first input is greater than second input
  • GreaterOrEqual (binary) - first input is greater than or equal to second input


  • Log (unary)
  • Sin (unary)
  • Cos (unary)
  • Tan (unary)
  • Arcsin (unary)
  • Arccos (unary)
  • Arctan (unary)
  • Sqrt (unary)
  • Abs (unary)
  • Exp (unary)
  • Erf (unary)

Financial Calculations

  • RSI - Relative Strength Index
  • MACD - Moving Average Convergence Divergence


  • Int (unary)
  • Float (unary)
  • Bool (unary)
  • Str (unary)

Basket Functions

  • Len (unary)
  • Count (unary)
  • Min (unary)
  • Max (unary)
  • Sum (unary)
  • Average (unary)


  • RollingCount - Node to count inputs
  • RollingMin - Node to take rolling min of inputs
  • RollingMax - Node to take rolling max of inputs
  • RollingSum - Node to take rolling sum inputs
  • RollingAverage - Node to take the running average
  • SMA - Node to take the simple moving average over a window
  • EMA - Node to take an exponential moving average over a window

Node Type Converters

  • Lazy->Streaming

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for tributary, version 0.1.5
Filename, size File type Python version Upload date Hashes
Filename, size tributary-0.1.5.tar.gz (74.1 kB) File type Source Python version None Upload date Hashes View
Filename, size tributary-0.1.5-py2.py3-none-any.whl (106.6 kB) File type Wheel Python version py2.py3 Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page