From cf102172ec5a62ad644f5eeaff51011fc10bb75d Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 21 Jun 2026 14:40:15 +0200 Subject: [PATCH] feat(real-family): DataSource provider + shared real-data helpers Foundation for `aura sweep|walkforward --real` (#106, plan 0060 Task 1). - New CLI-side `DataChoice` (parsed) / `DataSource` (opened) provider: the synthetic-vs-real source choice as one type, supplying a member's source, pip, full/windowed sources, and walk-forward roller sizes. Not yet wired into the family builders (plan Tasks 2-5); `#[allow(dead_code)]` marks the wired-later items, to be removed once the builders thread `&DataSource`. - DRY: `run_sample_real`'s inline pip-refusal / no-data refusal / probe-window logic is extracted to shared `instrument_spec_or_refuse` / `no_real_data` / `probe_window` helpers, now reused by `DataSource::from_choice` / `full_window` instead of duplicated. - Real walk-forward roller-size constants (90d / 30d / 30d in ns, Fork D/F). The synthetic provider is consumer-dependent by design (full-window consumers draw showcase_prices; windowed consumers draw walkforward_prices) so the byte-unchanged guarantee holds across both family paths. Verified: cargo test -p aura-cli (all green incl. byte-unchanged synthetic path), cargo clippy -p aura-cli --all-targets -D warnings (clean). Full-workspace gate at iter close after Tasks 2-5. refs #106 --- crates/aura-cli/src/main.rs | 241 +++++++++++++++++++++++++++++++----- 1 file changed, 208 insertions(+), 33 deletions(-) diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index acc74f4..9ec0f96 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -37,6 +37,23 @@ use std::collections::HashSet; /// The real path threads the looked-up `spec.pip_size` instead. const SYNTHETIC_PIP_SIZE: f64 = 0.0001; +/// Real walk-forward roller sizes (Fork D/F). `WindowRoller` takes sizes in the +/// stream's epoch-unit; for real M1 that is nanoseconds. A classic 3-month +/// in-sample / 1-month out-of-sample / 1-month step (contiguous OOS tiling). +// +// `dead_code` allow: these feed `DataSource::wf_window_sizes`'s real arm, which is +// itself unreachable until Task 3 threads `&DataSource` into `walkforward_family`. +// The allow is removed implicitly once that wiring lands (Task 3 makes them live); +// Task 1 is intentionally additive, so the clippy `-D warnings` gate is Task 5. +#[allow(dead_code)] +const WF_DAY_NS: i64 = 86_400_000_000_000; +#[allow(dead_code)] +const WF_REAL_IS_NS: i64 = 90 * WF_DAY_NS; +#[allow(dead_code)] +const WF_REAL_OOS_NS: i64 = 30 * WF_DAY_NS; +#[allow(dead_code)] +const WF_REAL_STEP_NS: i64 = 30 * WF_DAY_NS; + /// The built-in synthetic price stream: rises through t=4 then reverses, so the /// demo trace carries one exposure sign flip and a real drawdown (C22 populated /// trace). Deterministic and fixed (C1). @@ -320,46 +337,21 @@ fn run_sample(trace: Option<&str>) -> RunReport { fn run_sample_real(symbol: &str, from_ms: Option, to_ms: Option, trace: Option<&str>) -> RunReport { // Per-instrument pip: look up BEFORE any data access, so an un-specced symbol // refuses without touching local data. Honest by construction. - let spec = match aura_ingest::instrument_spec(symbol) { - Some(s) => s, - None => { - eprintln!("aura: no vetted pip/instrument spec for symbol '{symbol}' — refusing to run a real instrument with a guessed pip (add it to the instrument table)"); - std::process::exit(2); - } - }; + let spec = instrument_spec_or_refuse(symbol); let (mut h, rx_eq, rx_ex) = sample_harness(spec.pip_size); let server = std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH)); - let no_data = || -> ! { - eprintln!( - "aura: no local data for symbol '{symbol}' at {}", - data_server::DEFAULT_DATA_PATH - ); - std::process::exit(2) - }; if !server.has_symbol(symbol) { - no_data(); + no_real_data(symbol); } - let open = || aura_ingest::M1FieldSource::open(&server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close); // Manifest window: drain a separate probe (single-pass Source) for first/last ts. - let mut probe = match open() { - Some(p) => p, - None => no_data(), - }; - let first = match aura_engine::Source::peek(&probe) { - Some(t) => t, - None => no_data(), - }; - let mut last = first; - while let Some((t, _)) = aura_engine::Source::next(&mut probe) { - last = t; - } - let window = (first, last); + let window = probe_window(&server, symbol, from_ms, to_ms); - let source: Box = match open() { - Some(s) => Box::new(s), - None => no_data(), - }; + let source: Box = + match aura_ingest::M1FieldSource::open(&server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close) { + Some(s) => Box::new(s), + None => no_real_data(symbol), + }; h.run(vec![source]); let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); @@ -411,6 +403,167 @@ fn parse_real_args( Ok((symbol.to_string(), from, to, trace)) } +/// What `--real` parsing yields: the synthetic default, or a real symbol + an +/// optional window (parsed, not yet opened). Pure, so the grammar is unit-testable. +// +// `dead_code` allow: constructed by the `--real` parsers in Task 4; Task 1 only +// defines the type (additive). Clippy `-D warnings` is the Task 5 gate. +#[allow(dead_code)] +#[derive(Debug, Clone, PartialEq)] +enum DataChoice { + Synthetic, + Real { symbol: String, from_ms: Option, to_ms: Option }, +} + +/// The source provider threaded into the family builders: synthetic built-in +/// streams, or real M1 close bars from the data-server archive. Replaces the +/// hardcoded `VecSource` so a member's source, pip, window, and roller sizes come +/// from one place (Fork B/D/F). +/// +/// `Synthetic` denotes a *consumer-dependent* built-in stream, not one fixed +/// series: the full-window consumers (`full_window` / `run_sources`, used by +/// sweep / MC) draw the 18-bar `showcase_prices()`, while the windowed consumers +/// (`windowed_sources` / `wf_window_sizes`, used by walk-forward) draw the 60-bar +/// `walkforward_prices()` so the `(24,12,12)`-bar roller fits its span. The two +/// faces never reach one consumer (a family is either full-window or windowed), so +/// the split is invisible per call site but real across the type — read both +/// family builders to see it whole. +// +// `dead_code` allow: the `Real` variant is constructed by `from_choice` (live once +// Task 4's parsers feed it) and the provider is threaded into the family builders +// in Task 3; Task 1 is additive. Clippy `-D warnings` is the Task 5 gate. +#[allow(dead_code)] +enum DataSource { + Synthetic, + Real { + server: std::sync::Arc, + symbol: String, + from_ms: Option, + to_ms: Option, + pip: f64, + }, +} + +/// No-local-data refusal — stderr + exit(2), mirroring `run_sample_real`. +fn no_real_data(symbol: &str) -> ! { + eprintln!("aura: no local data for symbol '{symbol}' at {}", data_server::DEFAULT_DATA_PATH); + std::process::exit(2) +} + +/// Look up the vetted per-instrument pip/spec, or refuse (stderr + exit 2) on an +/// un-specced symbol — the single home of the guessed-pip refusal message, shared +/// by `run_sample_real` and `DataSource::from_choice`. Refusing BEFORE any data +/// access keeps the pip honest by construction (never guessed). +fn instrument_spec_or_refuse(symbol: &str) -> aura_ingest::InstrumentSpec { + match aura_ingest::instrument_spec(symbol) { + Some(s) => s, + None => { + eprintln!("aura: no vetted pip/instrument spec for symbol '{symbol}' — refusing to run a real instrument with a guessed pip (add it to the instrument table)"); + std::process::exit(2); + } + } +} + +/// Probe the full data window: open a single-pass probe `M1FieldSource`, drain it +/// for the first/last timestamp, and return `(first, last)`. Refuses (via +/// `no_real_data`) when the symbol/window yields no source or no bars. Shared by +/// `run_sample_real` (which needs the manifest window from a probe separate from +/// the run source) and `DataSource::full_window`. +fn probe_window( + server: &std::sync::Arc, + symbol: &str, + from_ms: Option, + to_ms: Option, +) -> (Timestamp, Timestamp) { + let mut probe = aura_ingest::M1FieldSource::open(server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close) + .unwrap_or_else(|| no_real_data(symbol)); + let first = aura_engine::Source::peek(&probe).unwrap_or_else(|| no_real_data(symbol)); + let mut last = first; + while let Some((t, _)) = aura_engine::Source::next(&mut probe) { + last = t; + } + (first, last) +} + +// `dead_code` allow: every method is reached only via the family builders Task 3 +// rewires; `from_choice` via the Task 4 dispatch. Task 1 is additive (see the +// type docs above). Clippy `-D warnings` is the Task 5 gate, by which point all +// these are live and the allow is a no-op. +#[allow(dead_code)] +impl DataSource { + /// Build a provider from a parsed choice, or refuse (stderr + exit 2) on an + /// un-vetted symbol / absent data — both BEFORE any member runs (Fork C/G), + /// via the same `instrument_spec_or_refuse` / `no_real_data` helpers as + /// `run_sample_real`. + fn from_choice(choice: DataChoice) -> DataSource { + match choice { + DataChoice::Synthetic => DataSource::Synthetic, + DataChoice::Real { symbol, from_ms, to_ms } => { + let spec = instrument_spec_or_refuse(&symbol); + let server = std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH)); + if !server.has_symbol(&symbol) { + no_real_data(&symbol); + } + DataSource::Real { server, symbol, from_ms, to_ms, pip: spec.pip_size } + } + } + } + + fn pip_size(&self) -> f64 { + match self { + DataSource::Synthetic => SYNTHETIC_PIP_SIZE, + DataSource::Real { pip, .. } => *pip, + } + } + + /// The full run window, probed once. Synthetic: the showcase span. Real: + /// `probe_window` drains a separate single-pass probe source for first/last ts + /// (the same helper `run_sample_real` uses for its manifest window). + fn full_window(&self) -> (Timestamp, Timestamp) { + match self { + DataSource::Synthetic => { + let s: Vec> = vec![Box::new(VecSource::new(showcase_prices()))]; + window_of(&s).expect("non-empty showcase stream") + } + DataSource::Real { server, symbol, from_ms, to_ms, .. } => { + probe_window(server, symbol, *from_ms, *to_ms) + } + } + } + + /// A fresh full-window source per member (single-pass). Synthetic: showcase. + fn run_sources(&self) -> Vec> { + match self { + DataSource::Synthetic => vec![Box::new(VecSource::new(showcase_prices()))], + DataSource::Real { server, symbol, from_ms, to_ms, .. } => vec![Box::new( + aura_ingest::M1FieldSource::open(server, symbol, *from_ms, *to_ms, aura_ingest::M1Field::Close) + .unwrap_or_else(|| no_real_data(symbol)), + )], + } + } + + /// A fresh windowed source for an IS/OOS sub-window (walk-forward). Synthetic: + /// `walkforward_window_source`. Real: `open_window` (ns-native `Timestamp`). + fn windowed_sources(&self, from: Timestamp, to: Timestamp) -> Vec> { + match self { + DataSource::Synthetic => vec![Box::new(walkforward_window_source(from, to))], + DataSource::Real { server, symbol, .. } => vec![Box::new( + aura_ingest::M1FieldSource::open_window(server, symbol, Some(from), Some(to), aura_ingest::M1Field::Close) + .unwrap_or_else(|| no_real_data(symbol)), + )], + } + } + + /// WindowRoller sizes per data kind (Fork F): bar-index for synthetic (24/12/12 + /// over the 60-bar span), calendar-ns for real. + fn wf_window_sizes(&self) -> (i64, i64, i64) { + match self { + DataSource::Synthetic => (24, 12, 12), + DataSource::Real { .. } => (WF_REAL_IS_NS, WF_REAL_OOS_NS, WF_REAL_STEP_NS), + } + } +} + /// The SMA-cross signal as a named composite (price -> fast/slow SMA -> spread). /// CLI-local sample builder; the engine ships no sample (the duplication with /// `blueprint.rs`'s test helper is the dedup tracked in #14). Value-empty: the SMA @@ -1184,6 +1337,28 @@ mod tests { const GER40_SEP2024_FROM_MS: i64 = 1_725_148_800_000; const GER40_SEP2024_TO_MS: i64 = 1_727_740_799_999; + #[test] + fn data_source_synthetic_pip_and_window_match_the_built_ins() { + let d = DataSource::Synthetic; + assert_eq!(d.pip_size(), SYNTHETIC_PIP_SIZE); + assert!(!d.run_sources().is_empty()); + assert_eq!(d.wf_window_sizes(), (24, 12, 12)); + // full_window equals window_of over the showcase stream (byte-unchanged source) + let s: Vec> = vec![Box::new(VecSource::new(showcase_prices()))]; + assert_eq!(d.full_window(), window_of(&s).unwrap()); + } + + #[test] + fn wf_real_roller_sizes_are_90_30_30_days_in_ns() { + // Independent expected value: a day reconstructed from its time units + // (24 h * 60 min * 60 s * 1e9 ns), not the constant's own `86_400_000_000_000` + // literal — so the test fails if either the literal or the day-count is wrong. + let day_ns: i64 = 24 * 60 * 60 * 1_000_000_000; + assert_eq!(WF_REAL_IS_NS, 90 * day_ns); + assert_eq!(WF_REAL_OOS_NS, 30 * day_ns); + assert_eq!(WF_REAL_STEP_NS, 30 * day_ns); + } + #[test] fn walkforward_report_is_deterministic() { // The built-in WFO render is byte-identical across two