feat(aura-engine): param-sweep grid primitive (the World, cycle C / iter 1)
Turn one value-empty blueprint into a family of disjoint instances over a
declared grid, run them in parallel, and collect per-point metrics. This is
C12.1's grid axis — the first "N instead of 1": #30/#31 made a blueprint
param-generic and bindable; this reaps it.
New module crates/aura-engine/src/sweep.rs, three parts:
- GridSpace: a validated cartesian grid over a blueprint's param_space() —
one discrete value-list per slot, kind/arity/non-empty checked at
construction (SweepError::{Arity, KindMismatch, EmptyAxis}, the typed gate
before any run). points() enumerates in odometer order (last axis fastest),
deterministic.
- sweep(): closure-driven. The engine owns enumeration + disjoint execution
+ collection; the author owns run_one: Fn(&[Scalar]) -> RunMetrics (build
fresh -> bootstrap_with_params -> run -> drain -> summarize). Metrics
reduction is harness-specific sink glue the engine cannot generically own
(C8/C18), and a fresh per-point build resolves #31's two structural facts:
bootstrap_with_params consumes the blueprint, and a Recorder bakes its
channel in — so each point gets its own drainable instance.
- SweepFamily / SweepPoint: ordered, self-describing (params carry the point
coordinate), in enumeration order independent of thread completion.
Execution is std::thread::scope over available_parallelism() workers pulling
point indices from a shared atomic cursor (work-stealing balances uneven
per-point cost); results are tagged by index and sorted after join, so the
family is bit-identical across thread counts and repeated runs (C1: disjoint,
lock-free; order = enumeration, not completion). No external dependency —
std only, never rayon (C16: the engine workspace stays zero-dep for deploy
purity). A private sweep_with_threads(nthreads) lets the tests pin
determinism at 1 vs N workers while the public sweep() derives the count.
Out of scope this iteration: the `aura sweep` CLI demonstrator (cycle 0028
iter 2); random enumeration, per-point manifest assembly (#33), value-domain
validation (C20) are deferred per the spec.
One deviation from the plan's verbatim code: GridSpace derives Debug
(load-bearing — the three .unwrap_err() fault tests need the Ok type to be
Debug to compile).
Verified: 8 new sweep:: tests green (enumeration/odometer order, the three
typed faults, sweep == N independent runs, determinism across thread counts,
distinct-points-distinct-metrics); cargo test --workspace green (92 engine
lib tests, no regression); cargo clippy --workspace --all-targets
-- -D warnings clean.
refs #32
This commit is contained in:
@@ -35,6 +35,7 @@ mod blueprint;
|
||||
mod graph_model;
|
||||
mod harness;
|
||||
mod report;
|
||||
mod sweep;
|
||||
|
||||
pub use blueprint::{
|
||||
aliases_on, signature_of, BlueprintNode, CompileError, Composite, OutField, ParamAlias, Role,
|
||||
@@ -42,6 +43,7 @@ pub use blueprint::{
|
||||
pub use graph_model::model_to_json;
|
||||
pub use harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};
|
||||
pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
|
||||
pub use sweep::{sweep, GridSpace, SweepError, SweepFamily, SweepPoint};
|
||||
// #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.
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
//! Grid param-sweep (C12.1): enumerate a cartesian grid over a blueprint's
|
||||
//! param-space and run each point disjointly (C1). This module owns enumeration
|
||||
//! (`GridSpace`), execution (`sweep`), and collection (`SweepFamily`); the
|
||||
//! per-point run-to-metrics closure is the author's (harness-specific sink glue
|
||||
//! the engine cannot generically own — C8/C18).
|
||||
|
||||
use aura_core::{ParamSpec, Scalar, ScalarKind};
|
||||
use crate::RunMetrics;
|
||||
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
|
||||
/// points (C12.1 grid axis).
|
||||
#[derive(Debug)]
|
||||
pub struct GridSpace {
|
||||
axes: Vec<Vec<Scalar>>,
|
||||
}
|
||||
|
||||
impl GridSpace {
|
||||
/// Validate `axes` against `space` (the blueprint's `param_space()`): one
|
||||
/// axis per slot (`Arity`), every value the slot's declared kind
|
||||
/// (`KindMismatch`), no empty axis (`EmptyAxis`, which would yield zero
|
||||
/// points). On success the grid enumerates `∏ |axis_i|` points.
|
||||
pub fn new(space: &[ParamSpec], axes: Vec<Vec<Scalar>>) -> Result<Self, SweepError> {
|
||||
if axes.len() != space.len() {
|
||||
return Err(SweepError::Arity { expected: space.len(), got: axes.len() });
|
||||
}
|
||||
for (slot, (axis, ps)) in axes.iter().zip(space).enumerate() {
|
||||
if axis.is_empty() {
|
||||
return Err(SweepError::EmptyAxis { slot });
|
||||
}
|
||||
for (value_index, v) in axis.iter().enumerate() {
|
||||
if v.kind() != ps.kind {
|
||||
return Err(SweepError::KindMismatch {
|
||||
slot,
|
||||
value_index,
|
||||
expected: ps.kind,
|
||||
got: v.kind(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Self { axes })
|
||||
}
|
||||
|
||||
/// The number of grid points (`∏ |axis_i|`), always `>= 1` (a valid grid has
|
||||
/// no empty axis; a zero-param grid is the single empty point).
|
||||
pub fn len(&self) -> usize {
|
||||
self.axes.iter().map(|a| a.len()).product()
|
||||
}
|
||||
|
||||
/// Always `false` — a valid `GridSpace` rejects empty axes, so it has at
|
||||
/// least one point. Present to satisfy clippy's `len_without_is_empty`.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// The cartesian product, in odometer order: the **last** axis varies
|
||||
/// fastest. Deterministic — the same grid yields the same point sequence.
|
||||
pub fn points(&self) -> Vec<Vec<Scalar>> {
|
||||
let mut out = Vec::with_capacity(self.len());
|
||||
let mut idx = vec![0usize; self.axes.len()];
|
||||
loop {
|
||||
out.push(self.axes.iter().zip(&idx).map(|(a, &i)| a[i]).collect());
|
||||
// odometer increment from the last axis
|
||||
let mut k = self.axes.len();
|
||||
loop {
|
||||
if k == 0 {
|
||||
return out;
|
||||
}
|
||||
k -= 1;
|
||||
idx[k] += 1;
|
||||
if idx[k] < self.axes[k].len() {
|
||||
break;
|
||||
}
|
||||
idx[k] = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A structural fault constructing a `GridSpace` — the typed gate before any run.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum SweepError {
|
||||
/// The number of axes does not equal the param-space slot count.
|
||||
Arity { expected: usize, got: usize },
|
||||
/// A grid value's kind does not match its slot's declared kind. `slot` is the
|
||||
/// flat param-space index; `value_index` is the position within that axis.
|
||||
KindMismatch { slot: usize, value_index: usize, expected: ScalarKind, got: ScalarKind },
|
||||
/// A slot was given no values (would collapse the product to zero points).
|
||||
EmptyAxis { slot: usize },
|
||||
}
|
||||
|
||||
/// One enumerated point and the metrics its run produced. Self-describing: the
|
||||
/// `params` vector is the point's coordinate in `param_space()` order.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct SweepPoint {
|
||||
pub params: Vec<Scalar>,
|
||||
pub metrics: RunMetrics,
|
||||
}
|
||||
|
||||
/// The ordered result family of a sweep — one `SweepPoint` per grid point, in
|
||||
/// enumeration (odometer) order, independent of thread completion order.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct SweepFamily {
|
||||
pub points: Vec<SweepPoint>,
|
||||
}
|
||||
|
||||
/// Run `run_one` over every grid point, 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,
|
||||
/// no external dependency (C16).
|
||||
pub fn sweep<F>(space: &GridSpace, run_one: F) -> SweepFamily
|
||||
where
|
||||
F: Fn(&[Scalar]) -> RunMetrics + Sync,
|
||||
{
|
||||
let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||||
sweep_with_threads(space, nthreads, run_one)
|
||||
}
|
||||
|
||||
/// 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). Workers pull point indices from a shared
|
||||
/// atomic cursor (work-stealing load-balances uneven per-point cost); each tags
|
||||
/// its result with the point index, and the family is assembled by sorting on
|
||||
/// that index after the scope joins — so the order is the enumeration order, not
|
||||
/// the completion order. Only the cursor is shared; the results side is lock-free.
|
||||
fn sweep_with_threads<F>(space: &GridSpace, nthreads: usize, run_one: F) -> SweepFamily
|
||||
where
|
||||
F: Fn(&[Scalar]) -> RunMetrics + Sync,
|
||||
{
|
||||
let points = space.points();
|
||||
// `points.len() >= 1` always (GridSpace rejects empty axes), so the clamp
|
||||
// lower bound never exceeds the upper; this also coerces a 0 to 1 worker.
|
||||
let nthreads = nthreads.clamp(1, points.len());
|
||||
let cursor = AtomicUsize::new(0);
|
||||
|
||||
let mut results: Vec<(usize, RunMetrics)> = std::thread::scope(|scope| {
|
||||
let handles: Vec<_> = (0..nthreads)
|
||||
.map(|_| {
|
||||
scope.spawn(|| {
|
||||
let mut local: Vec<(usize, RunMetrics)> = Vec::new();
|
||||
loop {
|
||||
let i = cursor.fetch_add(1, Ordering::Relaxed);
|
||||
if i >= points.len() {
|
||||
break;
|
||||
}
|
||||
local.push((i, run_one(&points[i])));
|
||||
}
|
||||
local
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
handles.into_iter().flat_map(|h| h.join().unwrap()).collect()
|
||||
});
|
||||
|
||||
results.sort_by_key(|&(i, _)| i);
|
||||
SweepFamily {
|
||||
points: results
|
||||
.into_iter()
|
||||
.map(|(i, metrics)| SweepPoint { params: points[i].clone(), metrics })
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{
|
||||
f64_field, summarize, BlueprintNode, Composite, Edge, OutField, ParamAlias, Role,
|
||||
RunMetrics, Target,
|
||||
};
|
||||
use aura_core::{Firing, ParamSpec, Scalar, ScalarKind, Timestamp};
|
||||
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
|
||||
use std::sync::mpsc;
|
||||
|
||||
fn i64_space(n: usize) -> Vec<ParamSpec> {
|
||||
(0..n)
|
||||
.map(|i| ParamSpec { name: format!("p{i}"), kind: ScalarKind::I64 })
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn points_enumerate_in_odometer_order() {
|
||||
let space = i64_space(2);
|
||||
let grid = GridSpace::new(
|
||||
&space,
|
||||
vec![
|
||||
vec![Scalar::I64(2), Scalar::I64(3)],
|
||||
vec![Scalar::I64(4), Scalar::I64(5)],
|
||||
],
|
||||
)
|
||||
.expect("valid grid");
|
||||
assert_eq!(
|
||||
grid.points(),
|
||||
vec![
|
||||
vec![Scalar::I64(2), Scalar::I64(4)],
|
||||
vec![Scalar::I64(2), Scalar::I64(5)],
|
||||
vec![Scalar::I64(3), Scalar::I64(4)],
|
||||
vec![Scalar::I64(3), Scalar::I64(5)],
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn len_is_product_of_axis_lengths() {
|
||||
let space = vec![
|
||||
ParamSpec { name: "a".into(), kind: ScalarKind::I64 },
|
||||
ParamSpec { name: "b".into(), kind: ScalarKind::I64 },
|
||||
ParamSpec { name: "c".into(), kind: ScalarKind::F64 },
|
||||
];
|
||||
let grid = GridSpace::new(
|
||||
&space,
|
||||
vec![
|
||||
vec![Scalar::I64(2), Scalar::I64(3)],
|
||||
vec![Scalar::I64(4), Scalar::I64(5)],
|
||||
vec![Scalar::F64(0.5)],
|
||||
],
|
||||
)
|
||||
.expect("valid grid");
|
||||
assert_eq!(grid.len(), 4);
|
||||
assert!(!grid.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn arity_mismatch_is_an_error() {
|
||||
let space = i64_space(2);
|
||||
let err = GridSpace::new(&space, vec![vec![Scalar::I64(2)]]).unwrap_err();
|
||||
assert_eq!(err, SweepError::Arity { expected: 2, got: 1 });
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wrong_kind_is_a_kind_mismatch() {
|
||||
let space = i64_space(1);
|
||||
let err = GridSpace::new(&space, vec![vec![Scalar::F64(1.0)]]).unwrap_err();
|
||||
assert_eq!(
|
||||
err,
|
||||
SweepError::KindMismatch {
|
||||
slot: 0,
|
||||
value_index: 0,
|
||||
expected: ScalarKind::I64,
|
||||
got: ScalarKind::F64,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_axis_is_an_error() {
|
||||
let space = i64_space(1);
|
||||
let err = GridSpace::new(&space, vec![vec![]]).unwrap_err();
|
||||
assert_eq!(err, SweepError::EmptyAxis { slot: 0 });
|
||||
}
|
||||
|
||||
fn synthetic_prices() -> Vec<(Timestamp, Scalar)> {
|
||||
[
|
||||
(1_i64, 1.0000_f64),
|
||||
(2, 1.0010),
|
||||
(3, 1.0030),
|
||||
(4, 1.0060),
|
||||
(5, 1.0040),
|
||||
(6, 1.0010),
|
||||
(7, 0.9990),
|
||||
]
|
||||
.iter()
|
||||
.map(|&(t, p)| (Timestamp(t), Scalar::F64(p)))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The SMA-cross signal as a reusable value-empty composite (duplicated from
|
||||
/// `blueprint.rs`'s test module — test fixtures stay module-local, the same
|
||||
/// way `report.rs` copies its harness helper).
|
||||
fn sma_cross() -> Composite {
|
||||
Composite::new(
|
||||
"sma_cross",
|
||||
vec![Sma::builder().into(), Sma::builder().into(), Sub::builder().into()],
|
||||
vec![
|
||||
Edge { from: 0, to: 2, slot: 0, from_field: 0 },
|
||||
Edge { from: 1, to: 2, slot: 1, from_field: 0 },
|
||||
],
|
||||
vec![Role {
|
||||
name: "price".into(),
|
||||
targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
|
||||
source: None,
|
||||
}],
|
||||
vec![
|
||||
ParamAlias { name: "fast".into(), node: 0, slot: 0 },
|
||||
ParamAlias { name: "slow".into(), node: 1, slot: 0 },
|
||||
],
|
||||
vec![OutField { node: 2, field: 0, name: "out".into() }],
|
||||
)
|
||||
}
|
||||
|
||||
/// The signal-quality harness as a value-empty composite blueprint with two
|
||||
/// recording sinks (duplicated from `blueprint.rs`'s test module).
|
||||
#[allow(clippy::type_complexity)]
|
||||
fn composite_sma_cross_harness() -> (
|
||||
Composite,
|
||||
mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
|
||||
mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
|
||||
) {
|
||||
let (tx_eq, rx_eq) = mpsc::channel();
|
||||
let (tx_ex, rx_ex) = mpsc::channel();
|
||||
let bp = Composite::new(
|
||||
"root",
|
||||
vec![
|
||||
BlueprintNode::Composite(sma_cross()),
|
||||
Exposure::builder().into(),
|
||||
SimBroker::builder(0.0001).into(),
|
||||
Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
|
||||
Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
|
||||
],
|
||||
vec![
|
||||
Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // composite out -> Exposure
|
||||
Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker slot 0
|
||||
Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity -> sink
|
||||
Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure -> sink
|
||||
],
|
||||
vec![Role {
|
||||
name: "src".into(),
|
||||
targets: vec![Target { node: 0, slot: 0 }, Target { node: 2, slot: 1 }],
|
||||
source: Some(ScalarKind::F64),
|
||||
}],
|
||||
vec![], // params: the interior sma_cross + Exposure carry the knobs
|
||||
vec![], // output: the root ends in sinks
|
||||
);
|
||||
(bp, rx_eq, rx_ex)
|
||||
}
|
||||
|
||||
/// Build + bootstrap + run + drain + summarize one grid point. A free `fn`
|
||||
/// (Copy + Sync) so it serves as the `sweep` closure AND a direct reference
|
||||
/// for the "sweep == N independent runs" comparison.
|
||||
fn run_point(point: &[Scalar]) -> RunMetrics {
|
||||
let (bp, rx_eq, rx_ex) = composite_sma_cross_harness();
|
||||
let mut h = bp
|
||||
.bootstrap_with_params(point.to_vec())
|
||||
.expect("grid points are kind-checked against param_space");
|
||||
h.run(vec![synthetic_prices()]);
|
||||
let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
|
||||
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
|
||||
summarize(&equity, &exposure)
|
||||
}
|
||||
|
||||
fn sma_cross_grid() -> GridSpace {
|
||||
let space = composite_sma_cross_harness().0.param_space();
|
||||
GridSpace::new(
|
||||
&space,
|
||||
vec![
|
||||
vec![Scalar::I64(2), Scalar::I64(3)], // fast ∈ {2, 3}
|
||||
vec![Scalar::I64(4), Scalar::I64(5)], // slow ∈ {4, 5}
|
||||
vec![Scalar::F64(0.5)], // scale ∈ {0.5}
|
||||
],
|
||||
)
|
||||
.expect("grid matches the sample param-space")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sweep_equals_n_independent_runs() {
|
||||
let grid = sma_cross_grid();
|
||||
let family = sweep(&grid, run_point);
|
||||
assert_eq!(family.points.len(), 4);
|
||||
// params carried in enumeration (odometer) order, self-describing
|
||||
assert_eq!(
|
||||
family.points[0].params,
|
||||
vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)],
|
||||
);
|
||||
assert_eq!(
|
||||
family.points[3].params,
|
||||
vec![Scalar::I64(3), Scalar::I64(5), Scalar::F64(0.5)],
|
||||
);
|
||||
// each point's metrics equal a direct, independent run of the same point:
|
||||
// the sweep adds enumeration + execution, never a metrics change (C1).
|
||||
for pt in &family.points {
|
||||
assert_eq!(pt.metrics, run_point(&pt.params));
|
||||
assert!(pt.metrics.total_pips.is_finite());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn family_is_deterministic_across_thread_counts() {
|
||||
let grid = sma_cross_grid();
|
||||
let one = sweep_with_threads(&grid, 1, run_point);
|
||||
let many = sweep_with_threads(&grid, 8, run_point);
|
||||
// same family at 1 worker and at N (C1: order = enumeration, not completion)
|
||||
assert_eq!(one, many);
|
||||
// and identical to the public `sweep`, which derives the worker count
|
||||
assert_eq!(one, sweep(&grid, run_point));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn distinct_points_produce_distinct_metrics() {
|
||||
let family = sweep(&sma_cross_grid(), run_point);
|
||||
// not a constant family: differing SMA lengths produce differing equity
|
||||
let first = family.points[0].metrics.total_pips;
|
||||
assert!(
|
||||
family.points.iter().any(|p| p.metrics.total_pips != first),
|
||||
"a 4-point SMA-length grid must not collapse to one metric",
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user