diff --git a/Cargo.lock b/Cargo.lock index 4ea51fb..c5693e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -56,6 +56,7 @@ dependencies = [ "aura-registry", "aura-std", "data-server", + "serde_json", ] [[package]] diff --git a/crates/aura-cli/Cargo.toml b/crates/aura-cli/Cargo.toml index 6e43078..32e583e 100644 --- a/crates/aura-cli/Cargo.toml +++ b/crates/aura-cli/Cargo.toml @@ -18,3 +18,6 @@ aura-ingest = { path = "../aura-ingest" } # data-server: the local M1 archive `aura run --real` streams from. Mirrors the # git line in crates/aura-ingest/Cargo.toml verbatim (same source of truth). data-server = { git = "http://192.168.178.103:3000/Brummel/data-server.git", branch = "main" } +# serde_json renders the walk-forward summary line (MetricStats + stitched total); +# admitted under the per-case dependency policy, same as aura-engine. +serde_json = { workspace = true } diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index e483deb..09683f4 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -15,10 +15,12 @@ mod render; use aura_core::{Firing, Scalar, ScalarKind, Timestamp}; use aura_engine::{ - f64_field, summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, - RunManifest, RunReport, SourceSpec, SweepFamily, Target, VecSource, + f64_field, param_stability, summarize, walk_forward, Composite, Edge, FlatGraph, + GraphBuilder, Harness, RollMode, RunManifest, RunReport, SourceSpec, SweepFamily, + SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds, WindowRoller, + WindowRun, }; -use aura_registry::{rank_by, Registry}; +use aura_registry::{optimize, rank_by, Registry}; use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub}; use std::sync::mpsc::{self, Receiver}; @@ -428,6 +430,157 @@ fn run_sweep() { } } +/// `aura walkforward`: run a built-in rolling walk-forward over the sample +/// blueprint + a synthetic windowed source. Per window: sweep the built-in grid on +/// the in-sample slice, optimize by total_pips (axis 2 inside axis 3, where +/// aura-cli bridges engine + registry), run the chosen params out-of-sample, +/// persist the OOS RunReport (C18), and print it; finally print the stitched +/// summary line. Deterministic (C1). +fn run_walkforward() { + let reg = default_registry(); + let result = walkforward_family(); + for w in &result.windows { + if let Err(e) = reg.append(&w.run.oos_report) { + eprintln!("aura: {e}"); + std::process::exit(2); + } + println!("{}", w.run.oos_report.to_json()); + } + println!("{}", walkforward_summary_json(&result)); +} + +/// The built-in rolling walk-forward: 24-bar in-sample, 12-bar out-of-sample, +/// 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() -> WalkForwardResult { + let prices = walkforward_prices(); + let span = ( + prices.first().expect("non-empty synthetic stream").0, + prices.last().expect("non-empty synthetic stream").0, + ); + let roller = WindowRoller::new(span, 24, 12, 12, RollMode::Rolling) + .expect("built-in walk-forward config fits the 60-bar synthetic span"); + walk_forward(roller, |w: WindowBounds| { + let is_family = sweep_over(w.is.0, w.is.1); + 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); + WindowRun { chosen_params: best.params, oos_equity, oos_report } + }) +} + +/// 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; + let space = bp.param_space(); + bp.axis("signals.trend.fast.length", [2, 3]) + .axis("signals.trend.slow.length", [4, 5]) + .axis("signals.momentum.fast.length", [2]) + .axis("signals.momentum.slow.length", [4]) + .axis("signals.momentum.signal.length", [3]) + .axis("signals.blend.weights[0]", [1.0]) + .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 mut h = bp + .bootstrap_with_params(point.to_vec()) + .expect("grid points are kind-checked against param_space"); + h.run(vec![Box::new(walkforward_window_source(from, to))]); + let equity = f64_field(&rx_eq.try_iter().collect::>(), 0); + let exposure = f64_field(&rx_ex.try_iter().collect::>(), 0); + let params = space + .iter() + .zip(point) + .map(|(ps, v)| (ps.name.clone(), scalar_as_param_f64(v))) + .collect(); + RunReport { + manifest: sim_optimal_manifest(params, (from, to), 0), + metrics: summarize(&equity, &exposure), + } + }) + .expect("the built-in named grid matches the sample param-space") +} + +/// Run the chosen params over an out-of-sample window; return the recorded +/// pip-equity segment (for stitching) and the OOS RunReport (the C18 record). +fn run_oos(params: &[Scalar], from: Timestamp, to: Timestamp) -> (Vec<(Timestamp, f64)>, RunReport) { + let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(); + let space = bp.param_space(); + let mut h = bp + .bootstrap_with_params(params.to_vec()) + .expect("chosen params are kind-checked against param_space"); + h.run(vec![Box::new(walkforward_window_source(from, to))]); + let equity = f64_field(&rx_eq.try_iter().collect::>(), 0); + let exposure = f64_field(&rx_ex.try_iter().collect::>(), 0); + let named = space + .iter() + .zip(params) + .map(|(ps, v)| (ps.name.clone(), scalar_as_param_f64(v))) + .collect(); + let report = RunReport { + manifest: sim_optimal_manifest(named, (from, to), 0), + metrics: summarize(&equity, &exposure), + }; + (equity, report) +} + +/// The walk-forward summary line: window count, stitched OOS total pips (the last +/// stitched-curve value), and the on-demand per-param stability. Canonical JSON +/// (C14). +fn walkforward_summary_json(result: &WalkForwardResult) -> String { + let total = result.stitched_oos_equity.last().map(|&(_, v)| v).unwrap_or(0.0); + serde_json::json!({ + "walkforward": { + "windows": result.windows.len(), + "stitched_total_pips": total, + "param_stability": param_stability(result), + } + }) + .to_string() +} + +/// A longer deterministic stream than `showcase_prices` — enough for several +/// IS/OOS windows with SMA warm-up. Seed-determined via `SyntheticSpec` (C1). +fn walkforward_prices() -> Vec<(Timestamp, Scalar)> { + let spec = SyntheticSpec { start: 1.0, len: 60, step: 1 }; + let mut src = spec.source(7); + let mut out = Vec::new(); + while let Some(item) = aura_engine::Source::next(&mut src) { + out.push(item); + } + out +} + +/// The in-memory windowed source the built-in demo uses (the firewall mapping to +/// `DataServer::stream_m1_windowed` is the real-data path; the demo stays in-memory, +/// mirroring `run_sweep`'s `showcase_prices`). Inclusive `[from, to]`. +fn walkforward_window_source(from: Timestamp, to: Timestamp) -> VecSource { + VecSource::new( + walkforward_prices() + .into_iter() + .filter(|&(t, _)| t >= from && t <= to) + .collect(), + ) +} + +/// Render the built-in walk-forward as the per-window OOS RunReport lines plus the +/// summary line — the `run_walkforward` shape minus registry persistence. Test +/// helper (mirrors `sweep_report`). +#[cfg(test)] +fn walkforward_report() -> String { + let result = walkforward_family(); + let mut out = String::new(); + for w in &result.windows { + out.push_str(&w.run.oos_report.to_json()); + out.push('\n'); + } + out.push_str(&walkforward_summary_json(&result)); + out.push('\n'); + out +} + /// `aura runs list`: print every stored run record, in store (over-time) order. fn runs_list() { for report in &load_runs_or_exit() { @@ -582,7 +735,7 @@ fn run_macd() -> RunReport { } const USAGE: &str = - "usage: aura run [--macd] | aura run --real [--from ] [--to ] | aura graph | aura sweep | aura runs list | aura runs rank "; + "usage: aura run [--macd] | aura run --real [--from ] [--to ] | aura graph | aura sweep | aura walkforward | aura runs list | aura runs rank "; fn main() { // Collect argv and match the whole vector: every accepted form is exhaustive, @@ -601,6 +754,7 @@ fn main() { }, ["graph"] => print!("{}", render::render_html(&sample_blueprint())), ["sweep"] => run_sweep(), + ["walkforward"] => run_walkforward(), ["runs", "list"] => runs_list(), ["runs", "rank", metric] => runs_rank(metric), ["--help"] | ["-h"] => println!("{USAGE}"), @@ -614,10 +768,28 @@ fn main() { #[cfg(test)] mod tests { use super::*; - // `SyntheticSpec` is used only by the seeded test vehicle below, so it is - // imported here rather than at module top (a top-level import would be - // `unused` in the non-test binary build under `clippy -D warnings`). - use aura_engine::SyntheticSpec; + + #[test] + fn walkforward_report_is_deterministic() { + // spec §Testing 10: the built-in WFO render is byte-identical across two + // calls (C1). + assert_eq!(walkforward_report(), walkforward_report()); + } + + #[test] + fn walkforward_report_has_one_oos_line_per_window_plus_summary() { + // spec §Testing 11: N per-window OOS RunReport lines + one summary line. + let out = walkforward_report(); + let lines: Vec<&str> = out.lines().collect(); + assert_eq!(lines.len(), 4); // built-in roll = 3 windows + 1 summary + assert!(lines[3].contains(r#""walkforward""#), "summary line: {}", lines[3]); + for line in &lines[..3] { + assert!( + line.contains(r#""manifest""#) && line.contains(r#""metrics""#), + "expected an OOS RunReport line, got: {line}", + ); + } + } /// The drained sink trace of a seeded run — the recorded rows of the equity /// and exposure sinks. Compared row-for-row so the C1 seed-determinism diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index 2331dda..d45ed26 100644 --- a/crates/aura-engine/src/lib.rs +++ b/crates/aura-engine/src/lib.rs @@ -23,8 +23,11 @@ //! recording sink's `Vec` rows to it. //! //! Orchestration families of the atomic sim unit -//! (`(topology + params + data-window + seed)`) ship along two of C12's axes: the -//! **grid** axis via [`sweep`] over a [`GridSpace`] into a [`SweepFamily`], and +//! (`(topology + params + data-window + seed)`) ship along three of C12's axes: the +//! **grid** axis via [`sweep`] over a [`GridSpace`] into a [`SweepFamily`], the +//! **window** axis via [`walk_forward`] over a [`WindowRoller`] into a +//! [`WalkForwardResult`] (C12 axis 3: rolling in-sample/out-of-sample splits, the +//! in-sample optimize closure-supplied), and //! the **seed** axis via [`monte_carlo`] over a seed set into an [`McFamily`] //! (C12 axis 4: Monte-Carlo *is* a sweep over seeds; both drive one shared //! disjoint-parallel executor). The **seed** input itself ships too: @@ -34,8 +37,8 @@ //! reduction ships via [`summarize`]. //! //! Still to come (subsequent cycles): the broker-independent position-event -//! output and downstream broker nodes (C10), the random param-sweep and -//! walk-forward orchestration axes, and registry lineage across a family. +//! output and downstream broker nodes (C10), the random param-sweep +//! orchestration axis, and registry lineage across a family. //! //! Visualization is never here: it is a downstream consumer node on the streams. @@ -46,6 +49,7 @@ mod harness; mod mc; mod report; mod sweep; +mod walkforward; pub use blueprint::{ BindError, Binder, BlueprintNode, CompileError, Composite, OutField, Role, @@ -59,6 +63,10 @@ pub use harness::{ pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport}; pub use sweep::{sweep, GridSpace, SweepError, SweepFamily, SweepPoint}; pub use mc::{monte_carlo, McAggregate, McDraw, McFamily, MetricStats}; +pub use walkforward::{ + param_stability, walk_forward, RollMode, WalkForwardError, WalkForwardResult, + WindowBounds, WindowOutcome, WindowRoller, WindowRun, +}; // #29: re-export the core scalar vocabulary a Blueprint builder needs // (SourceSpec.kind is a ScalarKind; sources/Recorder columns are Scalar / // Firing / Timestamp) so a graph builder has one import surface, not two. diff --git a/crates/aura-engine/src/mc.rs b/crates/aura-engine/src/mc.rs index 73a07a9..bc315b7 100644 --- a/crates/aura-engine/src/mc.rs +++ b/crates/aura-engine/src/mc.rs @@ -43,7 +43,7 @@ pub struct McAggregate { /// Mean + a fixed quantile set of one metric across the realizations. `p50` is /// the median (the two coincide by definition), so "mean/median/quantiles" is /// `mean` + `p50` + the surrounding quantiles, with no redundant field. -#[derive(Clone, Debug, PartialEq)] +#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)] pub struct MetricStats { pub mean: f64, pub p5: f64, @@ -53,6 +53,25 @@ pub struct MetricStats { pub p95: f64, } +impl MetricStats { + /// Mean + the fixed type-7 quantile set over a value set. Sorts a copy; + /// `values` must be finite and non-empty. The shared reduction behind both the + /// MC aggregate (a metric across draws) and the walk-forward param-stability + /// summary (a param across windows, [`crate::param_stability`]). + pub fn from_values(values: &[f64]) -> MetricStats { + let mut xs = values.to_vec(); + xs.sort_by(|a, b| a.partial_cmp(b).expect("values are finite")); + MetricStats { + mean: xs.iter().sum::() / xs.len() as f64, + p5: quantile(&xs, 0.05), + p25: quantile(&xs, 0.25), + p50: quantile(&xs, 0.50), + p75: quantile(&xs, 0.75), + p95: quantile(&xs, 0.95), + } + } +} + impl McAggregate { /// Pure reduction over the realizations — recomputable from `draws` alone (it /// is exactly what [`monte_carlo`] stored). `draws` must be non-empty (a @@ -60,16 +79,8 @@ impl McAggregate { /// non-empty slice every field is finite. pub fn from_draws(draws: &[McDraw]) -> McAggregate { let pick = |f: fn(&RunMetrics) -> f64| -> MetricStats { - let mut xs: Vec = draws.iter().map(|d| f(&d.report.metrics)).collect(); - xs.sort_by(|a, b| a.partial_cmp(b).expect("run metrics are finite")); - MetricStats { - mean: xs.iter().sum::() / xs.len() as f64, - p5: quantile(&xs, 0.05), - p25: quantile(&xs, 0.25), - p50: quantile(&xs, 0.50), - p75: quantile(&xs, 0.75), - p95: quantile(&xs, 0.95), - } + let xs: Vec = draws.iter().map(|d| f(&d.report.metrics)).collect(); + MetricStats::from_values(&xs) }; McAggregate { total_pips: pick(|m| m.total_pips), @@ -278,4 +289,26 @@ mod tests { assert_eq!(quantile(&xs, 0.0), 1.0); assert_eq!(quantile(&xs, 1.0), 5.0); } + + #[test] + fn metric_stats_from_values_matches_known_fixture() { + // spec §Testing 9: type-7 quantile + mean over [0,1,2,3,4], directly on the + // extracted reduction: mean=2.0, p50=2.0, p5≈0.2, p95≈3.8 (same numbers the + // aggregate fixture pins, now on from_values). + let s = MetricStats::from_values(&[0.0, 1.0, 2.0, 3.0, 4.0]); + assert_eq!(s.mean, 2.0); + assert_eq!(s.p50, 2.0); + assert!((s.p5 - 0.2).abs() < 1e-9, "p5 = {}", s.p5); + assert!((s.p95 - 3.8).abs() < 1e-9, "p95 = {}", s.p95); + } + + #[test] + fn metric_stats_serde_round_trips() { + // spec §Concrete code shapes: MetricStats gains serde (consistent with the + // report types) so the CLI summary renders it and #70 lineage can persist. + let s = MetricStats::from_values(&[1.0, 2.0, 3.0]); + let json = serde_json::to_string(&s).expect("serialize MetricStats"); + let back: MetricStats = serde_json::from_str(&json).expect("deserialize MetricStats"); + assert_eq!(back, s); + } } diff --git a/crates/aura-engine/src/walkforward.rs b/crates/aura-engine/src/walkforward.rs new file mode 100644 index 0000000..1ed2c29 --- /dev/null +++ b/crates/aura-engine/src/walkforward.rs @@ -0,0 +1,475 @@ +//! Walk-forward orchestration family (C12 axis 3): walk-forward varies the *data +//! window*, not a param (axis 1) or a seed (axis 4). A [`WindowRoller`] is a pure +//! iterator of [`WindowBounds`] — (in-sample, out-of-sample) time splits over a +//! span, holding NO tick data (the #71 eager-agnostic firewall: a 20y history is +//! the ~190GB eager object, so the surface carries bounds, never a materialized +//! stream). [`walk_forward`] runs a caller closure per split disjointly (C1, via +//! the shared [`run_indexed`](crate::sweep) core) and stitches the OOS pip-equity +//! segments into one continuous curve. C2 no-look-ahead is a pure bounds invariant +//! (`oos.0 > is.1`), checkable with zero ticks. The in-sample optimize (axis 2) is +//! closure-supplied: the engine cannot depend on the registry (C9), and C12 forbids +//! baking search policy into the primitive. Param stability is an on-demand +//! reduction ([`param_stability`]), not a stored field — the result is non-redundant. + +use crate::sweep::run_indexed; +use crate::{MetricStats, RunReport, Scalar, Timestamp}; + +/// One rolling split's time bounds: an in-sample window and the out-of-sample +/// window that follows it. Carries ONLY bounds — never tick data (#71). C2 +/// no-look-ahead is a pure invariant on these: `oos.0 > is.1`. Bounds are +/// inclusive `(from, to)` epoch-unit timestamps, matching `RunManifest.window`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct WindowBounds { + pub is: (Timestamp, Timestamp), + pub oos: (Timestamp, Timestamp), +} + +/// Rolling (fixed-length IS, start advances by `step`) vs anchored (IS always +/// starts at the span origin and grows; OOS positions identical to rolling). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RollMode { + Rolling, + Anchored, +} + +/// A structural fault configuring a walk-forward — the typed gate before any run, +/// analog to [`SweepError`](crate::SweepError). +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum WalkForwardError { + /// `is_len` / `oos_len` / `step` was `<= 0`. + NonPositiveLength { field: &'static str, value: i64 }, + /// The span is too short to fit even window 0. + SpanTooShort { span: (Timestamp, Timestamp), need: i64 }, +} + +/// A pure iterator of [`WindowBounds`] over an inclusive `[origin, end]` span — +/// holds no tick data (#71). For window `k` (`k = 0, 1, ...`): +/// +/// ```text +/// is_start_k = origin + k*step (Rolling) | origin (Anchored) +/// oos_start_k = origin + is_len + k*step +/// is_k = (is_start_k, oos_start_k - 1) +/// oos_k = (oos_start_k, oos_start_k + oos_len - 1) +/// ``` +/// +/// emitted while `oos_k.1 <= end`. Every split satisfies `oos.0 == is.1 + 1 > is.1` +/// (C2). OOS positions are identical in both modes; only the IS start differs. +#[derive(Debug)] +pub struct WindowRoller { + origin: i64, + end: i64, + is_len: i64, + oos_len: i64, + step: i64, + mode: RollMode, + k: i64, +} + +impl WindowRoller { + /// Validate config: `is_len`/`oos_len`/`step` must be `> 0`; the span must fit + /// at least window 0 (`origin + is_len + oos_len - 1 <= end`). Else + /// [`WalkForwardError`]. + pub fn new( + span: (Timestamp, Timestamp), + is_len: i64, + oos_len: i64, + step: i64, + mode: RollMode, + ) -> Result { + if is_len <= 0 { + return Err(WalkForwardError::NonPositiveLength { field: "is_len", value: is_len }); + } + if oos_len <= 0 { + return Err(WalkForwardError::NonPositiveLength { field: "oos_len", value: oos_len }); + } + if step <= 0 { + return Err(WalkForwardError::NonPositiveLength { field: "step", value: step }); + } + let Timestamp(origin) = span.0; + let Timestamp(end) = span.1; + let need = is_len + oos_len - 1; + if origin + need > end { + return Err(WalkForwardError::SpanTooShort { span, need }); + } + Ok(WindowRoller { origin, end, is_len, oos_len, step, mode, k: 0 }) + } +} + +impl Iterator for WindowRoller { + type Item = WindowBounds; + + fn next(&mut self) -> Option { + let oos_start = self.origin + self.is_len + self.k * self.step; + let oos_end = oos_start + self.oos_len - 1; + if oos_end > self.end { + return None; + } + let is_start = match self.mode { + RollMode::Rolling => self.origin + self.k * self.step, + RollMode::Anchored => self.origin, + }; + self.k += 1; + Some(WindowBounds { + is: (Timestamp(is_start), Timestamp(oos_start - 1)), + oos: (Timestamp(oos_start), Timestamp(oos_end)), + }) + } +} + +/// What a per-window run yields: the params chosen on the in-sample window, the +/// recorded OOS pip-equity segment (for stitching), and the OOS [`RunReport`] (the +/// per-window C18 record). Self-describing, analog to +/// [`SweepPoint`](crate::SweepPoint)/[`McDraw`](crate::McDraw). +#[derive(Clone, Debug, PartialEq)] +pub struct WindowRun { + pub chosen_params: Vec, + pub oos_equity: Vec<(Timestamp, f64)>, + pub oos_report: RunReport, +} + +/// One completed window: its bounds + the run the closure produced, in roll order. +#[derive(Clone, Debug, PartialEq)] +pub struct WindowOutcome { + pub bounds: WindowBounds, + pub run: WindowRun, +} + +/// The result family of a walk-forward. Analog to [`SweepFamily`](crate::SweepFamily) +/// — it stores ONLY the raw per-window outcomes plus the stitched curve; any +/// summary over the windows (param stability, etc.) is an on-demand reduction +/// ([`param_stability`]), NOT a stored field (just as a `SweepFamily` stores raw +/// points and `optimize`/`rank_by` are computed on demand). Non-redundant by +/// construction. +#[derive(Clone, Debug, PartialEq)] +pub struct WalkForwardResult { + /// Per-window outcomes in roll order (window 0 first), independent of thread + /// completion (C1). + pub windows: Vec, + /// The stitched continuous OOS pip-equity curve: each window's OOS segment + /// offset by the running sum of all prior segments' final cumulative values, + /// so equity carries forward across the IS/OOS boundaries (continuous, not a + /// per-window sawtooth). An empty OOS segment contributes `0.0` to the running + /// offset and no points, so an equity-less window leaves the curve unbroken. + /// For the canonical `step == oos_len` tiling the segment timestamps are also + /// contiguous. + pub stitched_oos_equity: Vec<(Timestamp, f64)>, +} + +/// Roll the windows, run `run_window` on each disjointly in parallel (C1, via the +/// shared [`run_indexed`](crate::sweep) core), then stitch the OOS equity into one +/// continuous curve. The varying dimension is the *data window* (C12 axis 3). +/// Eager-agnostic (#71): `run_window` receives only [`WindowBounds`]; its data +/// comes from a bounds-keyed producer in the closure body, never a materialized +/// stream `Vec` in this API. Precondition: the roller yields `>= 1` window +/// (`WindowRoller::new` rejects an empty roll). Param stability is a separate +/// on-demand reduction ([`param_stability`]), never computed here. +pub fn walk_forward(roller: WindowRoller, run_window: F) -> WalkForwardResult +where + F: Fn(WindowBounds) -> WindowRun + Sync, +{ + let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + walk_forward_with_threads(roller, nthreads, run_window) +} + +/// The thread-count-explicit core of [`walk_forward`]. Module-private: the public +/// `walk_forward` derives the count, while the tests drive it at 1 and at N to pin +/// determinism under concurrency (C1). A thin adapter over +/// [`run_indexed`](crate::sweep): it collects the roller's bounds (tiny — four +/// timestamps each, no tick data), runs each window disjointly, zips the runs back +/// onto their bounds in roll order, and stitches the OOS segments. +fn walk_forward_with_threads( + roller: WindowRoller, + nthreads: usize, + run_window: F, +) -> WalkForwardResult +where + F: Fn(WindowBounds) -> WindowRun + Sync, +{ + let bounds: Vec = roller.collect(); + debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll"); + let runs = run_indexed(bounds.len(), nthreads, |i| run_window(bounds[i])); + let windows: Vec = bounds + .into_iter() + .zip(runs) + .map(|(bounds, run)| WindowOutcome { bounds, run }) + .collect(); + let stitched_oos_equity = stitch(&windows); + WalkForwardResult { windows, stitched_oos_equity } +} + +/// Fold the per-window OOS equity segments into one continuous curve: each +/// segment's samples are offset by the running sum of all prior segments' final +/// cumulative values, so equity carries forward across windows (no per-window +/// reset). An empty segment adds `0.0` to the offset and no points. +fn stitch(windows: &[WindowOutcome]) -> Vec<(Timestamp, f64)> { + let mut out = Vec::new(); + let mut offset = 0.0_f64; + for w in windows { + let seg = &w.run.oos_equity; + for &(ts, v) in seg { + out.push((ts, v + offset)); + } + if let Some(&(_, last)) = seg.last() { + offset += last; + } + } + out +} + +/// On-demand param-stability summary: one [`MetricStats`](crate::MetricStats) per +/// param slot, over the chosen values (`Scalar` coerced to `f64`) across all +/// windows — reuses [`MetricStats::from_values`](crate::MetricStats::from_values). +/// A pure reduction over `result.windows[*].run.chosen_params`, NOT stored on the +/// result (sweep-precedent: a summary is computed on demand, not cached), so it +/// stays non-redundant and the caller can compute any other measure (std, IQR, +/// distinct-count) from the same raw substrate. One entry per param slot, in slot +/// order; empty if there are no windows or the strategy has no params. The slot +/// count is taken from the first window (all windows of one blueprint share the +/// param-space arity, C11). +pub fn param_stability(result: &WalkForwardResult) -> Vec { + let Some(first) = result.windows.first() else { + return Vec::new(); + }; + let nslots = first.run.chosen_params.len(); + (0..nslots) + .map(|slot| { + let vals: Vec = result + .windows + .iter() + .map(|w| scalar_as_f64(w.run.chosen_params[slot])) + .collect(); + MetricStats::from_values(&vals) + }) + .collect() +} + +/// Coerce a numeric `Scalar` to `f64` for stability statistics (mirrors the CLI's +/// `scalar_as_param_f64`). Params are `i64`/`f64` typed values; a non-numeric +/// scalar in a param slot is a wiring bug, surfaced like the engine's other +/// "checked at wiring" violations. +fn scalar_as_f64(s: Scalar) -> f64 { + match s { + Scalar::I64(n) => n as f64, + Scalar::F64(f) => f, + other => unreachable!("non-numeric param scalar: {other:?}"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{summarize, RunManifest}; + use aura_core::Scalar; + + fn dummy_report() -> RunReport { + RunReport { + manifest: RunManifest { + commit: "t".to_string(), + params: Vec::new(), + window: (Timestamp(0), Timestamp(0)), + seed: 0, + broker: "t".to_string(), + }, + metrics: summarize(&[], &[]), + } + } + + /// A WindowOutcome with explicit OOS equity + chosen params (zero bounds — the + /// stitch/stability reductions ignore bounds). + fn outcome(oos_equity: Vec<(Timestamp, f64)>, chosen: Vec) -> WindowOutcome { + WindowOutcome { + bounds: WindowBounds { + is: (Timestamp(0), Timestamp(0)), + oos: (Timestamp(0), Timestamp(0)), + }, + run: WindowRun { chosen_params: chosen, oos_equity, oos_report: dummy_report() }, + } + } + + /// A deterministic per-window run keyed by the OOS start, so distinct windows + /// differ. No real harness — these tests pin the orchestration (roll order, + /// stitch, stability); mc/sweep cover the harness path. Free `fn` => `Sync`. + fn run_window_fixture(w: WindowBounds) -> WindowRun { + let Timestamp(k) = w.oos.0; + WindowRun { + chosen_params: vec![Scalar::I64(k % 3), Scalar::F64(k as f64 * 0.5)], + oos_equity: vec![(w.oos.0, k as f64 * 0.1), (w.oos.1, k as f64 * 0.1 + 1.0)], + oos_report: dummy_report(), + } + } + + #[test] + fn walk_forward_runs_one_window_per_split_in_roll_order() { + // spec §Testing 5: N bounds -> N outcomes, bounds in roll order. + let roller = + WindowRoller::new((Timestamp(0), Timestamp(24)), 10, 5, 5, RollMode::Rolling) + .expect("valid"); + let expected: Vec = + WindowRoller::new((Timestamp(0), Timestamp(24)), 10, 5, 5, RollMode::Rolling) + .expect("valid") + .collect(); + let result = walk_forward(roller, run_window_fixture); + assert_eq!(result.windows.len(), expected.len()); + assert_eq!( + result.windows.iter().map(|w| w.bounds).collect::>(), + expected, + ); + } + + #[test] + fn walk_forward_is_deterministic_across_thread_counts() { + // spec §Testing 6: order is roll order, not completion (C1). + let cfg = || { + WindowRoller::new((Timestamp(0), Timestamp(100)), 20, 5, 5, RollMode::Rolling) + .expect("valid") + }; + let one = walk_forward_with_threads(cfg(), 1, run_window_fixture); + let many = walk_forward_with_threads(cfg(), 8, run_window_fixture); + assert_eq!(one, many); + assert_eq!(one, walk_forward(cfg(), run_window_fixture)); + } + + #[test] + fn stitched_curve_carries_equity_forward() { + // spec §Testing 7: segs [(t0,2),(t1,5)] then [(t2,1),(t3,3)] -> + // [(t0,2),(t1,5),(t2,6),(t3,8)] (offset by prior segment's final value 5). + let windows = vec![ + outcome(vec![(Timestamp(0), 2.0), (Timestamp(1), 5.0)], vec![]), + outcome(vec![(Timestamp(2), 1.0), (Timestamp(3), 3.0)], vec![]), + ]; + assert_eq!( + stitch(&windows), + vec![ + (Timestamp(0), 2.0), + (Timestamp(1), 5.0), + (Timestamp(2), 6.0), + (Timestamp(3), 8.0), + ], + ); + } + + #[test] + fn stitched_curve_passes_through_empty_segment() { + // spec §Testing 7b: an empty OOS segment adds 0.0 to the offset and no + // points: [(t0,2),(t1,5)], [], [(t2,1)] -> [(t0,2),(t1,5),(t2,6)]. + let windows = vec![ + outcome(vec![(Timestamp(0), 2.0), (Timestamp(1), 5.0)], vec![]), + outcome(vec![], vec![]), + outcome(vec![(Timestamp(2), 1.0)], vec![]), + ]; + assert_eq!( + stitch(&windows), + vec![(Timestamp(0), 2.0), (Timestamp(1), 5.0), (Timestamp(2), 6.0)], + ); + } + + #[test] + fn param_stability_reduces_chosen_params_per_slot() { + // spec §Testing 8: on-demand reduction over chosen params across windows. + // slot 0 [2,2,3,3] -> mean 2.5, p50 2.5; slot 1 [1,1,2,2] -> mean 1.5. + let mk = |a: i64, b: f64| WindowOutcome { + bounds: WindowBounds { + is: (Timestamp(0), Timestamp(0)), + oos: (Timestamp(0), Timestamp(0)), + }, + run: WindowRun { + chosen_params: vec![Scalar::I64(a), Scalar::F64(b)], + oos_equity: vec![], + oos_report: dummy_report(), + }, + }; + let result = WalkForwardResult { + windows: vec![mk(2, 1.0), mk(2, 1.0), mk(3, 2.0), mk(3, 2.0)], + stitched_oos_equity: vec![], + }; + let stab = param_stability(&result); + assert_eq!(stab.len(), 2); + assert_eq!(stab[0].mean, 2.5); + assert_eq!(stab[0].p50, 2.5); + assert_eq!(stab[1].mean, 1.5); + } + + #[test] + fn roller_rolling_emits_consecutive_oos_windows() { + // spec §Testing 1: rolling, origin=0, is_len=10, oos_len=5, step=5, end=24. + // w0 is(0,9) oos(10,14); w1 is(5,14) oos(15,19); w2 is(10,19) oos(20,24); + // w3 would need oos(25,29) > 24 -> stop. 3 windows; OOS tiles (step==oos_len). + let roller = + WindowRoller::new((Timestamp(0), Timestamp(24)), 10, 5, 5, RollMode::Rolling) + .expect("valid config"); + let ws: Vec = roller.collect(); + assert_eq!(ws.len(), 3); + assert_eq!( + ws[0], + WindowBounds { is: (Timestamp(0), Timestamp(9)), oos: (Timestamp(10), Timestamp(14)) }, + ); + assert_eq!( + ws[1], + WindowBounds { is: (Timestamp(5), Timestamp(14)), oos: (Timestamp(15), Timestamp(19)) }, + ); + assert_eq!( + ws[2], + WindowBounds { is: (Timestamp(10), Timestamp(19)), oos: (Timestamp(20), Timestamp(24)) }, + ); + } + + #[test] + fn roller_anchored_fixes_is_start_grows_is_end() { + // spec §Testing 2: anchored, same config — IS always starts at origin (0) + // and grows; OOS positions identical to rolling. + let roller = + WindowRoller::new((Timestamp(0), Timestamp(24)), 10, 5, 5, RollMode::Anchored) + .expect("valid config"); + let ws: Vec = roller.collect(); + assert_eq!(ws.len(), 3); + assert_eq!( + ws[0], + WindowBounds { is: (Timestamp(0), Timestamp(9)), oos: (Timestamp(10), Timestamp(14)) }, + ); + assert_eq!( + ws[1], + WindowBounds { is: (Timestamp(0), Timestamp(14)), oos: (Timestamp(15), Timestamp(19)) }, + ); + assert_eq!( + ws[2], + WindowBounds { is: (Timestamp(0), Timestamp(19)), oos: (Timestamp(20), Timestamp(24)) }, + ); + assert!(ws.iter().all(|w| w.is.0 == Timestamp(0))); + } + + #[test] + fn roller_every_split_has_no_lookahead() { + // spec §Testing 3 (C2): every emitted split has oos.0 > is.1, zero ticks. + for mode in [RollMode::Rolling, RollMode::Anchored] { + let roller = WindowRoller::new((Timestamp(0), Timestamp(100)), 20, 7, 3, mode) + .expect("valid config"); + for w in roller { + assert!(w.oos.0 > w.is.1, "look-ahead: oos.0={:?} !> is.1={:?}", w.oos.0, w.is.1); + } + } + } + + #[test] + fn roller_rejects_nonpositive_lengths_and_short_span() { + // spec §Testing 4: typed config faults before any run. + let span = (Timestamp(0), Timestamp(100)); + assert_eq!( + WindowRoller::new(span, 0, 5, 5, RollMode::Rolling).unwrap_err(), + WalkForwardError::NonPositiveLength { field: "is_len", value: 0 }, + ); + assert_eq!( + WindowRoller::new(span, 10, 0, 5, RollMode::Rolling).unwrap_err(), + WalkForwardError::NonPositiveLength { field: "oos_len", value: 0 }, + ); + assert_eq!( + WindowRoller::new(span, 10, 5, 0, RollMode::Rolling).unwrap_err(), + WalkForwardError::NonPositiveLength { field: "step", value: 0 }, + ); + // span too short for window 0: need is_len+oos_len-1 = 14, end=10 < 0+14. + assert_eq!( + WindowRoller::new((Timestamp(0), Timestamp(10)), 10, 5, 5, RollMode::Rolling) + .unwrap_err(), + WalkForwardError::SpanTooShort { span: (Timestamp(0), Timestamp(10)), need: 14 }, + ); + } +}