Files
Aura/crates/aura-engine/src/walkforward.rs
T
claude b048923f1c feat(engine): shared rayon pool for the disjoint-parallel core
Replace the per-call std::thread::scope fan-out in run_indexed with one
process-wide rayon pool. run_indexed's callers nest — sweep (grid
points), monte_carlo (seeds), and walk_forward (window bounds), where a
walk_forward window runs an inner sweep — and each std::thread::scope
spawned its own available_parallelism() OS-thread batch, so the two
levels multiplied to ~available_parallelism()^2 threads (measured
150-240 concurrent on a 24-core box during a 42-cell campaign, ~10x
oversubscription). rayon's pool is process-wide and persistent: a nested
par_iter work-steals within the same N workers instead of spawning, so
live workers never exceed the pool size.

Determinism (C1) is preserved bit-for-bit: run_indexed collects via
IndexedParallelIterator in enumeration order (not completion order), and
the float aggregation (summarize_r / McAggregate / stitch) runs after
the ordered collect, so IEEE-754 non-associativity introduces no
variance. The existing _with_threads(1) == _with_threads(8) == public
determinism tests stay green, now driving local rayon pools of 1 and N
workers via ThreadPool::install.

Structural inversion: the public sweep/monte_carlo/walk_forward become
the core (calling run_indexed on the ambient pool through a shared
assemble_* helper); the _with_threads variants collapse into thin,
test-only wrappers (#[cfg(test)]) that run the same assembly inside a
local pool. run_one stays Fn + Sync (no + Send): run_indexed borrows it
in the map closure and the wrappers borrow it into install — passing it
by value would move it into rayon's Map and force F: Send, a bound the
public callers do not carry (hence the load-bearing
#[allow(clippy::redundant_closure)] on run_indexed).

Adds a nested-oversubscription regression test (asserts nested
run_indexed never exceeds the pool size — the pre-rayon shape ran
POOL^2) and an ignored wall-time benchmark. rayon admitted under the
amended-C16 per-case dependency policy; it stays within C1 (parallelism
across sims, never within one).

This is the shared pool only — the first of three steps toward
saturating the cores. The diagnosis found the larger loss is the idle
valleys (~42% of the box idle: the sequential cell loop and the
single-threaded bootstrap), which the pool alone does not fill. Two
follow-ups are filed and recorded on #268: the Registry write path is
not concurrency-safe (append_family races on a shared families.jsonl),
the prerequisite for parallelizing the C1-disjoint cell loop with a RAM
budget bounding concurrently-active instruments (the FileCache retains a
symbol's parsed files while any reference is live).

Verified: cargo test -p aura-engine (276 passed, the C1 determinism
tests green), the new oversubscription test green, the benchmark runs
(no overflow), clippy -D warnings clean, cargo test --workspace no
failures.

closes #268
2026-07-16 00:30:10 +02:00

604 lines
25 KiB
Rust

//! 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<WindowRoller, WalkForwardError> {
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<WindowBounds> {
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<Cell>,
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<ParamSpec>,
/// Per-window outcomes in roll order (window 0 first), independent of thread
/// completion (C1).
pub windows: Vec<WindowOutcome>,
/// 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)
}
}
/// The single assembly path behind both [`walk_forward`] and the thread-count-
/// explicit test wrapper: run `run_window` over each already-rolled `bounds` via
/// `run_indexed` (on whichever pool is ambient when called), stitch the OOS
/// equity, and assemble the result. One source of truth so the two callers can
/// never drift on assembly logic while their pool selection differs.
fn assemble_walk_forward<F>(
space: Vec<ParamSpec>,
bounds: Vec<WindowBounds>,
run_window: &F,
) -> WalkForwardResult
where
F: Fn(WindowBounds) -> WindowRun + Sync,
{
let runs = run_indexed(bounds.len(), |i| run_window(bounds[i]));
let windows: Vec<WindowOutcome> = 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 }
}
/// 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<F>(roller: WindowRoller, space: Vec<ParamSpec>, run_window: F) -> WalkForwardResult
where
F: Fn(WindowBounds) -> WindowRun + Sync,
{
let bounds: Vec<WindowBounds> = roller.collect();
debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll");
assemble_walk_forward(space, bounds, &run_window)
}
/// The thread-count-explicit wrapper of [`walk_forward`]. Module-private: the
/// public `walk_forward` runs on the ambient pool; the tests drive this at 1
/// and at N to pin determinism (C1). The roll enumeration runs outside the
/// pool; `assemble_walk_forward` (run_indexed + stitch + result assembly) runs
/// inside a local rayon pool of `nthreads` workers, sharing the exact assembly
/// path `walk_forward` uses — no second copy to drift. Bounds match
/// `walk_forward` (`F: Sync`): the install closure only borrows `run_window`,
/// never moves it.
#[cfg(test)]
fn walk_forward_with_threads<F>(
roller: WindowRoller,
space: Vec<ParamSpec>,
nthreads: usize,
run_window: F,
) -> WalkForwardResult
where
F: Fn(WindowBounds) -> WindowRun + Sync,
{
let bounds: Vec<WindowBounds> = roller.collect();
debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll");
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(nthreads)
.build()
.expect("rayon thread pool");
pool.install(|| assemble_walk_forward(space, bounds, &run_window))
}
/// 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<MetricStats> {
if result.windows.is_empty() {
return Vec::new();
}
let coerce: Vec<fn(Cell) -> 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<f64> =
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(),
defaults: Vec::new(),
window: (Timestamp(0), Timestamp(0)),
seed: 0,
broker: "t".to_string(),
selection: None,
instrument: None,
topology_hash: None,
project: None,
},
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<ParamSpec> {
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<Cell>) -> 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<WindowBounds> =
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::<Vec<_>>(),
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<WindowBounds> = 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<WindowBounds> = 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 },
);
}
}