diff --git a/crates/aura-ingest/Cargo.toml b/crates/aura-ingest/Cargo.toml index bed3f28..3ec5626 100644 --- a/crates/aura-ingest/Cargo.toml +++ b/crates/aura-ingest/Cargo.toml @@ -11,8 +11,10 @@ aura-core = { path = "../aura-core" } # ingestion edge (transitively chrono + regex + zip). Dependencies are admitted # by the amended C16 per-case policy (INDEX.md), not a blanket zero-dep wall. data-server = { git = "http://192.168.178.103:3000/Brummel/data-server.git", branch = "main" } +# M1FieldSource impls aura_engine::Source in the lib, so the engine is a regular +# (non-dev) dependency. +aura-engine = { path = "../aura-engine" } [dev-dependencies] -# the integration test bootstraps a sample harness + folds a RunReport -aura-engine = { path = "../aura-engine" } +# the integration tests bootstrap a sample harness + fold a RunReport aura-std = { path = "../aura-std" } diff --git a/crates/aura-ingest/src/lib.rs b/crates/aura-ingest/src/lib.rs index d9a4204..9135e5b 100644 --- a/crates/aura-ingest/src/lib.rs +++ b/crates/aura-ingest/src/lib.rs @@ -14,7 +14,7 @@ use aura_core::{Scalar, Timestamp}; use data_server::records::M1Parsed; -use data_server::DataServer; +use data_server::{DataServer, SymbolChunkIter}; use std::sync::Arc; /// Normalize data-server's Unix-millisecond time to aura's canonical epoch-ns @@ -122,6 +122,106 @@ pub fn load_m1_window( Some(transpose_m1(&bars)) } +/// Which base column of an M1 bar a source streams (C7: a composite window is a +/// bundle of base columns; one `Source` per consumed field). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum M1Field { + Open, + High, + Low, + Close, + Spread, + Volume, +} + +/// Project one M1 bar into `(normalized ts, scalar)` for `field`. Pure: a +/// function of `(field, bar)` only — no iterator, no clock — so it is testable +/// in isolation. Time is normalized ms→epoch-ns at this one seam (C3). +fn decode(field: M1Field, bar: &M1Parsed) -> (Timestamp, Scalar) { + let value = match field { + M1Field::Open => Scalar::F64(bar.open), + M1Field::High => Scalar::F64(bar.high), + M1Field::Low => Scalar::F64(bar.low), + M1Field::Close => Scalar::F64(bar.close), + M1Field::Spread => Scalar::F64(bar.spread), + M1Field::Volume => Scalar::I64(bar.volume), + }; + (unix_ms_to_epoch_ns(bar.time_ms), value) +} + +/// A streaming [`Source`](aura_engine::Source) over a data-server M1 window. Holds +/// at most one `Arc` chunk (a zero-copy clone of the cache's chunk) and a cursor; +/// constructs each `Scalar` per-pull; refills via `next_chunk()` when the chunk +/// drains. Resident footprint is O(one chunk), independent of window length. +pub struct M1FieldSource { + iter: SymbolChunkIter, + chunk: Option>, + pos: usize, + field: M1Field, + head: Option<(Timestamp, Scalar)>, +} + +impl M1FieldSource { + /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract). + /// `None` when no archived file overlaps the window (unknown symbol / + /// far-future window) — propagating `stream_m1_windowed`'s file-level Option. + /// A window that overlaps a file but holds zero matching bars yields a source + /// whose first `peek` is `None` (immediately exhausted), not `None` here. + pub fn open( + server: &Arc, + symbol: &str, + from_ms: Option, + to_ms: Option, + field: M1Field, + ) -> Option { + let iter = server.stream_m1_windowed(symbol, from_ms, to_ms)?; + let mut s = Self { iter, chunk: None, pos: 0, field, head: None }; + s.advance(); + Some(s) + } + + /// Decode the head at the cursor, refilling chunks as needed. Sets + /// `self.head = None` at exhaustion. + fn advance(&mut self) { + loop { + if self.chunk.is_none() { + self.chunk = self.iter.next_chunk(); + self.pos = 0; + if self.chunk.is_none() { + self.head = None; + return; + } + } + let chunk = self.chunk.as_ref().unwrap(); + if self.pos < chunk.len() { + self.head = Some(decode(self.field, &chunk[self.pos])); + return; + } + self.chunk = None; // current chunk drained — loop to refill + } + } + + /// Records resident in this source right now: the current chunk's length (or + /// 0 at exhaustion). The residency probe — bounded by one chunk length by + /// construction (the source holds at most one `Arc` chunk and no accumulating + /// field), so it can never grow with window length. + pub fn resident_records(&self) -> usize { + self.chunk.as_ref().map_or(0, |c| c.len()) + } +} + +impl aura_engine::Source for M1FieldSource { + fn peek(&self) -> Option { + self.head.map(|(t, _)| t) + } + fn next(&mut self) -> Option<(Timestamp, Scalar)> { + let item = self.head?; + self.pos += 1; + self.advance(); + Some(item) + } +} + #[cfg(test)] mod tests { use super::*; @@ -205,4 +305,17 @@ mod tests { fn close_stream_empty_on_empty_columns() { assert!(transpose_m1(&[]).close_stream().is_empty()); } + + #[test] + fn decode_projects_each_field_to_its_scalar_kind() { + // full_bar: open 1.0, high 2.0, low 0.5, close 1.5, spread 0.1, volume 100. + let bar = full_bar(1_000); + assert_eq!(decode(M1Field::Open, &bar), (Timestamp(1_000_000_000), Scalar::F64(1.0))); + assert_eq!(decode(M1Field::High, &bar), (Timestamp(1_000_000_000), Scalar::F64(2.0))); + assert_eq!(decode(M1Field::Low, &bar), (Timestamp(1_000_000_000), Scalar::F64(0.5))); + assert_eq!(decode(M1Field::Close, &bar), (Timestamp(1_000_000_000), Scalar::F64(1.5))); + assert_eq!(decode(M1Field::Spread, &bar), (Timestamp(1_000_000_000), Scalar::F64(0.1))); + // volume is the one i64 column. + assert_eq!(decode(M1Field::Volume, &bar), (Timestamp(1_000_000_000), Scalar::I64(100))); + } } diff --git a/crates/aura-ingest/tests/streaming_seam.rs b/crates/aura-ingest/tests/streaming_seam.rs new file mode 100644 index 0000000..849f1f0 --- /dev/null +++ b/crates/aura-ingest/tests/streaming_seam.rs @@ -0,0 +1,191 @@ +//! Gated integration test for the Source ingestion seam: a real data-server M1 +//! close stream driven LAZILY through the signal-quality sample harness via +//! `M1FieldSource`, plus the residency predicate (peak resident records ≤ one +//! chunk, independent of window length) and the N-field sharing property. Skips +//! where the local Pepperstone data directory is absent. + +use std::sync::mpsc; +use std::sync::Arc; + +use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp}; +use aura_engine::{ + f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, Source, SourceSpec, + Target, +}; +use aura_ingest::{M1Field, M1FieldSource}; +use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub}; +use data_server::loader::CHUNK_SIZE; +use data_server::{DataServer, DEFAULT_DATA_PATH}; + +const SYMBOL: &str = "AAPL.US"; +// A bounded 2006-08 window in inclusive Unix-ms — enough bars to run the e2e +// backtest and the two-field sharing test end-to-end (it need not be +// multi-chunk; the residency test drives the unbounded archive instead). +const FROM_MS: i64 = 1_154_390_400_000; +const TO_MS: i64 = 1_157_068_799_999; + +/// Bootstrap the cycle-0007 two-sink signal-quality harness and run it on a +/// single boxed source, folding the recorded equity + exposure into a RunReport. +fn run_sample(window: (Timestamp, Timestamp), source: Box) -> RunReport { + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_ex, rx_ex) = mpsc::channel(); + let f64_recorder_sig = || NodeSchema { + inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }], + output: vec![], + params: vec![], + }; + let mut h = Harness::bootstrap(FlatGraph { + nodes: vec![ + Box::new(Sma::new(2)), + Box::new(Sma::new(4)), + Box::new(Sub::new()), + Box::new(Exposure::new(0.5)), + Box::new(SimBroker::new(0.0001)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), + ], + 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, + targets: vec![ + Target { node: 0, slot: 0 }, + Target { node: 1, slot: 0 }, + Target { node: 4, slot: 1 }, + ], + }], + edges: vec![ + Edge { from: 0, to: 2, slot: 0, from_field: 0 }, + Edge { from: 1, to: 2, slot: 1, from_field: 0 }, + Edge { from: 2, to: 3, slot: 0, from_field: 0 }, + Edge { from: 3, to: 4, slot: 0, from_field: 0 }, + Edge { from: 4, to: 5, slot: 0, from_field: 0 }, + Edge { from: 3, to: 6, slot: 0, from_field: 0 }, + ], + }) + .expect("valid signal-quality DAG"); + + h.run(vec![source]); + + 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); + let exposure = f64_field(&ex_rows, 0); + RunReport { + manifest: RunManifest { + commit: "streaming-seam-test".to_string(), + params: vec![ + ("sma_fast".to_string(), 2.0), + ("sma_slow".to_string(), 4.0), + ("exposure_scale".to_string(), 0.5), + ], + window, + seed: 0, + broker: "sim-optimal(pip_size=0.0001)".to_string(), + }, + metrics: summarize(&equity, &exposure), + } +} + +fn skip_if_no_data(server: &Arc) -> bool { + if !server.has_symbol(SYMBOL) { + eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)"); + return true; + } + false +} + +#[test] +fn streaming_close_source_backtests_end_to_end_deterministically() { + let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + if skip_if_no_data(&server) { + return; + } + let open_close = || { + M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close) + .expect("AAPL.US has data in the 2006-08 window") + }; + + // window bounds for the manifest: drain a source once to read first/last ts. + let mut probe = open_close(); + let first = probe.peek().expect("non-empty window"); + let mut last = first; + while let Some((t, _)) = Source::next(&mut probe) { + last = t; + } + let window = (first, last); + + let r1 = run_sample(window, Box::new(open_close())); + assert!(r1.metrics.total_pips.is_finite(), "a backtest ran over the streamed window"); + + // same window streamed again ⇒ bit-identical report (C1). + let r2 = run_sample(window, Box::new(open_close())); + assert_eq!(r1.to_json(), r2.to_json()); +} + +#[test] +fn residency_is_bounded_by_one_chunk_independent_of_window_length() { + let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + if skip_if_no_data(&server) { + return; + } + // The whole archive (unbounded both ends), not the narrow e2e window: the + // non-vacuity assertion below needs > CHUNK_SIZE bars, and early AAPL.US M1 + // coverage is sparse (a single month can hold far fewer). Driving the full + // span is also the strongest form of the "residency independent of window + // length" property — the longest window the local data offers. + let mut src = M1FieldSource::open(&server, SYMBOL, None, None, M1Field::Close) + .expect("AAPL.US has archived data"); + + // Drive the source the way `run` does (peek to pick, next to pop), sampling + // resident records at every pull. + let mut peak = 0usize; + let mut total = 0usize; + while src.peek().is_some() { + peak = peak.max(src.resident_records()); + Source::next(&mut src); + total += 1; + } + + // multi-chunk window: more records than a single chunk holds, so the bound + // below is non-vacuous (a O(window) source would have peaked at `total`). + assert!(total > CHUNK_SIZE, "window must span multiple chunks (got {total} records)"); + // O(one chunk), NOT O(window): the per-pull ceiling never exceeds one chunk, + // regardless of how many chunks the window spans. + assert!(peak <= CHUNK_SIZE, "resident records {peak} exceeded one chunk ({CHUNK_SIZE})"); + assert!(peak > 0, "a non-empty window resided at least one record"); +} + +#[test] +fn two_field_sources_share_one_window() { + let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + if skip_if_no_data(&server) { + return; + } + // close + volume over the same window: each is an independent Source pulling + // its field; both stream to exhaustion (the N-sources-share-the-ts-axis shape). + let mut close = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close) + .expect("close source"); + let mut volume = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Volume) + .expect("volume source"); + + let mut n_close = 0usize; + while let Some((_, v)) = Source::next(&mut close) { + assert!(matches!(v, Scalar::F64(_)), "close is an f64 column"); + n_close += 1; + } + let mut n_volume = 0usize; + while let Some((_, v)) = Source::next(&mut volume) { + assert!(matches!(v, Scalar::I64(_)), "volume is an i64 column"); + n_volume += 1; + } + assert_eq!(n_close, n_volume, "both fields share the same ts axis ⇒ same count"); + assert!(n_close > 0); +}