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
This commit is contained in:
@@ -19,6 +19,14 @@ serde = { workspace = true }
|
||||
# serde_json renders RunReport JSON for both the registry (disk) and stdout
|
||||
# (RunReport::to_json) — one shape, no hand-rolled writer (amended C16).
|
||||
serde_json = { workspace = true }
|
||||
# rayon is admitted under the amended C16 per-case dependency policy (INDEX.md):
|
||||
# it backs `run_indexed`'s work-stealing pool for the sweep/campaign executor.
|
||||
# It stays within C1 (parallelism across sims, never within one) — each sim
|
||||
# still runs its own deterministic, synchronous event loop on one thread; rayon
|
||||
# only fans the *disjoint* sims of a family out across threads, replacing
|
||||
# `std::thread::scope`. No dep on a workspace precedent (chrono's pattern:
|
||||
# direct versioned dep).
|
||||
rayon = "1"
|
||||
|
||||
[dev-dependencies]
|
||||
# aura-std is a TEST-ONLY dependency: the engine's own integration tests
|
||||
|
||||
@@ -110,6 +110,26 @@ fn quantile(sorted: &[f64], p: f64) -> f64 {
|
||||
}
|
||||
}
|
||||
|
||||
/// The single assembly path behind both [`monte_carlo`] and the thread-count-
|
||||
/// explicit test wrapper: run `run_one` over every seed via `run_indexed` (on
|
||||
/// whichever pool is ambient when called), build the per-seed draws in
|
||||
/// seed-input order, and reduce them into the aggregate. One source of truth so
|
||||
/// the two callers can never drift on assembly logic while their pool selection
|
||||
/// differs.
|
||||
fn assemble_mc<F>(base_point: &[Scalar], seeds: &[u64], run_one: &F) -> McFamily
|
||||
where
|
||||
F: Fn(u64, &[Scalar]) -> RunReport + Sync,
|
||||
{
|
||||
let reports = run_indexed(seeds.len(), |i| run_one(seeds[i], base_point));
|
||||
let draws: Vec<McDraw> = seeds
|
||||
.iter()
|
||||
.zip(reports)
|
||||
.map(|(&seed, report)| McDraw { seed, report })
|
||||
.collect();
|
||||
let aggregate = McAggregate::from_draws(&draws);
|
||||
McFamily { draws, aggregate }
|
||||
}
|
||||
|
||||
/// Run `run_one(seed, base_point)` over every seed, disjointly in parallel (C1),
|
||||
/// and collect the family in seed-input order plus the aggregate. The varying
|
||||
/// dimension is the *seed* (C12 axis 4: MC = sweep over seeds), not a tuning
|
||||
@@ -124,15 +144,18 @@ pub fn monte_carlo<F>(base_point: &[Scalar], seeds: &[u64], run_one: F) -> McFam
|
||||
where
|
||||
F: Fn(u64, &[Scalar]) -> RunReport + Sync,
|
||||
{
|
||||
let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||||
monte_carlo_with_threads(base_point, seeds, nthreads, run_one)
|
||||
debug_assert!(!seeds.is_empty(), "monte_carlo requires a non-empty seed set");
|
||||
assemble_mc(base_point, seeds, &run_one)
|
||||
}
|
||||
|
||||
/// The thread-count-explicit core of [`monte_carlo`]. Module-private: the public
|
||||
/// `monte_carlo` 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 runs each seed disjointly and zips the
|
||||
/// reports back onto their seeds in seed-input order.
|
||||
/// The thread-count-explicit wrapper of [`monte_carlo`]. Module-private: the
|
||||
/// public `monte_carlo` runs on the ambient pool; the tests drive this at 1 and
|
||||
/// at N to pin determinism (C1). `assemble_mc` (run_indexed + draw assembly +
|
||||
/// aggregate) runs inside a local rayon pool of `nthreads` workers, sharing the
|
||||
/// exact assembly path `monte_carlo` uses — no second copy to drift. Bounds
|
||||
/// match `monte_carlo` (`F: Sync`): the install closure only borrows `run_one`,
|
||||
/// never moves it.
|
||||
#[cfg(test)]
|
||||
fn monte_carlo_with_threads<F>(
|
||||
base_point: &[Scalar],
|
||||
seeds: &[u64],
|
||||
@@ -143,14 +166,11 @@ where
|
||||
F: Fn(u64, &[Scalar]) -> RunReport + Sync,
|
||||
{
|
||||
debug_assert!(!seeds.is_empty(), "monte_carlo requires a non-empty seed set");
|
||||
let reports = run_indexed(seeds.len(), nthreads, |i| run_one(seeds[i], base_point));
|
||||
let draws: Vec<McDraw> = seeds
|
||||
.iter()
|
||||
.zip(reports)
|
||||
.map(|(&seed, report)| McDraw { seed, report })
|
||||
.collect();
|
||||
let aggregate = McAggregate::from_draws(&draws);
|
||||
McFamily { draws, aggregate }
|
||||
let pool = rayon::ThreadPoolBuilder::new()
|
||||
.num_threads(nthreads)
|
||||
.build()
|
||||
.expect("rayon thread pool");
|
||||
pool.install(|| assemble_mc(base_point, seeds, &run_one))
|
||||
}
|
||||
|
||||
/// Distribution of `E[R]` under a moving-block bootstrap of an OOS per-trade R series
|
||||
|
||||
+140
-68
@@ -12,7 +12,6 @@
|
||||
use aura_core::{zip_params, Cell, ParamSpec, Scalar, ScalarKind};
|
||||
use crate::RunReport;
|
||||
use crate::harness::SplitMix64;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
/// A validated cartesian grid over a blueprint's param-space: one discrete
|
||||
/// value-list per param slot, in `param_space()` order. Enumerates a family of
|
||||
@@ -366,77 +365,20 @@ impl SweepFamily {
|
||||
}
|
||||
}
|
||||
|
||||
/// Run `run_one` over every point the `space` produces (`Space::points` —
|
||||
/// `GridSpace`'s cartesian product or `RandomSpace`'s seeded draws), disjointly
|
||||
/// in parallel (C1), and collect the family in enumeration order. `run_one`
|
||||
/// builds + bootstraps + runs +
|
||||
/// summarizes one point; it shares nothing mutable, so it is `Sync` and the runs
|
||||
/// are lock-free. Parallelism is `available_parallelism()` workers — `std` only,
|
||||
/// via `std::thread::scope`.
|
||||
pub fn sweep<S, F>(space: &S, run_one: F) -> SweepFamily
|
||||
/// The single assembly path behind both [`sweep`] and the thread-count-explicit
|
||||
/// test wrapper: run `run_one` over each already-enumerated `point` via
|
||||
/// `run_indexed` (on whichever pool is ambient when called — the process-wide
|
||||
/// global pool, or a locally `install`ed one) and collect the family in
|
||||
/// enumeration order. Kept as one source of truth so the two callers can never
|
||||
/// drift on assembly logic (grid points, `RandomSpace` draws) while their pool
|
||||
/// selection differs.
|
||||
fn assemble_sweep<F>(space: Vec<ParamSpec>, points: Vec<Vec<Cell>>, run_one: &F) -> SweepFamily
|
||||
where
|
||||
S: Space,
|
||||
F: Fn(&[Cell]) -> RunReport + Sync,
|
||||
{
|
||||
let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||||
sweep_with_threads(space, nthreads, run_one)
|
||||
}
|
||||
|
||||
/// Run `n` disjoint jobs in parallel and collect their results in **job-index
|
||||
/// order** — the shared disjoint-parallel core both `sweep` (over grid points)
|
||||
/// and `monte_carlo` (over seeds) drive (C1: order is the input order, not the
|
||||
/// completion order). Workers pull job indices from a shared atomic cursor
|
||||
/// (work-stealing load-balances uneven per-job cost); each tags its result with
|
||||
/// the index, and the results are sorted on that index after the scope joins.
|
||||
/// Only the cursor is shared; the results side is lock-free. `nthreads` is
|
||||
/// clamped to `[1, n.max(1)]` (a 0-job call yields an empty vec; a 0 thread
|
||||
/// count coerces to 1).
|
||||
pub(crate) fn run_indexed<T, F>(n: usize, nthreads: usize, run_one: F) -> Vec<T>
|
||||
where
|
||||
T: Send,
|
||||
F: Fn(usize) -> T + Sync,
|
||||
{
|
||||
let nthreads = nthreads.clamp(1, n.max(1));
|
||||
let cursor = AtomicUsize::new(0);
|
||||
|
||||
let mut results: Vec<(usize, T)> = std::thread::scope(|scope| {
|
||||
let handles: Vec<_> = (0..nthreads)
|
||||
.map(|_| {
|
||||
scope.spawn(|| {
|
||||
let mut local: Vec<(usize, T)> = Vec::new();
|
||||
loop {
|
||||
let i = cursor.fetch_add(1, Ordering::Relaxed);
|
||||
if i >= n {
|
||||
break;
|
||||
}
|
||||
local.push((i, run_one(i)));
|
||||
}
|
||||
local
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
handles.into_iter().flat_map(|h| h.join().unwrap()).collect()
|
||||
});
|
||||
|
||||
results.sort_by_key(|&(i, _)| i);
|
||||
results.into_iter().map(|(_, t)| t).collect()
|
||||
}
|
||||
|
||||
/// The thread-count-explicit core of [`sweep`]. Module-private: the public
|
||||
/// `sweep` 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`]: it
|
||||
/// enumerates the `space`'s param-space points (`Space::points`), runs each
|
||||
/// disjointly, and zips the reports back onto their points in enumeration
|
||||
/// (odometer) order.
|
||||
fn sweep_with_threads<S, F>(space: &S, nthreads: usize, run_one: F) -> SweepFamily
|
||||
where
|
||||
S: Space,
|
||||
F: Fn(&[Cell]) -> RunReport + Sync,
|
||||
{
|
||||
let points = space.points();
|
||||
let reports = run_indexed(points.len(), nthreads, |i| run_one(&points[i]));
|
||||
let reports = run_indexed(points.len(), |i| run_one(&points[i]));
|
||||
SweepFamily {
|
||||
space: space.param_specs().to_vec(),
|
||||
space,
|
||||
points: points
|
||||
.into_iter()
|
||||
.zip(reports)
|
||||
@@ -445,6 +387,74 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
/// Run `run_one` over every point the `space` produces (`Space::points` —
|
||||
/// `GridSpace`'s cartesian product or `RandomSpace`'s seeded draws), disjointly
|
||||
/// in parallel (C1), and collect the family in enumeration order. `run_one`
|
||||
/// builds + bootstraps + runs + summarizes one point; it shares nothing
|
||||
/// mutable, so it is `Sync` and the runs are lock-free. Parallelism is the
|
||||
/// shared rayon pool (`run_indexed`): a nested sweep / walk-forward shares its
|
||||
/// workers rather than each spawning its own `available_parallelism()`
|
||||
/// OS-thread batch.
|
||||
pub fn sweep<S, F>(space: &S, run_one: F) -> SweepFamily
|
||||
where
|
||||
S: Space,
|
||||
F: Fn(&[Cell]) -> RunReport + Sync,
|
||||
{
|
||||
let points = space.points();
|
||||
assemble_sweep(space.param_specs().to_vec(), points, &run_one)
|
||||
}
|
||||
|
||||
/// Run `n` disjoint jobs in parallel and collect their results in **job-index
|
||||
/// order** — the shared disjoint-parallel core that `sweep` (over grid points),
|
||||
/// `monte_carlo` (over seeds), and `walk_forward` (over window bounds) drive
|
||||
/// (C1: order is the input order, not the completion order). Jobs run on the
|
||||
/// ambient rayon pool: the process-wide global pool at top level, or — inside a
|
||||
/// `ThreadPool::install` scope (the `*_with_threads` test wrappers) — that local
|
||||
/// pool. Nested `run_indexed` calls (a `walk_forward` window running an inner
|
||||
/// `sweep`) share the same workers via work-stealing, so total live workers
|
||||
/// never exceed the pool size — no per-call OS-thread batch, no multiplicative
|
||||
/// oversubscription. `IndexedParallelIterator::collect` restores enumeration
|
||||
/// order regardless of which worker ran which job; a 0-job call yields an empty
|
||||
/// vec. `run_one` is shared by reference across workers (`Sync`), never moved.
|
||||
// `run_one` is captured by reference in the `map` closure below (needs only
|
||||
// `F: Sync`, not `F: Send`): the seemingly-redundant closure is load-bearing —
|
||||
// passing `run_one` directly would move it into `Map` and require `F: Send`, a
|
||||
// bound the `_with_threads` wrappers' `install`-borrowed callers don't carry.
|
||||
#[allow(clippy::redundant_closure)]
|
||||
pub(crate) fn run_indexed<T, F>(n: usize, run_one: F) -> Vec<T>
|
||||
where
|
||||
T: Send,
|
||||
F: Fn(usize) -> T + Sync,
|
||||
{
|
||||
use rayon::prelude::*;
|
||||
(0..n).into_par_iter().map(|i| run_one(i)).collect()
|
||||
}
|
||||
|
||||
/// The thread-count-explicit wrapper of [`sweep`]. Module-private: the public
|
||||
/// `sweep` runs on the ambient pool, while the tests drive this at 1 and at N
|
||||
/// to pin determinism under concurrency (C1). The point enumeration runs
|
||||
/// outside the pool; `assemble_sweep` (run_indexed + family assembly) runs
|
||||
/// inside a local rayon pool of `nthreads` workers via `ThreadPool::install`,
|
||||
/// sharing the exact assembly path `sweep` uses — no second copy to drift. A
|
||||
/// 1-worker pool runs the jobs inline (no deadlock) — the sequential-equivalent
|
||||
/// reference. Bounds match `sweep` (`F: Sync`): the install closure only
|
||||
/// *borrows* `run_one` and moves the already-owned `space`/`points`, so no
|
||||
/// `Send` bound on `F` is needed.
|
||||
#[cfg(test)]
|
||||
fn sweep_with_threads<S, F>(space: &S, nthreads: usize, run_one: F) -> SweepFamily
|
||||
where
|
||||
S: Space,
|
||||
F: Fn(&[Cell]) -> RunReport + Sync,
|
||||
{
|
||||
let points = space.points();
|
||||
let param_specs = space.param_specs().to_vec();
|
||||
let pool = rayon::ThreadPoolBuilder::new()
|
||||
.num_threads(nthreads)
|
||||
.build()
|
||||
.expect("rayon thread pool");
|
||||
pool.install(|| assemble_sweep(param_specs, points, &run_one))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -803,6 +813,68 @@ mod tests {
|
||||
assert_eq!(one, sweep(&grid, run_point));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nested_run_indexed_shares_the_pool() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
// A shared gauge: live leaf jobs on entry, max seen. Under one shared pool,
|
||||
// nested run_indexed (outer jobs each spawning an inner run_indexed) must
|
||||
// never exceed the pool size — the pre-rayon per-call scope would have run
|
||||
// POOL*POOL threads here. The sleep lets concurrency actually build; the
|
||||
// assertion is an upper bound, so the test is not timing-flaky.
|
||||
const POOL: usize = 4;
|
||||
let live = AtomicUsize::new(0);
|
||||
let max = AtomicUsize::new(0);
|
||||
let pool = rayon::ThreadPoolBuilder::new()
|
||||
.num_threads(POOL)
|
||||
.build()
|
||||
.expect("rayon thread pool");
|
||||
pool.install(|| {
|
||||
run_indexed(POOL * 2, |_outer| {
|
||||
run_indexed(POOL * 2, |_inner| {
|
||||
let now = live.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
max.fetch_max(now, Ordering::SeqCst);
|
||||
std::thread::sleep(std::time::Duration::from_millis(2));
|
||||
live.fetch_sub(1, Ordering::SeqCst);
|
||||
0u8
|
||||
})
|
||||
});
|
||||
});
|
||||
let peak = max.load(Ordering::SeqCst);
|
||||
assert!(
|
||||
peak <= POOL,
|
||||
"nested run_indexed peaked at {peak} live workers, exceeding the {POOL}-worker pool"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore = "benchmark: cargo test -p aura-engine bench_nested_pool -- --ignored --nocapture"]
|
||||
fn bench_nested_pool_walltime() {
|
||||
use std::time::Instant;
|
||||
// Synthetic nested workload: an outer fan-out whose each job runs an inner
|
||||
// fan-out of CPU-bound leaves — the walk_forward(windows) -> sweep(members)
|
||||
// shape, without any archive. Measures wall time under the shared pool; run
|
||||
// before/after the rayon swap to see the oversubscription collapse. Not a
|
||||
// committed baseline (that is #251).
|
||||
fn busy(rounds: u64) -> u64 {
|
||||
(0..rounds).fold(0u64, |a, x| a.wrapping_add(x.wrapping_mul(x).rotate_left(7)))
|
||||
}
|
||||
const OUTER: usize = 48;
|
||||
const INNER: usize = 12;
|
||||
let start = Instant::now();
|
||||
let checksum: u64 = run_indexed(OUTER, |_w| {
|
||||
run_indexed(INNER, |i| busy(4_000_000 + i as u64))
|
||||
.into_iter()
|
||||
.fold(0u64, u64::wrapping_add)
|
||||
})
|
||||
.into_iter()
|
||||
.fold(0u64, u64::wrapping_add);
|
||||
let elapsed = start.elapsed();
|
||||
println!(
|
||||
"nested run_indexed ({OUTER}x{INNER} CPU-bound leaves): {elapsed:?} \
|
||||
(checksum {checksum})"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn distinct_points_produce_distinct_metrics() {
|
||||
let family = sweep(&sma_cross_grid(), run_point);
|
||||
|
||||
@@ -174,6 +174,33 @@ impl WalkForwardResult {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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).
|
||||
@@ -186,16 +213,20 @@ pub fn walk_forward<F>(roller: WindowRoller, space: Vec<ParamSpec>, run_window:
|
||||
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)
|
||||
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 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.
|
||||
/// 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>,
|
||||
@@ -207,18 +238,11 @@ where
|
||||
{
|
||||
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);
|
||||
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 }
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user