//! 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, ParamSpec, RunReport, ScalarKind, Timestamp}; use aura_core::{zip_params, Cell, Scalar}; /// 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 { /// The chosen params as the tag-free coordinate [`Cell`]s the inner sweep /// produced (C7: the kind lives once on [`WalkForwardResult::space`], not at the /// value). The named/typed view is `zip_params(&result.space, &chosen_params)` — /// the same idiom as [`SweepFamily::named_params`](crate::SweepFamily). 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 { /// The param-space schema (kinds + slot names), once for the whole family — /// the kinds the per-window [`WindowRun::chosen_params`] cells are read against /// (C7). Identical for every window (same blueprint). Mirrors /// [`SweepFamily::space`](crate::SweepFamily); the two stay distinct types. pub space: Vec, /// 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)>, } impl WalkForwardResult { /// The chosen params of the `window`-th outcome paired with their names — a /// derived view over the carried param-space (reuses [`zip_params`]); the /// walk-forward mirror of [`SweepFamily::named_params`](crate::SweepFamily). pub fn named_params(&self, window: usize) -> Vec<(String, Scalar)> { zip_params(&self.space, &self.windows[window].run.chosen_params) } } /// 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, space: Vec, 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, space, 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, space: Vec, 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); debug_assert!( windows.iter().all(|w| w.run.chosen_params.len() == space.len()), "every window's chosen point must match the param-space arity (same blueprint)" ); WalkForwardResult { space, 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 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). One entry per /// param slot, in slot order; empty if there are no windows or the strategy has no /// params. The numeric coercion is resolved once per slot against the schema /// (`result.space`, C7) — never per window-value: an `i64`/`f64` slot maps to its /// value, a `bool` slot to `0.0`/`1.0` (so `mean` is the fraction of windows that /// chose `true`). A `timestamp` slot is structurally impossible (C20: a timestamp /// is a structural axis, never a numeric knob). pub fn param_stability(result: &WalkForwardResult) -> Vec { if result.windows.is_empty() { return Vec::new(); } let coerce: Vec f64> = result .space .iter() .map(|ps| match ps.kind { ScalarKind::I64 => (|c: Cell| c.i64() as f64) as fn(Cell) -> f64, ScalarKind::F64 => |c: Cell| c.f64(), ScalarKind::Bool => |c: Cell| c.bool() as i64 as f64, ScalarKind::Timestamp => { unreachable!("timestamp is a structural axis (C20), never a param knob") } }) .collect(); coerce .iter() .enumerate() .map(|(slot, f)| { let vals: Vec = result.windows.iter().map(|w| f(w.run.chosen_params[slot])).collect(); MetricStats::from_values(&vals) }) .collect() } #[cfg(test)] mod tests { use super::*; use crate::{summarize, RunManifest}; 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(&[], &[]), } } /// The 2-slot param-space the run-window fixtures choose into: an i64 length /// and an f64 scale. Matches `run_window_fixture` / `mk` arity (debug_assert). fn fixture_space() -> Vec { vec![ ParamSpec { name: "len".to_string(), kind: ScalarKind::I64 }, ParamSpec { name: "scale".to_string(), kind: ScalarKind::F64 }, ] } /// 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![Cell::from_i64(k % 3), Cell::from_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() { // 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, fixture_space(), 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() { // 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(), fixture_space(), 1, run_window_fixture); let many = walk_forward_with_threads(cfg(), fixture_space(), 8, run_window_fixture); assert_eq!(one, many); assert_eq!(one, walk_forward(cfg(), fixture_space(), run_window_fixture)); } #[test] fn stitched_curve_carries_equity_forward() { // 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() { // 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() { // 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![Cell::from_i64(a), Cell::from_f64(b)], oos_equity: vec![], oos_report: dummy_report(), }, }; let result = WalkForwardResult { space: fixture_space(), 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 param_stability_reduces_bool_slot_to_fraction_true() { // a Bool param slot coerces 0/1, so mean = fraction of // windows that chose `true`. 3 of 4 true -> 0.75. let mk = |flag: bool| WindowOutcome { bounds: WindowBounds { is: (Timestamp(0), Timestamp(0)), oos: (Timestamp(0), Timestamp(0)), }, run: WindowRun { chosen_params: vec![Cell::from_bool(flag)], oos_equity: vec![], oos_report: dummy_report(), }, }; let result = WalkForwardResult { space: vec![ParamSpec { name: "flag".to_string(), kind: ScalarKind::Bool }], windows: vec![mk(true), mk(false), mk(true), mk(true)], stitched_oos_equity: vec![], }; let stab = param_stability(&result); assert_eq!(stab.len(), 1); assert_eq!(stab[0].mean, 0.75); } #[test] fn named_params_pairs_each_name_with_the_chosen_window_value() { // #76: WalkForwardResult::named_params(w) is the typed/named view — it // pairs each param-space NAME with window `w`'s chosen VALUE, in slot // order, reading the right window. Mirror of SweepFamily::named_params. // Expected vectors are spelled out explicitly (not via zip_params) so the // assertion pins the observable pairing, not the implementation. let mk = |fast: i64, scale: f64| WindowOutcome { bounds: WindowBounds { is: (Timestamp(0), Timestamp(0)), oos: (Timestamp(0), Timestamp(0)), }, run: WindowRun { chosen_params: vec![Cell::from_i64(fast), Cell::from_f64(scale)], oos_equity: vec![], oos_report: dummy_report(), }, }; let result = WalkForwardResult { space: vec![ ParamSpec { name: "sma.fast".to_string(), kind: ScalarKind::I64 }, ParamSpec { name: "bias.scale".to_string(), kind: ScalarKind::F64 }, ], windows: vec![mk(8, 0.25), mk(13, 0.75)], stitched_oos_equity: vec![], }; assert_eq!( result.named_params(0), vec![ ("sma.fast".to_string(), Scalar::i64(8)), ("bias.scale".to_string(), Scalar::f64(0.25)), ], ); assert_eq!( result.named_params(1), vec![ ("sma.fast".to_string(), Scalar::i64(13)), ("bias.scale".to_string(), Scalar::f64(0.75)), ], ); } #[test] fn roller_rolling_emits_consecutive_oos_windows() { // 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() { // 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() { // (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() { // 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 }, ); } }