feat(aura-engine): walk-forward family — WindowRoller + walk_forward (C12 axis 3)
The third C12 orchestration axis: walk-forward varies the data window. A
WindowRoller is a pure iterator of WindowBounds { is, oos } — bounds only, never
tick data (#71 firewall). walk_forward runs a caller closure per split disjointly
over the shared run_indexed core (C1), and stitches the OOS pip-equity segments
into one continuous curve (each segment offset by the running sum of prior
segments' finals; an empty segment contributes 0.0, mirroring summarize). C2
no-look-ahead is structural: the roller only ever emits oos.0 == is.1 + 1, pinned
by a zero-tick bounds test. The in-sample optimize (axis 2) is closure-supplied,
not called by the engine: aura-engine has no aura-registry dependency (C9), and
C12 forbids baking search policy into the primitive — the CLI bridges both crates.
Param-stability is on-demand, not stored (the R2 decision, recorded with
provenance on #69): WalkForwardResult stores only the raw per-window outcomes +
the stitched curve, and param_stability(&result) -> Vec<MetricStats> is a public
reduction over the retained per-window chosen params. A stored summary would be
recomputable from the windows (redundant) and would force one statistic canonical
when "stability" admits several; this mirrors SweepFamily (raw points + on-demand
optimize), not McFamily (stored aggregate). MetricStats::from_values is extracted
from McAggregate::from_draws (behaviour-preserving — the 7 mc tests stay green)
and gains serde so the CLI summary renders it and #70 lineage can persist it.
`aura walkforward` runs a built-in rolling walk-forward over a synthetic windowed
source: per window it sweeps the built-in grid in-sample, optimizes by total_pips,
runs the chosen params out-of-sample, persists each OOS RunReport (C18), and prints
a stitched summary line.
Two plan deviations, both compiler-forced and consistent with siblings: added
#[derive(Debug)] to WindowRoller (the RED tests' unwrap_err needs Ok: Debug; the
sibling types already derive Debug); removed a now-redundant test-module
SyntheticSpec import after Step 2 hoisted it to the top-level use.
Verification: cargo test --workspace green (aura-engine 152 incl. walkforward 9 +
mc 9, aura-cli 14); clippy --workspace --all-targets -D warnings exit 0; cargo doc
--workspace --no-deps clean.
refs #69
This commit is contained in:
Generated
+1
@@ -56,6 +56,7 @@ dependencies = [
|
||||
"aura-registry",
|
||||
"aura-std",
|
||||
"data-server",
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -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 }
|
||||
|
||||
+180
-8
@@ -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::<Vec<_>>(), 0);
|
||||
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 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::<Vec<_>>(), 0);
|
||||
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 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 <SYMBOL> [--from <ms>] [--to <ms>] | aura graph | aura sweep | aura runs list | aura runs rank <metric>";
|
||||
"usage: aura run [--macd] | aura run --real <SYMBOL> [--from <ms>] [--to <ms>] | aura graph | aura sweep | aura walkforward | aura runs list | aura runs rank <metric>";
|
||||
|
||||
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
|
||||
|
||||
@@ -23,8 +23,11 @@
|
||||
//! recording sink's `Vec<Scalar>` 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.
|
||||
|
||||
@@ -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,15 +53,14 @@ pub struct MetricStats {
|
||||
pub p95: f64,
|
||||
}
|
||||
|
||||
impl McAggregate {
|
||||
/// Pure reduction over the realizations — recomputable from `draws` alone (it
|
||||
/// is exactly what [`monte_carlo`] stored). `draws` must be non-empty (a
|
||||
/// Monte-Carlo over zero realizations has no defined mean/quantile); on a
|
||||
/// 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<f64> = draws.iter().map(|d| f(&d.report.metrics)).collect();
|
||||
xs.sort_by(|a, b| a.partial_cmp(b).expect("run metrics are finite"));
|
||||
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::<f64>() / xs.len() as f64,
|
||||
p5: quantile(&xs, 0.05),
|
||||
@@ -70,6 +69,18 @@ impl McAggregate {
|
||||
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
|
||||
/// Monte-Carlo over zero realizations has no defined mean/quantile); on a
|
||||
/// non-empty slice every field is finite.
|
||||
pub fn from_draws(draws: &[McDraw]) -> McAggregate {
|
||||
let pick = |f: fn(&RunMetrics) -> f64| -> MetricStats {
|
||||
let xs: Vec<f64> = 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<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 {
|
||||
pub chosen_params: Vec<Scalar>,
|
||||
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<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)>,
|
||||
}
|
||||
|
||||
/// 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, 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<F>(
|
||||
roller: WindowRoller,
|
||||
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 runs = run_indexed(bounds.len(), nthreads, |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);
|
||||
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<MetricStats> {
|
||||
let Some(first) = result.windows.first() else {
|
||||
return Vec::new();
|
||||
};
|
||||
let nslots = first.run.chosen_params.len();
|
||||
(0..nslots)
|
||||
.map(|slot| {
|
||||
let vals: Vec<f64> = 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<Scalar>) -> 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<WindowBounds> =
|
||||
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::<Vec<_>>(),
|
||||
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<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() {
|
||||
// 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<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() {
|
||||
// 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 },
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user