Brummel/Aura
1
0
Fork 0
Code Issues 26 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e304dbaae1fc33f48e1ee26bf8d608b295602ed3
Aura/crates/aura-engine/src/harness.rs
T
Brummel e304dbaae1 feat(aura-core): name input ports 
PortSpec gains a non-load-bearing `name: String`, so an input port is named just
as FieldSpec.name (output) and ParamSpec.name (param) already are — input ports
were the lone unnamed member of the node signature. Identity stays positional by
slot (C23); the name is render/debug only, never read by bootstrap or the run
loop. PortSpec drops Copy (String is not Copy), exactly as ParamSpec already does.

- Every aura-std node names its input slots: SMA/EMA "series", Sub/Add "lhs"/"rhs",
  Exposure "signal", SimBroker "exposure"/"price" (the slots become
  self-documenting), LinComb "term[i]" and Recorder "col[i]" generated in their
  build loops (mirroring LinComb's existing weights[i] param loop).
- derive_signature carries a composite's Role.name into the derived input port
  (it was dropped before — the output side already carried FieldSpec.name), so the
  graph model is homogeneously named at both levels.
- model_to_json (port_json + the composite-header inputs in scope_json) emits the
  name as a third tuple element: ["f64","any","exposure"]. The byte golden was
  re-captured (machine bytes) and its substring twins updated; the model is now
  fully named across inputs/outputs/params.
- All 16 PortSpec construction sites threaded in one compile-gate change; test
  fixtures carry fixture names. C8 realization note added to the design ledger.

Why name-only, no validation: the name is a pure debug symbol. Wire-by-name was
rejected (it would be a C23 contract change). Bootstrap slot-wiring validation
(which would close #21's same-kind swap footgun) is deferred to its own cycle —
a name alone does not catch the swap; it makes the slots self-documenting and
gives a future validation something to check against.

Verified: cargo test --workspace 168 green; clippy --all-targets -D warnings
clean; cargo build --workspace clean. Read-only render path (C9), no serde (C14),
scalar kinds unchanged (C4).

closes #50
refs #21
refs #51
2026-06-10 16:19:08 +02:00

1938 lines
80 KiB
Rust
Raw Blame History

//! The harness — the closed root graph that runs (glossary: `harness`, C20) —
//! and its deterministic run loop. A `Harness` is a bootstrapped, frozen root
//! graph: a flat node array plus an index edge table, topologically ordered,
//! driven cycle by cycle by a k-way merge of timestamped sources (C3/C4). It is
//! the flat, monomorphized sharpening of RustAst's reference-counted,
//! interior-mutable observer push graph: no reference counting, no interior
//! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns
//! (the cycle-0002 shape), so `Ctx` borrows them read-only and additionally
//! carries the cycle timestamp (`ctx.now()`, C4). A node's `eval` returns a
//! borrowed record (`Option<&[Scalar]>`); each out-edge forwards one field of it
//! (`Edge::from_field`) into a consumer slot, so the K fields are co-fresh (C6).
//!
//! Firing (C5/C6) gates re-evaluation. Two read clocks drive it, both stamped per
//! input slot on every push: `fresh_at` (the `cycle_id` of the last push —
//! freshness epoch, C5) and `last_ts` (the data timestamp of that push — barrier
//! token, C6). A node fires when any `Firing::Any` input is fresh this cycle, or
//! when a `Firing::Barrier` group has every member at the current timestamp; a
//! node that does not fire pushes nothing, so consumers keep seeing the held
//! value via `window[0]` (sample-and-hold falls out of the push model). The
//! barrier token is the timestamp, not the `cycle_id`: under C4 four same-time
//! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never
//! fire across sources — the timestamp generalizes it faithfully.
use aura_core::{AnyColumn, Ctx, Firing, Node, NodeSchema, Scalar, ScalarKind, Timestamp};
/// Forwards one field (`from_field`) of a producer's output record into a
/// consumer's input slot. Consuming a whole record is N such edges, one per field
/// (there is no "bind whole record" mechanism).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Edge {
pub from: usize,
pub to: usize,
pub slot: usize,
pub from_field: usize,
}
/// An input slot the source value is forwarded into each cycle.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Target {
pub node: usize,
pub slot: usize,
}
/// A declared source: the scalar kind it produces and the input slots each of
/// its records is forwarded into. The multi-source generalization of cycle
/// 0003's `(source_targets, source_kind)` pair — sources are k-way-merged by
/// timestamp at ingestion (C3); there is no merge inside the graph.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SourceSpec {
pub kind: ScalarKind,
pub targets: Vec<Target>,
}
/// The flat, type-erased, index-wired output of `Composite::compile` — the target
/// `Harness::bootstrap` consumes (C23's "compilat", now a named type). `signatures[i]`
/// is the static signature of `nodes[i]` (gathered from each primitive's builder at
/// lowering); `bootstrap` reads kinds/output from it and `nodes[i].lookbacks()` for
/// buffer depth. `sources` are the lowered bound roles, in role-declaration order.
pub struct FlatGraph {
pub nodes: Vec<Box<dyn Node>>,
pub signatures: Vec<NodeSchema>,
pub sources: Vec<SourceSpec>,
pub edges: Vec<Edge>,
}
/// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
/// generalized to the whole topology.
#[derive(Debug, PartialEq, Eq)]
pub enum BootstrapError {
/// An edge or source-target connects mismatched scalar kinds.
KindMismatch { producer: ScalarKind, consumer: ScalarKind },
/// A node or slot index in an edge or target is out of range.
BadIndex,
/// The wiring contains a directed cycle (only an explicit delay node may close
/// a loop; that node does not exist yet).
Cycle,
}
/// Per-input engine bookkeeping (invisible to nodes): the two read clocks of the
/// firing machinery. `fresh_at` is the `cycle_id` of the last push into this slot
/// (freshness epoch, C5: fresh this cycle iff `fresh_at == cycle_id`); `last_ts`
/// is the data timestamp of that push (barrier token, C6: a barrier group is
/// complete iff all members carry `last_ts == T`). A never-pushed slot keeps the
/// cold sentinel `Timestamp(i64::MIN)`, which no real timestamp equals.
struct SlotState {
fresh_at: u64,
last_ts: Timestamp,
}
struct NodeBox {
node: Box<dyn Node>,
inputs: Vec<AnyColumn>,
firing: Vec<Firing>,
slots: Vec<SlotState>,
out_len: usize,
}
/// A bootstrapped, frozen root graph instance plus its deterministic run loop.
pub struct Harness {
nodes: Vec<NodeBox>,
topo: Vec<usize>,
out_edges: Vec<Vec<Edge>>,
sources: Vec<SourceSpec>,
}
// Manual, node-opaque `Debug`: `Box<dyn Node>` is not `Debug`, so the struct
// cannot derive it. This summary form is enough for `Result::unwrap_err` (which
// formats the `Ok` arm on a failed bootstrap-rejection assertion) without
// printing node internals.
impl core::fmt::Debug for Harness {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Harness")
.field("nodes", &self.nodes.len())
.field("topo", &self.topo)
.field("out_edges", &self.out_edges)
.field("sources", &self.sources)
.finish()
}
}
impl Harness {
/// Bind a flat graph into a frozen, runnable graph. Sizes each node's input
/// columns — KIND/firing from its carried signature, DEPTH from the built node's
/// `lookbacks()` — lifts each input's firing policy, initializes per-slot
/// freshness state, kind-checks every source target and edge, and topologically
/// orders the nodes (Kahn), rejecting any directed cycle.
pub fn bootstrap(flat: FlatGraph) -> Result<Harness, BootstrapError> {
let FlatGraph { nodes, signatures, sources, edges } = flat;
let n = nodes.len();
// size each node's input columns: KIND/firing from the carried signature,
// DEPTH from the built node's lookbacks() (the one param-dependent quantity)
let mut boxes: Vec<NodeBox> = Vec::with_capacity(n);
for (nd, sig) in nodes.into_iter().zip(signatures.iter()) {
let depths = nd.lookbacks();
debug_assert_eq!(depths.len(), sig.inputs.len(), "lookbacks() arity == signature inputs");
let inputs: Vec<AnyColumn> = sig
.inputs
.iter()
.zip(depths)
.map(|(spec, depth)| AnyColumn::with_capacity(spec.kind, depth))
.collect();
let firing: Vec<Firing> = sig.inputs.iter().map(|spec| spec.firing).collect();
let slots: Vec<SlotState> = sig
.inputs
.iter()
.map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) })
.collect();
let out_len = sig.output.len();
boxes.push(NodeBox { node: nd, inputs, firing, slots, out_len });
}
// source targets: each source's value must match each of its target slots' kind
for src in &sources {
for t in &src.targets {
let s = signatures.get(t.node).ok_or(BootstrapError::BadIndex)?;
let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
if slot.kind != src.kind {
return Err(BootstrapError::KindMismatch {
producer: src.kind,
consumer: slot.kind,
});
}
}
}
// edges: indices in range, producer output field kind == consumer slot kind
let mut out_edges: Vec<Vec<Edge>> = vec![Vec::new(); n];
for &e in &edges {
let from = signatures.get(e.from).ok_or(BootstrapError::BadIndex)?;
let to = signatures.get(e.to).ok_or(BootstrapError::BadIndex)?;
let field = from.output.get(e.from_field).ok_or(BootstrapError::BadIndex)?;
let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?;
if field.kind != slot.kind {
return Err(BootstrapError::KindMismatch {
producer: field.kind,
consumer: slot.kind,
});
}
out_edges[e.from].push(e);
}
// Kahn topological sort; a leftover node means a cycle
let mut indeg = vec![0usize; n];
for &e in &edges {
indeg[e.to] += 1;
}
let mut queue: Vec<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
let mut topo: Vec<usize> = Vec::with_capacity(n);
let mut head = 0;
while head < queue.len() {
let u = queue[head];
head += 1;
topo.push(u);
for e in &out_edges[u] {
indeg[e.to] -= 1;
if indeg[e.to] == 0 {
queue.push(e.to);
}
}
}
if topo.len() != n {
return Err(BootstrapError::Cycle);
}
Ok(Harness {
nodes: boxes,
topo,
out_edges,
sources,
})
}
/// Drive the sources, k-way-merged in timestamp order (ties by source index,
/// C4). One stream per source, each ascending in timestamp (C3 ingestion
/// precondition). Recording is a node-side concern: a recording node pushes
/// its record to a destination it holds (out of graph) inside `eval`; the
/// engine only routes in-graph edges and is oblivious to the side effect.
/// Allocates nothing per cycle beyond the reused scratch buffer.
pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) {
assert_eq!(
streams.len(),
self.sources.len(),
"run: one stream per source required (got {} streams for {} sources)",
streams.len(),
self.sources.len()
);
// disjoint field borrows so the topo walk can read topo/out_edges/sources
// while mutating nodes
let Harness { nodes, topo, out_edges, sources } = self;
let mut cursor: Vec<usize> = vec![0; streams.len()];
let mut cycle_id: u64 = 0;
let mut scratch: Vec<Scalar> = Vec::new();
loop {
// pick the live source head with the smallest (timestamp, source index)
let mut pick: Option<usize> = None;
for (s, stream) in streams.iter().enumerate() {
if cursor[s] < stream.len() {
match pick {
None => pick = Some(s),
Some(p) => {
if stream[cursor[s]].0 < streams[p][cursor[p]].0 {
pick = Some(s);
}
}
}
}
}
let s = match pick {
Some(s) => s,
None => break, // all streams exhausted
};
let (ts, value) = streams[s][cursor[s]];
cursor[s] += 1;
cycle_id += 1;
// forward the source value into its target slots, stamping freshness
for t in sources[s].targets.iter() {
let nb = &mut nodes[t.node];
nb.inputs[t.slot].push(value).expect("source kind checked at wiring");
nb.slots[t.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
}
// evaluate in topological order; gate by firing; forward Some outputs
for &nidx in topo.iter() {
let out_len = nodes[nidx].out_len;
let fired = {
let nb = &nodes[nidx];
fires(&nb.firing, &nb.slots, cycle_id, ts)
};
if !fired {
continue; // hold: no eval, no push
}
let result: Option<&[Scalar]> = {
let nb = &mut nodes[nidx];
nb.node.eval(Ctx::new(&nb.inputs, ts))
};
if let Some(row) = result {
debug_assert_eq!(row.len(), out_len, "node returned a row of the wrong width");
scratch.clear();
scratch.extend_from_slice(row);
for e in out_edges[nidx].iter() {
let nb = &mut nodes[e.to];
nb.inputs[e.slot]
.push(scratch[e.from_field])
.expect("edge kind checked at wiring");
nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
}
}
}
}
}
}
/// The firing predicate (C5/C6): does this node re-evaluate this cycle? A node
/// fires when *any* of its input groups fires (OR). A `Firing::Any` input fires
/// the node when it is fresh this cycle (`fresh_at == cycle_id`). A barrier group
/// fires when >=1 member is fresh this cycle AND every member carries
/// `last_ts == ts` — the ">=1 fresh" clause is the once-per-timestamp guard (a
/// group completed in a prior cycle has no fresh member now, so it does not
/// re-fire).
fn fires(firing: &[Firing], slots: &[SlotState], cycle_id: u64, ts: Timestamp) -> bool {
// mode A: any fire-on-any-fresh input that is fresh this cycle
for (i, f) in firing.iter().enumerate() {
if matches!(f, Firing::Any) && slots[i].fresh_at == cycle_id {
return true;
}
}
// mode B: each distinct barrier group fires when complete this cycle
if let Some(max_group) = firing.iter().filter_map(group_id).max() {
for g in 0..=max_group {
let mut has_member = false;
let mut any_fresh = false;
let mut all_at_ts = true;
for (i, f) in firing.iter().enumerate() {
if group_id(f) == Some(g) {
has_member = true;
if slots[i].fresh_at == cycle_id {
any_fresh = true;
}
if slots[i].last_ts != ts {
all_at_ts = false;
}
}
}
if has_member && any_fresh && all_at_ts {
return true;
}
}
}
false
}
/// The barrier group id of a firing policy, or `None` for mode A.
fn group_id(f: &Firing) -> Option<u8> {
match f {
Firing::Any => None,
Firing::Barrier(g) => Some(*g),
}
}
#[cfg(test)]
mod tests {
use super::*;
// PortSpec / NodeSchema name the fixtures' declared signatures, brought in here
// (production code reads them via the carried signatures, never naming the types).
use aura_core::{FieldSpec, NodeSchema, PortSpec};
use aura_std::{Exposure, Recorder, Sma, SimBroker, Sub};
use std::sync::mpsc;
/// Build an f64 source stream from (timestamp, value) points.
fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect()
}
/// One f64 input port with the given firing policy (lookback 1 is the bootstrap
/// default for these test fixtures, supplied by their `lookbacks()`).
fn f64_port(firing: Firing) -> PortSpec {
PortSpec { kind: ScalarKind::F64, firing, name: "in".into() }
}
/// Bootstrap a hand-wired graph from boxed nodes + their declared signatures.
/// The signatures parallel `nodes` (one per node, in order); bootstrap reads
/// kinds/firing from them and depth from each node's `lookbacks()`.
fn boot(
nodes: Vec<Box<dyn Node>>,
signatures: Vec<NodeSchema>,
sources: Vec<SourceSpec>,
edges: Vec<Edge>,
) -> Result<Harness, BootstrapError> {
Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges })
}
/// The declared signature of a `Recorder` over `kinds` with the given firing.
fn recorder_sig(kinds: &[ScalarKind], firing: Firing) -> NodeSchema {
NodeSchema {
inputs: kinds.iter().enumerate().map(|(i, &kind)| PortSpec { kind, firing, name: format!("col[{i}]") }).collect(),
output: vec![],
params: vec![],
}
}
// --- firing-policy fixtures (test-local; not library nodes — C9: examples
// for the engine's own tests, no speculative aura-std surface) ---
/// Mode A as-of join: a 2-input f64 sum that fires whenever either input is
/// fresh, holding the other. Warm-up returns None until both have a value.
struct AsOfSum {
out: [Scalar; 1],
}
impl AsOfSum {
fn sig() -> NodeSchema {
NodeSchema {
inputs: vec![f64_port(Firing::Any), f64_port(Firing::Any)],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
}
}
}
impl Node for AsOfSum {
fn lookbacks(&self) -> Vec<usize> {
vec![1, 1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
let a = ctx.f64_in(0);
let b = ctx.f64_in(1);
if a.is_empty() || b.is_empty() {
return None;
}
self.out[0] = Scalar::F64(a[0] + b[0]);
Some(&self.out)
}
}
/// Mode B barrier join: a 2-input f64 sum that fires only when both inputs
/// share the current cycle timestamp (both warm by construction when it fires).
struct BarrierSum {
out: [Scalar; 1],
}
impl BarrierSum {
fn sig() -> NodeSchema {
NodeSchema {
inputs: vec![f64_port(Firing::Barrier(0)), f64_port(Firing::Barrier(0))],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
}
}
}
impl Node for BarrierSum {
fn lookbacks(&self) -> Vec<usize> {
vec![1, 1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
self.out[0] = Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]);
Some(&self.out)
}
}
/// Mixed A+B: barrier pair (inputs 0,1 in group 0) plus an as-of input
/// (input 2). Fires when the pair completes (holding input 2) OR when input 2
/// ticks (holding the pair) — the OR-combine.
struct MixedSum {
out: [Scalar; 1],
}
impl MixedSum {
fn sig() -> NodeSchema {
NodeSchema {
inputs: vec![
f64_port(Firing::Barrier(0)),
f64_port(Firing::Barrier(0)),
f64_port(Firing::Any),
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
}
}
}
impl Node for MixedSum {
fn lookbacks(&self) -> Vec<usize> {
vec![1, 1, 1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
let a = ctx.f64_in(0);
let b = ctx.f64_in(1);
let c = ctx.f64_in(2);
if a.is_empty() || b.is_empty() || c.is_empty() {
return None;
}
self.out[0] = Scalar::F64(a[0] + b[0] + c[0]);
Some(&self.out)
}
}
/// A neutral multi-field producer: five f64 inputs bundled into one 5-field
/// record. No trading-domain logic — it proves the K > 1 output mechanism in
/// isolation. The five inputs are a Barrier(0) group, so the node emits one
/// complete bar only when all five share the cycle's timestamp.
struct Ohlcv {
out: [Scalar; 5],
}
impl Ohlcv {
fn sig() -> NodeSchema {
NodeSchema {
inputs: vec![f64_port(Firing::Barrier(0)); 5],
output: vec![
FieldSpec { name: "open", kind: ScalarKind::F64 },
FieldSpec { name: "high", kind: ScalarKind::F64 },
FieldSpec { name: "low", kind: ScalarKind::F64 },
FieldSpec { name: "close", kind: ScalarKind::F64 },
FieldSpec { name: "volume", kind: ScalarKind::F64 },
],
params: vec![],
}
}
}
impl Node for Ohlcv {
fn lookbacks(&self) -> Vec<usize> {
vec![1; 5]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
for i in 0..5 {
let w = ctx.f64_in(i);
if w.is_empty() {
return None; // not yet warmed
}
self.out[i] = Scalar::F64(w[0]);
}
Some(&self.out) // one 5-field record, all fields co-fresh
}
}
/// A producer whose output record mixes kinds: field 0 is f64, field 1 is i64.
/// Used only to prove the bootstrap kind check is per-field (field 0 would bind
/// into an f64 slot; field 1 would not). Its `eval` never runs in these tests —
/// bootstrap rejects the wiring first.
struct TwoField {
out: [Scalar; 2],
}
impl TwoField {
fn sig() -> NodeSchema {
NodeSchema {
inputs: vec![f64_port(Firing::Any)],
output: vec![
FieldSpec { name: "f", kind: ScalarKind::F64 },
FieldSpec { name: "i", kind: ScalarKind::I64 },
],
params: vec![],
}
}
}
impl Node for TwoField {
fn lookbacks(&self) -> Vec<usize> {
vec![1]
}
fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
self.out[0] = Scalar::F64(0.0);
self.out[1] = Scalar::I64(0);
Some(&self.out)
}
}
/// A node that records AND forwards: it sends `(now, value)` out of the graph
/// (sink side effect) and returns its value as a one-field output the engine
/// forwards downstream (producer). Proves the C8 "both" role.
struct TapForward {
out: [Scalar; 1],
tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
}
impl TapForward {
fn sig() -> NodeSchema {
NodeSchema {
inputs: vec![f64_port(Firing::Any)],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
}
}
}
impl Node for TapForward {
fn lookbacks(&self) -> Vec<usize> {
vec![1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
let w = ctx.f64_in(0);
if w.is_empty() {
return None;
}
let v = w[0];
let _ = self.tx.send((ctx.now(), vec![Scalar::F64(v)])); // sink side effect
self.out[0] = Scalar::F64(v);
Some(&self.out) // producer output: engine forwards it
}
}
#[test]
fn chain_source_sma_runs() {
// node 0 = SMA(3); source -> SMA(3).in0; node 1 = Recorder taps node 0.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Sma::new(3)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid");
h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]);
let got: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// SMA(3) warms at cycle 3; the recorder captures only fired cycles, each
// tagged with the cycle's timestamp (sparse — no None hold-rows).
assert_eq!(
got,
vec![
(Timestamp(3), vec![Scalar::F64(2.0)]),
(Timestamp(4), vec![Scalar::F64(3.0)]),
(Timestamp(5), vec![Scalar::F64(4.0)]),
]
);
}
#[test]
fn fan_out_join_dag_runs_deterministically() {
// 0 = SMA(2), 1 = SMA(4), 2 = Sub; source fans into both SMAs; SMAs join
// into Sub; node 3 = Recorder taps Sub — the 0003 baseline on the new API.
let build = |tx| {
boot(
vec![
Box::new(Sma::new(2)),
Box::new(Sma::new(4)),
Box::new(Sub::new()),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
}],
vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, from_field: 0 },
Edge { from: 2, to: 3, slot: 0, from_field: 0 },
],
)
.expect("valid DAG")
};
let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![prices.clone()]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
assert_eq!(
out,
vec![
(Timestamp(4), vec![Scalar::F64(2.0)]),
(Timestamp(5), vec![Scalar::F64(2.0)]),
(Timestamp(6), vec![Scalar::F64(2.0)]),
]
);
// determinism (C1): a second identical run drains a bit-identical stream.
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![prices]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out);
}
#[test]
fn mode_a_as_of_fires_on_any_fresh_and_holds() {
// AsOfSum @0; node 1 = Recorder taps it. source 0 ticks t=1..4; source 1
// ticks t=2,4 (slower); both AsOfSum inputs Any.
let build = |tx| {
boot(
vec![
Box::new(AsOfSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
AsOfSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid")
};
let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![s0.clone(), s1.clone()]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm.
assert_eq!(
out,
vec![
(Timestamp(2), vec![Scalar::F64(120.0)]),
(Timestamp(3), vec![Scalar::F64(130.0)]),
(Timestamp(4), vec![Scalar::F64(140.0)]),
(Timestamp(4), vec![Scalar::F64(240.0)]),
]
);
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![s0, s1]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out); // deterministic
}
#[test]
fn mode_b_barrier_fires_only_on_timestamp_coincidence() {
// identical wiring to mode A, but both BarrierSum inputs are Barrier(0).
let build = |tx| {
boot(
vec![
Box::new(BarrierSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
BarrierSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid")
};
let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![s0.clone(), s1.clone()]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// records ONLY at t=2 and t=4 where both inputs share the timestamp.
assert_eq!(
out,
vec![
(Timestamp(2), vec![Scalar::F64(120.0)]),
(Timestamp(4), vec![Scalar::F64(240.0)]),
]
);
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![s0, s1]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out); // deterministic
}
#[test]
fn within_source_diamond_rejoin_barrier_fires() {
// One source fans out through SMA(2), SMA(4) that rejoin at a Barrier(0)
// node; node 3 = Recorder taps the barrier. Every push in a cycle carries
// that cycle's timestamp, so once both SMAs warm and emit in the same
// cycle, both barrier inputs share the timestamp and the barrier fires.
let build = |tx| {
boot(
vec![
Box::new(Sma::new(2)),
Box::new(Sma::new(4)),
Box::new(BarrierSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
BarrierSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
}],
vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, from_field: 0 },
Edge { from: 2, to: 3, slot: 0, from_field: 0 },
],
)
.expect("valid DAG")
};
let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![prices.clone()]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// SMA(4) warms at cycle 4; from then both paths emit each cycle at the same
// timestamp, so the barrier fires: SMA(2)+SMA(4) = 15+13, 17+15, 19+17.
assert_eq!(
out,
vec![
(Timestamp(4), vec![Scalar::F64(28.0)]),
(Timestamp(5), vec![Scalar::F64(32.0)]),
(Timestamp(6), vec![Scalar::F64(36.0)]),
]
);
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![prices]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out);
}
#[test]
fn mixed_a_and_b_or_combine_on_one_node() {
// MixedSum @0 (in0,in1 barrier group 0; in2 as-of); node 1 = Recorder taps it.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![
Box::new(MixedSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
MixedSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 2 }] },
],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid");
let s0 = f64_stream(&[(2, 20.0), (5, 50.0)]); // in0 (barrier)
let s1 = f64_stream(&[(2, 200.0)]); // in1 (barrier)
let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]); // in2 (as-of)
h.run(vec![s0, s1, s2]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// c3: barrier pair completes at t=2, holds c=1 -> 221. c4: as-of input
// ticks at t=3, holds the pair -> 223. c1 filters; c2,c5 hold (no record).
assert_eq!(
out,
vec![
(Timestamp(2), vec![Scalar::F64(221.0)]),
(Timestamp(3), vec![Scalar::F64(223.0)]),
]
);
}
#[test]
fn bootstrap_rejects_a_cycle() {
// two SMA(1) nodes wired a -> b -> a
let err = boot(
vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
],
vec![],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 1, to: 0, slot: 0, from_field: 0 }],
)
.unwrap_err();
assert_eq!(err, BootstrapError::Cycle);
}
#[test]
fn bootstrap_rejects_a_kind_mismatch() {
// SMA(1) declares an f64 input; an i64 source mismatches
let err = boot(
vec![Box::new(Sma::new(1))],
vec![
Sma::builder().schema().clone(),
],
vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![],
)
.unwrap_err();
assert_eq!(
err,
BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
);
}
#[test]
fn bootstrap_rejects_a_bad_index() {
// an edge target node (9) that does not exist -> BadIndex. (The old trigger
// — an out-of-range observe index — is gone with `observe`; BadIndex itself
// is unchanged, only the path that reaches it.)
let err = boot(
vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }],
)
.unwrap_err();
assert_eq!(err, BootstrapError::BadIndex);
}
/// Build five timestamp-aligned f64 sources feeding Ohlcv's five barrier slots.
fn ohlcv_streams() -> Vec<Vec<(Timestamp, Scalar)>> {
vec![
f64_stream(&[(1, 10.0), (2, 20.0)]), // open
f64_stream(&[(1, 15.0), (2, 25.0)]), // high
f64_stream(&[(1, 8.0), (2, 19.0)]), // low
f64_stream(&[(1, 12.0), (2, 22.0)]), // close
f64_stream(&[(1, 100.0), (2, 200.0)]), // volume
]
}
fn ohlcv_sources() -> Vec<SourceSpec> {
(0..5)
.map(|slot| SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot }],
})
.collect()
}
#[test]
fn ohlcv_bundles_five_field_record() {
// node 0 = Ohlcv; five sources feed O/H/L/C/V; node 1 = a 5-input Recorder
// taps all five fields via five edges. The barrier fires once all five share
// the timestamp, so each bar is recorded once, on the fifth cycle of its ts.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Recorder::new(
&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
Firing::Any,
tx,
)),
],
vec![
Ohlcv::sig(),
recorder_sig(&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any),
],
ohlcv_sources(),
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // open
Edge { from: 0, to: 1, slot: 1, from_field: 1 }, // high
Edge { from: 0, to: 1, slot: 2, from_field: 2 }, // low
Edge { from: 0, to: 1, slot: 3, from_field: 3 }, // close
Edge { from: 0, to: 1, slot: 4, from_field: 4 }, // volume
],
)
.expect("valid");
h.run(ohlcv_streams());
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
assert_eq!(
out,
vec![
(Timestamp(1), vec![
Scalar::F64(10.0),
Scalar::F64(15.0),
Scalar::F64(8.0),
Scalar::F64(12.0),
Scalar::F64(100.0),
]),
(Timestamp(2), vec![
Scalar::F64(20.0),
Scalar::F64(25.0),
Scalar::F64(19.0),
Scalar::F64(22.0),
Scalar::F64(200.0),
]),
]
);
}
#[test]
fn edge_binds_single_field_high_minus_low() {
// [Ohlcv (0), Sub (1), Recorder (2)]; Sub binds high (field 1) and low
// (field 2) of the Ohlcv record -> high - low; the Recorder taps Sub.
// Proves from_field routes the right columns (not field 0) and the two
// bound fields are co-fresh (Sub's Any inputs both fire in the bar's cycle).
let build = |tx| {
boot(
vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Sub::new()),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Ohlcv::sig(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
ohlcv_sources(),
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 1 }, // high
Edge { from: 0, to: 1, slot: 1, from_field: 2 }, // low
Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // Sub -> Recorder
],
)
.expect("valid DAG")
};
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(ohlcv_streams());
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// bar1: 15 - 8 = 7; bar2: 25 - 19 = 6 (each on the bar's fifth cycle).
assert_eq!(
out,
vec![
(Timestamp(1), vec![Scalar::F64(7.0)]),
(Timestamp(2), vec![Scalar::F64(6.0)]),
]
);
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(ohlcv_streams());
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out);
}
#[test]
fn distinct_edges_read_distinct_fields() {
// Same Ohlcv, a different consumer: Sub binds close (field 3) and open
// (field 0) -> close - open; the Recorder taps Sub. Proves two edges on one
// record read two different fields (3 and 0, not the high/low pair above).
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Sub::new()),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Ohlcv::sig(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
ohlcv_sources(),
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 3 }, // close
Edge { from: 0, to: 1, slot: 1, from_field: 0 }, // open
Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // Sub -> Recorder
],
)
.expect("valid DAG");
h.run(ohlcv_streams());
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// bar1: 12 - 10 = 2; bar2: 22 - 20 = 2.
assert_eq!(
out,
vec![
(Timestamp(1), vec![Scalar::F64(2.0)]),
(Timestamp(2), vec![Scalar::F64(2.0)]),
]
);
}
#[test]
fn bootstrap_rejects_from_field_out_of_range() {
// Sma(0) has a 1-field output (index 0 only); an edge reading field 9 is
// out of range -> BadIndex (caught before any kind check).
let err = boot(
vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }],
)
.unwrap_err();
assert_eq!(err, BootstrapError::BadIndex);
}
#[test]
fn bootstrap_rejects_per_field_kind_mismatch() {
// TwoField(0) output: field 0 f64, field 1 i64. Binding field 1 (i64) into
// Sma(1)'s f64 input slot is a per-field kind mismatch -> KindMismatch. (The
// mismatch is field-specific: from_field 0 would have matched.)
let err = boot(
vec![Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }), Box::new(Sma::new(1))],
vec![
TwoField::sig(),
Sma::builder().schema().clone(),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
)
.unwrap_err();
assert_eq!(
err,
BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
);
}
#[test]
fn multi_sink_records_distinct_interior_streams() {
// Two recorders tap SMA(2) and SMA(4) in ONE run -> one run records many
// streams (the #2 headline). Each drained stream is individually correct.
let (tx_fast, rx_fast) = mpsc::channel();
let (tx_slow, rx_slow) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Sma::new(2)),
Box::new(Sma::new(4)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_fast)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_slow)),
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
}],
vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> recorder fast
Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow
],
)
.expect("valid DAG");
h.run(vec![f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])]);
let fast: Vec<(Timestamp, Vec<Scalar>)> = rx_fast.try_iter().collect();
let slow: Vec<(Timestamp, Vec<Scalar>)> = rx_slow.try_iter().collect();
// SMA(2) warms at cycle 2, SMA(4) at cycle 4 — two different-rate streams.
assert_eq!(
fast,
vec![
(Timestamp(2), vec![Scalar::F64(11.0)]),
(Timestamp(3), vec![Scalar::F64(13.0)]),
(Timestamp(4), vec![Scalar::F64(15.0)]),
(Timestamp(5), vec![Scalar::F64(17.0)]),
]
);
assert_eq!(
slow,
vec![
(Timestamp(4), vec![Scalar::F64(13.0)]),
(Timestamp(5), vec![Scalar::F64(15.0)]),
]
);
}
#[test]
fn recorder_taps_all_fields_of_a_record() {
// A 5-input Recorder taps all five OHLCV fields via five field-wise edges
// (0005: N edges, no whole-record bind); its recorded row is the whole bar.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Recorder::new(
&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
Firing::Any,
tx,
)),
],
vec![
Ohlcv::sig(),
recorder_sig(&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any),
],
ohlcv_sources(),
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 },
Edge { from: 0, to: 1, slot: 1, from_field: 1 },
Edge { from: 0, to: 1, slot: 2, from_field: 2 },
Edge { from: 0, to: 1, slot: 3, from_field: 3 },
Edge { from: 0, to: 1, slot: 4, from_field: 4 },
],
)
.expect("valid");
h.run(vec![
f64_stream(&[(1, 10.0)]),
f64_stream(&[(1, 15.0)]),
f64_stream(&[(1, 8.0)]),
f64_stream(&[(1, 12.0)]),
f64_stream(&[(1, 100.0)]),
]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
assert_eq!(out.len(), 1);
assert_eq!(out[0].1.len(), 5); // all five fields recorded as one row
assert_eq!(
out,
vec![(Timestamp(1), vec![
Scalar::F64(10.0),
Scalar::F64(15.0),
Scalar::F64(8.0),
Scalar::F64(12.0),
Scalar::F64(100.0),
])]
);
}
#[test]
fn recorder_records_mixed_scalar_kinds() {
// A recorder with i64 + f64 + bool + timestamp inputs records a four-field
// mixed-kind row -> recording is not f64-only. Four sources tick once each
// at t=1,2,3,4; only on cycle 4 are all slots warm, so it records once,
// holding the earlier-ticked values.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![Box::new(Recorder::new(
&[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp],
Firing::Any,
tx,
))],
vec![
recorder_sig(&[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp], Firing::Any),
],
vec![
SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
SourceSpec { kind: ScalarKind::Bool, targets: vec![Target { node: 0, slot: 2 }] },
SourceSpec { kind: ScalarKind::Timestamp, targets: vec![Target { node: 0, slot: 3 }] },
],
vec![],
)
.expect("valid");
h.run(vec![
vec![(Timestamp(1), Scalar::I64(7))],
vec![(Timestamp(2), Scalar::F64(1.5))],
vec![(Timestamp(3), Scalar::Bool(true))],
vec![(Timestamp(4), Scalar::Ts(Timestamp(99)))],
]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
assert_eq!(
out,
vec![(Timestamp(4), vec![
Scalar::I64(7),
Scalar::F64(1.5),
Scalar::Bool(true),
Scalar::Ts(Timestamp(99)),
])]
);
}
#[test]
fn node_is_producer_and_sink_at_once() {
// TapForward records its input AND forwards it downstream; a second
// Recorder taps the forwarded output. Both channels see the same stream ->
// one node is producer and sink at once (C8 "both").
let (tx_tap, rx_tap) = mpsc::channel();
let (tx_down, rx_down) = mpsc::channel();
let mut h = boot(
vec![
Box::new(TapForward { out: [Scalar::F64(0.0)], tx: tx_tap }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_down)),
],
vec![
TapForward::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid");
h.run(vec![f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])]);
let tapped: Vec<(Timestamp, Vec<Scalar>)> = rx_tap.try_iter().collect();
let downstream: Vec<(Timestamp, Vec<Scalar>)> = rx_down.try_iter().collect();
let expected = vec![
(Timestamp(1), vec![Scalar::F64(10.0)]),
(Timestamp(2), vec![Scalar::F64(20.0)]),
(Timestamp(3), vec![Scalar::F64(30.0)]),
];
assert_eq!(tapped, expected); // it recorded (sink side effect)
assert_eq!(downstream, expected); // and forwarded (producer output)
}
#[test]
fn recording_is_deterministic() {
// Two fresh harnesses, two channels, identical input -> bit-identical
// recorded streams (C1).
let build = |tx| {
boot(
vec![
Box::new(Sma::new(3)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid")
};
let prices = f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)]);
let (tx_a, rx_a) = mpsc::channel();
let mut a = build(tx_a);
a.run(vec![prices.clone()]);
let run_a: Vec<(Timestamp, Vec<Scalar>)> = rx_a.try_iter().collect();
let (tx_b, rx_b) = mpsc::channel();
let mut b = build(tx_b);
b.run(vec![prices]);
let run_b: Vec<(Timestamp, Vec<Scalar>)> = rx_b.try_iter().collect();
assert_eq!(run_a, run_b);
assert!(!run_a.is_empty()); // and it actually recorded something
}
#[test]
fn recorder_barrier_firing_records_only_on_coincidence() {
// A 2-input Barrier(0) recorder records only on cycles where both inputs
// share the timestamp — the recorder's OWN firing policy gates recording.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![Box::new(Recorder::new(
&[ScalarKind::F64, ScalarKind::F64],
Firing::Barrier(0),
tx,
))],
vec![
recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Barrier(0)),
],
vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
],
vec![],
)
.expect("valid");
let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
h.run(vec![s0, s1]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// records ONLY at t=2 and t=4 (both inputs coincide); holds otherwise.
assert_eq!(
out,
vec![
(Timestamp(2), vec![Scalar::F64(20.0), Scalar::F64(100.0)]),
(Timestamp(4), vec![Scalar::F64(40.0), Scalar::F64(200.0)]),
]
);
}
#[test]
fn recorder_any_firing_records_on_each_fresh() {
// A 2-input Any recorder records on any-fresh once both are warm (as-of),
// holding the stale input.
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![Box::new(Recorder::new(
&[ScalarKind::F64, ScalarKind::F64],
Firing::Any,
tx,
))],
vec![
recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Any),
],
vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
],
vec![],
)
.expect("valid");
let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
h.run(vec![s0, s1]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// from t=2 on, records every cycle holding the stale input; two cycles fall
// on t=4 (the s0 tick then the s1 tick).
assert_eq!(
out,
vec![
(Timestamp(2), vec![Scalar::F64(20.0), Scalar::F64(100.0)]),
(Timestamp(3), vec![Scalar::F64(30.0), Scalar::F64(100.0)]),
(Timestamp(4), vec![Scalar::F64(40.0), Scalar::F64(100.0)]),
(Timestamp(4), vec![Scalar::F64(40.0), Scalar::F64(200.0)]),
]
);
}
#[test]
fn bootstrap_rejects_kind_mismatched_recorder_edge() {
// TwoField output: field 0 f64, field 1 i64. Binding field 1 (i64) into a
// Recorder's f64 input slot is a per-field kind mismatch -> KindMismatch
// (0005's check already covers recorder edges; recording adds no new hole).
let (tx, _rx) = mpsc::channel();
let err = boot(
vec![
Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
TwoField::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
)
.unwrap_err();
assert_eq!(
err,
BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
);
}
// --- #3 stress matrix: M-producer x N-consumer x K-sink DAGs ---
// The closed `Harness::bootstrap` + `run` substrate must carry arbitrary
// node-level fan-out / fan-in / depth / width deterministically (C1) and
// compute every recorded stream correctly. Earlier cycles proved *source*
// fan-out, single chains and firing in isolation; these close the node
// fan-out / deep-chain / wide-layer holes #3 names. No trading domain.
#[test]
fn node_fan_out_identical_taps_record_identical_streams() {
// PROPERTY: one PRODUCING node read by several consumers via distinct
// edges from the same `from` node feeds every consumer the identical,
// correct stream — node fan-out (not source fan-out: a single SMA(3)
// output is forwarded down three edges to three recorders).
let build = |t1, t2, t3| {
boot(
vec![
Box::new(Sma::new(3)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t1)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t2)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)),
],
vec![
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }],
}],
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 },
Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 0, to: 3, slot: 0, from_field: 0 },
],
)
.expect("valid fan-out DAG")
};
let prices = f64_stream(&[(1, 2.0), (2, 4.0), (3, 6.0), (4, 8.0), (5, 10.0)]);
let (a1, ra) = mpsc::channel();
let (b1, rb) = mpsc::channel();
let (c1, rc) = mpsc::channel();
let mut h = build(a1, b1, c1);
h.run(vec![prices.clone()]);
let s1: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
let s2: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
let s3: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
// SMA(3) warms at cycle 3: mean(2,4,6)=4, mean(4,6,8)=6, mean(6,8,10)=8.
let expected = vec![
(Timestamp(3), vec![Scalar::F64(4.0)]),
(Timestamp(4), vec![Scalar::F64(6.0)]),
(Timestamp(5), vec![Scalar::F64(8.0)]),
];
assert_eq!(s1, expected);
assert_eq!(s2, expected); // every tap sees the identical shared stream
assert_eq!(s3, expected);
// determinism (C1): a fresh harness drains bit-identical.
let (a2, ra2) = mpsc::channel();
let (b2, rb2) = mpsc::channel();
let (c2, rc2) = mpsc::channel();
let mut h2 = build(a2, b2, c2);
h2.run(vec![prices]);
assert_eq!(ra2.try_iter().collect::<Vec<_>>(), expected);
assert_eq!(rb2.try_iter().collect::<Vec<_>>(), expected);
assert_eq!(rc2.try_iter().collect::<Vec<_>>(), expected);
}
#[test]
fn node_fan_out_divergent_consumers_each_compute_their_own() {
// PROPERTY: one producer (SMA(2)) read by consumers that do DIFFERENT
// things — recorded raw by one sink AND fed as input0 of a Sub whose
// other input is a second producer (SMA(4)) — yields both the raw tap
// and the downstream-combined stream, each independently correct.
let build = |t_raw, t_sub| {
boot(
vec![
Box::new(Sma::new(2)), // 0: shared producer
Box::new(Sma::new(4)), // 1: second producer
Box::new(Sub::new()), // 2: SMA(2) - SMA(4)
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_raw)), // 3: raw tap of 0
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_sub)), // 4: tap of Sub
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
}],
vec![
Edge { from: 0, to: 3, slot: 0, from_field: 0 }, // SMA(2) -> raw recorder
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0
Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA(4) -> Sub.in1
Edge { from: 2, to: 4, slot: 0, from_field: 0 }, // Sub -> Sub recorder
],
)
.expect("valid divergent-fan-out DAG")
};
let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);
let (tr, rr) = mpsc::channel();
let (ts, rs) = mpsc::channel();
let mut h = build(tr, ts);
h.run(vec![prices.clone()]);
let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
// SMA(2): 11,13,15,17,19 at cycles 2..6 (raw tap).
let raw_expected = vec![
(Timestamp(2), vec![Scalar::F64(11.0)]),
(Timestamp(3), vec![Scalar::F64(13.0)]),
(Timestamp(4), vec![Scalar::F64(15.0)]),
(Timestamp(5), vec![Scalar::F64(17.0)]),
(Timestamp(6), vec![Scalar::F64(19.0)]),
];
// Sub warms once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
let sub_expected = vec![
(Timestamp(4), vec![Scalar::F64(2.0)]),
(Timestamp(5), vec![Scalar::F64(2.0)]),
(Timestamp(6), vec![Scalar::F64(2.0)]),
];
assert_eq!(raw, raw_expected);
assert_eq!(sub, sub_expected);
let (tr2, rr2) = mpsc::channel();
let (ts2, rs2) = mpsc::channel();
let mut h2 = build(tr2, ts2);
h2.run(vec![prices]);
assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
}
#[test]
fn node_fan_out_under_mixed_firing_each_consumer_records_per_policy() {
// PROPERTY: one producer (SMA(2)) read simultaneously by an as-of
// (Firing::Any) consumer and a Firing::Barrier consumer — each records
// per ITS OWN firing policy off the SAME shared upstream value. The Any
// tap fires on every SMA push; the BarrierSum fires only on the cycles
// where the held SMA output and a second source coincide on a timestamp.
let build = |t_any, t_bar| {
boot(
vec![
Box::new(Sma::new(2)), // 0: shared producer (src A)
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_any)), // 1: as-of tap of 0
Box::new(BarrierSum { out: [Scalar::F64(0.0)] }), // 2: SMA(2) + src B (barrier)
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_bar)), // 3: tap of barrier
],
vec![
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
BarrierSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, // A
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 2, slot: 1 }] }, // B
],
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // SMA(2) -> as-of recorder
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> barrier.in0
Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // barrier -> recorder
],
)
.expect("valid mixed-firing fan-out DAG")
};
let a = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0)]); // drives SMA(2)
let b = f64_stream(&[(2, 100.0), (4, 200.0)]); // barrier.in1
let (ta, rany) = mpsc::channel();
let (tb, rbar) = mpsc::channel();
let mut h = build(ta, tb);
h.run(vec![a.clone(), b.clone()]);
let any: Vec<(Timestamp, Vec<Scalar>)> = rany.try_iter().collect();
let bar: Vec<(Timestamp, Vec<Scalar>)> = rbar.try_iter().collect();
// As-of tap records every SMA(2) fire: 11@t2, 13@t3, 15@t4.
let any_expected = vec![
(Timestamp(2), vec![Scalar::F64(11.0)]),
(Timestamp(3), vec![Scalar::F64(13.0)]),
(Timestamp(4), vec![Scalar::F64(15.0)]),
];
// Barrier fires only where held SMA output and src B share a timestamp:
// t=2 (SMA held=11 + B=100 = 111), t=4 (SMA held=15 + B=200 = 215).
let bar_expected = vec![
(Timestamp(2), vec![Scalar::F64(111.0)]),
(Timestamp(4), vec![Scalar::F64(215.0)]),
];
assert_eq!(any, any_expected);
assert_eq!(bar, bar_expected);
let (ta2, rany2) = mpsc::channel();
let (tb2, rbar2) = mpsc::channel();
let mut h2 = build(ta2, tb2);
h2.run(vec![a, b]);
assert_eq!(rany2.try_iter().collect::<Vec<_>>(), any_expected); // deterministic
assert_eq!(rbar2.try_iter().collect::<Vec<_>>(), bar_expected);
}
#[test]
fn deep_transform_chain_propagates_end_to_end() {
// PROPERTY: a linear chain of several transform nodes (SMA(2) -> SMA(2)
// -> SMA(2) -> recorder) propagates values correctly through depth, with
// each stage's warm-up delaying the tail — closes "deep chains untested".
let build = |tx| {
boot(
vec![
Box::new(Sma::new(2)), // 0
Box::new(Sma::new(2)), // 1
Box::new(Sma::new(2)), // 2
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), // 3 tail
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }],
}],
vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 0, from_field: 0 },
Edge { from: 2, to: 3, slot: 0, from_field: 0 },
],
)
.expect("valid deep chain")
};
let prices = f64_stream(&[(1, 2.0), (2, 4.0), (3, 6.0), (4, 8.0), (5, 10.0)]);
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![prices.clone()]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// stage1 (c2..c5): 3,5,7,9. stage2 (c3..): 4,6,8. stage3 (c4..): 5,7.
let expected = vec![
(Timestamp(4), vec![Scalar::F64(5.0)]),
(Timestamp(5), vec![Scalar::F64(7.0)]),
];
assert_eq!(out, expected);
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![prices]);
assert_eq!(rx2.try_iter().collect::<Vec<_>>(), expected); // deterministic
}
#[test]
fn wide_parallel_layer_multi_sink_records_each_stream() {
// PROPERTY: one source fanned to several PARALLEL producers of different
// params (SMA(2), SMA(3), SMA(4)), each recorded by its own sink in one
// run, yields each parallel stream correctly — closes "wide layers
// untested" and exercises multi-sink at width.
let build = |t2, t3, t4| {
boot(
vec![
Box::new(Sma::new(2)), // 0
Box::new(Sma::new(3)), // 1
Box::new(Sma::new(4)), // 2
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t2)), // 3
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)), // 4
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t4)), // 5
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![
Target { node: 0, slot: 0 },
Target { node: 1, slot: 0 },
Target { node: 2, slot: 0 },
],
}],
vec![
Edge { from: 0, to: 3, slot: 0, from_field: 0 },
Edge { from: 1, to: 4, slot: 0, from_field: 0 },
Edge { from: 2, to: 5, slot: 0, from_field: 0 },
],
)
.expect("valid wide layer")
};
let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)]);
let (a, ra) = mpsc::channel();
let (b, rb) = mpsc::channel();
let (c, rc) = mpsc::channel();
let mut h = build(a, b, c);
h.run(vec![prices.clone()]);
let w2: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
let w3: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
let w4: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
let e2 = vec![
(Timestamp(2), vec![Scalar::F64(11.0)]),
(Timestamp(3), vec![Scalar::F64(13.0)]),
(Timestamp(4), vec![Scalar::F64(15.0)]),
(Timestamp(5), vec![Scalar::F64(17.0)]),
];
let e3 = vec![
(Timestamp(3), vec![Scalar::F64(12.0)]),
(Timestamp(4), vec![Scalar::F64(14.0)]),
(Timestamp(5), vec![Scalar::F64(16.0)]),
];
let e4 = vec![
(Timestamp(4), vec![Scalar::F64(13.0)]),
(Timestamp(5), vec![Scalar::F64(15.0)]),
];
assert_eq!(w2, e2);
assert_eq!(w3, e3);
assert_eq!(w4, e4);
let (a2, ra2) = mpsc::channel();
let (b2, rb2) = mpsc::channel();
let (c2, rc2) = mpsc::channel();
let mut h2 = build(a2, b2, c2);
h2.run(vec![prices]);
assert_eq!(ra2.try_iter().collect::<Vec<_>>(), e2); // deterministic
assert_eq!(rb2.try_iter().collect::<Vec<_>>(), e3);
assert_eq!(rc2.try_iter().collect::<Vec<_>>(), e4);
}
#[test]
fn milestone_end_to_end_mixed_dag_records_every_stream_deterministically() {
// PROPERTY (#3 headline): a single richer DAG combining node fan-out
// (SMA(2) -> raw recorder AND Sub) + source fan-out (source -> SMA(2),
// SMA(4)) + fan-in (Sub) + multiple sinks of MIXED scalar kinds (two f64
// streams + one i64 stream) records every stream correctly AND is fully
// deterministic — the "pure compute substrate carries arbitrary
// M-producer x N-consumer x K-sink DAGs" gate.
let build = |t_raw, t_sub, t_i64| {
boot(
vec![
Box::new(Sma::new(2)), // 0: shared f64 producer
Box::new(Sma::new(4)), // 1: second f64 producer
Box::new(Sub::new()), // 2: SMA(2) - SMA(4)
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_raw)), // 3
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_sub)), // 4
Box::new(Recorder::new(&[ScalarKind::I64], Firing::Any, t_i64)), // 5: i64 sink
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::F64], Firing::Any),
recorder_sig(&[ScalarKind::I64], Firing::Any),
],
vec![
SourceSpec {
kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
},
SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 5, slot: 0 }] },
],
vec![
Edge { from: 0, to: 3, slot: 0, from_field: 0 }, // SMA(2) raw
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0
Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA(4) -> Sub.in1
Edge { from: 2, to: 4, slot: 0, from_field: 0 }, // Sub -> recorder
],
)
.expect("valid milestone DAG")
};
let prices =
f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);
let counts: Vec<(Timestamp, Scalar)> = vec![
(Timestamp(1), Scalar::I64(7)),
(Timestamp(2), Scalar::I64(8)),
(Timestamp(3), Scalar::I64(9)),
];
let (tr, rr) = mpsc::channel();
let (ts, rs) = mpsc::channel();
let (ti, ri) = mpsc::channel();
let mut h = build(tr, ts, ti);
h.run(vec![prices.clone(), counts.clone()]);
let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
let i64s: Vec<(Timestamp, Vec<Scalar>)> = ri.try_iter().collect();
let raw_expected = vec![
(Timestamp(2), vec![Scalar::F64(11.0)]),
(Timestamp(3), vec![Scalar::F64(13.0)]),
(Timestamp(4), vec![Scalar::F64(15.0)]),
(Timestamp(5), vec![Scalar::F64(17.0)]),
(Timestamp(6), vec![Scalar::F64(19.0)]),
];
let sub_expected = vec![
(Timestamp(4), vec![Scalar::F64(2.0)]),
(Timestamp(5), vec![Scalar::F64(2.0)]),
(Timestamp(6), vec![Scalar::F64(2.0)]),
];
let i64_expected = vec![
(Timestamp(1), vec![Scalar::I64(7)]),
(Timestamp(2), vec![Scalar::I64(8)]),
(Timestamp(3), vec![Scalar::I64(9)]),
];
assert_eq!(raw, raw_expected);
assert_eq!(sub, sub_expected);
assert_eq!(i64s, i64_expected);
let (tr2, rr2) = mpsc::channel();
let (ts2, rs2) = mpsc::channel();
let (ti2, ri2) = mpsc::channel();
let mut h2 = build(tr2, ts2, ti2);
h2.run(vec![prices, counts]);
assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
assert_eq!(ri2.try_iter().collect::<Vec<_>>(), i64_expected);
}
#[test]
fn signal_quality_loop_records_pip_equity() {
let (tx_eq, rx_eq) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Sma::new(2)), // 0 fast
Box::new(Sma::new(4)), // 1 slow
Box::new(Sub::new()), // 2 raw signal
Box::new(Exposure::new(0.5)), // 3 exposure
Box::new(SimBroker::new(0.0001)), // 4 pip equity
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 sink
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
Exposure::builder().schema().clone(),
SimBroker::builder(0.0001).schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![
Target { node: 0, slot: 0 }, // price -> SMA fast
Target { node: 1, slot: 0 }, // price -> SMA slow
Target { node: 4, slot: 1 }, // price -> broker price input
],
}],
vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast -> Sub.0
Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // slow -> Sub.1
Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // signal -> Exposure
Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // exposure -> broker.0
Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> recorder
],
)
.expect("valid signal-quality harness");
h.run(vec![f64_stream(&[
(1, 1.0000),
(2, 1.0010),
(3, 1.0025),
(4, 1.0020),
(5, 1.0040),
])]);
let equity: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
// broker fires every cycle (price fresh each tick): five records, ts 1..=5
assert_eq!(equity.len(), 5);
assert_eq!(
equity.iter().map(|(t, _)| t.0).collect::<Vec<_>>(),
vec![1, 2, 3, 4, 5]
);
// flat until SMA(4) warms (cycle 4) and the held exposure meets the next
// price move (cycle 5): the first four equities are exactly 0.
for (_, row) in &equity[0..4] {
assert_eq!(row, &vec![Scalar::F64(0.0)]);
}
// cycle 5: prev_exposure 0.00175 * (1.0040 - 1.0020) / 0.0001 = 0.035 pips
let Scalar::F64(last) = equity[4].1[0] else {
panic!("equity is f64");
};
assert!((last - 0.035).abs() < 1e-9, "final equity = {last}, want ~0.035");
}
#[test]
fn signal_quality_loop_is_deterministic() {
let build = || {
let (tx, rx) = mpsc::channel();
let mut h = boot(
vec![
Box::new(Sma::new(2)),
Box::new(Sma::new(4)),
Box::new(Sub::new()),
Box::new(Exposure::new(0.5)),
Box::new(SimBroker::new(0.0001)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
Exposure::builder().schema().clone(),
SimBroker::builder(0.0001).schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![
Target { node: 0, slot: 0 },
Target { node: 1, slot: 0 },
Target { node: 4, slot: 1 },
],
}],
vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, from_field: 0 },
Edge { from: 2, to: 3, slot: 0, from_field: 0 },
Edge { from: 3, to: 4, slot: 0, from_field: 0 },
Edge { from: 4, to: 5, slot: 0, from_field: 0 },
],
)
.expect("valid harness");
h.run(vec![f64_stream(&[
(1, 1.0000),
(2, 1.0010),
(3, 1.0025),
(4, 1.0020),
(5, 1.0040),
])]);
rx.try_iter().collect::<Vec<(Timestamp, Vec<Scalar>)>>()
};
assert_eq!(build(), build());
}
}
Reference in New Issue View Git Blame Copy Permalink