Brummel/Aura
1
0
Fork 0
Code Issues 28 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(aura): node signature lives in the blueprint; collapse Blueprint into Composite

Browse Source 
Consolidate the node data structure so every node's signature (NodeSchema:
inputs/output/params) is declared once and exists in the blueprint pre-build,
and dissolve the special "root graph" type. Behaviour-preserving (C1).

Signature vs sizing
- NodeSchema is now the static signature only: InputSpec -> PortSpec{kind,firing},
  with lookback removed. The signature is fully static per blueprint (input
  kinds/firing, output fields, params); LinComb's variable arity is a builder arg,
  not an injected param.
- The one param-dependent quantity, an input's buffer lookback (e.g. Sma's window =
  its injected length), moves out of the signature to Node::lookbacks() -> Vec<usize>,
  read only by bootstrap for sizing. Node::schema() is removed.
- LeafFactory -> PrimitiveBuilder, which carries the full NodeSchema. The built node
  no longer re-declares it: closes the params-declared-twice drift (#36, the 8
  per-node factory_params_match_built_node_schema lockstep tests are deleted — their
  subject is now structurally impossible) and a value-empty recipe exposes its full
  I/O interface pre-build (#43).

Root is just a bound composite
- struct Blueprint is deleted; its compile/bootstrap/param_space methods move onto
  Composite. Role gains source: Option<ScalarKind> (None = open interior port,
  Some = bound ingestion feed). A composite is runnable iff every root role is bound;
  the "main graph" is no longer a category, only the fully-source-bound composite.
  New error CompileError::UnboundRootRole for an open root role.
- BlueprintNode::signature() answers uniformly for both arms: Primitive returns the
  builder's declared schema, Composite derives it from the interior (role kinds in,
  OutField kinds out, aggregated params), pre-build, no build.

compile -> FlatGraph -> bootstrap
- compile validates structure pre-build via signature() (validate_wiring: range +
  kind, returning the same variants as before, so an edge kind fault is now caught
  before any build closure fires) and emits FlatGraph{nodes,signatures,sources,edges}.
- bootstrap consumes the FlatGraph: kinds/firing/output from the carried signatures,
  buffer depth from node.lookbacks(). SourceSpec survives as the flat descriptor.

Renames: BlueprintNode::Leaf -> Primitive, LeafFactory -> PrimitiveBuilder.

Render (aura-cli/src/graph.rs) is migrated compile-only: it takes &Composite, maps
bound roles to the same source-entry shape, so both render goldens reproduce
byte-identical output (no re-capture needed). Render-fidelity tuning is the next cycle.

Verification (orchestrator-run, not agent-reported): cargo build --workspace green;
cargo test --workspace 150 passed / 0 failed; cargo clippy --workspace --all-targets
-D warnings clean. All pinned determinism/run-output tests pass with values unchanged;
no behavioural assertion was altered to go green. 5 new tests assert the signature is
pre-build and uniform, that compile rejects a kind mismatch without building (via a
panicking builder), UnboundRootRole, and lookbacks()/signature arity agreement.

Deferred to cycle-close audit (per plan): docs/design/INDEX.md and some aura-std
module docs still name the old Node::schema()/LeafFactory/BlueprintNode::Leaf/
Blueprint::param_space contracts; prose reconciliation is the architect's at audit.

closes #43 #36
This commit is contained in:
Brummel
2026-06-09 12:49:22 +02:00
parent c7fc2f6a37
commit 1b3909316e
17 changed files with 1398 additions and 775 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/graph.rs
+26 -20
View File
@@ -15,10 +15,10 @@
use ascii_dag::graph::{Graph, RenderMode}; use ascii_dag::graph::{Graph, RenderMode};
use ascii_dag::render::colors::Palette; use ascii_dag::render::colors::Palette;
use aura_core::{LeafFactory, Node, ScalarKind}; use aura_core::{Node, PrimitiveBuilder, ScalarKind};
use aura_engine::{ use aura_engine::{
aliases_on, signature_of, Blueprint, BlueprintNode, Composite, Edge, OutField, ParamAlias, aliases_on, signature_of, BlueprintNode, Composite, Edge, OutField, ParamAlias, SourceSpec,
SourceSpec, Target, Target,
}; };
/// Where a leaf's param **names** come from in the shared leaf-label path — the one /// Where a leaf's param **names** come from in the shared leaf-label path — the one
@@ -58,18 +58,24 @@ pub enum Color {
/// Blueprint view: the authored structure (#38). A flat main graph wires the /// Blueprint view: the authored structure (#38). A flat main graph wires the
/// harness with each composite shown as a single opaque node; a `where:` section /// harness with each composite shown as a single opaque node; a `where:` section
/// defines each distinct composite type once. Flat layout only (no subgraphs). /// defines each distinct composite type once. Flat layout only (no subgraphs).
pub fn render_blueprint(bp: &Blueprint, color: Color) -> String { pub fn render_blueprint(bp: &Composite, color: Color) -> String {
// the main graph is the shared graph core (`render_graph`) over the blueprint's // the main graph is the shared graph core (`render_graph`) over the root
// top-level (nodes, edges): leaves enriched exactly as `where:` interior leaves // composite's top-level (nodes, edges): leaves enriched exactly as `where:`
// are (param names folded in, a `<field> →` prefix for any input fed by a // interior leaves are (param names folded in, a `<field> →` prefix for any input
// multi-output producer — the headline: macd's `histogram` driving Exposure), // fed by a multi-output producer — the headline: macd's `histogram` driving
// composites opaque. The deltas vs a composite definition: param names come from // Exposure), composites opaque. The deltas vs a composite definition: param names
// the factory (no alias overlay at the root); the entries are sources, not roles; // come from the builder (no alias overlay at the root); the entries are the
// there is no output record (terminals are sinks) and no title. // root's source-bound roles, not interior roles; there is no output record
// (terminals are sinks) and no title.
let entries: Vec<Entry> = bp let entries: Vec<Entry> = bp
.sources() .input_roles()
.iter() .iter()
.map(|src| Entry { name: format!("source:{:?}", src.kind), targets: &src.targets }) .filter_map(|role| {
role.source.map(|kind| Entry {
name: format!("source:{kind:?}"),
targets: &role.targets,
})
})
.collect(); .collect();
let main = render_graph( let main = render_graph(
bp.nodes(), bp.nodes(),
@@ -94,7 +100,7 @@ pub fn render_blueprint(bp: &Blueprint, color: Color) -> String {
/// The shared graph core both views render through (#48): a flat ascii-dag of the /// The shared graph core both views render through (#48): a flat ascii-dag of the
/// computation DAG plus external-entry markers. `nodes`/`edges` are a borrowed /// computation DAG plus external-entry markers. `nodes`/`edges` are a borrowed
/// (blueprint-root or composite-interior) graph slice — never an owned /// (blueprint-root or composite-interior) graph slice — never an owned
/// `Composite` (`LeafFactory` is not `Clone`). Each leaf gets the unified /// `Composite` (`PrimitiveBuilder` is not `Clone`). Each leaf gets the unified
/// [`leaf_label`]; each composite is opaque (`[name]`); a producing node carrying an /// [`leaf_label`]; each composite is opaque (`[name]`); a producing node carrying an
/// `OutField` re-export gets the `name := ` binding prefix folded on (empty list → /// `OutField` re-export gets the `name := ` binding prefix folded on (empty list →
/// no binding, the blueprint case); every [`Entry`] is drawn as a marker wired to its /// no binding, the blueprint case); every [`Entry`] is drawn as a marker wired to its
@@ -115,7 +121,7 @@ fn render_graph(
let mut labels: Vec<String> = Vec::with_capacity(nodes.len()); let mut labels: Vec<String> = Vec::with_capacity(nodes.len());
for (i, item) in nodes.iter().enumerate() { for (i, item) in nodes.iter().enumerate() {
let base = match item { let base = match item {
BlueprintNode::Leaf(factory) => { BlueprintNode::Primitive(factory) => {
leaf_label(nodes, edges, entries, i, factory, &param_names, stub_ctx) leaf_label(nodes, edges, entries, i, factory, &param_names, stub_ctx)
} }
BlueprintNode::Composite(c) => c.name().to_string(), BlueprintNode::Composite(c) => c.name().to_string(),
@@ -166,7 +172,7 @@ fn render_flat(labels: &[String], edges: &[(usize, usize)], color: Color) -> Str
/// `name()` (the authoring type identity — same name implies same structure). /// `name()` (the authoring type identity — same name implies same structure).
/// Recurses into a composite's interior on first sight so nested composites get /// Recurses into a composite's interior on first sight so nested composites get
/// their own definition; a later same-name occurrence is skipped (deduped). /// their own definition; a later same-name occurrence is skipped (deduped).
fn collect_distinct_composites(bp: &Blueprint) -> Vec<&Composite> { fn collect_distinct_composites(bp: &Composite) -> Vec<&Composite> {
fn walk<'a>(items: &'a [BlueprintNode], seen: &mut Vec<&'a str>, out: &mut Vec<&'a Composite>) { fn walk<'a>(items: &'a [BlueprintNode], seen: &mut Vec<&'a str>, out: &mut Vec<&'a Composite>) {
for item in items { for item in items {
if let BlueprintNode::Composite(c) = item if let BlueprintNode::Composite(c) = item
@@ -209,7 +215,7 @@ fn signature(c: &Composite) -> String {
.nodes() .nodes()
.get(a.node) .get(a.node)
.and_then(|n| match n { .and_then(|n| match n {
BlueprintNode::Leaf(f) => f.params().get(a.slot).map(|p| kind_str(p.kind)), BlueprintNode::Primitive(f) => f.params().get(a.slot).map(|p| kind_str(p.kind)),
BlueprintNode::Composite(_) => None, BlueprintNode::Composite(_) => None,
}) })
.unwrap_or("?"); .unwrap_or("?");
@@ -247,7 +253,7 @@ fn leaf_label(
edges: &[Edge], edges: &[Edge],
entries: &[Entry], entries: &[Entry],
index: usize, index: usize,
factory: &LeafFactory, factory: &PrimitiveBuilder,
param_names: &ParamNames, param_names: &ParamNames,
stub_ctx: Option<&Composite>, stub_ctx: Option<&Composite>,
) -> String { ) -> String {
@@ -334,7 +340,7 @@ fn multi_output_field_name(nodes: &[BlueprintNode], from: usize, from_field: usi
None None
} }
} }
BlueprintNode::Leaf(_) => (from_field > 0).then(|| from_field.to_string()), BlueprintNode::Primitive(_) => (from_field > 0).then(|| from_field.to_string()),
} }
} }
@@ -396,7 +402,7 @@ fn slot_source(c: &Composite, index: usize, slot: usize) -> (String, usize, bool
/// rendered identifier never goes below. /// rendered identifier never goes below.
fn signature_base_len(c: &Composite, node: usize) -> usize { fn signature_base_len(c: &Composite, node: usize) -> usize {
match &c.nodes()[node] { match &c.nodes()[node] {
BlueprintNode::Leaf(_) => 1 + aliases_on(c.params(), node).count(), BlueprintNode::Primitive(_) => 1 + aliases_on(c.params(), node).count(),
BlueprintNode::Composite(_) => 1, BlueprintNode::Composite(_) => 1,
} }
} }
crates/aura-cli/src/main.rs
+129 -87
View File
@@ -10,8 +10,8 @@ mod graph;
use aura_core::{Firing, Scalar, ScalarKind, Timestamp}; use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{ use aura_engine::{
f64_field, summarize, Blueprint, BlueprintNode, Composite, Edge, Harness, OutField, f64_field, summarize, BlueprintNode, Composite, Edge, FlatGraph, Harness, OutField, ParamAlias,
ParamAlias, Role, RunManifest, RunReport, SourceSpec, Target, Role, RunManifest, RunReport, SourceSpec, Target,
}; };
use aura_std::{Ema, Exposure, Recorder, SimBroker, Sma, Sub}; use aura_std::{Ema, Exposure, Recorder, SimBroker, Sma, Sub};
use std::io::IsTerminal; use std::io::IsTerminal;
@@ -50,8 +50,13 @@ fn sample_harness() -> (
) { ) {
let (tx_eq, rx_eq) = mpsc::channel(); let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel(); let (tx_ex, rx_ex) = mpsc::channel();
let h = Harness::bootstrap( let f64_recorder_sig = || aura_engine::NodeSchema {
vec![ inputs: vec![aura_engine::PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
output: vec![],
params: vec![],
};
let h = Harness::bootstrap(FlatGraph {
nodes: vec![
Box::new(Sma::new(2)), // 0 fast SMA Box::new(Sma::new(2)), // 0 fast SMA
Box::new(Sma::new(4)), // 1 slow SMA Box::new(Sma::new(4)), // 1 slow SMA
Box::new(Sub::new()), // 2 spread Box::new(Sub::new()), // 2 spread
@@ -60,7 +65,16 @@ fn sample_harness() -> (
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
], ],
vec![SourceSpec { signatures: vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
Exposure::builder().schema().clone(),
SimBroker::builder(0.0001).schema().clone(),
f64_recorder_sig(),
f64_recorder_sig(),
],
sources: vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![ targets: vec![
Target { node: 0, slot: 0 }, Target { node: 0, slot: 0 },
@@ -68,7 +82,7 @@ fn sample_harness() -> (
Target { node: 4, slot: 1 }, // price into the broker's price slot Target { node: 4, slot: 1 }, // price into the broker's price slot
], ],
}], }],
vec![ edges: vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, 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: 2, to: 3, slot: 0, from_field: 0 },
@@ -76,7 +90,7 @@ fn sample_harness() -> (
Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5 Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6 Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
], ],
) })
.expect("valid sample signal-quality DAG"); .expect("valid sample signal-quality DAG");
(h, rx_eq, rx_ex) (h, rx_eq, rx_ex)
} }
@@ -122,7 +136,7 @@ fn run_sample() -> RunReport {
fn sma_cross(name: &str) -> Composite { fn sma_cross(name: &str) -> Composite {
Composite::new( Composite::new(
name, name,
vec![Sma::factory().into(), Sma::factory().into(), Sub::factory().into()], vec![Sma::builder().into(), Sma::builder().into(), Sub::builder().into()],
vec![ vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 },
@@ -130,6 +144,7 @@ fn sma_cross(name: &str) -> Composite {
vec![Role { vec![Role {
name: "price".into(), name: "price".into(),
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
source: None,
}], }],
vec![ vec![
ParamAlias { name: "fast".into(), node: 0, slot: 0 }, // fast SMA length ParamAlias { name: "fast".into(), node: 0, slot: 0 }, // fast SMA length
@@ -143,35 +158,39 @@ fn sma_cross(name: &str) -> Composite {
/// exposure scale are injected at compile via the point vector (see /// exposure scale are injected at compile via the point vector (see
/// `sample_point`). Recorders need a channel to construct; the receivers are /// `sample_point`). Recorders need a channel to construct; the receivers are
/// dropped because the render never runs the graph. /// dropped because the render never runs the graph.
fn build_sample() -> Blueprint { fn build_sample() -> Composite {
let (tx_eq, _rx_eq) = mpsc::channel(); let (tx_eq, _rx_eq) = mpsc::channel();
let (tx_ex, _rx_ex) = mpsc::channel(); let (tx_ex, _rx_ex) = mpsc::channel();
Blueprint::new( Composite::new(
"sample",
vec![ vec![
BlueprintNode::Composite(sma_cross("sma_cross")), BlueprintNode::Composite(sma_cross("sma_cross")),
Exposure::factory().into(), Exposure::builder().into(),
SimBroker::factory(0.0001).into(), SimBroker::builder(0.0001).into(),
Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_eq).into(), Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_ex).into(), Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
], ],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![
Target { node: 0, slot: 0 }, // price -> sma_cross role 0
Target { node: 2, slot: 1 }, // price -> SimBroker price slot
],
}],
vec![ vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // spread -> Exposure Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // spread -> Exposure
Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker slot 0 Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker slot 0
Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity -> sink Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity -> sink
Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure -> sink Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure -> sink
], ],
vec![Role {
name: "price".into(),
targets: vec![
Target { node: 0, slot: 0 }, // price -> sma_cross role 0
Target { node: 2, slot: 1 }, // price -> SimBroker price slot
],
source: Some(ScalarKind::F64),
}],
vec![], // params: the interior sma_cross carries the aliases
vec![], // output: the root ends in sinks, no re-export
) )
} }
/// The built-in sample rendered by `aura graph`. /// The built-in sample rendered by `aura graph`.
fn sample_blueprint() -> Blueprint { fn sample_blueprint() -> Composite {
build_sample() build_sample()
} }
@@ -195,11 +214,11 @@ fn macd(name: &str) -> Composite {
Composite::new( Composite::new(
name, name,
vec![ vec![
Ema::factory().into(), // 0 fast EMA Ema::builder().into(), // 0 fast EMA
Ema::factory().into(), // 1 slow EMA Ema::builder().into(), // 1 slow EMA
Sub::factory().into(), // 2 MACD line = fast − slow Sub::builder().into(), // 2 MACD line = fast − slow
Ema::factory().into(), // 3 signal EMA of the MACD line Ema::builder().into(), // 3 signal EMA of the MACD line
Sub::factory().into(), // 4 histogram = MACD line − signal Sub::builder().into(), // 4 histogram = MACD line − signal
], ],
vec![ vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast → line[0] Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast → line[0]
@@ -214,6 +233,7 @@ fn macd(name: &str) -> Composite {
Target { node: 0, slot: 0 }, // price → fast EMA Target { node: 0, slot: 0 }, // price → fast EMA
Target { node: 1, slot: 0 }, // price → slow EMA Target { node: 1, slot: 0 }, // price → slow EMA
], ],
source: None,
}], }],
vec![ vec![
ParamAlias { name: "fast".into(), node: 0, slot: 0 }, // fast EMA length ParamAlias { name: "fast".into(), node: 0, slot: 0 }, // fast EMA length
@@ -234,34 +254,38 @@ fn macd(name: &str) -> Composite {
fn macd_strategy_blueprint( fn macd_strategy_blueprint(
tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>, tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>, tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
) -> Blueprint { ) -> Composite {
Blueprint::new( Composite::new(
"macd_strategy",
vec![ vec![
BlueprintNode::Composite(macd("macd")), BlueprintNode::Composite(macd("macd")),
Exposure::factory().into(), Exposure::builder().into(),
SimBroker::factory(0.0001).into(), SimBroker::builder(0.0001).into(),
Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_eq).into(), Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx_ex).into(), Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
], ],
vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![
Target { node: 0, slot: 0 }, // price → macd role 0
Target { node: 2, slot: 1 }, // price → SimBroker price slot
],
}],
vec![ vec![
Edge { from: 0, to: 1, slot: 0, from_field: 2 }, // histogram → Exposure Edge { from: 0, to: 1, slot: 0, from_field: 2 }, // histogram → Exposure
Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure → broker slot 0 Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure → broker slot 0
Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity → sink Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity → sink
Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure → sink Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure → sink
], ],
vec![Role {
name: "price".into(),
targets: vec![
Target { node: 0, slot: 0 }, // price → macd role 0
Target { node: 2, slot: 1 }, // price → SimBroker price slot
],
source: Some(ScalarKind::F64),
}],
vec![], // params: the interior macd carries the aliases
vec![], // output: the root ends in sinks, no re-export
) )
} }
/// The MACD strategy blueprint for the structural render (receivers dropped, as /// The MACD strategy blueprint for the structural render (receivers dropped, as
/// the render never runs the graph). /// the render never runs the graph).
fn macd_blueprint() -> Blueprint { fn macd_blueprint() -> Composite {
let (tx_eq, _rx_eq) = mpsc::channel(); let (tx_eq, _rx_eq) = mpsc::channel();
let (tx_ex, _rx_ex) = mpsc::channel(); let (tx_ex, _rx_ex) = mpsc::channel();
macd_strategy_blueprint(tx_eq, tx_ex) macd_strategy_blueprint(tx_eq, tx_ex)
@@ -296,10 +320,10 @@ fn macd_prices() -> Vec<(Timestamp, Scalar)> {
fn run_macd() -> RunReport { fn run_macd() -> RunReport {
let (tx_eq, rx_eq) = mpsc::channel(); let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel(); let (tx_ex, rx_ex) = mpsc::channel();
let (nodes, sources, edges) = macd_strategy_blueprint(tx_eq, tx_ex) let flat = macd_strategy_blueprint(tx_eq, tx_ex)
.compile_with_params(&macd_point()) .compile_with_params(&macd_point())
.expect("valid macd blueprint"); .expect("valid macd blueprint");
let mut h = Harness::bootstrap(nodes, sources, edges).expect("valid macd harness"); let mut h = Harness::bootstrap(flat).expect("valid macd harness");
let prices = macd_prices(); let prices = macd_prices();
let window = ( let window = (
@@ -333,9 +357,9 @@ fn run_macd() -> RunReport {
/// Compile a blueprint under its point vector and render the flat post-inline /// Compile a blueprint under its point vector and render the flat post-inline
/// (C23) view — the shared body of the `--compiled` paths. /// (C23) view — the shared body of the `--compiled` paths.
fn render_compiled(bp: Blueprint, point: &[Scalar], color: graph::Color) -> String { fn render_compiled(bp: Composite, point: &[Scalar], color: graph::Color) -> String {
let (nodes, sources, edges) = bp.compile_with_params(point).expect("valid blueprint"); let flat = bp.compile_with_params(point).expect("valid blueprint");
graph::render_compilat(&nodes, &sources, &edges, color) graph::render_compilat(&flat.nodes, &flat.sources, &flat.edges, color)
} }
const USAGE: &str = "usage: aura run [--macd] | aura graph [--compiled | --macd [--compiled]]"; const USAGE: &str = "usage: aura run [--macd] | aura graph [--compiled | --macd [--compiled]]";
@@ -415,25 +439,30 @@ mod tests {
// structure only (no compile/validate), so a minimal fixture suffices. // structure only (no compile/validate), so a minimal fixture suffices.
let inner = Composite::new( let inner = Composite::new(
"inner", "inner",
vec![Sma::factory().into()], vec![Sma::builder().into()],
vec![], vec![],
vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }], vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
vec![], vec![],
vec![OutField { node: 0, field: 0, name: "out".into() }], vec![OutField { node: 0, field: 0, name: "out".into() }],
); );
let outer = Composite::new( let outer = Composite::new(
"outer", "outer",
vec![BlueprintNode::Composite(inner), Sub::factory().into()], vec![BlueprintNode::Composite(inner), Sub::builder().into()],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }] }], vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: None }],
vec![], vec![],
vec![OutField { node: 1, field: 0, name: "out".into() }], vec![OutField { node: 1, field: 0, name: "out".into() }],
); );
let bp = Blueprint::new( let bp = Composite::new(
vec![BlueprintNode::Composite(outer)], "root",
vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], vec![BlueprintNode::Composite(outer)],
vec![], vec![],
); vec![
Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) },
],
vec![], // params
vec![], // output
);
let out = graph::render_blueprint(&bp, graph::Color::Plain); // must not panic (no unimplemented!) let out = graph::render_blueprint(&bp, graph::Color::Plain); // must not panic (no unimplemented!)
// outer shows the inner composite as an opaque node, and both get a definition // outer shows the inner composite as an opaque node, and both get a definition
assert!(out.contains("[outer]"), "missing opaque outer node:\n{out}"); assert!(out.contains("[outer]"), "missing opaque outer node:\n{out}");
@@ -445,21 +474,23 @@ mod tests {
#[test] #[test]
fn reused_composite_defined_once() { fn reused_composite_defined_once() {
// the same composite type used twice: two opaque nodes, one definition. // the same composite type used twice: two opaque nodes, one definition.
let bp = Blueprint::new( let bp = Composite::new(
vec![ "root",
vec![
BlueprintNode::Composite(sma_cross("dup")), BlueprintNode::Composite(sma_cross("dup")),
BlueprintNode::Composite(sma_cross("dup")), BlueprintNode::Composite(sma_cross("dup")),
Exposure::factory().into(), Exposure::builder().into(),
], ],
vec![SourceSpec { vec![
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: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 0, from_field: 0 },
], ],
); vec![
Role { name: "src".into(), targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], source: Some(ScalarKind::F64) },
],
vec![], // params
vec![], // output
);
let out = graph::render_blueprint(&bp, graph::Color::Plain); let out = graph::render_blueprint(&bp, graph::Color::Plain);
assert_eq!(out.matches("[dup]").count(), 2, "two opaque uses expected:\n{out}"); assert_eq!(out.matches("[dup]").count(), 2, "two opaque uses expected:\n{out}");
assert_eq!(out.matches("dup(").count(), 1, "body defined once:\n{out}"); assert_eq!(out.matches("dup(").count(), 1, "body defined once:\n{out}");
@@ -468,9 +499,8 @@ mod tests {
#[test] #[test]
fn compiled_view_dissolves_the_composite_boundary() { fn compiled_view_dissolves_the_composite_boundary() {
let bp = sample_blueprint(); let bp = sample_blueprint();
let (nodes, sources, edges) = let flat = bp.compile_with_params(&sample_point()).expect("valid sample");
bp.compile_with_params(&sample_point()).expect("valid sample"); let out = graph::render_compilat(&flat.nodes, &flat.sources, &flat.edges, graph::Color::Plain);
let out = graph::render_compilat(&nodes, &sources, &edges, graph::Color::Plain);
// node labels survive inlining... // node labels survive inlining...
assert!(out.contains("SMA(2)") && out.contains("SMA(4)"), "labels lost:\n{out}"); assert!(out.contains("SMA(2)") && out.contains("SMA(4)"), "labels lost:\n{out}");
// ...but the composite cluster name does NOT (boundary dissolved, C23) // ...but the composite cluster name does NOT (boundary dissolved, C23)
@@ -483,12 +513,12 @@ mod tests {
// The blueprint is now param-generic (identical for both orderings), so the // The blueprint is now param-generic (identical for both orderings), so the
// swap is observable only after the vector is injected — in the COMPILED // swap is observable only after the vector is injected — in the COMPILED
// view, where the flat nodes carry their valued labels (SMA(2)/SMA(4)). // view, where the flat nodes carry their valued labels (SMA(2)/SMA(4)).
let (cn, cs, ce) = let cflat =
sample_blueprint().compile_with_params(&sample_point()).expect("valid sample"); sample_blueprint().compile_with_params(&sample_point()).expect("valid sample");
let correct = graph::render_compilat(&cn, &cs, &ce, graph::Color::Plain); let correct = graph::render_compilat(&cflat.nodes, &cflat.sources, &cflat.edges, graph::Color::Plain);
let (sn, ss, se) = let sflat =
sample_blueprint().compile_with_params(&swapped_point()).expect("valid sample"); sample_blueprint().compile_with_params(&swapped_point()).expect("valid sample");
let swapped = graph::render_compilat(&sn, &ss, &se, graph::Color::Plain); let swapped = graph::render_compilat(&sflat.nodes, &sflat.sources, &sflat.edges, graph::Color::Plain);
assert_ne!(correct, swapped, "a fast/slow SMA swap must change the compiled render"); assert_ne!(correct, swapped, "a fast/slow SMA swap must change the compiled render");
} }
@@ -534,9 +564,8 @@ sma_cross(fast:i64, slow:i64) -> (cross):
#[test] #[test]
fn compiled_view_golden() { fn compiled_view_golden() {
let bp = sample_blueprint(); let bp = sample_blueprint();
let (nodes, sources, edges) = let flat = bp.compile_with_params(&sample_point()).expect("valid sample");
bp.compile_with_params(&sample_point()).expect("valid sample"); let out = graph::render_compilat(&flat.nodes, &flat.sources, &flat.edges, graph::Color::Plain);
let out = graph::render_compilat(&nodes, &sources, &edges, graph::Color::Plain);
let expected = r#" [source:F64] let expected = r#" [source:F64]
┌────────└─┐──────┐ ┌────────└─┐──────┐
↓ ↓ │ ↓ ↓ │
@@ -615,13 +644,20 @@ sma_cross(fast:i64, slow:i64) -> (cross):
// #price (role name) and #Es // #price (role name) and #Es
let c = Composite::new( let c = Composite::new(
"roles", "roles",
vec![Ema::factory().into(), Sub::factory().into()], vec![Ema::builder().into(), Sub::builder().into()],
vec![Edge { from: 0, to: 1, slot: 1, from_field: 0 }], vec![Edge { from: 0, to: 1, slot: 1, from_field: 0 }],
vec![Role { name: "price".into(), targets: vec![Target { node: 1, slot: 0 }] }], vec![Role { name: "price".into(), targets: vec![Target { node: 1, slot: 0 }], source: None }],
vec![ParamAlias { name: "slow".into(), node: 0, slot: 0 }], vec![ParamAlias { name: "slow".into(), node: 0, slot: 0 }],
vec![OutField { node: 1, field: 0, name: "o".into() }], vec![OutField { node: 1, field: 0, name: "o".into() }],
); );
let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]); let bp = Composite::new(
"root",
vec![BlueprintNode::Composite(c)],
vec![],
vec![],
vec![], // params
vec![], // output
);
let out = graph::render_blueprint(&bp, graph::Color::Plain); let out = graph::render_blueprint(&bp, graph::Color::Plain);
assert!(out.contains("[o := Sub(#price,#Es)]"), "role name verbatim + source-derived, bound as output `o`: {out}"); assert!(out.contains("[o := Sub(#price,#Es)]"), "role name verbatim + source-derived, bound as output `o`: {out}");
} }
@@ -634,13 +670,13 @@ sma_cross(fast:i64, slow:i64) -> (cross):
let c = Composite::new( let c = Composite::new(
"nest", "nest",
vec![ vec![
Ema::factory().into(), // 0 fast Ema::builder().into(), // 0 fast
Ema::factory().into(), // 1 slow Ema::builder().into(), // 1 slow
Ema::factory().into(), // 2 up Ema::builder().into(), // 2 up
Ema::factory().into(), // 3 down Ema::builder().into(), // 3 down
Sub::factory().into(), // 4 = Sub(0,1) Sub::builder().into(), // 4 = Sub(0,1)
Sub::factory().into(), // 5 = Sub(2,3) Sub::builder().into(), // 5 = Sub(2,3)
Sub::factory().into(), // 6 = Sub(4,5) (the outer fan-in) Sub::builder().into(), // 6 = Sub(4,5) (the outer fan-in)
], ],
vec![ vec![
Edge { from: 0, to: 4, slot: 0, from_field: 0 }, Edge { from: 0, to: 4, slot: 0, from_field: 0 },
@@ -657,8 +693,7 @@ sma_cross(fast:i64, slow:i64) -> (cross):
Target { node: 1, slot: 0 }, Target { node: 1, slot: 0 },
Target { node: 2, slot: 0 }, Target { node: 2, slot: 0 },
Target { node: 3, slot: 0 }, Target { node: 3, slot: 0 },
], ], source: None, }],
}],
vec![ vec![
ParamAlias { name: "fast".into(), node: 0, slot: 0 }, ParamAlias { name: "fast".into(), node: 0, slot: 0 },
ParamAlias { name: "slow".into(), node: 1, slot: 0 }, ParamAlias { name: "slow".into(), node: 1, slot: 0 },
@@ -667,7 +702,14 @@ sma_cross(fast:i64, slow:i64) -> (cross):
], ],
vec![OutField { node: 6, field: 0, name: "o".into() }], vec![OutField { node: 6, field: 0, name: "o".into() }],
); );
let bp = Blueprint::new(vec![BlueprintNode::Composite(c)], vec![], vec![]); let bp = Composite::new(
"root",
vec![BlueprintNode::Composite(c)],
vec![],
vec![],
vec![], // params
vec![], // output
);
let out = graph::render_blueprint(&bp, graph::Color::Plain); let out = graph::render_blueprint(&bp, graph::Color::Plain);
assert!(out.contains("[o := Sub(#SEf,#SEu)]"), "outer Sub descends into inner Subs, bound as output `o`: {out}"); assert!(out.contains("[o := Sub(#SEf,#SEu)]"), "outer Sub descends into inner Subs, bound as output `o`: {out}");
assert!(out.contains("[Sub(#Ef,#Es)]"), "inner Sub uses EMA aliases: {out}"); assert!(out.contains("[Sub(#Ef,#Es)]"), "inner Sub uses EMA aliases: {out}");
crates/aura-core/src/lib.rs
+5 -4
View File
@@ -16,9 +16,10 @@
//! //!
//! Delivered in cycle 0002 — the node contract: //! Delivered in cycle 0002 — the node contract:
//! //!
//! - [`Node`] — the `schema`/`eval` contract every node implements (C8), with //! - [`Node`] — the `lookbacks`/`eval` contract every node implements (C8); the
//! [`NodeSchema`] / [`InputSpec`] declaring inputs (kind + lookback) and the //! static signature ([`NodeSchema`] / [`PortSpec`] declaring input ports by kind
//! output record ([`FieldSpec`] columns; length 1 = scalar); //! and the output record of [`FieldSpec`] columns; length 1 = scalar) is declared
//! on the node's [`PrimitiveBuilder`], pre-build;
//! - [`Ctx`] — the per-`eval` read-side: zero-copy, financial-indexed [`Window`] //! - [`Ctx`] — the per-`eval` read-side: zero-copy, financial-indexed [`Window`]
//! access into each input (closing the cycle-0001 read-side gap on //! access into each input (closing the cycle-0001 read-side gap on
//! [`AnyColumn`]). //! [`AnyColumn`]).
@@ -39,5 +40,5 @@ pub use any::AnyColumn;
pub use column::{Column, Window}; pub use column::{Column, Window};
pub use ctx::Ctx; pub use ctx::Ctx;
pub use error::KindMismatch; pub use error::KindMismatch;
pub use node::{FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec}; pub use node::{FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder};
pub use scalar::{Scalar, ScalarKind, Timestamp}; pub use scalar::{Scalar, ScalarKind, Timestamp};
crates/aura-core/src/node.rs
+91 -62
View File
@@ -1,12 +1,14 @@
//! The node contract (C8): the interface every node implements. A node declares //! The node contract (C8): the interface every node implements. A node's static
//! its inputs (each with its scalar kind, lookback depth, and firing policy, C6) //! signature (its input ports — each a scalar kind + firing policy, C6 — its output
//! and its output record (0..K base columns, C7; an empty output declares a //! record of 0..K base columns, C7, and its tunable params) is declared once on the
//! pure consumer / sink role, C8) via `schema`, and computes one cycle's row //! node's `PrimitiveBuilder` (`NodeSchema`), pre-build. The one signature-adjacent
//! via `eval`. //! quantity that may depend on an injected param — the per-input buffer lookback —
//! Tunable params (C12/C19) are declared via `params` (cycle 0015): each node's //! is answered by `Node::lookbacks()`, read only by bootstrap. `eval` computes one
//! typed knobs, which `Blueprint::param_space` aggregates into the sweep's flat, //! cycle's row.
//! path-qualified param-space (C8/C23). Identity is positional (slot); the name is //! Tunable params (C12/C19) are declared via the builder's `NodeSchema.params`
//! a non-load-bearing debug symbol. //! (cycle 0015): each node's typed knobs, which `Composite::param_space` aggregates
//! into the sweep's flat, path-qualified param-space (C8/C23). Identity is
//! positional (slot); the name is a non-load-bearing debug symbol.
use crate::{Ctx, Scalar, ScalarKind}; use crate::{Ctx, Scalar, ScalarKind};
@@ -24,12 +26,13 @@ pub enum Firing {
Barrier(u8), Barrier(u8),
} }
/// One declared input of a node: its scalar kind, the lookback depth the engine /// One declared input **port** of a node: its scalar kind and firing policy (C6).
/// must pre-size for it (must be >= 1), and its firing policy (C6). /// The lookback depth is NOT here — it is a build-time *sizing* concern answered by
/// `Node::lookbacks()`, not part of the static signature (a node's lookback can
/// depend on an injected param, e.g. `Sma`'s window = its `length`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)] #[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct InputSpec { pub struct PortSpec {
pub kind: ScalarKind, pub kind: ScalarKind,
pub lookback: usize,
pub firing: Firing, pub firing: Firing,
} }
@@ -57,49 +60,50 @@ pub struct ParamSpec {
pub kind: ScalarKind, pub kind: ScalarKind,
} }
/// A param-generic blueprint leaf (C19): a node's declared tunable params plus a /// A param-generic blueprint **primitive** recipe (C19): a node's full declared
/// closure that builds a sized instance through the node's own constructor (the /// signature (`NodeSchema` — inputs/output/params) plus a closure that builds a
/// single sizing/validation gate). A blueprint holds these recipes, never built /// sized instance through the node's own constructor (the single sizing/validation
/// instances, so it stays value-empty until a param-set is injected (C19/C23). /// gate). A blueprint holds these recipes, never built instances, so it stays
pub struct LeafFactory { /// value-empty until a param-set is injected (C19/C23). The signature is declared
/// here ONCE — the built node no longer re-declares it (closes the param-declared-
/// twice drift, #36) and a value-empty recipe now exposes its full I/O interface
/// pre-build (#43).
pub struct PrimitiveBuilder {
name: &'static str, name: &'static str,
params: Vec<ParamSpec>, schema: NodeSchema,
// The build closure's type is exactly the recipe contract (a param slice in, a // The build closure's type is exactly the recipe contract (a param slice in, a
// boxed node out); a type alias would not clarify it. // boxed node out); a type alias would not clarify it.
#[allow(clippy::type_complexity)] #[allow(clippy::type_complexity)]
build: Box<dyn Fn(&[Scalar]) -> Box<dyn Node>>, build: Box<dyn Fn(&[Scalar]) -> Box<dyn Node>>,
} }
impl LeafFactory { impl PrimitiveBuilder {
/// `name` is the param-generic render label (the node type, e.g. `"SMA"`); /// `name` is the param-generic render label (the node type, e.g. `"SMA"`);
/// `params` the declared knobs; `build` constructs a sized node from a /// `schema` the full declared signature; `build` constructs a sized node from a
/// kind-checked param slice. /// kind-checked param slice.
pub fn new( pub fn new(
name: &'static str, name: &'static str,
params: Vec<ParamSpec>, schema: NodeSchema,
build: impl Fn(&[Scalar]) -> Box<dyn Node> + 'static, build: impl Fn(&[Scalar]) -> Box<dyn Node> + 'static,
) -> Self { ) -> Self {
Self { name, params, build: Box::new(build) } Self { name, schema, build: Box::new(build) }
} }
/// The declared tunable params (read by `Blueprint::param_space`, pre-build). /// The full declared signature (read pre-build by `Composite::param_space`,
/// `BlueprintNode::signature`, and the renderer).
pub fn schema(&self) -> &NodeSchema {
&self.schema
}
/// The declared tunable params (a view into the signature; pre-build).
pub fn params(&self) -> &[ParamSpec] { pub fn params(&self) -> &[ParamSpec] {
&self.params &self.schema.params
} }
/// Build a sized node from its param slice (the slice is kind-checked by the /// Build a sized node from its param slice (the slice is kind-checked by the
/// caller before this runs). /// caller before this runs).
pub fn build(&self, params: &[Scalar]) -> Box<dyn Node> { pub fn build(&self, params: &[Scalar]) -> Box<dyn Node> {
(self.build)(params) (self.build)(params)
} }
/// The param-generic render label for the blueprint view (C22 "structure /// The param-generic render label for the blueprint view (C22): just the node
/// before"): just the node type, e.g. `SMA`. A value-empty recipe has no values /// type, e.g. `SMA`.
/// to show; the tunable knobs are surfaced by `Blueprint::param_space`. The
/// factory label stays the bare type because alias / handle names are a
/// *composite-level* concept — the renderer folds them into the leaf label at
/// the composite boundary (`render_definition`'s `leaf_label`), where the
/// `(node, slot)` → name mapping lives, not on the standalone factory. (Wide
/// labels are safe: both `aura graph` views are flat since cycle 0017 — no
/// cluster subgraph, so no sibling-overlap garble.) The compiled view labels the
/// built node valued (`SMA(2)`) via `Node::label`, unaffected.
pub fn label(&self) -> String { pub fn label(&self) -> String {
self.name.to_string() self.name.to_string()
} }
@@ -112,7 +116,7 @@ impl LeafFactory {
/// the hot path — the `Vec`s are fine here. /// the hot path — the `Vec`s are fine here.
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct NodeSchema { pub struct NodeSchema {
pub inputs: Vec<InputSpec>, pub inputs: Vec<PortSpec>,
pub output: Vec<FieldSpec>, pub output: Vec<FieldSpec>,
pub params: Vec<ParamSpec>, pub params: Vec<ParamSpec>,
} }
@@ -120,15 +124,22 @@ pub struct NodeSchema {
/// The universal composable dataflow unit (C8): a **producer, consumer, or both**. /// The universal composable dataflow unit (C8): a **producer, consumer, or both**.
/// A producer/transformer exposes one output port carrying a record of 1..K base /// A producer/transformer exposes one output port carrying a record of 1..K base
/// columns; a **pure consumer (sink)** declares `output: vec![]` and records via /// columns; a **pure consumer (sink)** declares `output: vec![]` and records via
/// an out-of-graph side effect in `eval` (cycle 0006). `schema` declares the /// an out-of-graph side effect in `eval` (cycle 0006). The static signature is
/// interface; `eval` computes one cycle's row, returning a borrowed slice into a /// declared on the node's `PrimitiveBuilder`; `eval` computes one cycle's row,
/// buffer the node owns. `None` = filter / not-yet-warmed-up / pure sink; a /// returning a borrowed slice into a buffer the node owns. `None` = filter /
/// returned `Some(row)` must satisfy `row.len() == schema().output.len()` (so a /// not-yet-warmed-up / pure sink; a returned `Some(row)` must satisfy
/// `row.len() == signature().output.len()` (so a
/// pure consumer returns `None` or an empty `Some(&[])`) — the engine debug-asserts /// pure consumer returns `None` or an empty `Some(&[])`) — the engine debug-asserts
/// the width and forwards the row field-wise along the node's out-edges. `&mut self` /// the width and forwards the row field-wise along the node's out-edges. `&mut self`
/// because a node may keep its own derived state (and its output buffer). /// because a node may keep its own derived state (and its output buffer).
pub trait Node { pub trait Node {
fn schema(&self) -> NodeSchema; /// The per-input buffer **lookback** depth (each `>= 1`), in input-slot order —
/// the only signature-adjacent quantity that may depend on an injected param
/// (e.g. `Sma`'s window = its `length`). Read once by `Harness::bootstrap` to
/// size each input column; never on the hot path. Length MUST equal the node's
/// declared `signature().inputs.len()`.
fn lookbacks(&self) -> Vec<usize>;
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]>; fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]>;
/// A one-line, **non-load-bearing** render label (C23): a debug symbol for /// A one-line, **non-load-bearing** render label (C23): a debug symbol for
/// tracing / graph rendering (#13), never read by the run loop and never /// tracing / graph rendering (#13), never read by the run loop and never
@@ -147,11 +158,11 @@ pub trait Node {
mod tests { mod tests {
use super::*; use super::*;
/// A minimal node with an empty schema, reused across the label / factory tests. /// A minimal node with no inputs, reused across the label / builder tests.
struct Bare; struct Bare;
impl Node for Bare { impl Node for Bare {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { inputs: vec![], output: vec![], params: vec![] } vec![]
} }
fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
None None
@@ -159,9 +170,9 @@ mod tests {
} }
#[test] #[test]
fn input_spec_carries_firing() { fn port_spec_carries_firing() {
let a = InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }; let a = PortSpec { kind: ScalarKind::F64, firing: Firing::Any };
let b = InputSpec { kind: ScalarKind::F64, lookback: 2, firing: Firing::Barrier(0) }; let b = PortSpec { kind: ScalarKind::F64, firing: Firing::Barrier(0) };
assert_eq!(a.firing, Firing::Any); assert_eq!(a.firing, Firing::Any);
assert_eq!(b.firing, Firing::Barrier(0)); assert_eq!(b.firing, Firing::Barrier(0));
assert_ne!(a.firing, b.firing); assert_ne!(a.firing, b.firing);
@@ -174,34 +185,52 @@ mod tests {
} }
#[test] #[test]
fn leaf_factory_label_is_the_bare_type() { fn primitive_builder_label_is_the_bare_type() {
// param-generic: the label is just the node type, no knob suffix (the // param-generic: the label is just the node type, no knob suffix (the
// ascii-dag blueprint view cannot render wide cluster-sibling labels). // ascii-dag blueprint view cannot render wide cluster-sibling labels).
let with = LeafFactory::new( let with = PrimitiveBuilder::new(
"SMA", "SMA",
vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], NodeSchema {
inputs: vec![],
output: vec![],
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
},
|_| Box::new(Bare), |_| Box::new(Bare),
); );
assert_eq!(with.label(), "SMA"); assert_eq!(with.label(), "SMA");
let none = LeafFactory::new("Sub", vec![], |_| Box::new(Bare)); let none = PrimitiveBuilder::new(
"Sub",
NodeSchema { inputs: vec![], output: vec![], params: vec![] },
|_| Box::new(Bare),
);
assert_eq!(none.label(), "Sub"); assert_eq!(none.label(), "Sub");
} }
#[test] #[test]
fn leaf_factory_build_runs_the_closure() { fn primitive_builder_build_runs_the_closure() {
let f = LeafFactory::new("Bare", vec![], |_| Box::new(Bare)); let f = PrimitiveBuilder::new(
assert_eq!(f.build(&[]).schema().params, Vec::<ParamSpec>::new()); "Bare",
NodeSchema { inputs: vec![], output: vec![], params: vec![] },
|_| Box::new(Bare),
);
assert_eq!(f.params(), Vec::<ParamSpec>::new());
assert_eq!(f.build(&[]).lookbacks(), Vec::<usize>::new());
} }
#[test] #[test]
fn schema_carries_declared_params() { fn schema_carries_declared_params() {
let s = NodeSchema { let b = PrimitiveBuilder::new(
inputs: vec![], "SMA",
output: vec![], NodeSchema {
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], inputs: vec![],
}; output: vec![],
assert_eq!(s.params.len(), 1); params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
assert_eq!(s.params[0].name, "length"); },
assert_eq!(s.params[0].kind, ScalarKind::I64); |_| Box::new(Bare),
);
assert_eq!(
b.schema().params,
vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }]
);
} }
} }
crates/aura-engine/src/blueprint.rs
+647 -295
View File
File diff suppressed because it is too large Load Diff
crates/aura-engine/src/harness.rs
+293 -88
View File
@@ -21,7 +21,7 @@
//! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never //! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never
//! fire across sources — the timestamp generalizes it faithfully. //! fire across sources — the timestamp generalizes it faithfully.
use aura_core::{AnyColumn, Ctx, Firing, Node, Scalar, ScalarKind, Timestamp}; use aura_core::{AnyColumn, Ctx, Firing, Node, NodeSchema, Scalar, ScalarKind, Timestamp};
/// Forwards one field (`from_field`) of a producer's output record into a /// 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 /// consumer's input slot. Consuming a whole record is N such edges, one per field
@@ -51,6 +51,18 @@ pub struct SourceSpec {
pub targets: Vec<Target>, 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" /// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
/// generalized to the whole topology. /// generalized to the whole topology.
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
@@ -107,41 +119,41 @@ impl core::fmt::Debug for Harness {
} }
impl Harness { impl Harness {
/// Bind nodes + wiring into a frozen, runnable graph. Sizes each node's input /// Bind a flat graph into a frozen, runnable graph. Sizes each node's input
/// columns from its `schema`, lifts each input's firing policy, initializes /// columns — KIND/firing from its carried signature, DEPTH from the built node's
/// per-slot freshness state, kind-checks every source target and edge, and /// `lookbacks()` — lifts each input's firing policy, initializes per-slot
/// topologically orders the nodes (Kahn), rejecting any directed cycle. /// freshness state, kind-checks every source target and edge, and topologically
pub fn bootstrap( /// orders the nodes (Kahn), rejecting any directed cycle.
nodes: Vec<Box<dyn Node>>, pub fn bootstrap(flat: FlatGraph) -> Result<Harness, BootstrapError> {
sources: Vec<SourceSpec>, let FlatGraph { nodes, signatures, sources, edges } = flat;
edges: Vec<Edge>,
) -> Result<Harness, BootstrapError> {
let n = nodes.len(); let n = nodes.len();
let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect(); // size each node's input columns: KIND/firing from the carried signature,
// DEPTH from the built node's lookbacks() (the one param-dependent quantity)
// size each node's input columns from its schema; lift firing; init slots
let mut boxes: Vec<NodeBox> = Vec::with_capacity(n); let mut boxes: Vec<NodeBox> = Vec::with_capacity(n);
for (nd, schema) in nodes.into_iter().zip(schemas.iter()) { for (nd, sig) in nodes.into_iter().zip(signatures.iter()) {
let inputs: Vec<AnyColumn> = schema let depths = nd.lookbacks();
debug_assert_eq!(depths.len(), sig.inputs.len(), "lookbacks() arity == signature inputs");
let inputs: Vec<AnyColumn> = sig
.inputs .inputs
.iter() .iter()
.map(|spec| AnyColumn::with_capacity(spec.kind, spec.lookback)) .zip(depths)
.map(|(spec, depth)| AnyColumn::with_capacity(spec.kind, depth))
.collect(); .collect();
let firing: Vec<Firing> = schema.inputs.iter().map(|spec| spec.firing).collect(); let firing: Vec<Firing> = sig.inputs.iter().map(|spec| spec.firing).collect();
let slots: Vec<SlotState> = schema let slots: Vec<SlotState> = sig
.inputs .inputs
.iter() .iter()
.map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) }) .map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) })
.collect(); .collect();
let out_len = schema.output.len(); let out_len = sig.output.len();
boxes.push(NodeBox { node: nd, inputs, firing, slots, out_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 // source targets: each source's value must match each of its target slots' kind
for src in &sources { for src in &sources {
for t in &src.targets { for t in &src.targets {
let s = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?; let s = signatures.get(t.node).ok_or(BootstrapError::BadIndex)?;
let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?; let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
if slot.kind != src.kind { if slot.kind != src.kind {
return Err(BootstrapError::KindMismatch { return Err(BootstrapError::KindMismatch {
@@ -155,8 +167,8 @@ impl Harness {
// edges: indices in range, producer output field 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]; let mut out_edges: Vec<Vec<Edge>> = vec![Vec::new(); n];
for &e in &edges { for &e in &edges {
let from = schemas.get(e.from).ok_or(BootstrapError::BadIndex)?; let from = signatures.get(e.from).ok_or(BootstrapError::BadIndex)?;
let to = schemas.get(e.to).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 field = from.output.get(e.from_field).ok_or(BootstrapError::BadIndex)?;
let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?; let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?;
if field.kind != slot.kind { if field.kind != slot.kind {
@@ -333,10 +345,9 @@ fn group_id(f: &Firing) -> Option<u8> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
// InputSpec / NodeSchema are not imported by harness.rs production code (it // PortSpec / NodeSchema name the fixtures' declared signatures, brought in here
// only reads `nd.schema()` fields, never naming the types), so they are not // (production code reads them via the carried signatures, never naming the types).
// brought in by `use super::*` — the fixtures construct them, so import here. use aura_core::{FieldSpec, NodeSchema, PortSpec};
use aura_core::{FieldSpec, InputSpec, NodeSchema};
use aura_std::{Exposure, Recorder, Sma, SimBroker, Sub}; use aura_std::{Exposure, Recorder, Sma, SimBroker, Sub};
use std::sync::mpsc; use std::sync::mpsc;
@@ -345,6 +356,33 @@ mod tests {
points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect() 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 }
}
/// 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().map(|&kind| PortSpec { kind, firing }).collect(),
output: vec![],
params: vec![],
}
}
// --- firing-policy fixtures (test-local; not library nodes — C9: examples // --- firing-policy fixtures (test-local; not library nodes — C9: examples
// for the engine's own tests, no speculative aura-std surface) --- // for the engine's own tests, no speculative aura-std surface) ---
@@ -353,17 +391,19 @@ mod tests {
struct AsOfSum { struct AsOfSum {
out: [Scalar; 1], out: [Scalar; 1],
} }
impl Node for AsOfSum { impl AsOfSum {
fn schema(&self) -> NodeSchema { fn sig() -> NodeSchema {
NodeSchema { NodeSchema {
inputs: vec![ inputs: vec![f64_port(Firing::Any), f64_port(Firing::Any)],
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![], params: vec![],
} }
} }
}
impl Node for AsOfSum {
fn lookbacks(&self) -> Vec<usize> {
vec![1, 1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
let a = ctx.f64_in(0); let a = ctx.f64_in(0);
let b = ctx.f64_in(1); let b = ctx.f64_in(1);
@@ -380,17 +420,19 @@ mod tests {
struct BarrierSum { struct BarrierSum {
out: [Scalar; 1], out: [Scalar; 1],
} }
impl Node for BarrierSum { impl BarrierSum {
fn schema(&self) -> NodeSchema { fn sig() -> NodeSchema {
NodeSchema { NodeSchema {
inputs: vec![ inputs: vec![f64_port(Firing::Barrier(0)), f64_port(Firing::Barrier(0))],
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![], params: vec![],
} }
} }
}
impl Node for BarrierSum {
fn lookbacks(&self) -> Vec<usize> {
vec![1, 1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
self.out[0] = Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]); self.out[0] = Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]);
Some(&self.out) Some(&self.out)
@@ -403,18 +445,23 @@ mod tests {
struct MixedSum { struct MixedSum {
out: [Scalar; 1], out: [Scalar; 1],
} }
impl Node for MixedSum { impl MixedSum {
fn schema(&self) -> NodeSchema { fn sig() -> NodeSchema {
NodeSchema { NodeSchema {
inputs: vec![ inputs: vec![
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) }, f64_port(Firing::Barrier(0)),
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) }, f64_port(Firing::Barrier(0)),
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, f64_port(Firing::Any),
], ],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![], params: vec![],
} }
} }
}
impl Node for MixedSum {
fn lookbacks(&self) -> Vec<usize> {
vec![1, 1, 1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
let a = ctx.f64_in(0); let a = ctx.f64_in(0);
let b = ctx.f64_in(1); let b = ctx.f64_in(1);
@@ -434,16 +481,10 @@ mod tests {
struct Ohlcv { struct Ohlcv {
out: [Scalar; 5], out: [Scalar; 5],
} }
impl Node for Ohlcv { impl Ohlcv {
fn schema(&self) -> NodeSchema { fn sig() -> NodeSchema {
NodeSchema { NodeSchema {
inputs: vec![ inputs: vec![f64_port(Firing::Barrier(0)); 5],
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
],
output: vec![ output: vec![
FieldSpec { name: "open", kind: ScalarKind::F64 }, FieldSpec { name: "open", kind: ScalarKind::F64 },
FieldSpec { name: "high", kind: ScalarKind::F64 }, FieldSpec { name: "high", kind: ScalarKind::F64 },
@@ -454,6 +495,11 @@ mod tests {
params: vec![], params: vec![],
} }
} }
}
impl Node for Ohlcv {
fn lookbacks(&self) -> Vec<usize> {
vec![1; 5]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
for i in 0..5 { for i in 0..5 {
let w = ctx.f64_in(i); let w = ctx.f64_in(i);
@@ -473,10 +519,10 @@ mod tests {
struct TwoField { struct TwoField {
out: [Scalar; 2], out: [Scalar; 2],
} }
impl Node for TwoField { impl TwoField {
fn schema(&self) -> NodeSchema { fn sig() -> NodeSchema {
NodeSchema { NodeSchema {
inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }], inputs: vec![f64_port(Firing::Any)],
output: vec![ output: vec![
FieldSpec { name: "f", kind: ScalarKind::F64 }, FieldSpec { name: "f", kind: ScalarKind::F64 },
FieldSpec { name: "i", kind: ScalarKind::I64 }, FieldSpec { name: "i", kind: ScalarKind::I64 },
@@ -484,6 +530,11 @@ mod tests {
params: vec![], params: vec![],
} }
} }
}
impl Node for TwoField {
fn lookbacks(&self) -> Vec<usize> {
vec![1]
}
fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> {
self.out[0] = Scalar::F64(0.0); self.out[0] = Scalar::F64(0.0);
self.out[1] = Scalar::I64(0); self.out[1] = Scalar::I64(0);
@@ -498,14 +549,19 @@ mod tests {
out: [Scalar; 1], out: [Scalar; 1],
tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>, tx: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
} }
impl Node for TapForward { impl TapForward {
fn schema(&self) -> NodeSchema { fn sig() -> NodeSchema {
NodeSchema { NodeSchema {
inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }], inputs: vec![f64_port(Firing::Any)],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![], params: vec![],
} }
} }
}
impl Node for TapForward {
fn lookbacks(&self) -> Vec<usize> {
vec![1]
}
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
let w = ctx.f64_in(0); let w = ctx.f64_in(0);
if w.is_empty() { if w.is_empty() {
@@ -522,11 +578,15 @@ mod tests {
fn chain_source_sma_runs() { fn chain_source_sma_runs() {
// node 0 = SMA(3); source -> SMA(3).in0; node 1 = Recorder taps node 0. // node 0 = SMA(3); source -> SMA(3).in0; node 1 = Recorder taps node 0.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Sma::new(3)), Box::new(Sma::new(3)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
) )
@@ -550,13 +610,19 @@ mod tests {
// 0 = SMA(2), 1 = SMA(4), 2 = Sub; source fans into both SMAs; SMAs join // 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. // into Sub; node 3 = Recorder taps Sub — the 0003 baseline on the new API.
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), Box::new(Sma::new(2)),
Box::new(Sma::new(4)), Box::new(Sma::new(4)),
Box::new(Sub::new()), Box::new(Sub::new()),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -598,11 +664,15 @@ mod tests {
// AsOfSum @0; node 1 = Recorder taps it. source 0 ticks t=1..4; source 1 // AsOfSum @0; node 1 = Recorder taps it. source 0 ticks t=1..4; source 1
// ticks t=2,4 (slower); both AsOfSum inputs Any. // ticks t=2,4 (slower); both AsOfSum inputs Any.
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(AsOfSum { out: [Scalar::F64(0.0)] }), Box::new(AsOfSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
], ],
vec![
AsOfSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![ vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, 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: 1 }] },
@@ -640,11 +710,15 @@ mod tests {
fn mode_b_barrier_fires_only_on_timestamp_coincidence() { fn mode_b_barrier_fires_only_on_timestamp_coincidence() {
// identical wiring to mode A, but both BarrierSum inputs are Barrier(0). // identical wiring to mode A, but both BarrierSum inputs are Barrier(0).
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(BarrierSum { out: [Scalar::F64(0.0)] }), Box::new(BarrierSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
], ],
vec![
BarrierSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![ vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, 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: 1 }] },
@@ -683,13 +757,19 @@ mod tests {
// that cycle's timestamp, so once both SMAs warm and emit in the same // 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. // cycle, both barrier inputs share the timestamp and the barrier fires.
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), Box::new(Sma::new(2)),
Box::new(Sma::new(4)), Box::new(Sma::new(4)),
Box::new(BarrierSum { out: [Scalar::F64(0.0)] }), Box::new(BarrierSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -730,11 +810,15 @@ mod tests {
fn mixed_a_and_b_or_combine_on_one_node() { 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. // MixedSum @0 (in0,in1 barrier group 0; in2 as-of); node 1 = Recorder taps it.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(MixedSum { out: [Scalar::F64(0.0)] }), Box::new(MixedSum { out: [Scalar::F64(0.0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
], ],
vec![
MixedSum::sig(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
vec![ vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, 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: 1 }] },
@@ -763,8 +847,12 @@ mod tests {
#[test] #[test]
fn bootstrap_rejects_a_cycle() { fn bootstrap_rejects_a_cycle() {
// two SMA(1) nodes wired a -> b -> a // two SMA(1) nodes wired a -> b -> a
let err = Harness::bootstrap( let err = boot(
vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))], vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
],
vec![], vec![],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 1, to: 0, slot: 0, from_field: 0 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 1, to: 0, slot: 0, from_field: 0 }],
) )
@@ -775,8 +863,11 @@ mod tests {
#[test] #[test]
fn bootstrap_rejects_a_kind_mismatch() { fn bootstrap_rejects_a_kind_mismatch() {
// SMA(1) declares an f64 input; an i64 source mismatches // SMA(1) declares an f64 input; an i64 source mismatches
let err = Harness::bootstrap( let err = boot(
vec![Box::new(Sma::new(1))], vec![Box::new(Sma::new(1))],
vec![
Sma::builder().schema().clone(),
],
vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }], vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![], vec![],
) )
@@ -792,8 +883,12 @@ mod tests {
// an edge target node (9) that does not exist -> BadIndex. (The old trigger // 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 // — an out-of-range observe index — is gone with `observe`; BadIndex itself
// is unchanged, only the path that reaches it.) // is unchanged, only the path that reaches it.)
let err = Harness::bootstrap( let err = boot(
vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))], 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }], vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }],
) )
@@ -827,7 +922,7 @@ mod tests {
// taps all five fields via five edges. The barrier fires once all five share // 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. // the timestamp, so each bar is recorded once, on the fifth cycle of its ts.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Recorder::new( Box::new(Recorder::new(
@@ -836,6 +931,10 @@ mod tests {
tx, tx,
)), )),
], ],
vec![
Ohlcv::sig(),
recorder_sig(&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any),
],
ohlcv_sources(), ohlcv_sources(),
vec![ vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // open Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // open
@@ -876,12 +975,17 @@ mod tests {
// Proves from_field routes the right columns (not field 0) and the two // 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). // bound fields are co-fresh (Sub's Any inputs both fire in the bar's cycle).
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Sub::new()), Box::new(Sub::new()),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
], ],
vec![
Ohlcv::sig(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
ohlcv_sources(), ohlcv_sources(),
vec![ vec![
Edge { from: 0, to: 1, slot: 0, from_field: 1 }, // high Edge { from: 0, to: 1, slot: 0, from_field: 1 }, // high
@@ -917,12 +1021,17 @@ mod tests {
// (field 0) -> close - open; the Recorder taps Sub. Proves two edges on one // (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). // record read two different fields (3 and 0, not the high/low pair above).
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Sub::new()), Box::new(Sub::new()),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
], ],
vec![
Ohlcv::sig(),
Sub::builder().schema().clone(),
recorder_sig(&[ScalarKind::F64], Firing::Any),
],
ohlcv_sources(), ohlcv_sources(),
vec![ vec![
Edge { from: 0, to: 1, slot: 0, from_field: 3 }, // close Edge { from: 0, to: 1, slot: 0, from_field: 3 }, // close
@@ -947,8 +1056,12 @@ mod tests {
fn bootstrap_rejects_from_field_out_of_range() { fn bootstrap_rejects_from_field_out_of_range() {
// Sma(0) has a 1-field output (index 0 only); an edge reading field 9 is // 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). // out of range -> BadIndex (caught before any kind check).
let err = Harness::bootstrap( let err = boot(
vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))], 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }],
) )
@@ -961,8 +1074,12 @@ mod tests {
// TwoField(0) output: field 0 f64, field 1 i64. Binding field 1 (i64) into // 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 // Sma(1)'s f64 input slot is a per-field kind mismatch -> KindMismatch. (The
// mismatch is field-specific: from_field 0 would have matched.) // mismatch is field-specific: from_field 0 would have matched.)
let err = Harness::bootstrap( let err = boot(
vec![Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }), Box::new(Sma::new(1))], 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
) )
@@ -979,13 +1096,19 @@ mod tests {
// streams (the #2 headline). Each drained stream is individually correct. // streams (the #2 headline). Each drained stream is individually correct.
let (tx_fast, rx_fast) = mpsc::channel(); let (tx_fast, rx_fast) = mpsc::channel();
let (tx_slow, rx_slow) = mpsc::channel(); let (tx_slow, rx_slow) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Sma::new(2)), Box::new(Sma::new(2)),
Box::new(Sma::new(4)), Box::new(Sma::new(4)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_fast)), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_fast)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_slow)), 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -1023,7 +1146,7 @@ mod tests {
// A 5-input Recorder taps all five OHLCV fields via five field-wise edges // 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. // (0005: N edges, no whole-record bind); its recorded row is the whole bar.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }),
Box::new(Recorder::new( Box::new(Recorder::new(
@@ -1032,6 +1155,10 @@ mod tests {
tx, tx,
)), )),
], ],
vec![
Ohlcv::sig(),
recorder_sig(&[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any),
],
ohlcv_sources(), ohlcv_sources(),
vec![ vec![
Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 0, to: 1, slot: 0, from_field: 0 },
@@ -1071,12 +1198,15 @@ mod tests {
// at t=1,2,3,4; only on cycle 4 are all slots warm, so it records once, // at t=1,2,3,4; only on cycle 4 are all slots warm, so it records once,
// holding the earlier-ticked values. // holding the earlier-ticked values.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![Box::new(Recorder::new( vec![Box::new(Recorder::new(
&[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp], &[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp],
Firing::Any, Firing::Any,
tx, tx,
))], ))],
vec![
recorder_sig(&[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp], Firing::Any),
],
vec![ vec![
SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }, 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::F64, targets: vec![Target { node: 0, slot: 1 }] },
@@ -1111,11 +1241,15 @@ mod tests {
// one node is producer and sink at once (C8 "both"). // one node is producer and sink at once (C8 "both").
let (tx_tap, rx_tap) = mpsc::channel(); let (tx_tap, rx_tap) = mpsc::channel();
let (tx_down, rx_down) = mpsc::channel(); let (tx_down, rx_down) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(TapForward { out: [Scalar::F64(0.0)], tx: tx_tap }), Box::new(TapForward { out: [Scalar::F64(0.0)], tx: tx_tap }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_down)), 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
) )
@@ -1137,11 +1271,15 @@ mod tests {
// Two fresh harnesses, two channels, identical input -> bit-identical // Two fresh harnesses, two channels, identical input -> bit-identical
// recorded streams (C1). // recorded streams (C1).
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(3)), Box::new(Sma::new(3)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
) )
@@ -1168,12 +1306,15 @@ mod tests {
// A 2-input Barrier(0) recorder records only on cycles where both inputs // A 2-input Barrier(0) recorder records only on cycles where both inputs
// share the timestamp — the recorder's OWN firing policy gates recording. // share the timestamp — the recorder's OWN firing policy gates recording.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![Box::new(Recorder::new( vec![Box::new(Recorder::new(
&[ScalarKind::F64, ScalarKind::F64], &[ScalarKind::F64, ScalarKind::F64],
Firing::Barrier(0), Firing::Barrier(0),
tx, tx,
))], ))],
vec![
recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Barrier(0)),
],
vec![ vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, 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: 1 }] },
@@ -1200,12 +1341,15 @@ mod tests {
// A 2-input Any recorder records on any-fresh once both are warm (as-of), // A 2-input Any recorder records on any-fresh once both are warm (as-of),
// holding the stale input. // holding the stale input.
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![Box::new(Recorder::new( vec![Box::new(Recorder::new(
&[ScalarKind::F64, ScalarKind::F64], &[ScalarKind::F64, ScalarKind::F64],
Firing::Any, Firing::Any,
tx, tx,
))], ))],
vec![
recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Any),
],
vec![ vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, 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: 1 }] },
@@ -1236,11 +1380,15 @@ mod tests {
// Recorder's f64 input slot is a per-field kind mismatch -> KindMismatch // Recorder's f64 input slot is a per-field kind mismatch -> KindMismatch
// (0005's check already covers recorder edges; recording adds no new hole). // (0005's check already covers recorder edges; recording adds no new hole).
let (tx, _rx) = mpsc::channel(); let (tx, _rx) = mpsc::channel();
let err = Harness::bootstrap( let err = boot(
vec![ vec![
Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }), Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), 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![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }],
) )
@@ -1265,13 +1413,19 @@ mod tests {
// correct stream — node fan-out (not source fan-out: a single SMA(3) // correct stream — node fan-out (not source fan-out: a single SMA(3)
// output is forwarded down three edges to three recorders). // output is forwarded down three edges to three recorders).
let build = |t1, t2, t3| { let build = |t1, t2, t3| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(3)), Box::new(Sma::new(3)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t1)), 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, t2)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)), 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }],
@@ -1322,7 +1476,7 @@ mod tests {
// other input is a second producer (SMA(4)) — yields both the raw tap // other input is a second producer (SMA(4)) — yields both the raw tap
// and the downstream-combined stream, each independently correct. // and the downstream-combined stream, each independently correct.
let build = |t_raw, t_sub| { let build = |t_raw, t_sub| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), // 0: shared producer Box::new(Sma::new(2)), // 0: shared producer
Box::new(Sma::new(4)), // 1: second producer Box::new(Sma::new(4)), // 1: second producer
@@ -1330,6 +1484,13 @@ mod tests {
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_raw)), // 3: raw tap of 0 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 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
@@ -1384,13 +1545,19 @@ mod tests {
// tap fires on every SMA push; the BarrierSum fires only on the cycles // 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. // where the held SMA output and a second source coincide on a timestamp.
let build = |t_any, t_bar| { let build = |t_any, t_bar| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), // 0: shared producer (src A) 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(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(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 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![ vec![
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, // A SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, // A
SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 2, slot: 1 }] }, // B SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 2, slot: 1 }] }, // B
@@ -1441,13 +1608,19 @@ mod tests {
// -> SMA(2) -> recorder) propagates values correctly through depth, with // -> SMA(2) -> recorder) propagates values correctly through depth, with
// each stage's warm-up delaying the tail — closes "deep chains untested". // each stage's warm-up delaying the tail — closes "deep chains untested".
let build = |tx| { let build = |tx| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), // 0 Box::new(Sma::new(2)), // 0
Box::new(Sma::new(2)), // 1 Box::new(Sma::new(2)), // 1
Box::new(Sma::new(2)), // 2 Box::new(Sma::new(2)), // 2
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), // 3 tail 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![Target { node: 0, slot: 0 }], targets: vec![Target { node: 0, slot: 0 }],
@@ -1486,7 +1659,7 @@ mod tests {
// run, yields each parallel stream correctly — closes "wide layers // run, yields each parallel stream correctly — closes "wide layers
// untested" and exercises multi-sink at width. // untested" and exercises multi-sink at width.
let build = |t2, t3, t4| { let build = |t2, t3, t4| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), // 0 Box::new(Sma::new(2)), // 0
Box::new(Sma::new(3)), // 1 Box::new(Sma::new(3)), // 1
@@ -1495,6 +1668,14 @@ mod tests {
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)), // 4 Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t3)), // 4
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t4)), // 5 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![ targets: vec![
@@ -1559,7 +1740,7 @@ mod tests {
// deterministic — the "pure compute substrate carries arbitrary // deterministic — the "pure compute substrate carries arbitrary
// M-producer x N-consumer x K-sink DAGs" gate. // M-producer x N-consumer x K-sink DAGs" gate.
let build = |t_raw, t_sub, t_i64| { let build = |t_raw, t_sub, t_i64| {
Harness::bootstrap( boot(
vec![ vec![
Box::new(Sma::new(2)), // 0: shared f64 producer Box::new(Sma::new(2)), // 0: shared f64 producer
Box::new(Sma::new(4)), // 1: second f64 producer Box::new(Sma::new(4)), // 1: second f64 producer
@@ -1568,6 +1749,14 @@ mod tests {
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_sub)), // 4 Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_sub)), // 4
Box::new(Recorder::new(&[ScalarKind::I64], Firing::Any, t_i64)), // 5: i64 sink 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![ vec![
SourceSpec { SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
@@ -1634,7 +1823,7 @@ mod tests {
#[test] #[test]
fn signal_quality_loop_records_pip_equity() { fn signal_quality_loop_records_pip_equity() {
let (tx_eq, rx_eq) = mpsc::channel(); let (tx_eq, rx_eq) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Sma::new(2)), // 0 fast Box::new(Sma::new(2)), // 0 fast
Box::new(Sma::new(4)), // 1 slow Box::new(Sma::new(4)), // 1 slow
@@ -1643,6 +1832,14 @@ mod tests {
Box::new(SimBroker::new(0.0001)), // 4 pip equity Box::new(SimBroker::new(0.0001)), // 4 pip equity
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 sink 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![ targets: vec![
@@ -1692,7 +1889,7 @@ mod tests {
fn signal_quality_loop_is_deterministic() { fn signal_quality_loop_is_deterministic() {
let build = || { let build = || {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut h = Harness::bootstrap( let mut h = boot(
vec![ vec![
Box::new(Sma::new(2)), Box::new(Sma::new(2)),
Box::new(Sma::new(4)), Box::new(Sma::new(4)),
@@ -1701,6 +1898,14 @@ mod tests {
Box::new(SimBroker::new(0.0001)), Box::new(SimBroker::new(0.0001)),
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), 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 { vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![ targets: vec![
crates/aura-engine/src/lib.rs
+3 -4
View File
@@ -36,15 +36,14 @@ mod harness;
mod report; mod report;
pub use blueprint::{ pub use blueprint::{
aliases_on, signature_of, Blueprint, BlueprintNode, CompileError, Composite, OutField, aliases_on, signature_of, BlueprintNode, CompileError, Composite, OutField, ParamAlias, Role,
ParamAlias, Role,
}; };
pub use harness::{BootstrapError, Edge, Harness, SourceSpec, Target}; pub use harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};
pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport}; pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
// #29: re-export the core scalar vocabulary a Blueprint builder needs // #29: re-export the core scalar vocabulary a Blueprint builder needs
// (SourceSpec.kind is a ScalarKind; sources/Recorder columns are Scalar / // (SourceSpec.kind is a ScalarKind; sources/Recorder columns are Scalar /
// Firing / Timestamp) so a graph builder has one import surface, not two. // Firing / Timestamp) so a graph builder has one import surface, not two.
pub use aura_core::{Firing, ParamSpec, Scalar, ScalarKind, Timestamp}; pub use aura_core::{Firing, NodeSchema, ParamSpec, PortSpec, Scalar, ScalarKind, Timestamp};
#[cfg(test)] #[cfg(test)]
mod reexport_tests { mod reexport_tests {
crates/aura-engine/src/report.rs
+26 -7
View File
@@ -198,11 +198,21 @@ pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timest
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::{Edge, Harness, SourceSpec, Target}; use crate::{Edge, FlatGraph, Harness, SourceSpec, Target};
use aura_core::{Firing, ScalarKind}; use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub}; use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use std::sync::mpsc; use std::sync::mpsc;
/// The declared signature of a `Recorder` over one f64 column (the sink shape
/// the two-sink harness uses).
fn f64_recorder_sig() -> NodeSchema {
NodeSchema {
inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
output: vec![],
params: vec![],
}
}
/// Build an f64 source stream from (timestamp, value) points (mirrors the /// Build an f64 source stream from (timestamp, value) points (mirrors the
/// harness.rs test helper; the e2e test needs its own copy — the harness /// harness.rs test helper; the e2e test needs its own copy — the harness
/// test module's is private to that module). /// test module's is private to that module).
@@ -221,8 +231,8 @@ mod tests {
) { ) {
let (tx_eq, rx_eq) = mpsc::channel(); let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel(); let (tx_ex, rx_ex) = mpsc::channel();
let h = Harness::bootstrap( let h = Harness::bootstrap(FlatGraph {
vec![ nodes: vec![
Box::new(Sma::new(2)), // 0 Box::new(Sma::new(2)), // 0
Box::new(Sma::new(4)), // 1 Box::new(Sma::new(4)), // 1
Box::new(Sub::new()), // 2 Box::new(Sub::new()), // 2
@@ -231,7 +241,16 @@ mod tests {
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
], ],
vec![SourceSpec { signatures: vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
Exposure::builder().schema().clone(),
SimBroker::builder(0.0001).schema().clone(),
f64_recorder_sig(),
f64_recorder_sig(),
],
sources: vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![ targets: vec![
Target { node: 0, slot: 0 }, Target { node: 0, slot: 0 },
@@ -239,7 +258,7 @@ mod tests {
Target { node: 4, slot: 1 }, // price into the broker Target { node: 4, slot: 1 }, // price into the broker
], ],
}], }],
vec![ edges: vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, 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: 2, to: 3, slot: 0, from_field: 0 },
@@ -247,7 +266,7 @@ mod tests {
Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5 Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6 Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
], ],
) })
.expect("valid signal-quality DAG"); .expect("valid signal-quality DAG");
(h, rx_eq, rx_ex) (h, rx_eq, rx_ex)
} }
crates/aura-ingest/tests/real_bars.rs
+21 -7
View File
@@ -7,9 +7,9 @@
use std::sync::mpsc; use std::sync::mpsc;
use std::sync::Arc; use std::sync::Arc;
use aura_core::{Firing, Scalar, ScalarKind, Timestamp}; use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
use aura_engine::{ use aura_engine::{
f64_field, summarize, Edge, Harness, RunManifest, RunReport, SourceSpec, Target, f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target,
}; };
use aura_ingest::load_m1_window; use aura_ingest::load_m1_window;
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub}; use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
@@ -22,8 +22,13 @@ use data_server::{DataServer, DEFAULT_DATA_PATH};
fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport { fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
let (tx_eq, rx_eq) = mpsc::channel(); let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel(); let (tx_ex, rx_ex) = mpsc::channel();
let mut h = Harness::bootstrap( let f64_recorder_sig = || NodeSchema {
vec![ inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
output: vec![],
params: vec![],
};
let mut h = Harness::bootstrap(FlatGraph {
nodes: vec![
Box::new(Sma::new(2)), // 0 Box::new(Sma::new(2)), // 0
Box::new(Sma::new(4)), // 1 Box::new(Sma::new(4)), // 1
Box::new(Sub::new()), // 2 Box::new(Sub::new()), // 2
@@ -32,7 +37,16 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
], ],
vec![SourceSpec { signatures: vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
Exposure::builder().schema().clone(),
SimBroker::builder(0.0001).schema().clone(),
f64_recorder_sig(),
f64_recorder_sig(),
],
sources: vec![SourceSpec {
kind: ScalarKind::F64, kind: ScalarKind::F64,
targets: vec![ targets: vec![
Target { node: 0, slot: 0 }, Target { node: 0, slot: 0 },
@@ -40,7 +54,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
Target { node: 4, slot: 1 }, // price into the broker Target { node: 4, slot: 1 }, // price into the broker
], ],
}], }],
vec![ edges: vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, 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: 2, to: 3, slot: 0, from_field: 0 },
@@ -48,7 +62,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5 Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6 Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
], ],
) })
.expect("valid signal-quality DAG"); .expect("valid signal-quality DAG");
let window = ( let window = (
crates/aura-std/src/add.rs
+17 -20
View File
@@ -2,7 +2,7 @@
//! Combines two signal streams into one — the most basic combinator for the //! Combines two signal streams into one — the most basic combinator for the
//! north-star "combine one signal with another" research move (C10). //! north-star "combine one signal with another" research move (C10).
use aura_core::{Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind}; use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind};
/// Two-input f64 sum: input 0 plus input 1. Emits `None` until both inputs /// Two-input f64 sum: input 0 plus input 1. Emits `None` until both inputs
/// have a value. /// have a value.
@@ -26,10 +26,21 @@ impl Add {
Self { out: [Scalar::F64(0.0)] } Self { out: [Scalar::F64(0.0)] }
} }
/// The param-generic recipe for a blueprint leaf: paramless, builds through /// The param-generic recipe for a blueprint primitive: paramless, builds through
/// `Add::new`. /// `Add::new`.
pub fn factory() -> LeafFactory { pub fn builder() -> PrimitiveBuilder {
LeafFactory::new("Add", vec![], |_| Box::new(Add::new())) PrimitiveBuilder::new(
"Add",
NodeSchema {
inputs: vec![
PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
},
|_| Box::new(Add::new()),
)
} }
} }
@@ -40,15 +51,8 @@ impl Default for Add {
} }
impl Node for Add { impl Node for Add {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![1, 1]
inputs: vec![
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -71,13 +75,6 @@ mod tests {
use super::*; use super::*;
use aura_core::{AnyColumn, Timestamp}; use aura_core::{AnyColumn, Timestamp};
#[test]
fn factory_params_match_built_node_schema() {
let f = Add::factory();
let built = f.build(&[]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
fn add_is_sum_once_both_inputs_present() { fn add_is_sum_once_both_inputs_present() {
let mut add = Add::new(); let mut add = Add::new();
crates/aura-std/src/ema.rs
+20 -24
View File
@@ -18,7 +18,8 @@
//! nothing on the hot path (the output buffer is reused). //! nothing on the hot path (the output buffer is reused).
use aura_core::{ use aura_core::{
Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
ScalarKind,
}; };
/// Exponential moving average of one f64 input, smoothing `alpha = 2/(length+1)`, /// Exponential moving average of one f64 input, smoothing `alpha = 2/(length+1)`,
@@ -52,31 +53,27 @@ impl Ema {
} }
} }
/// The param-generic recipe for a blueprint leaf: declares `length` and builds /// The param-generic recipe for a blueprint primitive: declares `length` and builds
/// through `Ema::new` (the single sizing/validation gate; the slice is /// through `Ema::new` (the single sizing/validation gate; the slice is
/// kind-checked before `build` runs, so the typed read is total). /// kind-checked before `build` runs, so the typed read is total).
pub fn factory() -> LeafFactory { pub fn builder() -> PrimitiveBuilder {
LeafFactory::new( PrimitiveBuilder::new(
"EMA", "EMA",
vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], NodeSchema {
inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
},
|p| Box::new(Ema::new(p[0].as_i64().expect("length slot is I64") as usize)), |p| Box::new(Ema::new(p[0].as_i64().expect("length slot is I64") as usize)),
) )
} }
} }
impl Node for Ema { impl Node for Ema {
fn schema(&self) -> NodeSchema { // recursive: the running average lives in internal state, so only the newest
NodeSchema { // sample is read — `length` sizes alpha, not the window.
inputs: vec![InputSpec { fn lookbacks(&self) -> Vec<usize> {
kind: ScalarKind::F64, vec![1]
// recursive: the running average lives in internal state, so only
// the newest sample is read — `length` sizes alpha, not the window.
lookback: 1,
firing: Firing::Any,
}],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -157,12 +154,11 @@ mod tests {
} }
#[test] #[test]
fn factory_params_match_built_node_schema() { fn builder_declares_the_length_knob() {
let f = Ema::factory(); // the param-generic recipe carries the declared I64 `length` knob
let built = f.build(&[Scalar::I64(9)]); let b = Ema::builder();
assert_eq!(f.params(), built.schema().params.as_slice()); assert_eq!(b.params(), b.schema().params.as_slice());
// the built node carries the declared I64 `length` knob assert_eq!(b.schema().params[0].name, "length");
assert_eq!(built.schema().params[0].name, "length"); assert_eq!(b.schema().params[0].kind, ScalarKind::I64);
assert_eq!(built.schema().params[0].kind, ScalarKind::I64);
} }
} }
crates/aura-std/src/exposure.rs
+12 -18
View File
@@ -4,7 +4,8 @@
//! `scale` sets which signal magnitude maps to full exposure (sizing lives here). //! `scale` sets which signal magnitude maps to full exposure (sizing lives here).
use aura_core::{ use aura_core::{
Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
ScalarKind,
}; };
/// Bounded exposure from a raw signal score: `clamp(signal / scale, -1.0, +1.0)`. /// Bounded exposure from a raw signal score: `clamp(signal / scale, -1.0, +1.0)`.
@@ -21,25 +22,25 @@ impl Exposure {
Self { scale, out: [Scalar::F64(0.0)] } Self { scale, out: [Scalar::F64(0.0)] }
} }
/// The param-generic recipe for a blueprint leaf: declares `scale` and builds /// The param-generic recipe for a blueprint primitive: declares `scale` and builds
/// through `Exposure::new` (the single sizing/validation gate; the slice is /// through `Exposure::new` (the single sizing/validation gate; the slice is
/// kind-checked before `build` runs, so the typed read is total). /// kind-checked before `build` runs, so the typed read is total).
pub fn factory() -> LeafFactory { pub fn builder() -> PrimitiveBuilder {
LeafFactory::new( PrimitiveBuilder::new(
"Exposure", "Exposure",
vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }], NodeSchema {
inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
output: vec![FieldSpec { name: "exposure", kind: ScalarKind::F64 }],
params: vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
},
|p| Box::new(Exposure::new(p[0].as_f64().expect("scale slot is F64"))), |p| Box::new(Exposure::new(p[0].as_f64().expect("scale slot is F64"))),
) )
} }
} }
impl Node for Exposure { impl Node for Exposure {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![1]
inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
output: vec![FieldSpec { name: "exposure", kind: ScalarKind::F64 }],
params: vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -82,13 +83,6 @@ mod tests {
} }
} }
#[test]
fn factory_params_match_built_node_schema() {
let f = Exposure::factory();
let built = f.build(&[Scalar::F64(0.5)]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
fn exposure_is_none_until_input_present() { fn exposure_is_none_until_input_present() {
let mut e = Exposure::new(0.5); let mut e = Exposure::new(0.5);
crates/aura-std/src/lincomb.rs
+12 -25
View File
@@ -7,7 +7,8 @@
//! indexed `F64` knobs that `Blueprint::param_space` aggregates (C8/C12/C19). //! indexed `F64` knobs that `Blueprint::param_space` aggregates (C8/C12/C19).
use aura_core::{ use aura_core::{
Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
ScalarKind,
}; };
/// Weighted sum of `N` f64 inputs: `Σ weights[i] · input[i]`. The `weights` are /// Weighted sum of `N` f64 inputs: `Σ weights[i] · input[i]`. The `weights` are
@@ -39,16 +40,19 @@ impl LinComb {
Self { weights, out: [Scalar::F64(0.0)] } Self { weights, out: [Scalar::F64(0.0)] }
} }
/// The param-generic recipe for a blueprint leaf. The `arity` is topology /// The param-generic recipe for a blueprint primitive. The `arity` is topology
/// (fixed per blueprint, C19), taken as a factory arg; only the weight *values* /// (fixed per blueprint, C19), taken as a builder arg; only the weight *values*
/// are injected, slot by slot, through `LinComb::new` (the single sizing gate). /// are injected, slot by slot, through `LinComb::new` (the single sizing gate).
pub fn factory(arity: usize) -> LeafFactory { pub fn builder(arity: usize) -> PrimitiveBuilder {
let inputs = (0..arity)
.map(|_| PortSpec { kind: ScalarKind::F64, firing: Firing::Any })
.collect();
let params = (0..arity) let params = (0..arity)
.map(|i| ParamSpec { name: format!("weights[{i}]"), kind: ScalarKind::F64 }) .map(|i| ParamSpec { name: format!("weights[{i}]"), kind: ScalarKind::F64 })
.collect(); .collect();
LeafFactory::new( PrimitiveBuilder::new(
"LinComb", "LinComb",
params, NodeSchema { inputs, output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], params },
|p| Box::new(LinComb::new( |p| Box::new(LinComb::new(
p.iter().map(|s| s.as_f64().expect("weight slot is F64")).collect(), p.iter().map(|s| s.as_f64().expect("weight slot is F64")).collect(),
)), )),
@@ -57,18 +61,8 @@ impl LinComb {
} }
impl Node for LinComb { impl Node for LinComb {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![1; self.weights.len()]
inputs: self
.weights
.iter()
.map(|_| InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any })
.collect(),
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: (0..self.weights.len())
.map(|i| ParamSpec { name: format!("weights[{i}]"), kind: ScalarKind::F64 })
.collect(),
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -142,13 +136,6 @@ mod tests {
assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::F64(6.0)].as_slice())); assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::F64(6.0)].as_slice()));
} }
#[test]
fn factory_params_match_built_node_schema() {
let f = LinComb::factory(2);
let built = f.build(&[Scalar::F64(1.0), Scalar::F64(-1.0)]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
#[should_panic(expected = "LinComb needs at least one weight")] #[should_panic(expected = "LinComb needs at least one weight")]
fn lincomb_empty_weights_panics() { fn lincomb_empty_weights_panics() {
crates/aura-std/src/recorder.rs
+32 -37
View File
@@ -7,7 +7,7 @@
//! four base scalar kinds so any column can be persisted; returns `None` (filters) //! four base scalar kinds so any column can be persisted; returns `None` (filters)
//! until every input column is warm. //! until every input column is warm.
use aura_core::{Ctx, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; use aura_core::{Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
/// A recording sink over `kinds.len()` input columns. Each fired cycle it reads /// A recording sink over `kinds.len()` input columns. Each fired cycle it reads
@@ -16,42 +16,41 @@ use std::sync::mpsc::Sender;
/// value. /// value.
pub struct Recorder { pub struct Recorder {
kinds: Vec<ScalarKind>, kinds: Vec<ScalarKind>,
firing: Firing,
tx: Sender<(Timestamp, Vec<Scalar>)>, tx: Sender<(Timestamp, Vec<Scalar>)>,
} }
impl Recorder { impl Recorder {
/// A recorder over one input column per entry in `kinds`, each with the given /// A recorder over one input column per entry in `kinds`, each with the given
/// `firing` policy, sending recorded `(timestamp, row)` pairs to `tx`. /// `firing` policy, sending recorded `(timestamp, row)` pairs to `tx`. The
pub fn new(kinds: &[ScalarKind], firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self { /// `firing` policy is a property of the declared signature (carried by the
Self { kinds: kinds.to_vec(), firing, tx } /// `PrimitiveBuilder`'s `PortSpec`s), not of the built sink, so it is part of
/// the construction contract but not stored on the instance.
pub fn new(kinds: &[ScalarKind], _firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
Self { kinds: kinds.to_vec(), tx }
} }
/// The param-generic recipe for a blueprint leaf. The channel + kinds + firing /// The param-generic recipe for a blueprint primitive. The channel + kinds + firing
/// are non-param construction args (captured), not tunable params; the node /// are non-param construction args (captured), not tunable params; the node
/// declares none. `tx` is cloned per build (`mpsc::Sender: Clone`). /// declares none. The input ports (one per kind) are threaded into the schema
pub fn factory( /// statically; `tx`/`kinds` are cloned for the build closure.
pub fn builder(
kinds: Vec<ScalarKind>, kinds: Vec<ScalarKind>,
firing: Firing, firing: Firing,
tx: Sender<(Timestamp, Vec<Scalar>)>, tx: Sender<(Timestamp, Vec<Scalar>)>,
) -> LeafFactory { ) -> PrimitiveBuilder {
LeafFactory::new("Recorder", vec![], move |_| { let inputs = kinds.iter().map(|&kind| PortSpec { kind, firing }).collect();
Box::new(Recorder::new(&kinds, firing, tx.clone())) let build_kinds = kinds.clone();
}) PrimitiveBuilder::new(
"Recorder",
NodeSchema { inputs, output: vec![], params: vec![] }, // sink: empty output (C8)
move |_| Box::new(Recorder::new(&build_kinds, firing, tx.clone())),
)
} }
} }
impl Node for Recorder { impl Node for Recorder {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![1; self.kinds.len()]
inputs: self
.kinds
.iter()
.map(|&kind| InputSpec { kind, lookback: 1, firing: self.firing })
.collect(),
output: vec![],
params: vec![],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -81,26 +80,22 @@ mod tests {
use aura_core::{AnyColumn, Timestamp}; use aura_core::{AnyColumn, Timestamp};
use std::sync::mpsc; use std::sync::mpsc;
#[test]
fn factory_params_match_built_node_schema() {
let (tx, _rx) = mpsc::channel();
let f = Recorder::factory(vec![ScalarKind::F64], Firing::Any, tx);
let built = f.build(&[]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
fn recorder_captures_f64_stream_after_warmup() { fn recorder_captures_f64_stream_after_warmup() {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
let mut rec = Recorder::new(&[ScalarKind::F64], Firing::Any, tx); let mut rec = Recorder::new(&[ScalarKind::F64], Firing::Any, tx);
// size the one f64 input column from the schema, as the engine would. // a sink declares no output (C8) — asserted on the param-generic builder
let schema = rec.schema(); let (tx_b, _rx_b) = mpsc::channel();
assert!(schema.output.is_empty(), "a sink declares no output (C8)"); assert!(
let mut inputs = vec![AnyColumn::with_capacity( Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_b)
schema.inputs[0].kind, .schema()
schema.inputs[0].lookback, .output
)]; .is_empty(),
"a sink declares no output (C8)"
);
// size the one f64 input column from the node's lookback, as bootstrap would.
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, rec.lookbacks()[0])];
// cold: returns None and records nothing. // cold: returns None and records nothing.
assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None); assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None);
crates/aura-std/src/sim_broker.rs
+17 -20
View File
@@ -4,7 +4,7 @@
//! each cycle (decided at t-1) into a cumulative synthetic pip-equity output. //! each cycle (decided at t-1) into a cumulative synthetic pip-equity output.
//! Measures signal quality, not execution-modelled P&L. //! Measures signal quality, not execution-modelled P&L.
use aura_core::{Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind}; use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind};
/// Integrates `exposure * price-return` into cumulative pips. `pip_size` is /// Integrates `exposure * price-return` into cumulative pips. `pip_size` is
/// per-instrument reference metadata (beside the hot path, C7/C15), held here, /// per-instrument reference metadata (beside the hot path, C7/C15), held here,
@@ -55,24 +55,28 @@ impl SimBroker {
} }
} }
/// The param-generic recipe for a blueprint leaf. `pip_size` is per-instrument /// The param-generic recipe for a blueprint primitive. `pip_size` is per-instrument
/// metadata (C10/C15), not a tunable param — it is captured by the closure, not /// metadata (C10/C15), not a tunable param — it is captured by the closure, not
/// injected; the node declares no params. /// injected; the node declares no params.
pub fn factory(pip_size: f64) -> LeafFactory { pub fn builder(pip_size: f64) -> PrimitiveBuilder {
LeafFactory::new("SimBroker", vec![], move |_| Box::new(SimBroker::new(pip_size))) PrimitiveBuilder::new(
"SimBroker",
NodeSchema {
inputs: vec![
PortSpec { kind: ScalarKind::F64, firing: Firing::Any }, // 0 exposure
PortSpec { kind: ScalarKind::F64, firing: Firing::Any }, // 1 price
],
output: vec![FieldSpec { name: "equity", kind: ScalarKind::F64 }],
params: vec![],
},
move |_| Box::new(SimBroker::new(pip_size)),
)
} }
} }
impl Node for SimBroker { impl Node for SimBroker {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![1, 1]
inputs: vec![
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, // 0 exposure
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, // 1 price
],
output: vec![FieldSpec { name: "equity", kind: ScalarKind::F64 }],
params: vec![],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -122,13 +126,6 @@ mod tests {
} }
} }
#[test]
fn factory_params_match_built_node_schema() {
let f = SimBroker::factory(0.0001);
let built = f.build(&[]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
fn sim_broker_integrates_lagged_exposure_times_return() { fn sim_broker_integrates_lagged_exposure_times_return() {
let mut b = SimBroker::new(1.0); let mut b = SimBroker::new(1.0);
crates/aura-std/src/sma.rs
+30 -37
View File
@@ -4,7 +4,8 @@
//! engine present (the test drives it by hand, as the sim loop later will). //! engine present (the test drives it by hand, as the sim loop later will).
use aura_core::{ use aura_core::{
Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, ParamSpec, Scalar, ScalarKind, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar,
ScalarKind,
}; };
/// Simple moving average over the last `length` values of one f64 input. /// Simple moving average over the last `length` values of one f64 input.
@@ -20,29 +21,25 @@ impl Sma {
Self { length, out: [Scalar::F64(0.0)] } Self { length, out: [Scalar::F64(0.0)] }
} }
/// The param-generic recipe for a blueprint leaf: declares `length` and builds /// The param-generic recipe for a blueprint primitive: declares `length` and builds
/// through `Sma::new` (the single sizing/validation gate; the slice is /// through `Sma::new` (the single sizing/validation gate; the slice is
/// kind-checked before `build` runs, so the typed read is total). /// kind-checked before `build` runs, so the typed read is total).
pub fn factory() -> LeafFactory { pub fn builder() -> PrimitiveBuilder {
LeafFactory::new( PrimitiveBuilder::new(
"SMA", "SMA",
vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], NodeSchema {
inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any }],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
},
|p| Box::new(Sma::new(p[0].as_i64().expect("length slot is I64") as usize)), |p| Box::new(Sma::new(p[0].as_i64().expect("length slot is I64") as usize)),
) )
} }
} }
impl Node for Sma { impl Node for Sma {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![self.length]
inputs: vec![InputSpec {
kind: ScalarKind::F64,
lookback: self.length,
firing: Firing::Any,
}],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -70,14 +67,12 @@ mod tests {
#[test] #[test]
fn sma_warms_up_then_tracks_the_window_mean() { fn sma_warms_up_then_tracks_the_window_mean() {
let mut sma = Sma::new(3); let sma_for_depth = Sma::new(3);
let schema = sma.schema();
// size the input column from the schema, as the engine will at wiring // size the input column from the node's lookback, as bootstrap will at wiring
let mut inputs = vec![AnyColumn::with_capacity( let mut inputs =
schema.inputs[0].kind, vec![AnyColumn::with_capacity(ScalarKind::F64, sma_for_depth.lookbacks()[0])];
schema.inputs[0].lookback, let mut sma = sma_for_depth;
)];
let feed = [1.0_f64, 2.0, 3.0, 4.0, 5.0]; let feed = [1.0_f64, 2.0, 3.0, 4.0, 5.0];
// means of [1,2,3], [2,3,4], [3,4,5] once warmed up // means of [1,2,3], [2,3,4], [3,4,5] once warmed up
@@ -124,37 +119,35 @@ mod tests {
assert_eq!(Recorder::new(&[ScalarKind::F64], Firing::Any, tx).label(), "Recorder"); assert_eq!(Recorder::new(&[ScalarKind::F64], Firing::Any, tx).label(), "Recorder");
} }
#[test]
fn factory_params_match_built_node_schema() {
let f = Sma::factory();
let built = f.build(&[Scalar::I64(3)]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
fn nodes_declare_expected_params() { fn nodes_declare_expected_params() {
use crate::{Add, Exposure, LinComb, Recorder, SimBroker, Sub}; use crate::{Add, Exposure, LinComb, Recorder, SimBroker, Sub};
use aura_core::{Firing, ParamSpec, ScalarKind}; use aura_core::{Firing, ParamSpec, ScalarKind};
// single scalar knobs // single scalar knobs (declared on the param-generic builder, pre-build)
assert_eq!( assert_eq!(
Sma::new(3).schema().params, Sma::builder().schema().params,
vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
); );
assert_eq!( assert_eq!(
Exposure::new(0.5).schema().params, Exposure::builder().schema().params,
vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }], vec![ParamSpec { name: "scale".into(), kind: ScalarKind::F64 }],
); );
// vector knob expands flat to N indexed F64 entries // vector knob expands flat to N indexed F64 entries
let lc = LinComb::new(vec![1.0, -1.0]).schema().params; let lc = LinComb::builder(2).schema().params.clone();
assert_eq!(lc.len(), 2); assert_eq!(lc.len(), 2);
assert_eq!(lc[0].name, "weights[0]"); assert_eq!(lc[0].name, "weights[0]");
assert_eq!(lc[1].name, "weights[1]"); assert_eq!(lc[1].name, "weights[1]");
assert!(lc.iter().all(|p| p.kind == ScalarKind::F64)); assert!(lc.iter().all(|p| p.kind == ScalarKind::F64));
// param-less nodes declare empty // param-less nodes declare empty
assert!(Sub::new().schema().params.is_empty()); assert!(Sub::builder().schema().params.is_empty());
assert!(Add::new().schema().params.is_empty()); assert!(Add::builder().schema().params.is_empty());
assert!(SimBroker::new(0.0001).schema().params.is_empty()); assert!(SimBroker::builder(0.0001).schema().params.is_empty());
let (tx, _rx) = std::sync::mpsc::channel(); let (tx, _rx) = std::sync::mpsc::channel();
assert!(Recorder::new(&[ScalarKind::F64], Firing::Any, tx).schema().params.is_empty()); assert!(
Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx)
.schema()
.params
.is_empty()
);
} }
} }
crates/aura-std/src/sub.rs
+17 -20
View File
@@ -3,7 +3,7 @@
//! real fan-out + join to run (two SMAs joining into one node), exercising //! real fan-out + join to run (two SMAs joining into one node), exercising
//! multi-input `Ctx` access inside a running graph. //! multi-input `Ctx` access inside a running graph.
use aura_core::{Ctx, FieldSpec, Firing, InputSpec, LeafFactory, Node, NodeSchema, Scalar, ScalarKind}; use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind};
/// Two-input f64 difference: input 0 minus input 1. Emits `None` until both /// Two-input f64 difference: input 0 minus input 1. Emits `None` until both
/// inputs have a value. /// inputs have a value.
@@ -17,10 +17,21 @@ impl Sub {
Self { out: [Scalar::F64(0.0)] } Self { out: [Scalar::F64(0.0)] }
} }
/// The param-generic recipe for a blueprint leaf: paramless, builds through /// The param-generic recipe for a blueprint primitive: paramless, builds through
/// `Sub::new`. /// `Sub::new`.
pub fn factory() -> LeafFactory { pub fn builder() -> PrimitiveBuilder {
LeafFactory::new("Sub", vec![], |_| Box::new(Sub::new())) PrimitiveBuilder::new(
"Sub",
NodeSchema {
inputs: vec![
PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
PortSpec { kind: ScalarKind::F64, firing: Firing::Any },
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
},
|_| Box::new(Sub::new()),
)
} }
} }
@@ -31,15 +42,8 @@ impl Default for Sub {
} }
impl Node for Sub { impl Node for Sub {
fn schema(&self) -> NodeSchema { fn lookbacks(&self) -> Vec<usize> {
NodeSchema { vec![1, 1]
inputs: vec![
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
],
output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }],
params: vec![],
}
} }
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
@@ -62,13 +66,6 @@ mod tests {
use super::*; use super::*;
use aura_core::{AnyColumn, Timestamp}; use aura_core::{AnyColumn, Timestamp};
#[test]
fn factory_params_match_built_node_schema() {
let f = Sub::factory();
let built = f.build(&[]);
assert_eq!(f.params(), built.schema().params.as_slice());
}
#[test] #[test]
fn sub_is_difference_once_both_inputs_present() { fn sub_is_difference_once_both_inputs_present() {
let mut sub = Sub::new(); let mut sub = Sub::new();