diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 9480ed8..b6d1763 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -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)> = rx_eq.try_iter().collect(); - let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); - let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); - let req_rows: Vec<(Timestamp, Vec)> = 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)> = 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)> = rx_eq.try_iter().collect(); + let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let req_rows: Vec<(Timestamp, Vec)> = 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, slow_len: Option, 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"); diff --git a/crates/aura-cli/tests/cli_run.rs b/crates/aura-cli/tests/cli_run.rs index 068204a..92dce18 100644 --- a/crates/aura-cli/tests/cli_run.rs +++ b/crates/aura-cli/tests/cli_run.rs @@ -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); +}