Custom nodes with the Operator trait

The built-in operators cover a lot. They don’t cover everything.

Maybe you need a domain-specific calculation that doesn’t fit SMA/EMA/RSI — an RF dwell-time estimator, a custom anomaly score, a protocol-specific health metric. Maybe it’s a proprietary detector your team doesn’t want to open-source. Maybe you just want to prototype a new operator idea without waiting for it to land in tflo-ops.

Custom nodes are how you extend the graph. The Operator trait defines the contract — implement it, and your logic gets wired into the same execution engine that runs built-in nodes. This is the exact mechanism the tflo-fintech crate uses to ship ADX, ATR, and KAMA without any finance-specific code living in tflo-core.

The trait

pub trait Operator: Send + Sync + 'static {
    /// Evaluate the node against the current record's resolved inputs and timestamp.
    fn eval(&mut self, inputs: &[Computed], ts: i64) -> NodeOutput;

    /// Reset internal state to the freshly-constructed condition.
    fn reset(&mut self) {}

    /// Human-readable name, used in graph-plan and debug output.
    fn name(&self) -> &str { "operator" }

    /// Serialize state for checkpointing (default: not checkpointable).
    fn save(&self) -> Option<Vec<u8>> { None }

    /// Restore state from bytes produced by `save()`.
    fn load(&mut self, bytes: &[u8]) -> Result<(), OperatorLoadError> { /* ... */ }
}

Only eval is mandatory. It receives one Computed per wired input — in declaration order — plus the record timestamp ts, and returns a NodeOutput. Wrap a Result<f64, Absent> result with NodeOutput::computed(...). Return NodeOutput::computed(Err(Absent::WarmingUp)) while the node is still warming up. reset, name, save, and load all have defaults — overriding save/load makes the node checkpointable; the default leaves it out of snapshots.

The free require(inputs, idx) helper reads an input and returns Computed — because eval returns NodeOutput (not Computed), the ? operator no longer applies. Use match instead: match require(inputs, 0) { Ok(v) => v, Err(e) => return NodeOutput::computed(Err(e)) }.

Build a Rate-of-Change node

Rate of Change measures how much a value has changed relative to N records ago: (current - N_records_ago) / N_records_ago. (tflo-ops actually ships this as the rate_of_change operator — but it makes a clean, self-contained example of the trait.) Here’s the full implementation:

use std::collections::VecDeque;
use tflo_core::compile::{Absent, Computed, NodeOutput};
use tflo_core::operator::{require, Operator};

struct RateOfChange {
    period: usize,
    buffer: VecDeque<f64>,
}

impl RateOfChange {
    pub fn new(period: usize) -> Self {
        Self {
            period,
            buffer: VecDeque::with_capacity(period + 1),
        }
    }
}

impl Operator for RateOfChange {
    fn eval(&mut self, inputs: &[Computed], _ts: i64) -> NodeOutput {
        let current = match require(inputs, 0) {
            Ok(v) => v,
            Err(e) => return NodeOutput::computed(Err(e)),
        };
        self.buffer.push_back(current);

        if self.buffer.len() > self.period + 1 {
            self.buffer.pop_front();
        }

        if self.buffer.len() < self.period + 1 {
            return NodeOutput::computed(Err(Absent::WarmingUp)); // not enough data yet
        }

        let n_records_ago = self.buffer.front().copied().unwrap_or(current);
        if n_records_ago == 0.0 {
            return NodeOutput::computed(Err(Absent::DivideByZero));
        }

        NodeOutput::computed(Ok((current - n_records_ago) / n_records_ago))
    }

    fn reset(&mut self) {
        self.buffer.clear();
    }

    fn name(&self) -> &str {
        "rate_of_change"
    }
}

That’s it. eval reads its single input with require — which returns Err(Absent::WarmingUp) if the node is mis-wired — and wraps the result in NodeOutput::computed(...). The trait carries no step/context plumbing: an operator sees only resolved Computed inputs plus the record timestamp, which keeps the contract small.

Wiring it into the graph

An Operator drops straight into a .tflo() closure through Comp::custom_node or its single-input shorthand Comp::custom_node1:

use tflo_core::prelude::*;
use tflo_core::operator::Operator;

#[derive(Clone)]
struct Reading {
    ts: i64,
    value: f64,
}

let roc: Vec<f64> = readings.into_iter()
    .tflo(|t| {
        t.timestamp(|x| x.ts);
        let value = t.prop(|x| x.value);
        // single-input node: consumes `value`
        value.custom_node1(|| RateOfChange::new(14))
    })
    .collect();

custom_node1 takes a factory closure — || RateOfChange::new(14) — not a node instance. The graph stores the factory so it can mint a fresh, independent node for every compiled graph, including each per-key graph in keyed execution. That’s what keeps state from leaking between keys.

For a node that reads more than one input, use the multi-input Comp::custom_node. The inputs arrive in eval in exactly the order you list them:

use tflo_core::compile::{Computed, NodeOutput};
use tflo_core::operator::{require, Operator};

struct Spread;

impl Operator for Spread {
    fn eval(&mut self, inputs: &[Computed], _ts: i64) -> NodeOutput {
        // Absent in either input short-circuits the whole node.
        let bid = match require(inputs, 0) { Ok(v) => v, Err(e) => return NodeOutput::computed(Err(e)) };
        let ask = match require(inputs, 1) { Ok(v) => v, Err(e) => return NodeOutput::computed(Err(e)) };
        NodeOutput::computed(Ok(ask - bid))
    }
}

let spreads: Vec<f64> = readings.into_iter()
    .tflo(|t| {
        t.timestamp(|x| x.ts);
        let high = t.prop(|x| x.high);
        let low = t.prop(|x| x.low);
        // inputs reach eval() as [high, low]
        Comp::custom_node(&high, &[&low], || Spread)
    })
    .collect();

custom_node panics if you pass an empty input slice — every node needs at least one input to attach to.

Use constants via the builder

Need a constant fed into your node? TFlowBuilder::constant() registers a Node::Const and hands back a Comp<R> you can chain or pass as an input:

use tflo_core::prelude::*;

// Inside a .tflo() closure:
let threshold = t.constant(0.05);
let signal = value.cross_above(&threshold);

Operators are first-class

Once attached, an operator compiles into the same execution pipeline as everything else. Same topological ordering guarantees. Same absence propagation — returning NodeOutput::computed(Err(Absent::WarmingUp)) holds downstream nodes back exactly like a built-in window that hasn’t filled. Same snapshot/restore lifecycle — implement save/load and the node checkpoints with the rest of the graph. The only difference is that you wrote the eval body.

That’s the whole extension story: implement one trait, attach it with one method, and your logic runs shoulder to shoulder with the built-ins.

Source code

The full working example is available on GitHub: tflo-examples/examples/custom-nodes

For composite operator patterns (extension traits), see: tflo-examples/examples/custom-composite

— Matt