From eec129ee51c93cfb230d0db1ad08543a481d0ee8 Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 15 Jun 2026 10:23:09 +0200 Subject: [PATCH] =?UTF-8?q?feat(aura-engine):=20Source=20ingestion=20seam?= =?UTF-8?q?=20=E2=80=94=20run()=20takes=20Vec>?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine's ingestion input was the only type encoding an eager-dataflow assumption: `Harness::run(Vec>)` — 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`, `next(&mut) -> Option<(Timestamp, Scalar)>`), object-safe so the merge holds `Vec>`. 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>`; 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 --- crates/aura-cli/src/main.rs | 10 +- crates/aura-engine/src/blueprint.rs | 21 +-- crates/aura-engine/src/harness.rs | 197 +++++++++++++++++--------- crates/aura-engine/src/lib.rs | 5 +- crates/aura-engine/src/report.rs | 6 +- crates/aura-engine/src/sweep.rs | 4 +- crates/aura-ingest/tests/real_bars.rs | 3 +- 7 files changed, 156 insertions(+), 90 deletions(-) diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 7c5971b..6ecd0d1 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = 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::>(), 0); let exposure = f64_field(&rx_ex.try_iter().collect::>(), 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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = 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::>().is_empty(), "equity sink drained empty"); assert!(!rx_ex.try_iter().collect::>().is_empty(), "exposure sink drained empty"); } diff --git a/crates/aura-engine/src/blueprint.rs b/crates/aura-engine/src/blueprint.rs index 6ef4906..9893e64 100644 --- a/crates/aura-engine/src/blueprint.rs +++ b/crates/aura-engine/src/blueprint.rs @@ -812,6 +812,7 @@ fn slot_kind(t: Target, flat_signatures: &[NodeSchema]) -> Result>(); let named_ex = comp_ex.try_iter().collect::>(); @@ -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::>(); let pos_ex_v = pos_ex.try_iter().collect::>(); @@ -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::>(); let flat_ex_v = flat_ex.try_iter().collect::>(); @@ -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::>(); let b = rx_b.try_iter().collect::>(); @@ -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::>(); 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::>(); 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::>(), eq2.try_iter().collect::>()); } @@ -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::>(); assert!(!ex.is_empty(), "the cured, by-name-bound cross must run to a populated exposure trace"); } diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index 981253d..fa08953 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -51,6 +51,45 @@ pub struct SourceSpec { pub targets: Vec, } +/// 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>`. +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; + + /// 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 { + 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>) { + pub fn run(&mut self, mut sources_in: Vec>) { 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 = vec![0; streams.len()]; let mut cycle_id: u64 = 0; let mut scratch: Vec = Vec::new(); loop { - // pick the live source head with the smallest (timestamp, source index) - let mut pick: Option = 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 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)> = 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)> = 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)> = 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)> = 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)> = 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)> = 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)> = 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)> = 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)> = 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)> = 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> { + fn ohlcv_streams() -> Vec> { 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)> = rx_fast.try_iter().collect(); let slow: Vec<(Timestamp, Vec)> = 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)> = 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)> = 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)> = rx_tap.try_iter().collect(); let downstream: Vec<(Timestamp, Vec)> = 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)> = 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)> = 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)> = 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)> = 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)> = ra.try_iter().collect(); let s2: Vec<(Timestamp, Vec)> = rb.try_iter().collect(); let s3: Vec<(Timestamp, Vec)> = 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::>(), expected); assert_eq!(rb2.try_iter().collect::>(), expected); assert_eq!(rc2.try_iter().collect::>(), 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)> = rr.try_iter().collect(); let sub: Vec<(Timestamp, Vec)> = 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::>(), raw_expected); // deterministic assert_eq!(rs2.try_iter().collect::>(), 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)> = rany.try_iter().collect(); let bar: Vec<(Timestamp, Vec)> = 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::>(), any_expected); // deterministic assert_eq!(rbar2.try_iter().collect::>(), 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)> = 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::>(), 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)> = ra.try_iter().collect(); let w3: Vec<(Timestamp, Vec)> = rb.try_iter().collect(); let w4: Vec<(Timestamp, Vec)> = 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::>(), e2); // deterministic assert_eq!(rb2.try_iter().collect::>(), e3); assert_eq!(rc2.try_iter().collect::>(), 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)> = rr.try_iter().collect(); let sub: Vec<(Timestamp, Vec)> = rs.try_iter().collect(); let i64s: Vec<(Timestamp, Vec)> = 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::>(), raw_expected); // deterministic assert_eq!(rs2.try_iter().collect::>(), sub_expected); assert_eq!(ri2.try_iter().collect::>(), 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)> = 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::)>>() }; assert_eq!(build(), build()); diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index 5f97c1d..4b5d7b4 100644 --- a/crates/aura-engine/src/lib.rs +++ b/crates/aura-engine/src/lib.rs @@ -22,8 +22,7 @@ //! pip-equity + exposure streams into [`RunMetrics`]; [`f64_field`] bridges a //! recording sink's `Vec` 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 diff --git a/crates/aura-engine/src/report.rs b/crates/aura-engine/src/report.rs index 838a8b0..cea3189 100644 --- a/crates/aura-engine/src/report.rs +++ b/crates/aura-engine/src/report.rs @@ -145,7 +145,7 @@ pub fn f64_field(rows: &[(Timestamp, Vec)], 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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); let equity = f64_field(&eq_rows, 0); diff --git a/crates/aura-engine/src/sweep.rs b/crates/aura-engine/src/sweep.rs index b1dabb2..b384a65 100644 --- a/crates/aura-engine/src/sweep.rs +++ b/crates/aura-engine/src/sweep.rs @@ -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 { @@ -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::>(), 0); let exposure = f64_field(&rx_ex.try_iter().collect::>(), 0); RunReport { diff --git a/crates/aura-ingest/tests/real_bars.rs b/crates/aura-ingest/tests/real_bars.rs index ad1893c..6674509 100644 --- a/crates/aura-ingest/tests/real_bars.rs +++ b/crates/aura-ingest/tests/real_bars.rs @@ -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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect();