Crate wingfoil

Expand description

§Wingfoil

Wingfoil is a blazingly fast, highly scalable stream processing framework designed for latency-critical use cases such as electronic trading and real-time AI systems.

Wingfoil simplifies receiving, processing and distributing streaming data across your entire stack.

§Features

Fast: Ultra-low latency and high throughput with an efficient DAG based execution engine.
Simple and obvious to use: Define your graph of calculations; Wingfoil manages its execution.
Multi-language: currently available as a Rust crate and as a beta release, python package with plans to add WASM/JavaScript/TypeScript support.
Backtesting: Replay historical data to backtest and optimise strategies.
Async/Tokio: seamless integration, allows you to leverage async at your graph edges.
Multi-threading: distribute graph execution across cores.
I/O Adapters: production-ready KDB+ integration for tick data, CSV, Kafka (coming soon), and more.

§Quick Start

use wingfoil::*;
use std::time::Duration;
fn main() {
    let period = Duration::from_secs(1);
    ticker(period)
        .count()
        .map(|i| format!("hello, world {:}", i))
        .print()
        .run(RunMode::RealTime, RunFor::Duration(period*3)
    );
}

This output is produced:

hello, world 1
hello, world 2
hello, world 3

You can download from crates.io, read the documentation, review the benchmarks or jump straight into one of the examples. You can download the wingfoil Python module from pypi.

§Order Book Example

Load a CSV of AAPL limit orders, maintain an order book using the lobster crate, derive trades and two-way prices, and export back to CSV — all in a few lines:

let book = RefCell::new(lobster::OrderBook::default());
let get_time = |msg: &Message| NanoTime::new((msg.seconds * 1e9) as u64);
let (fills, prices) = csv_read_vec("aapl.csv", get_time, true)
    .map(move |chunk| process_orders(chunk, &book))
    .split();
let prices_export = prices
    .filter_value(|price: &Option<TwoWayPrice>| !price.is_none())
    .map(|price| price.unwrap())
    .distinct()
    .csv_write("prices.csv");
let fills_export = fills.csv_write_vec("fills.csv");
Graph::new(vec![prices_export, fills_export], RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
    .print()
    .run()
    .unwrap();

§Get Involved!

We want to hear from you! Especially if you:

are interested in contributing
know of a project that wingfoil would be well-suited for
would like to request a feature or report a bug
have any feedback

Please do get in touch:

ping us on discord
email us at hello@wingfoil.io
submit an issue
get involved in the discussion

§Graph Execution

Wingfoil abstracts away the details of how to co-ordinate the calculation of your application, parts of which may be executing at different frequencies. Only the nodes that actually require cycling are executed which allows wingfoil to efficiently scale to very large graphs. Consider the following example:

use wingfoil::*;
use std::time::Duration;

fn main() {
    let period = Duration::from_millis(10);
    let source = ticker(period).count(); // 1, 2, 3 etc
    let is_even = source.map(|i| i % 2 == 0);
    let odds = source
        .filter(is_even.not())
        .map(|i| format!("{:} is odd", i));
    let evens = source
        .filter(is_even)
        .map(|i| format!("{:} is even", i));
    merge(vec![odds, evens])
        .print()
        .run(
            RunMode::HistoricalFrom(NanoTime::ZERO),
            RunFor::Duration(period * 5),
        );
}

This output is produced:

1 is odd
2 is even
3 is odd
4 is even
5 is odd
6 is even

We can visualise the graph like this:

The input and output nodes tick 6 times each and the evens and odds nodes tick 3 times.

§Historical vs RealTime

Time is a first-class citizen in wingfoil. Engine time is measured in nanoseconds from the UNIX epoch and represented by a NanoTime.

In this example we compare and contrast RealTime vs Historical RunMode. RealTime is used for production deployment. Historical is used for development, unit-testing, integration-testing and back-testing.

use wingfoil::*;
use std::time::Duration;
use log::Level::Info;

pub fn main() {
    env_logger::init();
    for run_mode in vec![
        RunMode::RealTime,
        RunMode::HistoricalFrom(NanoTime::ZERO)
    ] {
        println!("\nUsing RunMode::{:?}", run_mode);
        ticker(Duration::from_secs(1))
            .count()
            .logged("tick", Info)
            .run(run_mode, RunFor::Cycles(3));
    }
}

This output is produced:

Using RunMode::RealTime
[17:34:46Z INFO wingfoil] 0.000_001 tick 1
[17:34:47Z INFO wingfoil] 1.000_131 tick 2
[17:34:48Z INFO wingfoil] 2.000_381 tick 3

Using RunMode::HistoricalFrom(NanoTime(0))
[17:34:48Z INFO  wingfoil] 0.000_000 tick 1
[17:34:48Z INFO  wingfoil] 1.000_000 tick 2
[17:34:48Z INFO  wingfoil] 2.000_000 tick 3

In Realtime mode the log statements are written every second. In Historical mode the log statements are written immediately. In both cases engine time advances by 1 second between each tick.

§Order Book Example

In this example, we load a CSV file of limit orders for AAPL stock ticker trading on the NASDAQ exchange. The data is sourced from lobsterdata samples

We use the coincidentally named lobster rust library to maintain an order book over time.

Trades and two-way prices are derived, exported back out to CSV and plotted below.

The frequencies of the inputs and outputs are all different to each other.

In addition to the csv output, we also get a performance summary and pretty-print of the graph. One hours worth of data was processed in 287 milliseconds.

8 nodes wired in 10.326µs
Completed 91998 cycles in 287.125397ms. 3.12µs average.
[00] CsvReaderStream
[01]   FilterStream
[02]      MapStream
[03]        MapStream
[04]          DistinctStream
[05]            CsvWriterNode
[06]        MapStream
[07]          CsvVecWriterNode

env_logger::init();
let book = RefCell::new(lobster::OrderBook::default());
let source_path = "examples/lobster/data/aapl.csv";
let fills_path = "examples/lobster/data/fills.csv";
let prices_path = "examples/lobster/data/prices.csv";
// map from seconds from midnight to NanoTime time
let get_time = |msg: &Message| NanoTime::new((msg.seconds * 1e9) as u64);
let (fills, prices) = csv_read_vec(source_path, get_time, true)
    .map(move |chunk| process_orders(chunk, &book))
    .split();
let prices_export = prices
    .filter_value(|price: &Option<TwoWayPrice>| !price.is_none())
    .map(|price| price.unwrap())
    .distinct()
    .csv_write(prices_path);
let fills_export = fills.csv_write_vec(fills_path);
let run_mode = RunMode::HistoricalFrom(NanoTime::ZERO);
let run_for = RunFor::Forever;
Graph::new(vec![prices_export, fills_export], run_mode, run_for)
    .print()
    .run()
    .unwrap();

See the order book example for more details.

§async / tokio integration

In this example, we demonstrate wingfoil’s integration with tokio / async. This makes building IO adapters at the graph edge a breeze, giving the best of sync and async worlds.

Async streams are a useful abstraction for IO but are not as powerful or as easy to use for implementing complex business logic as wingfoil. Async uses an implicit, depth first graph execution strategy. In contrast wingfoil’s explicit, breadth first graph execution algorithm, is easier to reason about, encourages re-use and provides a more structured and productive development environment. Wingfoil’s explicit modelling of time with support for historical and realtime modes, is also huge win in terms of productivity and ability to backtest strategies over async streams.

The key methods here are produce_async which maps an async futures stream into a wingfoil stream and consume_async which takes a wingfoil stream and consumes it with a supplied async closure.

This hybrid sync-async approach enforces clear separation between IO and business logic, which can often be problematic in async oriented systems.

RUST_LOG=INFO cargo run --example async

use async_stream::stream;
use std::time::Duration;
use std::pin::Pin;
use futures::StreamExt;
use wingfoil::*;

let period = Duration::from_millis(10);
let run_for = RunFor::Duration(period * 5);
let run_mode = RunMode::RealTime;
let producer = move |_ctx: RunParams| async move {
    Ok(stream! {
        for i in 0.. {
            tokio::time::sleep(period).await; // simulate waiting IO
            let time = NanoTime::now();
            yield Ok((time, i * 10));
        }
    })
};

let consumer = async move |mut source: Pin<Box<dyn FutStream<u32>>>| {
    while let Some((time, value)) = source.next().await {
        println!("{time:?}, {value:?}");
    }
    Ok(())
};

produce_async(producer)
    .logged("on-graph", log::Level::Info)
    .collapse()
    .consume_async(Box::new(consumer))
    .run(run_mode, run_for);

See the async example for more details.

§Graph Execution

In this example we illustrate the power of wingfoil’s breadth-first graph execution algorithm and compare with ReactiveX’s depth-first approach. Depth first execution is also problematic for async streams.

Also note that wingfoil’s breadth first approach, by construction, eliminates “reactive glitches” (potential logic defects due to inconsistent intermediate state). See StackOverflow and Wikipedia for more details.

In wingfoil we build an example with a depth of 127 branch / recombine operations:

use wingfoil::*;

let mut source = constant(1_u128);
for _ in 0..127 {
    source = add(&source, &source);
}
let cycles = source.count();
cycles.run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
    .unwrap();
println!("cycles {:?}", cycles.peek_value());
println!("value {:?}", source.peek_value());

It produces the correct ouput of 2^127 in 1 engine cycle that takes less than half a millisecond to complete.

384 nodes wired in 212.904µs
Completed 1 cycles in 463.793µs. 463.684µs average.
cycles 1
value 170141183460469231731687303715884105728

If we try and implement the same logic in ReactiveX we quickly realise that it is unfeasible. The issue is that each branch / recombine operation doubles the number of downstream ticks, resulting in an explosive O(2^N) time complexity. Extrapolating from a smaller depths, we can estimate that it would of the order of 10^24 years to run the example with a depth of 127. See the breadth first example for more details.

§RFQ Example

In this example, we sketch out a Request For Quote (RFQ) responder, receiving messages from paradigm over an async websocket and using the demux pattern to route RFQs to pool of statically wired sub-graphs, where business logic can be implemented to respond to them.

This example can be run with:

RUST_LOG=INFO cargo run --example rfq

See the rfq example for more details.

§Multithreading

Wingfoil supports multi-threading to distribute workloads across cores. The approach is to compose sub-graphs, each running in its own dedicated thread.

§Messaging

Wingfoil is agnostic to messaging and we plan to add support for shared memory IPC adapters such as Aeron for ultra-low latency use cases, ZeroMq for low latency server to server communication and Kafka for high latency, fault tolerant messaging.

§Graph Dynamism

Some use cases require graph-dynamism - consider for example Request for Quote (RFQ) markets, where the graph needs to be able to adapt to an incoming RFQ. Conceptually this could be done by changing the shape of the graph at run time. However, we take a simpler approach by virtualising this, following a demux pattern.

§Performance

We take performance seriously, and ongoing work is focused on making Wingfoil even closer to a zero-cost abstraction, Currently, the overhead of cycling a node in the graph is less than 10 nanoseconds.

For best performance we recommend using cheaply cloneable types:

For small strings: arraystring
For small vectors: tinyvec
For larger or heap-allocated types:
- Use Rc<T> for single threaded contexts.
- Use Arc<T> for multithreaded contexts.

Modules§

adapters: A library of input and output adapters

Macros§

burst: Macro to create a Burst<T> with type inference.

Structs§

CallBackStream: A queue of values that are emitted at specified times. Useful for unit tests. Can also be used to feed stream output back into the Graph as input on later cycles.
DemuxMap: Maintains map from Key k, to output node for demux in order to demux a source. Used by StreamOperators::demux_it.
FeedbackSink: Write end of a feedback channel. Clone-able and can be moved into closures. Calling send pushes a value onto the shared queue and schedules the paired source stream to cycle.
Graph: Engine for co-ordinating execution of Nodes
GraphState: Maintains the parts of the graph state that is accessible to Nodes.
NanoTime: A time in nanoseconds since the unix epoch.
Overflow: Represents a Stream of values that failed to be demuxed because the the demux capacity was exceeded. Output of StreamOperators::demux and StreamOperators::demux_it.
RunParams: Context passed to async producer closures during graph setup.
UpStreams: The graph can ask a Node what it’s upstreams sources are. The node replies with a UpStreams for passive and active sources. All sources are wired upstream. Active nodes trigger Node.cycle() when they tick. Passive Nodes do not.

Enums§

DemuxEvent: A message used to signal that a demuxed child stream can be closed. Used by StreamOperators::demux and StreamOperators::demux_it
Dep: Wraps a Stream to indicate whether it is an active or passive dependency. Active dependencies trigger downstream nodes when they tick. Passive dependencies are read but don’t trigger execution.
RunFor: Defines how long the graph should run for. Can be a Duration, number of cycles or forever.
RunMode: Whether the Graph should run RealTime or Historical mode.

Traits§

AsNode: Used to cast Rc<dyn Stream> to Rc<dyn Node>
AsStream: Used co cast Rc of concrete stream into Rc of dyn Stream.
BenchBuilder: A function that accepts a trigger node and wires downstream logic to be benchmarked.
FutStream: A convenience alias for futures::Stream with items of type (NanoTime, T). used by StreamOperators::consume_async.
IntoNode: Used to consume a concrete MutableNode and return an Rc<dyn Node>>.
IntoStream: Used to consume a concrete Stream and return an Rc<dyn Stream>>.
MutableNode: Implement this trait create your own Node.
Node: A wiring point in the Graph.
NodeFlowOperators: Flow-control operators for Nodes. These mirror the same-named methods on StreamOperators but operate on tick signals rather than values.
NodeOperators: A trait containing operators that can be applied to Nodes. Used to support method chaining syntax.
Stream: A Node which has some state that can peeked at.
StreamOperators: A trait containing operators that can be applied to Streams. Used to support method chaining syntax.
StreamPeek: The trait through which a Streams can current value can be peeked at.
StreamPeekRef: A trait through which a reference to Stream’s value can be peeked at.

Functions§

add: Returns a Stream that adds both it’s source Streams. Ticks when either of it’s sources ticks.
add_bench: Used to add wingfoil bench to criterion.
always: Produces a Node that ticks on every engine cycle.
bimap: Maps two Streams into one using the supplied function. Use Dep::Active and Dep::Passive to control which upstreams trigger execution.
combine: Collects a Vec of Streams into a Stream of Vec.
constant: Returns a stream that ticks once with the specified value, on the first cycle.
feedback: Creates a feedback channel. Returns a (FeedbackSink, Stream) pair. The sink pushes values onto a shared [TimeQueue]; the source stream pops them on the next engine cycle.
feedback_node: Creates a feedback channel carrying (). Returns a (FeedbackSink, Node) pair suitable for signalling ticks without carrying a value.
merge: Returns a stream that merges it’s sources into one. Ticks when either of it’s sources ticks. If more than one source ticks at the same time, the first one that was supplied is used.
never: Produces a Node that never ticks.
produce_async: Create a Stream from a fallible async function that produces a futures::Stream.
producer: Creates a Stream emitting values on this thread but produced on a worker thread.
ticker: Returns a Node that ticks with the specified period.
trimap: Maps three Streams into one using the supplied function. Use Dep::Active and Dep::Passive to control which upstreams trigger execution.
try_bimap: Maps two Streams into one using a fallible closure. Use Dep::Active and Dep::Passive to control which upstreams trigger execution. Errors propagate to graph execution.
try_trimap: Maps three Streams into one using a fallible closure. Use Dep::Active and Dep::Passive to control which upstreams trigger execution. Errors propagate to graph execution.

Type Aliases§

Burst: A small vector optimised for single-element bursts.

Crate wingfoil

Crate wingfoil Copy item path

§Wingfoil

§Features

§Quick Start

§Order Book Example

§Get Involved!

§Graph Execution

§Historical vs RealTime

§Order Book Example

§async / tokio integration

§Graph Execution

§RFQ Example

§Multithreading

§Messaging

§Graph Dynamism

§Performance

Modules§

Macros§

Structs§

Enums§

Traits§

Functions§

Type Aliases§

Crate wingfoil