11.0.0: Build feature pipelines at runtime.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "sliding_features"
-version = "10.0.1"
+version = "11.0.0"
 authors = ["MathisWellmann <[email protected]>"]
 edition = "2024"
 license-file = "LICENSE"
@@ -82,6 +82,10 @@ harness = false
 name = "noise_elimination_technology"
 harness = false
 
+[[bench]]
+name = "pipeline"
+harness = false
+
 [[bench]]
 name = "polarized_fractal_efficiency"
 harness = false

README.md

@@ -52,6 +52,45 @@ for v in &test_values {
 Each View will first call it's chained View to get it's last value, which will then be used to update the state of the View.
 Some Views have additional parameters such as ALMA. 
 
+### Features at runtime: the `pipeline` module
+Chained views are the zero-cost choice when the features are known at compile
+time. When they are only known at runtime, write them as text and compile
+them with the `pipeline` module, the crate's one way to assemble features at
+runtime. Features can then be stored, compared and generated (e.g. by a
+search or an LLM agent) as data, without compiling code. A pipeline computes
+many features over several input series in one flat program:
+
+```rust
+use sliding_features::pipeline::{Domain, Input, Limits, Pipeline};
+
+let source = "
+    let r = ln_return(price)
+    vol_ratio = std(r, 20) / std(r, 100)
+    trend = zscore(ema(price, 10) - ema(price, 50), 200)
+";
+let inputs = vec![Input::new("price", Domain::positive())];
+let mut pipeline = Pipeline::compile(source, inputs, Limits::default()).unwrap();
+pipeline.update(&[100.0]);
+assert_eq!(pipeline.output(0), None, "warming up");
+```
+
+- `name = expr` defines an output, `let name = expr` a helper, and every
+  function takes an expression and then numeric parameters, e.g. `pfe(x, 20, 5)`.
+  `pipeline::reference()` renders the language guide and the table of all
+  functions, ready for a prompt.
+- Equal subexpressions are computed once, and an update runs a flat list of
+  operations with no allocation, recursion or dynamic dispatch.
+- Before any data is seen, the compiler derives every node's value range from
+  the inputs' domains: it rejects calls that can never be ready (`ln` of a
+  negative value) and constant outputs, and warns about calls that are not
+  always ready, with line, column and a hint how to fix it.
+- Outputs are only ever finite: a division by zero is "not ready" instead of
+  infinite. `Limits` bounds windows, node count and nesting, so text from an
+  untrusted author cannot make a pipeline arbitrarily slow or large.
+- The `Display` form is canonical, so equal strings mean equal features.
+- `pipeline::Builder` assembles the same pipelines from code, and
+  `pipeline::Feature` turns a one-input, one-output pipeline into a `View`.
+
 ### Examples
 See examples folder for some code ideas
 ```shell 
@@ -92,6 +131,8 @@ A View defines the function which processes value updates. They currently includ
     * Multiply
     * Divide
 * Math functions
+    * Abs
+    * Ln
     * Tanh
     * GTE - Greater Than or Equal clipping function
     * LTE - Lower Than or Equal clipping function

benches/pipeline.rs

@@ -0,0 +1,133 @@
+//! A pipeline against the same feature as a hand-written generic chain, and
+//! a feature set as one pipeline against one pipeline per feature.
+
+use std::{
+    hint::black_box,
+    num::NonZeroUsize,
+};
+
+use criterion::{
+    Criterion,
+    criterion_group,
+    criterion_main,
+};
+use sliding_features::{
+    View,
+    pipeline::{
+        Feature,
+        Pipeline,
+    },
+    pure_functions::{
+        Echo,
+        Subtract,
+    },
+    sliding_windows::{
+        Ema,
+        Sma,
+        ZScoreStandardization,
+    },
+};
+
+const N: usize = 100_000;
+
+fn random_walk() -> Vec<f64> {
+    let rng = romu::Rng::from_seed_with_64bit(1);
+    let mut price = 100.0;
+    Vec::from_iter((0..N).map(|_| {
+        price += rng.f64() - 0.5;
+        price
+    }))
+}
+
+fn feature(source: &str) -> Feature {
+    let pipeline: Pipeline = source.parse().expect("compiles");
+    Feature::try_from(pipeline).expect("one input and one output")
+}
+
+/// Push every value through a view and read it, as a feature loop does.
+fn drive(view: &mut impl View<f64>, vals: &[f64]) {
+    for v in vals {
+        view.update(*v);
+        let _ = black_box(view.last());
+    }
+}
+
+/// One feature: the cost of interpreting the program against code the
+/// compiler monomorphized.
+fn one_feature(c: &mut Criterion) {
+    let vals = random_walk();
+    let source = "zscore(ema(x, 10) - sma(x, 50), 200)";
+    let window = |n| NonZeroUsize::new(n).expect("non zero");
+    let mut group = c.benchmark_group("pipeline_one_feature_100k");
+    group.bench_function("static_chain", |b| {
+        b.iter(|| {
+            let mut view = ZScoreStandardization::new(
+                Subtract::new(
+                    Ema::new(Echo::new(), window(10)),
+                    Sma::new(Echo::new(), window(50)),
+                ),
+                window(200),
+            );
+            drive(&mut view, &vals);
+        })
+    });
+    group.bench_function("pipeline", |b| {
+        b.iter(|| drive(&mut feature(source), &vals))
+    });
+    group.finish();
+}
+
+/// Mostly stateless nodes: the per-operation overhead of the interpreter.
+fn stateless_heavy(c: &mut Criterion) {
+    let vals = random_walk();
+    let source = "tanh(abs(-(-(-(-(x * 2 + 1)))) - ema(x, 5)) * 0.01)";
+    let mut group = c.benchmark_group("pipeline_stateless_heavy_100k");
+    group.bench_function("pipeline", |b| {
+        b.iter(|| drive(&mut feature(source), &vals))
+    });
+    group.finish();
+}
+
+/// A feature set with shared subexpressions: one pipeline computes each of
+/// them once, one pipeline per feature computes them per feature.
+fn feature_set(c: &mut Criterion) {
+    let vals = random_walk();
+    let sources = [
+        "std(ln_return(x), 20) / std(ln_return(x), 100)",
+        "std(ln_return(x), 20)",
+        "zscore(ema(x, 10) - ema(x, 50), 200)",
+        "mad(ema(x, 10) - ema(x, 50), 100)",
+        "roc(ema(x, 10), 10)",
+        "rsi(x, 14)",
+        "ema(rsi(x, 14), 5)",
+        "tanh(zscore(ln_return(x), 100))",
+    ];
+    let mut group = c.benchmark_group("pipeline_feature_set_100k");
+    group.bench_function("pipeline_per_feature", |b| {
+        b.iter(|| {
+            let mut features = Vec::from_iter(sources.iter().map(|s| feature(s)));
+            for v in &vals {
+                for feature in &mut features {
+                    feature.update(*v);
+                    let _ = black_box(feature.last());
+                }
+            }
+        })
+    });
+    group.bench_function("one_pipeline", |b| {
+        let source = sources.join("\n");
+        b.iter(|| {
+            let mut pipeline: Pipeline = source.parse().expect("compiles");
+            for v in &vals {
+                pipeline.update(&[*v]);
+                for out in pipeline.outputs() {
+                    let _ = black_box(out);
+                }
+            }
+        })
+    });
+    group.finish();
+}
+
+criterion_group!(benches, one_feature, stateless_heavy, feature_set);
+criterion_main!(benches);

src/lib.rs

@@ -5,6 +5,7 @@
 //! The sliding_features crate provides modular, chainable sliding windows
 //! for various signal processing function and technical indicators
 
+pub mod pipeline;
 pub mod pure_functions;
 pub mod rolling;
 pub mod sliding_windows;