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

feat(aura-engine): Source ingestion seam — run() takes Vec<Box<dyn Source>>

Browse Source 
The engine's ingestion input was the only type encoding an eager-dataflow
assumption: `Harness::run(Vec<Vec<(Timestamp, Scalar)>>)` — one fully
materialized stream per source. At realistic research scale (20y, 3 tick
streams, ~7.9e9 ticks) that layout is ~190 GB resident, non-residable. Yet
the merge loop never needs the whole stream: it only ever peeks the head
timestamp of each live source and pops one record. So the eager Vec is
gratuitous.

This lands the first half of the Source seam (plan Tasks 1-2):

  - A `Source` trait (`peek(&self) -> Option<Timestamp>`,
    `next(&mut) -> Option<(Timestamp, Scalar)>`), object-safe so the merge
    holds `Vec<Box<dyn Source>>`. The peek/next split is the producer contract
    the k-way merge drives.
  - `VecSource`, an adapter over the old `Vec<(Timestamp, Scalar)>` — the
    cycle-0011 eager shape, named. It IS the old behaviour.
  - `Harness::run` re-typed to `Vec<Box<dyn Source>>`; the merge loop's index
    arithmetic becomes peek/next. C4 tie-by-source-index is preserved (the
    strictly-`<` replace + source-order scan is byte-for-byte the old
    semantics). The forward+eval body is unchanged.
  - All 53 call sites threaded VecSource-wrapped in the same change, so the
    workspace compiles atomically and run output is byte-identical (C1).

Behaviour-preserving is the load-bearing guarantee: the full suite is green
and unchanged (the two-run C1 determinism pairs — real_bars r1/r2, blueprint
sweep-equality, harness h/h2 — stay byte-identical); aura-engine unit tests
129 -> 132 (+3 VecSource unit tests). No residency change yet: VecSource holds
the same Vec as before. The lazy data-server-backed streaming source and its
end-to-end residency proof are the second half (plan Tasks 3-4).

refs #71
This commit is contained in:
Brummel
2026-06-15 10:23:09 +02:00
parent 9337a8585d
commit eec129ee51
7 changed files with 156 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/main.rs
+5 -5
View File
@@ -11,7 +11,7 @@ mod render;
use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
f64_field, summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness,
RunManifest, RunReport, SourceSpec, SweepFamily, Target,
RunManifest, RunReport, SourceSpec, SweepFamily, Target, VecSource,
};
use aura_registry::{rank_by, Registry};
use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub};
@@ -137,7 +137,7 @@ fn run_sample() -> RunReport {
prices.first().expect("non-empty stream").0,
prices.last().expect("non-empty stream").0,
);
h.run(vec![prices]);
h.run(vec![Box::new(VecSource::new(prices))]);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
@@ -280,7 +280,7 @@ fn sweep_family() -> SweepFamily {
prices.first().expect("non-empty stream").0,
prices.last().expect("non-empty stream").0,
);
h.run(vec![prices]);
h.run(vec![Box::new(VecSource::new(prices))]);
let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
let params = space
@@ -459,7 +459,7 @@ fn run_macd() -> RunReport {
prices.first().expect("non-empty stream").0,
prices.last().expect("non-empty stream").0,
);
h.run(vec![prices]);
h.run(vec![Box::new(VecSource::new(prices))]);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
@@ -524,7 +524,7 @@ mod tests {
.with("exposure.scale", 0.5)
.bootstrap()
.expect("sample blueprint compiles under a valid point");
h.run(vec![showcase_prices()]);
h.run(vec![Box::new(VecSource::new(showcase_prices()))]);
assert!(!rx_eq.try_iter().collect::<Vec<_>>().is_empty(), "equity sink drained empty");
assert!(!rx_ex.try_iter().collect::<Vec<_>>().is_empty(), "exposure sink drained empty");
}
crates/aura-engine/src/blueprint.rs
+11 -10
View File
@@ -812,6 +812,7 @@ fn slot_kind(t: Target, flat_signatures: &[NodeSchema]) -> Result<ScalarKind, Co
mod tests {
use super::*;
use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices};
use crate::VecSource;
use aura_core::{Ctx, FieldSpec, Firing, NodeSchema, Timestamp};
use aura_std::{Ema, Exposure, Recorder, SimBroker, Sma, Sub};
use std::sync::mpsc;
@@ -1020,7 +1021,7 @@ mod tests {
.with("exposure.scale", 0.5)
.bootstrap()
.expect("named binding resolves and bootstraps");
named.run(vec![synthetic_prices()]);
named.run(vec![Box::new(VecSource::new(synthetic_prices()))]);
let named_eq = comp_eq.try_iter().collect::<Vec<_>>();
let named_ex = comp_ex.try_iter().collect::<Vec<_>>();
@@ -1028,7 +1029,7 @@ mod tests {
let mut positional = bp2
.bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
.expect("positional bootstrap");
positional.run(vec![synthetic_prices()]);
positional.run(vec![Box::new(VecSource::new(synthetic_prices()))]);
let pos_eq_v = pos_eq.try_iter().collect::<Vec<_>>();
let pos_ex_v = pos_ex.try_iter().collect::<Vec<_>>();
@@ -1739,14 +1740,14 @@ mod tests {
// (a) today's flat, hand-wired graph
let (mut flat, flat_eq, flat_ex) = hand_wired_sma_cross_harness();
flat.run(vec![prices.clone()]);
flat.run(vec![Box::new(VecSource::new(prices.clone()))]);
// (b) the same graph authored as a composite blueprint, compiled
let (bp, comp_eq, comp_ex) = composite_sma_cross_harness();
let mut composed = bp
.bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
.expect("composite blueprint compiles");
composed.run(vec![prices]);
composed.run(vec![Box::new(VecSource::new(prices))]);
let flat_eq_v = flat_eq.try_iter().collect::<Vec<_>>();
let flat_ex_v = flat_ex.try_iter().collect::<Vec<_>>();
@@ -1809,7 +1810,7 @@ mod tests {
let mut h = bp
.bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4)])
.expect("multi-output composite bootstraps");
h.run(vec![prices]);
h.run(vec![Box::new(VecSource::new(prices))]);
let a = rx_a.try_iter().collect::<Vec<_>>();
let b = rx_b.try_iter().collect::<Vec<_>>();
@@ -1826,13 +1827,13 @@ mod tests {
let (bp, eq, _ex) = composite_sma_cross_harness();
let mut a = bp.bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
.expect("compiles");
a.run(vec![prices.clone()]);
a.run(vec![Box::new(VecSource::new(prices.clone()))]);
let a_eq = eq.try_iter().collect::<Vec<_>>();
let (bp2, eq2, _ex2) = composite_sma_cross_harness();
let mut b = bp2.bootstrap_with_params(vec![Scalar::I64(5), Scalar::I64(20), Scalar::F64(1.0)])
.expect("compiles");
b.run(vec![prices]);
b.run(vec![Box::new(VecSource::new(prices))]);
let b_eq = eq2.try_iter().collect::<Vec<_>>();
assert!(!a_eq.is_empty() && !b_eq.is_empty(), "both traces populated");
@@ -1870,11 +1871,11 @@ mod tests {
let (bp, eq, _ex) = composite_sma_cross_harness();
let mut a = bp.bootstrap_with_params(vec![Scalar::I64(3), Scalar::I64(9), Scalar::F64(0.7)])
.expect("compiles");
a.run(vec![prices.clone()]);
a.run(vec![Box::new(VecSource::new(prices.clone()))]);
let (bp2, eq2, _ex2) = composite_sma_cross_harness();
let mut b = bp2.bootstrap_with_params(vec![Scalar::I64(3), Scalar::I64(9), Scalar::F64(0.7)])
.expect("compiles");
b.run(vec![prices]);
b.run(vec![Box::new(VecSource::new(prices))]);
assert_eq!(eq.try_iter().collect::<Vec<_>>(), eq2.try_iter().collect::<Vec<_>>());
}
@@ -2392,7 +2393,7 @@ mod tests {
.with("exposure.scale", 0.5)
.bootstrap()
.expect("named (injective) cross resolves by name and bootstraps");
h.run(vec![prices]);
h.run(vec![Box::new(VecSource::new(prices))]);
let ex = rx_ex.try_iter().collect::<Vec<_>>();
assert!(!ex.is_empty(), "the cured, by-name-bound cross must run to a populated exposure trace");
}
crates/aura-engine/src/harness.rs
+131 -66
View File
@@ -51,6 +51,45 @@ pub struct SourceSpec {
pub targets: Vec<Target>,
}
/// The ingestion-boundary producer the k-way merge drives (C3). A source yields
/// timestamped scalars in ascending timestamp order (the C3 ingestion
/// precondition). Object-safe: the merge holds `Vec<Box<dyn Source>>`.
pub trait Source {
/// The timestamp of the head record, without consuming it. `None` = the
/// source is exhausted. Cheap and side-effect-free: it reads a pre-decoded
/// head, never advancing the underlying stream.
fn peek(&self) -> Option<Timestamp>;
/// Pop the head `(timestamp, scalar)` and advance. `None` = exhausted.
/// After `next` returns `Some`, `peek` reflects the new head.
fn next(&mut self) -> Option<(Timestamp, Scalar)>;
}
/// A `Source` over a fully materialized stream — the cycle-0011 eager shape,
/// named. Every pre-seam call site wraps its `Vec<(Timestamp, Scalar)>` in this,
/// so run output is byte-identical (C1) and residency is unchanged.
pub struct VecSource {
stream: Vec<(Timestamp, Scalar)>,
cursor: usize,
}
impl VecSource {
pub fn new(stream: Vec<(Timestamp, Scalar)>) -> Self {
Self { stream, cursor: 0 }
}
}
impl Source for VecSource {
fn peek(&self) -> Option<Timestamp> {
self.stream.get(self.cursor).map(|&(t, _)| t)
}
fn next(&mut self) -> Option<(Timestamp, Scalar)> {
let item = self.stream.get(self.cursor).copied()?;
self.cursor += 1;
Some(item)
}
}
/// The flat, type-erased, index-wired output of `Composite::compile` — the target
/// `Harness::bootstrap` consumes (C23's "flat graph", now a named type). `signatures[i]`
/// is the static signature of `nodes[i]` (gathered from each primitive's builder at
@@ -217,12 +256,12 @@ impl Harness {
/// its record to a destination it holds (out of graph) inside `eval`; the
/// engine only routes in-graph edges and is oblivious to the side effect.
/// Allocates nothing per cycle beyond the reused scratch buffer.
pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) {
pub fn run(&mut self, mut sources_in: Vec<Box<dyn Source>>) {
assert_eq!(
streams.len(),
sources_in.len(),
self.sources.len(),
"run: one stream per source required (got {} streams for {} sources)",
streams.len(),
"run: one source per SourceSpec required (got {} sources for {} specs)",
sources_in.len(),
self.sources.len()
);
@@ -230,31 +269,31 @@ impl Harness {
// while mutating nodes
let Harness { nodes, topo, out_edges, sources } = self;
let mut cursor: Vec<usize> = vec![0; streams.len()];
let mut cycle_id: u64 = 0;
let mut scratch: Vec<Scalar> = Vec::new();
loop {
// pick the live source head with the smallest (timestamp, source index)
let mut pick: Option<usize> = None;
for (s, stream) in streams.iter().enumerate() {
if cursor[s] < stream.len() {
// pick the live source head with the smallest (timestamp, source index).
// strictly-`<` replace + source-order scan preserves C4 tie-breaking.
let mut pick: Option<(usize, Timestamp)> = None;
for (s, src) in sources_in.iter().enumerate() {
if let Some(ts) = src.peek() {
match pick {
None => pick = Some(s),
Some(p) => {
if stream[cursor[s]].0 < streams[p][cursor[p]].0 {
pick = Some(s);
}
}
None => pick = Some((s, ts)),
Some((_, best)) if ts < best => pick = Some((s, ts)),
_ => {}
}
}
}
let s = match pick {
Some(s) => s,
None => break, // all streams exhausted
Some((s, _)) => s,
None => break, // all sources exhausted
};
let (ts, value) = streams[s][cursor[s]];
cursor[s] += 1;
// `&mut *sources_in[s]`: deref the Box to `dyn Source` then re-borrow,
// so UFCS resolves Self = dyn Source (Box<dyn Source> does not itself
// impl Source). Fully-qualified to never collide with Iterator::next.
let (ts, value) =
Source::next(&mut *sources_in[s]).expect("peeked Some ⇒ next is Some");
cycle_id += 1;
// forward the source value into its target slots, stamping freshness
@@ -356,6 +395,32 @@ mod tests {
points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect()
}
#[test]
fn vec_source_peek_is_non_consuming_and_idempotent() {
let mut s = VecSource::new(vec![(Timestamp(1), Scalar::F64(10.0)), (Timestamp(2), Scalar::F64(20.0))]);
// peek twice: same head, no advance.
assert_eq!(s.peek(), Some(Timestamp(1)));
assert_eq!(s.peek(), Some(Timestamp(1)));
// next pops it; peek now reflects the new head.
assert_eq!(Source::next(&mut s), Some((Timestamp(1), Scalar::F64(10.0))));
assert_eq!(s.peek(), Some(Timestamp(2)));
}
#[test]
fn vec_source_exhaustion_yields_none() {
let mut s = VecSource::new(vec![(Timestamp(5), Scalar::I64(7))]);
assert_eq!(Source::next(&mut s), Some((Timestamp(5), Scalar::I64(7))));
assert_eq!(s.peek(), None);
assert_eq!(Source::next(&mut s), None);
}
#[test]
fn vec_source_empty_peeks_none() {
let mut s = VecSource::new(vec![]);
assert_eq!(s.peek(), None);
assert_eq!(Source::next(&mut s), None);
}
/// 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 {
@@ -591,7 +656,7 @@ mod tests {
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid");
h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]);
h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])))]);
let got: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// SMA(3) warms at cycle 3; the recorder captures only fired cycles, each
// tagged with the cycle's timestamp (sparse — no None hold-rows).
@@ -639,7 +704,7 @@ mod tests {
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![prices.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone()))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
assert_eq!(
@@ -654,7 +719,7 @@ mod tests {
// determinism (C1): a second identical run drains a bit-identical stream.
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![prices]);
h2.run(vec![Box::new(VecSource::new(prices))]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out);
}
@@ -686,7 +751,7 @@ mod tests {
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![s0.clone(), s1.clone()]);
h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm.
assert_eq!(
@@ -701,7 +766,7 @@ mod tests {
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![s0, s1]);
h2.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out); // deterministic
}
@@ -732,7 +797,7 @@ mod tests {
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![s0.clone(), s1.clone()]);
h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// records ONLY at t=2 and t=4 where both inputs share the timestamp.
assert_eq!(
@@ -745,7 +810,7 @@ mod tests {
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![s0, s1]);
h2.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out); // deterministic
}
@@ -786,7 +851,7 @@ mod tests {
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![prices.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone()))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// SMA(4) warms at cycle 4; from then both paths emit each cycle at the same
// timestamp, so the barrier fires: SMA(2)+SMA(4) = 15+13, 17+15, 19+17.
@@ -801,7 +866,7 @@ mod tests {
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![prices]);
h2.run(vec![Box::new(VecSource::new(prices))]);
let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
assert_eq!(out2, out);
}
@@ -831,7 +896,7 @@ mod tests {
let s1 = f64_stream(&[(2, 200.0)]); // in1 (barrier)
let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]); // in2 (as-of)
h.run(vec![s0, s1, s2]);
h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1)), Box::new(VecSource::new(s2))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// c3: barrier pair completes at t=2, holds c=1 -> 221. c4: as-of input
// ticks at t=3, holds the pair -> 223. c1 filters; c2,c5 hold (no record).
@@ -897,13 +962,13 @@ mod tests {
}
/// Build five timestamp-aligned f64 sources feeding Ohlcv's five barrier slots.
fn ohlcv_streams() -> Vec<Vec<(Timestamp, Scalar)>> {
fn ohlcv_streams() -> Vec<Box<dyn Source>> {
vec![
f64_stream(&[(1, 10.0), (2, 20.0)]), // open
f64_stream(&[(1, 15.0), (2, 25.0)]), // high
f64_stream(&[(1, 8.0), (2, 19.0)]), // low
f64_stream(&[(1, 12.0), (2, 22.0)]), // close
f64_stream(&[(1, 100.0), (2, 200.0)]), // volume
Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 20.0)]))), // open
Box::new(VecSource::new(f64_stream(&[(1, 15.0), (2, 25.0)]))), // high
Box::new(VecSource::new(f64_stream(&[(1, 8.0), (2, 19.0)]))), // low
Box::new(VecSource::new(f64_stream(&[(1, 12.0), (2, 22.0)]))), // close
Box::new(VecSource::new(f64_stream(&[(1, 100.0), (2, 200.0)]))), // volume
]
}
@@ -1119,7 +1184,7 @@ mod tests {
],
)
.expect("valid DAG");
h.run(vec![f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])]);
h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])))]);
let fast: Vec<(Timestamp, Vec<Scalar>)> = rx_fast.try_iter().collect();
let slow: Vec<(Timestamp, Vec<Scalar>)> = rx_slow.try_iter().collect();
// SMA(2) warms at cycle 2, SMA(4) at cycle 4 — two different-rate streams.
@@ -1170,11 +1235,11 @@ mod tests {
)
.expect("valid");
h.run(vec![
f64_stream(&[(1, 10.0)]),
f64_stream(&[(1, 15.0)]),
f64_stream(&[(1, 8.0)]),
f64_stream(&[(1, 12.0)]),
f64_stream(&[(1, 100.0)]),
Box::new(VecSource::new(f64_stream(&[(1, 10.0)]))),
Box::new(VecSource::new(f64_stream(&[(1, 15.0)]))),
Box::new(VecSource::new(f64_stream(&[(1, 8.0)]))),
Box::new(VecSource::new(f64_stream(&[(1, 12.0)]))),
Box::new(VecSource::new(f64_stream(&[(1, 100.0)]))),
]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
assert_eq!(out.len(), 1);
@@ -1217,10 +1282,10 @@ mod tests {
)
.expect("valid");
h.run(vec![
vec![(Timestamp(1), Scalar::I64(7))],
vec![(Timestamp(2), Scalar::F64(1.5))],
vec![(Timestamp(3), Scalar::Bool(true))],
vec![(Timestamp(4), Scalar::Ts(Timestamp(99)))],
Box::new(VecSource::new(vec![(Timestamp(1), Scalar::I64(7))])),
Box::new(VecSource::new(vec![(Timestamp(2), Scalar::F64(1.5))])),
Box::new(VecSource::new(vec![(Timestamp(3), Scalar::Bool(true))])),
Box::new(VecSource::new(vec![(Timestamp(4), Scalar::Ts(Timestamp(99)))])),
]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
assert_eq!(
@@ -1254,7 +1319,7 @@ mod tests {
vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
)
.expect("valid");
h.run(vec![f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])]);
h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])))]);
let tapped: Vec<(Timestamp, Vec<Scalar>)> = rx_tap.try_iter().collect();
let downstream: Vec<(Timestamp, Vec<Scalar>)> = rx_down.try_iter().collect();
let expected = vec![
@@ -1289,12 +1354,12 @@ mod tests {
let (tx_a, rx_a) = mpsc::channel();
let mut a = build(tx_a);
a.run(vec![prices.clone()]);
a.run(vec![Box::new(VecSource::new(prices.clone()))]);
let run_a: Vec<(Timestamp, Vec<Scalar>)> = rx_a.try_iter().collect();
let (tx_b, rx_b) = mpsc::channel();
let mut b = build(tx_b);
b.run(vec![prices]);
b.run(vec![Box::new(VecSource::new(prices))]);
let run_b: Vec<(Timestamp, Vec<Scalar>)> = rx_b.try_iter().collect();
assert_eq!(run_a, run_b);
@@ -1324,7 +1389,7 @@ mod tests {
.expect("valid");
let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
h.run(vec![s0, s1]);
h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// records ONLY at t=2 and t=4 (both inputs coincide); holds otherwise.
assert_eq!(
@@ -1359,7 +1424,7 @@ mod tests {
.expect("valid");
let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
h.run(vec![s0, s1]);
h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// from t=2 on, records every cycle holding the stale input; two cycles fall
// on t=4 (the s0 tick then the s1 tick).
@@ -1444,7 +1509,7 @@ mod tests {
let (b1, rb) = mpsc::channel();
let (c1, rc) = mpsc::channel();
let mut h = build(a1, b1, c1);
h.run(vec![prices.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone()))]);
let s1: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
let s2: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
let s3: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
@@ -1463,7 +1528,7 @@ mod tests {
let (b2, rb2) = mpsc::channel();
let (c2, rc2) = mpsc::channel();
let mut h2 = build(a2, b2, c2);
h2.run(vec![prices]);
h2.run(vec![Box::new(VecSource::new(prices))]);
assert_eq!(ra2.try_iter().collect::<Vec<_>>(), expected);
assert_eq!(rb2.try_iter().collect::<Vec<_>>(), expected);
assert_eq!(rc2.try_iter().collect::<Vec<_>>(), expected);
@@ -1509,7 +1574,7 @@ mod tests {
let (tr, rr) = mpsc::channel();
let (ts, rs) = mpsc::channel();
let mut h = build(tr, ts);
h.run(vec![prices.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone()))]);
let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
// SMA(2): 11,13,15,17,19 at cycles 2..6 (raw tap).
@@ -1532,7 +1597,7 @@ mod tests {
let (tr2, rr2) = mpsc::channel();
let (ts2, rs2) = mpsc::channel();
let mut h2 = build(tr2, ts2);
h2.run(vec![prices]);
h2.run(vec![Box::new(VecSource::new(prices))]);
assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
}
@@ -1576,7 +1641,7 @@ mod tests {
let (ta, rany) = mpsc::channel();
let (tb, rbar) = mpsc::channel();
let mut h = build(ta, tb);
h.run(vec![a.clone(), b.clone()]);
h.run(vec![Box::new(VecSource::new(a.clone())), Box::new(VecSource::new(b.clone()))]);
let any: Vec<(Timestamp, Vec<Scalar>)> = rany.try_iter().collect();
let bar: Vec<(Timestamp, Vec<Scalar>)> = rbar.try_iter().collect();
// As-of tap records every SMA(2) fire: 11@t2, 13@t3, 15@t4.
@@ -1597,7 +1662,7 @@ mod tests {
let (ta2, rany2) = mpsc::channel();
let (tb2, rbar2) = mpsc::channel();
let mut h2 = build(ta2, tb2);
h2.run(vec![a, b]);
h2.run(vec![Box::new(VecSource::new(a)), Box::new(VecSource::new(b))]);
assert_eq!(rany2.try_iter().collect::<Vec<_>>(), any_expected); // deterministic
assert_eq!(rbar2.try_iter().collect::<Vec<_>>(), bar_expected);
}
@@ -1637,7 +1702,7 @@ mod tests {
let (tx, rx) = mpsc::channel();
let mut h = build(tx);
h.run(vec![prices.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone()))]);
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
// stage1 (c2..c5): 3,5,7,9. stage2 (c3..): 4,6,8. stage3 (c4..): 5,7.
let expected = vec![
@@ -1648,7 +1713,7 @@ mod tests {
let (tx2, rx2) = mpsc::channel();
let mut h2 = build(tx2);
h2.run(vec![prices]);
h2.run(vec![Box::new(VecSource::new(prices))]);
assert_eq!(rx2.try_iter().collect::<Vec<_>>(), expected); // deterministic
}
@@ -1698,7 +1763,7 @@ mod tests {
let (b, rb) = mpsc::channel();
let (c, rc) = mpsc::channel();
let mut h = build(a, b, c);
h.run(vec![prices.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone()))]);
let w2: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
let w3: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
let w4: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
@@ -1725,7 +1790,7 @@ mod tests {
let (b2, rb2) = mpsc::channel();
let (c2, rc2) = mpsc::channel();
let mut h2 = build(a2, b2, c2);
h2.run(vec![prices]);
h2.run(vec![Box::new(VecSource::new(prices))]);
assert_eq!(ra2.try_iter().collect::<Vec<_>>(), e2); // deterministic
assert_eq!(rb2.try_iter().collect::<Vec<_>>(), e3);
assert_eq!(rc2.try_iter().collect::<Vec<_>>(), e4);
@@ -1785,7 +1850,7 @@ mod tests {
let (ts, rs) = mpsc::channel();
let (ti, ri) = mpsc::channel();
let mut h = build(tr, ts, ti);
h.run(vec![prices.clone(), counts.clone()]);
h.run(vec![Box::new(VecSource::new(prices.clone())), Box::new(VecSource::new(counts.clone()))]);
let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
let i64s: Vec<(Timestamp, Vec<Scalar>)> = ri.try_iter().collect();
@@ -1814,7 +1879,7 @@ mod tests {
let (ts2, rs2) = mpsc::channel();
let (ti2, ri2) = mpsc::channel();
let mut h2 = build(tr2, ts2, ti2);
h2.run(vec![prices, counts]);
h2.run(vec![Box::new(VecSource::new(prices)), Box::new(VecSource::new(counts))]);
assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
assert_eq!(ri2.try_iter().collect::<Vec<_>>(), i64_expected);
@@ -1858,13 +1923,13 @@ mod tests {
)
.expect("valid signal-quality harness");
h.run(vec![f64_stream(&[
h.run(vec![Box::new(VecSource::new(f64_stream(&[
(1, 1.0000),
(2, 1.0010),
(3, 1.0025),
(4, 1.0020),
(5, 1.0040),
])]);
])))]);
let equity: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
// broker fires every cycle (price fresh each tick): five records, ts 1..=5
@@ -1923,13 +1988,13 @@ mod tests {
],
)
.expect("valid harness");
h.run(vec![f64_stream(&[
h.run(vec![Box::new(VecSource::new(f64_stream(&[
(1, 1.0000),
(2, 1.0010),
(3, 1.0025),
(4, 1.0020),
(5, 1.0040),
])]);
])))]);
rx.try_iter().collect::<Vec<(Timestamp, Vec<Scalar>)>>()
};
assert_eq!(build(), build());
crates/aura-engine/src/lib.rs
+2 -3
View File
@@ -22,8 +22,7 @@
//! pip-equity + exposure streams into [`RunMetrics`]; [`f64_field`] bridges a
//! recording sink's `Vec<Scalar>` rows to it.
//!
//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
//! and source-native time normalization (C3/C11), the broker-independent
//! Still to come (subsequent cycles): the broker-independent
//! position-event output and downstream broker nodes (C10), and the param
//! injection + orchestration axes of the atomic sim unit
//! (`(topology + params + data-window + seed)`, swept by optimize / walk-forward
@@ -44,7 +43,7 @@ pub use blueprint::{
};
pub use builder::{BuildError, GraphBuilder, InPort, NodeHandle, OutPort, RoleHandle};
pub use graph_model::model_to_json;
pub use harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};
pub use harness::{BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, Target, VecSource};
pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
pub use sweep::{sweep, GridSpace, SweepError, SweepFamily, SweepPoint};
// #29: re-export the core scalar vocabulary a Blueprint builder needs
crates/aura-engine/src/report.rs
+3 -3
View File
@@ -145,7 +145,7 @@ pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timest
#[cfg(test)]
mod tests {
use super::*;
use crate::{Edge, FlatGraph, Harness, SourceSpec, Target};
use crate::{Edge, FlatGraph, Harness, SourceSpec, Target, VecSource};
use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use std::sync::mpsc;
@@ -220,13 +220,13 @@ mod tests {
fn run_once() -> RunReport {
let (mut h, rx_eq, rx_ex) = build_two_sink_harness();
h.run(vec![f64_stream(&[
h.run(vec![Box::new(VecSource::new(f64_stream(&[
(1, 1.0000),
(2, 1.0010),
(3, 1.0025),
(4, 1.0020),
(5, 1.0040),
])]);
])))]);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let equity = f64_field(&eq_rows, 0);
crates/aura-engine/src/sweep.rs
+2 -2
View File
@@ -170,7 +170,7 @@ where
mod tests {
use super::*;
use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices};
use crate::{f64_field, summarize, RunManifest};
use crate::{f64_field, summarize, RunManifest, VecSource};
use aura_core::{ParamSpec, Scalar, ScalarKind, Timestamp};
fn i64_space(n: usize) -> Vec<ParamSpec> {
@@ -260,7 +260,7 @@ mod tests {
let mut h = bp
.bootstrap_with_params(point.to_vec())
.expect("grid points are kind-checked against param_space");
h.run(vec![synthetic_prices()]);
h.run(vec![Box::new(VecSource::new(synthetic_prices()))]);
let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
RunReport {
crates/aura-ingest/tests/real_bars.rs
+2 -1
View File
@@ -10,6 +10,7 @@ use std::sync::Arc;
use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
use aura_engine::{
f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target,
VecSource,
};
use aura_ingest::load_m1_window;
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
@@ -69,7 +70,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
prices.first().map(|&(t, _)| t).unwrap_or(Timestamp(0)),
prices.last().map(|&(t, _)| t).unwrap_or(Timestamp(0)),
);
h.run(vec![prices]);
h.run(vec![Box::new(VecSource::new(prices))]);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();