wingfoil 8.0.0

graph based stream processing framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

//! Streaming statistics: EWMA, rolling windows, and time-weighting.
//!
//! Demonstrates the [`StatisticsOperators`] trait — a family of numeric
//! aggregations that chain onto any `Stream<T: ToPrimitive>` and emit `f64`.
//! Every operator comes in a count-based flavour and, where it makes sense, a
//! time-weighted one (via [`Weighting`]) so irregular tick spacing is handled
//! correctly.
//!
//! One price stream feeds a spread of statistics.  [`combine`] bundles them into
//! a single `Stream<Burst<f64>>` — one row of values per tick — which we sample
//! twice a second and print as a table row as it ticks.
//!
//! ```bash
//! cargo run --example statistics
//! ```

use std::rc::Rc;
use std::time::Duration;
use wingfoil::adapters::statistics::*;
use wingfoil::*;

fn main() {
    // A single synthetic price stream: a gentle sine wave around 100, one
    // sample every 100ms. Every statistic below is derived from this one stream.
    let price = ticker(Duration::from_millis(100))
        .count()
        .map(|n: u64| 100.0 + ((n as f64) * 0.6).sin() * 5.0);

    // One statistic per column, all derived from the shared price stream.
    let labels = ["price", "ewma", "sma", "std", "min", "max", "twap"];
    let columns: Vec<Rc<dyn Stream<f64>>> = vec![
        price.clone(),
        price.ewma(EwmaSpan::PerTick(0.3)),
        price.mean(Window::Count(10), Weighting::Count),
        price.std(Window::Count(10), Weighting::Count),
        price.min(Window::Count(10)),
        price.max(Window::Count(10)),
        price.mean(Window::Unbounded, Weighting::Time),
    ];

    // Header first, then stream: `combine` collects the columns (in order) into
    // one Stream<Burst<f64>> whose value each tick is a row holding every
    // column's latest value. Sample that row twice a second and print it live.
    print!("{:>6}", "time");
    for label in labels {
        print!(" {label:>7}");
    }
    println!();

    let trigger = ticker(Duration::from_millis(500));
    let table = combine(columns).sample(trigger).for_each(|row, time| {
        print!("{:>5.1}s", f64::from(time) / 1e9);
        for value in &row {
            print!(" {value:>7.2}");
        }
        println!();
    });

    Graph::new(
        vec![table],
        RunMode::HistoricalFrom(NanoTime::ZERO),
        RunFor::Duration(Duration::from_secs(5)),
    )
    .run()
    .unwrap();
}