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

feat(real-family): thread DataSource + pip through the family builders

Browse Source 
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
This commit is contained in:
Brummel
2026-06-21 14:55:39 +02:00
parent cf102172ec
commit 2a6ca47acd
1 changed files with 82 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/main.rs
+82 -55
View File
@@ -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<Box<dyn aura_engine::Source>> = 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<Box<dyn aura_engine::Source>> {
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<Scalar>)>,
Receiver<(Timestamp, Vec<Scalar>)>,
@@ -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<String> = 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<Box<dyn aura_engine::Source>> =
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::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
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<Scalar>)>,
Receiver<(Timestamp, Vec<Scalar>)>,
@@ -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/<name>/<member_key>/`.
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<String> = 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<Box<dyn aura_engine::Source>> =
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::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
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/<n>/<member_key>/` (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/<n>/oos<ns>/` (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<Box<dyn aura_engine::Source>> =
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<Box<dyn aura_engine::Source>> =
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::<Vec<_>>(), 0);
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 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<Box<dyn aura_engine::Source>> =
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::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
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<String> =
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");
}