feat(engine): the deterministic single-source sim loop
Cycle 0003: aura-engine's first real content — a Sim that runs a wired DAG of
nodes deterministically, cycle by cycle. First point in the project where
authored nodes execute against data, not just under a hand-fed Ctx.
- `Sim` (aura-engine) — a bootstrapped, frozen root graph: a flat Vec<NodeBox>
(each node owns its input columns, the cycle-0002 shape) + an index edge
table, topologically ordered (Kahn). `bootstrap` sizes every input column from
its node's schema, kind-checks each edge and source target, and rejects
directed cycles — C7's "type check paid once at wiring" generalized to the
whole topology (`BootstrapError::{KindMismatch, BadIndex, Cycle}`).
- `run` — the deterministic loop: per record, forward the source value into its
target slots, evaluate nodes in topo order, capture the observed node's output
at eval time, and forward each `Some` output into its consumers' input columns
(a `None` forwards nothing — the structural seed of sample-and-hold). The loop
destructures `&mut self` into disjoint field borrows and allocates nothing on
the per-cycle path. This is the flat, monomorphized sharpening of RustAst's
reference-counted, interior-mutable observer push graph (C1/C7).
- `Edge` / `Target` (aura-engine) — producer->consumer and source->consumer wiring.
- `Sub` (aura-std) — a 2-input f64-difference node, so the loop is proven on a
real fan-out + join DAG (source -> {SMA(2), SMA(4)} -> Sub), with a determinism
assertion (C1: a second identical run is bit-identical).
Engine library depends only on aura-core; a test-only dev-dependency on aura-std
lets the integration test wire real Sma/Sub nodes. Sim carries a hand-written,
node-opaque `Debug` impl (Box<dyn Node> is not Debug; the impl prints
nodes.len() + the index/topology fields, needed by the bootstrap-rejection tests'
`unwrap_err`).
Deliberately deferred (recorded as decisions): freshness-gated recompute /
sample-and-hold (C5 -> cycle 0004, with the second source that makes it testable);
the explicit monotonic cycle_id counter (its first reader is freshness, 0004);
the Source trait + data-server ingestion + k-way merge (C3/C11); the builder API
(C19); the real sink + run registry (C18/C22).
Gates green: cargo build/test (26: 18 aura-core + 3 aura-std + 5 aura-engine)/
clippy -D warnings clean; surface-purity grep (no dyn-Any / Rc / RefCell) clean.
refs walking-skeleton
This commit is contained in:
@@ -9,3 +9,6 @@ publish.workspace = true
|
||||
aura-core = { path = "../aura-core" }
|
||||
# data-server (the first data source) enters when the ingestion task starts;
|
||||
# kept out of the bare skeleton so the workspace compiles without the network.
|
||||
|
||||
[dev-dependencies]
|
||||
aura-std = { path = "../aura-std" }
|
||||
|
||||
@@ -1,13 +1,23 @@
|
||||
//! `aura-engine` — the headless, UI-agnostic reactive SoA engine.
|
||||
//!
|
||||
//! Home of (to be specified): the ingestion boundary (k-way merge of timestamped
|
||||
//! sources into one chronological cycle stream), the deterministic synchronous
|
||||
//! single-threaded sim loop (one unique state per input tick), freshness-gated
|
||||
//! recompute, the strategy's broker-independent position-event output and the
|
||||
//! downstream broker nodes that project it into equity (sim-optimal pip /
|
||||
//! realistic currency; several attachable at once for comparable curves), and
|
||||
//! the atomic sim unit
|
||||
//! `(topology + params + data-window + seed) -> metrics` that the sweep /
|
||||
//! optimize / walk-forward / Monte-Carlo axes orchestrate.
|
||||
//! Delivered in cycle 0003 — the deterministic single-source sim loop:
|
||||
//!
|
||||
//! - [`Sim`] — a bootstrapped, frozen root graph (a flat node array + an index
|
||||
//! edge table, topologically ordered) and its deterministic `run` loop: one
|
||||
//! source driven through a wired DAG of nodes, cycle by cycle, each output
|
||||
//! forwarded into its consumers' input columns (C1/C4/C7/C8/C9).
|
||||
//! - [`Edge`] / [`Target`] — producer->consumer and source->consumer wiring.
|
||||
//! - [`BootstrapError`] — wiring faults caught once, at bootstrap (kind
|
||||
//! mismatch, bad index, directed cycle).
|
||||
//!
|
||||
//! Still to come (subsequent cycles): freshness-gated recompute / sample-and-hold
|
||||
//! (C5), the ingestion boundary (k-way merge of timestamped sources, C3), the
|
||||
//! broker-independent position-event output and downstream broker nodes (C10),
|
||||
//! and the atomic sim unit `(topology + params + data-window + seed) -> metrics`
|
||||
//! that the sweep / optimize / walk-forward / Monte-Carlo axes orchestrate.
|
||||
//!
|
||||
//! Visualization is never here: it is a downstream consumer node on the streams.
|
||||
|
||||
mod sim;
|
||||
|
||||
pub use sim::{BootstrapError, Edge, Sim, Target};
|
||||
|
||||
@@ -0,0 +1,327 @@
|
||||
//! The deterministic single-source sim loop. A `Sim` is a bootstrapped, frozen
|
||||
//! root graph — a flat node array plus an index edge table, topologically ordered
|
||||
//! — driven cycle by cycle by one source. It is the flat, monomorphized sharpening
|
||||
//! of RustAst's reference-counted, interior-mutable observer push graph: no
|
||||
//! reference counting, no interior mutability, no per-cycle allocation (C1/C7).
|
||||
//! Each node owns its input
|
||||
//! columns (the cycle-0002 shape), so `Ctx` is unchanged; a producer's `eval`
|
||||
//! output is forwarded into its consumers' input columns, and a `None` forwards
|
||||
//! nothing (the structural seed of sample-and-hold, realized with freshness in a
|
||||
//! later cycle).
|
||||
|
||||
use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
|
||||
|
||||
/// A producer-output -> consumer-input-slot forwarding edge.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub struct Edge {
|
||||
pub from: usize,
|
||||
pub to: usize,
|
||||
pub slot: usize,
|
||||
}
|
||||
|
||||
/// An input slot the source value is forwarded into each cycle.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub struct Target {
|
||||
pub node: usize,
|
||||
pub slot: usize,
|
||||
}
|
||||
|
||||
/// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
|
||||
/// generalized to the whole topology.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum BootstrapError {
|
||||
/// An edge or source-target connects mismatched scalar kinds.
|
||||
KindMismatch { producer: ScalarKind, consumer: ScalarKind },
|
||||
/// A node or slot index in an edge or target (or the observe index) is out of range.
|
||||
BadIndex,
|
||||
/// The wiring contains a directed cycle (only an explicit delay node may close
|
||||
/// a loop; that node does not exist yet).
|
||||
Cycle,
|
||||
}
|
||||
|
||||
struct NodeBox {
|
||||
node: Box<dyn Node>,
|
||||
inputs: Vec<AnyColumn>,
|
||||
}
|
||||
|
||||
/// A bootstrapped, frozen root graph instance plus its deterministic run loop.
|
||||
pub struct Sim {
|
||||
nodes: Vec<NodeBox>,
|
||||
topo: Vec<usize>,
|
||||
out_edges: Vec<Vec<Edge>>,
|
||||
source_targets: Vec<Target>,
|
||||
observe: usize,
|
||||
}
|
||||
|
||||
// Manual, node-opaque `Debug`: `Box<dyn Node>` is not `Debug`, so the struct
|
||||
// cannot derive it. This summary form is enough for `Result::unwrap_err` (which
|
||||
// formats the `Ok` arm on a failed bootstrap-rejection assertion) without
|
||||
// printing node internals.
|
||||
impl core::fmt::Debug for Sim {
|
||||
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
|
||||
f.debug_struct("Sim")
|
||||
.field("nodes", &self.nodes.len())
|
||||
.field("topo", &self.topo)
|
||||
.field("out_edges", &self.out_edges)
|
||||
.field("source_targets", &self.source_targets)
|
||||
.field("observe", &self.observe)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl Sim {
|
||||
/// Bind nodes + wiring into a frozen, runnable graph. Sizes each node's input
|
||||
/// columns from its `schema`, kind-checks every edge and source target, and
|
||||
/// topologically orders the nodes (Kahn), rejecting any directed cycle.
|
||||
pub fn bootstrap(
|
||||
nodes: Vec<Box<dyn Node>>,
|
||||
source_targets: Vec<Target>,
|
||||
edges: Vec<Edge>,
|
||||
source_kind: ScalarKind,
|
||||
observe: usize,
|
||||
) -> Result<Sim, BootstrapError> {
|
||||
let n = nodes.len();
|
||||
if observe >= n {
|
||||
return Err(BootstrapError::BadIndex);
|
||||
}
|
||||
|
||||
let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect();
|
||||
|
||||
// size each node's own input columns from its schema
|
||||
let mut boxes: Vec<NodeBox> = Vec::with_capacity(n);
|
||||
for (nd, schema) in nodes.into_iter().zip(schemas.iter()) {
|
||||
let inputs: Vec<AnyColumn> = schema
|
||||
.inputs
|
||||
.iter()
|
||||
.map(|spec| AnyColumn::with_capacity(spec.kind, spec.lookback))
|
||||
.collect();
|
||||
boxes.push(NodeBox { node: nd, inputs });
|
||||
}
|
||||
|
||||
// source targets: the source value must match each target slot's kind
|
||||
for t in &source_targets {
|
||||
let s = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?;
|
||||
let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
|
||||
if slot.kind != source_kind {
|
||||
return Err(BootstrapError::KindMismatch {
|
||||
producer: source_kind,
|
||||
consumer: slot.kind,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// edges: indices in range, producer output kind == consumer slot kind
|
||||
let mut out_edges: Vec<Vec<Edge>> = vec![Vec::new(); n];
|
||||
for &e in &edges {
|
||||
let from = schemas.get(e.from).ok_or(BootstrapError::BadIndex)?;
|
||||
let to = schemas.get(e.to).ok_or(BootstrapError::BadIndex)?;
|
||||
let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?;
|
||||
if from.output != slot.kind {
|
||||
return Err(BootstrapError::KindMismatch {
|
||||
producer: from.output,
|
||||
consumer: slot.kind,
|
||||
});
|
||||
}
|
||||
out_edges[e.from].push(e);
|
||||
}
|
||||
|
||||
// Kahn topological sort; a leftover node means a cycle
|
||||
let mut indeg = vec![0usize; n];
|
||||
for &e in &edges {
|
||||
indeg[e.to] += 1;
|
||||
}
|
||||
let mut queue: Vec<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
|
||||
let mut topo: Vec<usize> = Vec::with_capacity(n);
|
||||
let mut head = 0;
|
||||
while head < queue.len() {
|
||||
let u = queue[head];
|
||||
head += 1;
|
||||
topo.push(u);
|
||||
for e in &out_edges[u] {
|
||||
indeg[e.to] -= 1;
|
||||
if indeg[e.to] == 0 {
|
||||
queue.push(e.to);
|
||||
}
|
||||
}
|
||||
}
|
||||
if topo.len() != n {
|
||||
return Err(BootstrapError::Cycle);
|
||||
}
|
||||
|
||||
Ok(Sim {
|
||||
nodes: boxes,
|
||||
topo,
|
||||
out_edges,
|
||||
source_targets,
|
||||
observe,
|
||||
})
|
||||
}
|
||||
|
||||
/// Drive the records in timestamp order; returns the observed node's per-cycle
|
||||
/// output. Allocates nothing on the per-cycle eval/forward path.
|
||||
pub fn run(
|
||||
&mut self,
|
||||
records: impl Iterator<Item = (Timestamp, Scalar)>,
|
||||
) -> Vec<Option<Scalar>> {
|
||||
// disjoint field borrows so the topo walk can read `topo`/`out_edges`
|
||||
// while mutating `nodes`
|
||||
let Sim { nodes, topo, out_edges, source_targets, observe } = self;
|
||||
let observe = *observe;
|
||||
|
||||
let mut out = Vec::new();
|
||||
for (_ts, value) in records {
|
||||
// forward the source value into its target input slots
|
||||
for t in source_targets.iter() {
|
||||
nodes[t.node].inputs[t.slot]
|
||||
.push(value)
|
||||
.expect("source kind checked at wiring");
|
||||
}
|
||||
|
||||
// evaluate in topological order; capture the observed output; forward
|
||||
let mut observed = None;
|
||||
for &nidx in topo.iter() {
|
||||
let result = {
|
||||
let nb = &mut nodes[nidx];
|
||||
nb.node.eval(Ctx::new(&nb.inputs))
|
||||
};
|
||||
if nidx == observe {
|
||||
observed = result;
|
||||
}
|
||||
if let Some(v) = result {
|
||||
for e in out_edges[nidx].iter() {
|
||||
nodes[e.to].inputs[e.slot]
|
||||
.push(v)
|
||||
.expect("edge kind checked at wiring");
|
||||
}
|
||||
}
|
||||
}
|
||||
out.push(observed);
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aura_std::{Sma, Sub};
|
||||
|
||||
fn f64_records(prices: &[f64]) -> impl Iterator<Item = (Timestamp, Scalar)> + '_ {
|
||||
prices
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, &p)| (Timestamp(i as i64), Scalar::F64(p)))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chain_source_sma_runs() {
|
||||
// node 0 = SMA(3); source -> SMA(3).in0; observe node 0
|
||||
let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(3))];
|
||||
let mut sim = Sim::bootstrap(
|
||||
nodes,
|
||||
vec![Target { node: 0, slot: 0 }],
|
||||
vec![],
|
||||
ScalarKind::F64,
|
||||
0,
|
||||
)
|
||||
.expect("valid");
|
||||
let out = sim.run(f64_records(&[1.0, 2.0, 3.0, 4.0, 5.0]));
|
||||
assert_eq!(
|
||||
out,
|
||||
vec![
|
||||
None,
|
||||
None,
|
||||
Some(Scalar::F64(2.0)),
|
||||
Some(Scalar::F64(3.0)),
|
||||
Some(Scalar::F64(4.0)),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fan_out_join_dag_runs_deterministically() {
|
||||
// 0 = SMA(2), 1 = SMA(4), 2 = Sub; source fans into both SMAs; SMAs join into Sub
|
||||
let build = || -> Sim {
|
||||
let nodes: Vec<Box<dyn Node>> =
|
||||
vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())];
|
||||
Sim::bootstrap(
|
||||
nodes,
|
||||
vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
|
||||
vec![Edge { from: 0, to: 2, slot: 0 }, Edge { from: 1, to: 2, slot: 1 }],
|
||||
ScalarKind::F64,
|
||||
2,
|
||||
)
|
||||
.expect("valid DAG")
|
||||
};
|
||||
|
||||
let prices = [10.0_f64, 12.0, 14.0, 16.0, 18.0, 20.0];
|
||||
|
||||
let mut sim = build();
|
||||
let out = sim.run(f64_records(&prices));
|
||||
// Sub fires only once SMA(4) is warm (cycle index 3): 15-13, 17-15, 19-17 -> 2.
|
||||
assert_eq!(
|
||||
out,
|
||||
vec![
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
Some(Scalar::F64(2.0)),
|
||||
Some(Scalar::F64(2.0)),
|
||||
Some(Scalar::F64(2.0)),
|
||||
]
|
||||
);
|
||||
|
||||
// determinism (C1): a second identical run is bit-identical
|
||||
let mut sim2 = build();
|
||||
let out2 = sim2.run(f64_records(&prices));
|
||||
assert_eq!(out, out2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bootstrap_rejects_a_cycle() {
|
||||
// two SMA(1) nodes wired a -> b -> a
|
||||
let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))];
|
||||
let err = Sim::bootstrap(
|
||||
nodes,
|
||||
vec![],
|
||||
vec![Edge { from: 0, to: 1, slot: 0 }, Edge { from: 1, to: 0, slot: 0 }],
|
||||
ScalarKind::F64,
|
||||
0,
|
||||
)
|
||||
.unwrap_err();
|
||||
assert_eq!(err, BootstrapError::Cycle);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bootstrap_rejects_a_kind_mismatch() {
|
||||
// SMA(1) declares an f64 input; feeding an i64 source mismatches
|
||||
let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(1))];
|
||||
let err = Sim::bootstrap(
|
||||
nodes,
|
||||
vec![Target { node: 0, slot: 0 }],
|
||||
vec![],
|
||||
ScalarKind::I64,
|
||||
0,
|
||||
)
|
||||
.unwrap_err();
|
||||
assert_eq!(
|
||||
err,
|
||||
BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 }
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bootstrap_rejects_a_bad_index() {
|
||||
// observe node 5 does not exist
|
||||
let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(1))];
|
||||
let err = Sim::bootstrap(
|
||||
nodes,
|
||||
vec![Target { node: 0, slot: 0 }],
|
||||
vec![],
|
||||
ScalarKind::F64,
|
||||
5,
|
||||
)
|
||||
.unwrap_err();
|
||||
assert_eq!(err, BootstrapError::BadIndex);
|
||||
}
|
||||
}
|
||||
@@ -16,4 +16,6 @@
|
||||
//! average — a worked producer node proving the `aura-core` `Node` contract.
|
||||
|
||||
mod sma;
|
||||
mod sub;
|
||||
pub use sma::Sma;
|
||||
pub use sub::Sub;
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
//! `Sub` — two-input f64 difference (input 0 minus input 1), e.g. a fast/slow
|
||||
//! spread. The walking skeleton's second worked node: it gives the sim loop a
|
||||
//! real fan-out + join to run (two SMAs joining into one node), exercising
|
||||
//! multi-input `Ctx` access inside a running graph.
|
||||
|
||||
use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
|
||||
|
||||
/// Two-input f64 difference: input 0 minus input 1. Emits `None` until both
|
||||
/// inputs have a value.
|
||||
#[derive(Default)]
|
||||
pub struct Sub;
|
||||
|
||||
impl Sub {
|
||||
/// Build a `Sub` node.
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
impl Node for Sub {
|
||||
fn schema(&self) -> NodeSchema {
|
||||
NodeSchema {
|
||||
inputs: vec![
|
||||
InputSpec { kind: ScalarKind::F64, lookback: 1 },
|
||||
InputSpec { kind: ScalarKind::F64, lookback: 1 },
|
||||
],
|
||||
output: ScalarKind::F64,
|
||||
}
|
||||
}
|
||||
|
||||
fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar> {
|
||||
let a = ctx.f64_in(0);
|
||||
let b = ctx.f64_in(1);
|
||||
if a.is_empty() || b.is_empty() {
|
||||
return None;
|
||||
}
|
||||
Some(Scalar::F64(a[0] - b[0]))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aura_core::AnyColumn;
|
||||
|
||||
#[test]
|
||||
fn sub_is_difference_once_both_inputs_present() {
|
||||
let mut sub = Sub::new();
|
||||
let mut inputs = vec![
|
||||
AnyColumn::with_capacity(ScalarKind::F64, 1),
|
||||
AnyColumn::with_capacity(ScalarKind::F64, 1),
|
||||
];
|
||||
|
||||
// only input 0 present -> None
|
||||
inputs[0].push(Scalar::F64(10.0)).unwrap();
|
||||
assert_eq!(sub.eval(Ctx::new(&inputs)), None);
|
||||
|
||||
// both present -> a - b
|
||||
inputs[1].push(Scalar::F64(4.0)).unwrap();
|
||||
assert_eq!(sub.eval(Ctx::new(&inputs)), Some(Scalar::F64(6.0)));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user