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

feat(0070): wire folding sink reductions into the stage1-r sweep (closes #138)

Browse Source 
Completes the M3 fix for BLOCKER #138. The no-trace stage1-r sweep now folds its
reductions online: `stage1_r_graph(reduce=true)` wires `SeriesReducer` (equity,
exposure -> one summary row each) and `GatedRecorder` (gated on
`closed_this_cycle` -> only the trade rows + the final row), and the folded
consumer rebuilds `RunMetrics` from those compact outputs. Per-member memory drops
from O(cycles) (~2 GiB over ~5.5M one-minute bars) to O(trades), so full-history
multi-member sweeps no longer exhaust RAM+swap. The `r_equity` tap is omitted when
reducing (it is consumed only by the `--trace` persistence path).

`reduce = trace.is_none()`: the `--trace` path and the single `run_stage1_r` keep
the raw recorders + `summarize`/`summarize_r` over full rows — the byte-for-byte
reference. A new CLI-seam test
(`sweep_strategy_stage1_r_folded_no_trace_metrics_equal_raw_trace_metrics`) proves
the folded no-trace sweep reports byte-identical per-member metrics to the raw
`--trace` path.

Verified: cargo test --workspace green (42 suites); cargo clippy --workspace
--all-targets -- -D warnings clean; stage1-r goldens byte-identical.

closes #138
This commit is contained in:
Brummel
2026-06-25 11:24:27 +02:00
parent cfe7bad0ff
commit eb8653488f
2 changed files with 169 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/main.rs
+70 -30
View File
@@ -19,7 +19,7 @@ use aura_composites::{risk_executor, risk_executor_vol_open, StopRule};
use aura_engine::{
f64_field, join_on_ts, monte_carlo, param_stability, summarize, summarize_r, walk_forward,
window_of, ColumnarTrace, Composite, Edge, FlatGraph, GraphBuilder, Harness, JoinedRow,
McAggregate, McFamily, RollMode, RunManifest, RunReport, SourceSpec, SweepFamily,
McAggregate, McFamily, RollMode, RunManifest, RunMetrics, RunReport, SourceSpec, SweepFamily,
SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds, WindowRoller, WindowRun,
};
use aura_registry::{
@@ -28,7 +28,8 @@ use aura_registry::{
WriteKind,
};
use aura_std::{
Bias, Ema, LinComb, LongOnly, Recorder, SimBroker, Sma, Sub, PM_FIELD_NAMES, PM_RECORD_KINDS,
Bias, Ema, GatedRecorder, LinComb, LongOnly, Recorder, SeriesReducer, SimBroker, Sma, Sub,
PM_FIELD_NAMES, PM_RECORD_KINDS,
};
use std::sync::mpsc::{self, Receiver};
use std::sync::LazyLock;
@@ -1176,7 +1177,7 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG
let (tx_ex, _) = mpsc::channel();
let (tx_r, _) = mpsc::channel();
let (tx_req, _) = mpsc::channel();
let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true);
let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true);
let space = bp.param_space();
// resolve the open stop slots' exact path-qualified names from the live param-space
// (the composite prefix is never hand-synced — match by the stable suffix).
@@ -1209,14 +1210,11 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG
let (tx_ex, rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, rx_req) = mpsc::channel();
let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true)
let reduce = trace.is_none();
let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, reduce)
.bootstrap_with_cells(point)
.expect("stage1-r grid points are kind-checked against param_space");
h.run(data.run_sources());
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let req_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_req.try_iter().collect();
// record the swept knobs PLUS the fixed R-defining bias scale, matching the
// single run: a family member must be reproducible from its own manifest
// (C18). The stop knobs are now real swept axes (they appear in `point`), so
@@ -1231,11 +1229,33 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG
let key = member_key(&named, &varying);
let mut manifest = sim_optimal_manifest(named, window, 0, pip);
manifest.broker = stage1_r_broker_label(pip);
if let Some(name) = trace {
persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows);
}
let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
metrics.r = Some(summarize_r(&r_rows, 0.0));
let metrics = if reduce {
// folded: GatedRecorder emits O(trades) R rows; each SeriesReducer
// emits one [last, max_drawdown, sign_flips] summary row.
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let (total_pips, max_drawdown) = rx_eq
.try_iter()
.next()
.map(|(_, row)| (row[0].as_f64(), row[1].as_f64()))
.unwrap_or((0.0, 0.0));
let bias_sign_flips =
rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0);
let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None };
m.r = Some(summarize_r(&r_rows, 0.0));
m
} else {
// trace path (--trace set): raw recorders, persist, summarize — today's code.
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let req_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_req.try_iter().collect();
if let Some(name) = trace {
persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows);
}
let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
m.r = Some(summarize_r(&r_rows, 0.0));
m
};
RunReport { manifest, metrics }
})
.expect("the stage1-r named grid matches the stage1-r param-space")
@@ -1929,6 +1949,7 @@ fn stage1_r_graph(
fast_len: Option<i64>,
slow_len: Option<i64>,
stop_open: bool,
reduce: bool,
) -> Composite {
let mut g = GraphBuilder::new("stage1_r");
// SMA-cross signal → Bias (the same signal the pip sample uses).
@@ -1946,23 +1967,36 @@ fn stage1_r_graph(
let exposure = g.add(Bias::builder().named("bias").bind("scale", Scalar::f64(0.5)));
// pip branch (verbatim from sample_blueprint_with_sinks).
let broker = g.add(SimBroker::builder(SYNTHETIC_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));
// R branch: bias + price → RiskExecutor(vol_stop) → dense R-record. The stop knobs
// are bound to the pinned constants (default) or left open as sweep axes (`stop_open`).
// In `reduce` mode the per-cycle taps fold online: SeriesReducer folds the eq/ex
// f64 series to one summary row, GatedRecorder retains only the gated R rows — the
// O(cycles)→O(trades) memory win. The raw `Recorder`s (and the r_equity tap) are the
// `--trace` path, where the full per-cycle series is persisted.
let gate_col = PM_FIELD_NAMES
.iter()
.position(|&n| n == "closed_this_cycle")
.expect("PM record has a closed_this_cycle column");
let eq = if reduce {
g.add(SeriesReducer::builder(Firing::Any, tx_eq))
} else {
g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq))
};
let ex = if reduce {
g.add(SeriesReducer::builder(Firing::Any, tx_ex))
} else {
g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex))
};
let exec = g.add(if stop_open {
risk_executor_vol_open(1.0)
} else {
risk_executor(StopRule::Vol { length: STAGE1_R_STOP_LENGTH, k: STAGE1_R_STOP_K }, 1.0)
});
let rrec = g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r));
// r_equity = cum_realized_r + unrealized_r — one tapped series for charting.
let r_equity = g.add(
LinComb::builder(2)
.bind("weights[0]", Scalar::f64(1.0))
.bind("weights[1]", Scalar::f64(1.0)),
);
let req = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_req));
let rrec = if reduce {
g.add(GatedRecorder::builder(PM_RECORD_KINDS.to_vec(), gate_col, Firing::Any, tx_r))
} else {
g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r))
};
let price = g.source_role("price", ScalarKind::F64);
g.feed(
price,
@@ -1971,19 +2005,25 @@ fn stage1_r_graph(
g.connect(fast.output("value"), spread.input("lhs"));
g.connect(slow.output("value"), spread.input("rhs"));
g.connect(spread.output("value"), exposure.input("signal"));
// bias fans to: broker (pip), exposure sink, and the RiskExecutor (R).
g.connect(exposure.output("bias"), broker.input("exposure"));
g.connect(exposure.output("bias"), ex.input("col[0]"));
g.connect(exposure.output("bias"), exec.input("bias"));
g.connect(broker.output("equity"), eq.input("col[0]"));
// tap the dense R-record (all PM fields) for summarize_r.
for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
g.connect(exec.output(field), rrec.input(COL_PORTS[i].as_str()));
}
// r_equity sum + tap.
g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]"));
g.connect(exec.output("unrealized_r"), r_equity.input("term[1]"));
g.connect(r_equity.output("value"), req.input("col[0]"));
if !reduce {
// r_equity = cum_realized_r + unrealized_r — one tapped series for charting.
let r_equity = g.add(
LinComb::builder(2)
.bind("weights[0]", Scalar::f64(1.0))
.bind("weights[1]", Scalar::f64(1.0)),
);
let req = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_req));
g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]"));
g.connect(exec.output("unrealized_r"), r_equity.input("term[1]"));
g.connect(r_equity.output("value"), req.input("col[0]"));
}
g.build().expect("stage1_r wiring resolves")
}
@@ -2003,7 +2043,7 @@ fn run_stage1_r(data: RunData, trace: Option<&str>) -> RunReport {
let (tx_ex, rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, rx_req) = mpsc::channel();
let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false)
let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false)
.compile_with_params(&[])
.expect("valid stage1-r blueprint");
let mut h = Harness::bootstrap(flat).expect("valid stage1-r harness");
crates/aura-cli/tests/cli_run.rs
+99
View File
@@ -1954,3 +1954,102 @@ fn sweep_strategy_stage1_r_no_flags_keeps_the_default_grid() {
);
let _ = std::fs::remove_dir_all(&cwd);
}
/// Extract the balanced `"metrics":{...}` object substring from one stdout member
/// line. The `family_id` prefix legitimately differs between a no-trace sweep
/// (`sweep-0`) and a `--trace n` sweep (`n-0`), so an equivalence check must
/// compare only the `metrics` block (the reduced/raw output under test), not the
/// whole line. Brace-balanced scan, not a JSON parser (no serde dep in this test
/// crate); the member lines are flat enough for this to be exact.
fn metrics_object(line: &str) -> &str {
let key = "\"metrics\":";
let start = line.find(key).expect("member line has a metrics block") + key.len();
let bytes = line.as_bytes();
assert_eq!(bytes[start], b'{', "metrics value must be an object: {line}");
let mut depth = 0usize;
for (i, &b) in bytes[start..].iter().enumerate() {
match b {
b'{' => depth += 1,
b'}' => {
depth -= 1;
if depth == 0 {
return &line[start..start + i + 1];
}
}
_ => {}
}
}
panic!("unbalanced metrics object: {line}")
}
/// Property (#138, task 8 — the streaming-reduction wiring invariant): a no-trace
/// `aura sweep --strategy stage1-r` (the *folded* path — `SeriesReducer` folds the
/// eq/ex series to one summary row each, `GatedRecorder` retains only the gated R
/// rows, so memory is O(trades) not O(cycles)) reports **byte-for-byte the same
/// `metrics` object per member** as the `--trace` path (the *raw-recorder*
/// reference that buffers every per-cycle row). This pins the actual M3 fix at the
/// CLI seam: the engine-level Task-7 equivalence tests prove the sink nodes agree
/// in isolation, but only this end-to-end check proves the CLI *wiring* —
/// `reduce = trace.is_none()` selecting the folding topology, and the folded
/// consumer reconstructing `RunMetrics` (`total_pips`/`max_drawdown` from the eq
/// `SeriesReducer` row, `bias_sign_flips` from the ex row, `r` from `summarize_r`
/// over the gated rows) — produces the identical observable report. Members are
/// matched in stdout order (both paths iterate the same deterministic grid, C1),
/// and the `metrics` block excludes the family_id, which is allowed to differ.
#[test]
fn sweep_strategy_stage1_r_folded_no_trace_metrics_equal_raw_trace_metrics() {
let cwd = temp_cwd("sweep-stage1-r-fold-vs-raw");
// folded path: no --trace -> reduce = true (SeriesReducer + GatedRecorder).
let folded = Command::new(BIN)
.args(["sweep", "--strategy", "stage1-r"])
.current_dir(&cwd)
.output()
.expect("spawn folded (no-trace) stage1-r sweep");
assert!(
folded.status.success(),
"folded no-trace sweep exit: {:?}; stderr: {}",
folded.status,
String::from_utf8_lossy(&folded.stderr)
);
let folded_out = String::from_utf8(folded.stdout).expect("utf-8");
// raw path: --trace -> reduce = false (the verbatim Recorder reference).
let raw = Command::new(BIN)
.args(["sweep", "--strategy", "stage1-r", "--trace", "t1"])
.current_dir(&cwd)
.output()
.expect("spawn raw (--trace) stage1-r sweep");
assert!(
raw.status.success(),
"raw --trace sweep exit: {:?}; stderr: {}",
raw.status,
String::from_utf8_lossy(&raw.stderr)
);
let raw_out = String::from_utf8(raw.stdout).expect("utf-8");
let folded_lines: Vec<&str> = folded_out.lines().collect();
let raw_lines: Vec<&str> = raw_out.lines().collect();
// both paths run the same 2×2 grid -> 4 members, in the same deterministic order.
assert_eq!(folded_lines.len(), 4, "folded sweep must print 4 members: {folded_out:?}");
assert_eq!(
raw_lines.len(),
folded_lines.len(),
"the two paths must yield the same member count: folded {} vs raw {}",
folded_lines.len(),
raw_lines.len()
);
for (i, (f, r)) in folded_lines.iter().zip(raw_lines.iter()).enumerate() {
let fm = metrics_object(f);
let rm = metrics_object(r);
// the metrics block is non-trivial: it must carry the R yardstick, so this
// is comparing a real reduced output, not two empty objects.
assert!(fm.contains("\"r\":{"), "folded member {i} must carry an r block: {fm}");
assert_eq!(
fm, rm,
"folded vs raw metrics diverge at member {i}\n folded: {fm}\n raw: {rm}"
);
}
let _ = std::fs::remove_dir_all(&cwd);
}