Crate indicator

Abstractions for stream aggregation that we call “Indicator” s.

This crate provides abstractions of different levels to build indicators:

• Indicators are the combinations of Operators. We can chain them by OperatorExt::then and apply them to the same input “simultaneously” by OperatorExt::facet.

• To handle Time series better, we introduced utils for ticked operators, which are defined in TickedOperatorExt, of which the input and output are both Tickable and sharing the same Tick. We provide many ticked version utils. For example, we can apply different TickedOperatorExts (using TickedOperatorExt::facet_t or ticked::FacetMap) to the same Tickable stream to get a “synced” result stream of the combined outputs of those operators sharing the same Tick of theirs input. Each item of a “synced” stream will have the form of TickValue<(O1, O2)>.

• TumblingOperators are defined by some operations on events of non-overlapping time windows (TumblingWindow). Moving average is one of the famous examples, which is defined by the average of the numbers from those events occur in the same tumbling window. We call those operations TumblingOperations. We can use tumbling function to create a TumblingOperator from a TumblingOperation.

• Finally, we can apply the indicators to Iterators or Streams by using IndicatorIteratorExt::indicator or IndicatorStreamExt::indicator accordingly.

Example

use indicator::*;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use time::macros::offset;
use arrayvec::ArrayVec;

/// Return an indicator that calculates `hl2` and `ohlc4` simultaneously.
fn hl2_ohlc4(period: Period) -> impl Operator<TickValue<Decimal>, Output = (Decimal, Decimal)> {
    tumbling(
        period,
        |_w: &ArrayVec<[Decimal; 4], 0>, y: &mut Option<[Decimal; 4]>, x| match y {
            Some(ohlc) => {
                ohlc[1] = ohlc[1].max(x);
                ohlc[2] = ohlc[2].min(x);
                ohlc[3] = x;
                *ohlc
            }
            None => {
                let ohlc = [x; 4];
                *y = Some(ohlc);
                ohlc
            }
        },
    )
    .then(facet_t(
        map_t(|ohlc: [Decimal; 4]| (ohlc[1] + ohlc[2]) / dec!(2)),
        map_t(|ohlc: [Decimal; 4]| (ohlc[0] + ohlc[1] + ohlc[2] + ohlc[3]) / dec!(4)),
    ))
    .map(|v| v.value)
}

Re-exports

pub use iter::IndicatorIteratorExt;

pub use operator::facet;

pub use operator::facet;

pub use operator::map;

pub use operator::map;

pub use operator::Operator;

pub use operator::OperatorExt;

pub use ticked::facet_t;

pub use ticked::map_t;

pub use ticked::tumbling::cached;

pub use ticked::tumbling::cached;

pub use ticked::tumbling::iterated;

pub use ticked::tumbling::iterated;

pub use ticked::tumbling::tumbling;

pub use ticked::tumbling::Cached;

pub use ticked::tumbling::CachedOperation;

pub use ticked::tumbling::Iterated;

pub use ticked::tumbling::IteratedOperation;

pub use ticked::tumbling::QueueCapAtLeast;

pub use ticked::tumbling::TumblingOperation;

pub use ticked::tumbling::TumblingOperator;

pub use ticked::tumbling::TumblingQueue;

pub use ticked::tuple_t;

pub use ticked::TickedOperatorExt;

pub use window::Period;

pub use window::PeriodKind;

pub use window::Tick;

pub use window::TickValue;

pub use window::Tickable;

pub use window::TumblingWindow;

pub use ticked::facet_map_t;

pub use facet::facet_map;

pub use ticked::shared;

pub use ticked::SharedMap;

pub use stream::IndicatorStreamExt;

pub use ticked::array_t;

pub use async_operator::AsyncOperator;

pub use async_operator::ServiceOperator;

pub use indicator_macros as macros;

Modules

async_operator

Async operator support.

context

Context Pattern.

gat

Operator using GAT.

iter

Iterator extension trait.

operator

Operator.

prelude

Prelude.

rayon

Rayon supported combinator.

reactive

Reactive streams pattern.

stream

Stream extension trait.

ticked

Ticked operators.

window

Time window.