From 2a6ca47acde27ea9a13e765df470559a3255b1b3 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 21 Jun 2026 14:55:39 +0200 Subject: [PATCH] feat(real-family): thread DataSource + pip through the family builders Plan 0060 Tasks 2-3 (#106): the family builders now take a &DataSource and draw their source, pip, window, and walk-forward roller sizes from it instead of the hardcoded synthetic VecSource / SYNTHETIC_PIP_SIZE. - sample_blueprint_with_sinks / momentum_blueprint_with_sinks take a pip_size (un-hardcode the SimBroker); every caller threads it. - sweep_family / momentum_sweep_family / walkforward_family / sweep_over / run_oos take &DataSource; run_sweep / run_walkforward take a DataSource and the dispatch passes DataSource::Synthetic (the --real parser is the next step). Synthetic behaviour is byte-unchanged. - Fix a design gap the implementer correctly bounced on: walk-forward is a windowed consumer whose span comes from the 60-bar walkforward_prices, NOT the 18-bar showcase that full_window() returns. Added DataSource::wf_full_span() (synthetic walk-forward span; real = the probed --from..--to) and pointed walkforward_family at it. Without it the synthetic roller (24,12,12) over span (1,18) exits SpanTooShort. Verified: cargo test -p aura-cli (all green, synthetic byte-unchanged), cargo clippy -p aura-cli --all-targets -D warnings (clean). refs #106 --- crates/aura-cli/src/main.rs | 137 +++++++++++++++++++++--------------- 1 file changed, 82 insertions(+), 55 deletions(-) diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 9ec0f96..fe4859c 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -531,6 +531,24 @@ impl DataSource { } } + /// The full walk-forward span. Synthetic draws the 60-bar `walkforward_prices` + /// span — NOT `showcase_prices` (which `full_window` uses): walk-forward is a + /// *windowed* consumer whose roller `(24,12,12)` needs 36 bars, so it uses the + /// longer built-in stream (byte-unchanged from the pre-`DataSource` + /// `walkforward_family`, which derived its span the same way). Real: the same + /// probed `--from..--to` window as `full_window`. + fn wf_full_span(&self) -> (Timestamp, Timestamp) { + match self { + DataSource::Synthetic => { + let s: Vec> = vec![Box::new(VecSource::new(walkforward_prices()))]; + window_of(&s).expect("non-empty walkforward 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 { @@ -615,7 +633,7 @@ fn signals(name: &str) -> Composite { /// signal is built by the nested `signals` composite; `build_sample` (the `aura /// graph` entry) is expressed on top of it. #[allow(clippy::type_complexity)] -fn sample_blueprint_with_sinks() -> ( +fn sample_blueprint_with_sinks(pip_size: f64) -> ( Composite, Receiver<(Timestamp, Vec)>, Receiver<(Timestamp, Vec)>, @@ -625,7 +643,7 @@ fn sample_blueprint_with_sinks() -> ( let mut g = GraphBuilder::new("sample"); let sig = g.add(signals("signals")); let exposure = g.add(Exposure::builder()); - let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE)); + let broker = g.add(SimBroker::builder(pip_size)); let eq = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq)); let ex = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex)); let price = g.source_role("price", ScalarKind::F64); @@ -641,7 +659,7 @@ fn sample_blueprint_with_sinks() -> ( /// The sample blueprint without its sink receivers — the `aura graph` render /// entry, which never runs the graph (so the receivers are dropped). fn build_sample() -> Composite { - sample_blueprint_with_sinks().0 + sample_blueprint_with_sinks(SYNTHETIC_PIP_SIZE).0 } /// The built-in sample rendered by `aura graph`. @@ -655,8 +673,10 @@ fn sample_blueprint() -> Composite { /// the same report. Each point builds a fresh blueprint (fresh sink channels), /// bootstraps it under the point vector, runs it, and folds the drained sinks to /// metrics — the per-point closure the engine `sweep` drives disjointly. -fn sweep_family(trace: Option<&str>) -> SweepFamily { - let bp = sample_blueprint_with_sinks().0; +fn sweep_family(trace: Option<&str>, data: &DataSource) -> SweepFamily { + let pip = data.pip_size(); + let window = data.full_window(); + let bp = sample_blueprint_with_sinks(pip).0; let space = bp.param_space(); let binder = bp .axis("signals.trend.fast.length", [2, 3]) @@ -670,19 +690,17 @@ fn sweep_family(trace: Option<&str>) -> SweepFamily { let varying: HashSet = binder.varying_axes().into_iter().collect(); binder .sweep(|point| { - let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(); + let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(pip); let mut h = bp .bootstrap_with_cells(point) .expect("grid points are kind-checked against param_space"); - let sources: Vec> = - vec![Box::new(VecSource::new(showcase_prices()))]; - let window = window_of(&sources).expect("non-empty showcase stream"); + let sources = data.run_sources(); h.run(sources); let eq_rows = rx_eq.try_iter().collect::>(); let ex_rows = rx_ex.try_iter().collect::>(); let named = zip_params(&space, point); let key = member_key(&named, &varying); - let manifest = sim_optimal_manifest(named, window, 0, SYNTHETIC_PIP_SIZE); + let manifest = sim_optimal_manifest(named, window, 0, pip); if let Some(name) = trace { persist_traces(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows); } @@ -701,7 +719,7 @@ fn sweep_family(trace: Option<&str>) -> SweepFamily { /// generic-sweep proof. The exposure sink taps the FINAL (gated) exposure so the /// bool's effect is visible in the trace. #[allow(clippy::type_complexity)] -fn momentum_blueprint_with_sinks() -> ( +fn momentum_blueprint_with_sinks(pip_size: f64) -> ( Composite, Receiver<(Timestamp, Vec)>, Receiver<(Timestamp, Vec)>, @@ -716,7 +734,7 @@ fn momentum_blueprint_with_sinks() -> ( let dist = g.add(Sub::builder()); // momentum = price - ema let expo = g.add(Exposure::builder().named("exposure")); // exposure.scale let gate = g.add(LongOnly::builder().named("longonly")); // longonly.enabled (the bool param) - let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE)); + let broker = g.add(SimBroker::builder(pip_size)); let eq = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq)); let ex = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex)); let price = g.source_role("price", ScalarKind::F64); @@ -736,8 +754,10 @@ fn momentum_blueprint_with_sinks() -> ( /// all three axes varying. Mirrors `sweep_family`: capture the binder's varying /// axes, key each member via the generic portable `member_key`. With `--trace`, /// persist each member under `runs/traces///`. -fn momentum_sweep_family(trace: Option<&str>) -> SweepFamily { - let bp = momentum_blueprint_with_sinks().0; +fn momentum_sweep_family(trace: Option<&str>, data: &DataSource) -> SweepFamily { + let pip = data.pip_size(); + let window = data.full_window(); + let bp = momentum_blueprint_with_sinks(pip).0; let space = bp.param_space(); let binder = bp .axis("ema.length", [5, 10]) @@ -746,19 +766,17 @@ fn momentum_sweep_family(trace: Option<&str>) -> SweepFamily { let varying: HashSet = binder.varying_axes().into_iter().collect(); binder .sweep(|point| { - let (bp, rx_eq, rx_ex) = momentum_blueprint_with_sinks(); + let (bp, rx_eq, rx_ex) = momentum_blueprint_with_sinks(pip); let mut h = bp .bootstrap_with_cells(point) .expect("grid points are kind-checked against param_space"); - let sources: Vec> = - vec![Box::new(VecSource::new(showcase_prices()))]; - let window = window_of(&sources).expect("non-empty showcase stream"); + let sources = data.run_sources(); h.run(sources); let eq_rows = rx_eq.try_iter().collect::>(); let ex_rows = rx_ex.try_iter().collect::>(); let named = zip_params(&space, point); let key = member_key(&named, &varying); - let manifest = sim_optimal_manifest(named, window, 0, SYNTHETIC_PIP_SIZE); + let manifest = sim_optimal_manifest(named, window, 0, pip); if let Some(name) = trace { persist_traces(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows); } @@ -774,7 +792,7 @@ fn momentum_sweep_family(trace: Option<&str>) -> SweepFamily { #[cfg(test)] fn sweep_report() -> String { let mut out = String::new(); - for pt in &sweep_family(None).points { + for pt in &sweep_family(None, &DataSource::Synthetic).points { out.push_str(&pt.report.to_json()); out.push('\n'); } @@ -831,11 +849,11 @@ fn parse_sweep_args(rest: &[&str]) -> Result<(Strategy, String, bool), String> { /// `family_id`, C18/C21) via `append_family`, and print each point's record line /// carrying the assigned id. With `--trace`, also persist each member's streams /// under `runs/traces///` (opt-in). -fn run_sweep(strategy: Strategy, name: &str, persist: bool) { +fn run_sweep(strategy: Strategy, name: &str, persist: bool, data: DataSource) { let reg = default_registry(); let family = match strategy { - Strategy::SmaCross => sweep_family(persist.then_some(name)), - Strategy::Momentum => momentum_sweep_family(persist.then_some(name)), + Strategy::SmaCross => sweep_family(persist.then_some(name), &data), + Strategy::Momentum => momentum_sweep_family(persist.then_some(name), &data), }; let id = match reg.append_family(name, FamilyKind::Sweep, &sweep_member_reports(&family)) { Ok(id) => id, @@ -857,9 +875,9 @@ fn run_sweep(strategy: Strategy, name: &str, persist: bool) { /// print each carrying the assigned id, then the stitched summary line. With /// `--trace`, also persist each OOS member's streams under /// `runs/traces//oos/` (opt-in). Deterministic (C1). -fn run_walkforward(name: &str, persist: bool) { +fn run_walkforward(name: &str, persist: bool, data: DataSource) { let reg = default_registry(); - let result = walkforward_family(persist.then_some(name)); + let result = walkforward_family(persist.then_some(name), &data); let id = match reg.append_family(name, FamilyKind::WalkForward, &walkforward_member_reports(&result)) { @@ -879,17 +897,21 @@ fn run_walkforward(name: &str, persist: bool) { /// stepping 12 (contiguous OOS tiling), over the 60-bar synthetic span -> 3 /// windows. Each window sweeps the built-in grid in-sample, optimizes by /// total_pips (axis 2), and runs the chosen params out-of-sample. -fn walkforward_family(trace: Option<&str>) -> WalkForwardResult { - let sources: Vec> = - vec![Box::new(VecSource::new(walkforward_prices()))]; - let span = window_of(&sources).expect("non-empty synthetic stream"); - let roller = WindowRoller::new(span, 24, 12, 12, RollMode::Rolling) - .expect("built-in walk-forward config fits the 60-bar synthetic span"); - let space = sample_blueprint_with_sinks().0.param_space(); +fn walkforward_family(trace: Option<&str>, data: &DataSource) -> WalkForwardResult { + let span = data.wf_full_span(); + let (is_len, oos_len, step) = data.wf_window_sizes(); + let roller = match WindowRoller::new(span, is_len, oos_len, step, RollMode::Rolling) { + Ok(r) => r, + Err(e) => { + eprintln!("aura: walk-forward window too short for one IS+OOS span: {e:?}"); + std::process::exit(2); + } + }; + let space = sample_blueprint_with_sinks(data.pip_size()).0.param_space(); walk_forward(roller, space, |w: WindowBounds| { - let is_family = sweep_over(w.is.0, w.is.1); + let is_family = sweep_over(w.is.0, w.is.1, data); let best = optimize(&is_family, "total_pips").expect("total_pips is a known metric"); - let (oos_equity, oos_report) = run_oos(&best.params, w.oos.0, w.oos.1, trace); + let (oos_equity, oos_report) = run_oos(&best.params, w.oos.0, w.oos.1, trace, data); WindowRun { // The tag-free sweep winner is the chosen point; its kinds live on // WalkForwardResult.space (computed once above from the same blueprint). @@ -902,8 +924,9 @@ fn walkforward_family(trace: Option<&str>) -> WalkForwardResult { /// Sweep the built-in named grid over an in-sample window, sourcing the in-memory /// windowed stream. Mirrors `sweep_family`, but windowed by `[from, to]`. -fn sweep_over(from: Timestamp, to: Timestamp) -> SweepFamily { - let bp = sample_blueprint_with_sinks().0; +fn sweep_over(from: Timestamp, to: Timestamp, data: &DataSource) -> SweepFamily { + let pip = data.pip_size(); + let bp = sample_blueprint_with_sinks(pip).0; let space = bp.param_space(); bp.axis("signals.trend.fast.length", [2, 3]) .axis("signals.trend.slow.length", [4, 5]) @@ -914,18 +937,17 @@ fn sweep_over(from: Timestamp, to: Timestamp) -> SweepFamily { .axis("signals.blend.weights[1]", [1.0]) .axis("exposure.scale", [0.5]) .sweep(|point| { - let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(); + let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(pip); let mut h = bp .bootstrap_with_cells(point) .expect("grid points are kind-checked against param_space"); - let sources: Vec> = - vec![Box::new(walkforward_window_source(from, to))]; + let sources = data.windowed_sources(from, to); let window = window_of(&sources).expect("non-empty in-sample window"); h.run(sources); let equity = f64_field(&rx_eq.try_iter().collect::>(), 0); let exposure = f64_field(&rx_ex.try_iter().collect::>(), 0); RunReport { - manifest: sim_optimal_manifest(zip_params(&space, point), window, 0, SYNTHETIC_PIP_SIZE), + manifest: sim_optimal_manifest(zip_params(&space, point), window, 0, pip), metrics: summarize(&equity, &exposure), } }) @@ -939,19 +961,20 @@ fn run_oos( from: Timestamp, to: Timestamp, trace: Option<&str>, + data: &DataSource, ) -> (Vec<(Timestamp, f64)>, RunReport) { - let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(); + let pip = data.pip_size(); + let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(pip); let space = bp.param_space(); let mut h = bp .bootstrap_with_cells(params) .expect("chosen params pre-validated by the in-sample GridSpace::new"); - let sources: Vec> = - vec![Box::new(walkforward_window_source(from, to))]; + let sources = data.windowed_sources(from, to); let window = window_of(&sources).expect("non-empty out-of-sample window"); h.run(sources); let eq_rows = rx_eq.try_iter().collect::>(); let ex_rows = rx_ex.try_iter().collect::>(); - let manifest = sim_optimal_manifest(zip_params(&space, params), window, 0, SYNTHETIC_PIP_SIZE); + let manifest = sim_optimal_manifest(zip_params(&space, params), window, 0, pip); if let Some(name) = trace { persist_traces(&format!("{name}/oos{}", from.0), &manifest, &eq_rows, &ex_rows); } @@ -1005,7 +1028,7 @@ fn walkforward_window_source(from: Timestamp, to: Timestamp) -> VecSource { /// helper (mirrors `sweep_report`). #[cfg(test)] fn walkforward_report() -> String { - let result = walkforward_family(None); + let result = walkforward_family(None, &DataSource::Synthetic); let mut out = String::new(); for w in &result.windows { out.push_str(&w.run.oos_report.to_json()); @@ -1301,15 +1324,15 @@ fn main() { ["chart", name, "--panels"] => emit_chart(name, ChartMode::Panels), ["graph"] => print!("{}", render::render_html(&sample_blueprint())), ["sweep", rest @ ..] => match parse_sweep_args(rest) { - Ok((strategy, name, persist)) => run_sweep(strategy, &name, persist), + Ok((strategy, name, persist)) => run_sweep(strategy, &name, persist, DataSource::Synthetic), Err(msg) => { eprintln!("aura: {msg}"); std::process::exit(2); } }, - ["walkforward"] => run_walkforward("walkforward", false), - ["walkforward", "--name", n] => run_walkforward(n, false), - ["walkforward", "--trace", n] => run_walkforward(n, true), + ["walkforward"] => run_walkforward("walkforward", false, DataSource::Synthetic), + ["walkforward", "--name", n] => run_walkforward(n, false, DataSource::Synthetic), + ["walkforward", "--trace", n] => run_walkforward(n, true, DataSource::Synthetic), ["mc"] => run_mc("mc", false), ["mc", "--name", n] => run_mc(n, false), ["mc", "--trace", n] => run_mc(n, true), @@ -1446,7 +1469,7 @@ mod tests { fn sample_blueprint_with_sinks_bootstraps_runs_and_drains() { // the factory returns the two Recorder receivers (build_sample drops them), // so a caller can bootstrap one point, run it, and drain both sinks. - let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(); + let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(SYNTHETIC_PIP_SIZE); let mut h = bp .with("signals.trend.fast.length", 2) .with("signals.trend.slow.length", 4) @@ -1521,7 +1544,11 @@ mod tests { // a fresh temp store (the run_* fns themselves bind `default_registry()`): // engine family -> per-kind extractor -> append_family. let sid = reg - .append_family("sweep", FamilyKind::Sweep, &sweep_member_reports(&sweep_family(None))) + .append_family( + "sweep", + FamilyKind::Sweep, + &sweep_member_reports(&sweep_family(None, &DataSource::Synthetic)), + ) .expect("sweep family"); let mid = reg .append_family("mc", FamilyKind::MonteCarlo, &mc_member_reports(&mc_family(None))) @@ -1530,7 +1557,7 @@ mod tests { .append_family( "walkforward", FamilyKind::WalkForward, - &walkforward_member_reports(&walkforward_family(None)), + &walkforward_member_reports(&walkforward_family(None, &DataSource::Synthetic)), ) .expect("walkforward family"); assert_eq!((sid.as_str(), mid.as_str(), wid.as_str()), ("sweep-0", "mc-0", "walkforward-0")); @@ -1832,7 +1859,7 @@ mod tests { // pins the default node-name path segments (ema / exposure / longonly) and // the param order/kinds the member key + sweep depend on. let names: Vec = - momentum_blueprint_with_sinks().0.param_space().into_iter().map(|p| p.name).collect(); + momentum_blueprint_with_sinks(SYNTHETIC_PIP_SIZE).0.param_space().into_iter().map(|p| p.name).collect(); assert_eq!( names, vec![ @@ -1845,8 +1872,8 @@ mod tests { #[test] fn momentum_sweep_is_deterministic_and_has_eight_points() { - let a = momentum_sweep_family(None); - let b = momentum_sweep_family(None); + let a = momentum_sweep_family(None, &DataSource::Synthetic); + let b = momentum_sweep_family(None, &DataSource::Synthetic); assert_eq!(a.points.len(), 8, "2x2x2 grid = 8 points"); assert_eq!(a, b, "C1: the momentum family is a pure function of the build"); }