feat(campaign): execute — cell loop, sweep/gate stages, walk-forward stage (0107 tasks 6-7)
The execution core: cells iterate strategy x instrument x window in doc order (sequential — C1 keeps parallelism across sims); the sweep stage enumerates the axes odometer (BTreeMap order, last axis fastest) through the engine sweep over a ListSpace with lowest-index-deterministic fault capture; selection dispatches optimize / optimize_plateau / optimize_deflated (deflation nulls seeded from the campaign doc's seed); gates filter per-member via member_metric (metrics.r == None fails an R predicate, conservative); an empty survivor set truncates the cell's realization and the run continues (exit-0 semantics at the caller). The walk-forward stage rolls the declared cell window in ms (WfMode -> RollMode), IS-sweeps ONLY the survivor points per window, stamps manifest.selection on OOS winner reports, and appends a WalkForward family; realization = CampaignRunRecord over untouched family records. The loop's task-7 BLOCKED was review-exhaustion on the task-6 stub wording the plan itself prescribed (task 7 deletes that stub); a fresh post-loop quality review on the final diff returned one Minor — the campaign_id prefix slice could panic on a malformed id — fixed as a typed PipelineShape refusal + test (execute_refuses_malformed_campaign_id). Known and accepted: a cross-cell member fault leaves earlier cells' families as unreferenced records in the append-only store; nested window/member parallelism can oversubscribe cores on large campaigns. Gates: aura-campaign 29/0, workspace 961+/0, clippy -D warnings clean. refs #198
This commit is contained in:
@@ -0,0 +1,603 @@
|
||||
//! Campaign execution (#198): the cell loop over strategy x instrument x
|
||||
//! window, the v1 stage semantics (sweep members -> family -> selection;
|
||||
//! per-member gate filtering; the walk-forward seam), and the realization
|
||||
//! record. Precondition: the CALLER has run the doc tiers (intrinsic +
|
||||
//! referential validation), so this module re-checks nothing a validated
|
||||
//! document guarantees (e.g. non-empty axes); execution-level refusals live
|
||||
//! in the crate's preflight, which `execute` runs before any member runs.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use aura_analysis::{FamilySelection, SelectionMode};
|
||||
use aura_core::{zip_params, Cell, ParamSpec, Scalar, Timestamp};
|
||||
use aura_engine::{
|
||||
sweep, walk_forward, GridSpace, ListSpace, RollMode, RunManifest, RunMetrics, RunReport,
|
||||
Space, SweepFamily, SweepPoint, WindowBounds, WindowRoller, WindowRun,
|
||||
};
|
||||
use aura_registry::{
|
||||
optimize, optimize_deflated, optimize_plateau, sweep_member_reports,
|
||||
walkforward_member_reports, CampaignRunRecord, CellRealization, FamilyKind, PlateauMode,
|
||||
Registry, StageRealization, StageSelection,
|
||||
};
|
||||
use aura_research::{Axis, CampaignDoc, DocRef, ProcessDoc, SelectRule, StageBlock, WfMode};
|
||||
|
||||
use crate::{
|
||||
member_metric, predicate_holds, preflight, CellSpec, ExecFault, MemberFault, MemberRunner,
|
||||
DEFLATION_BLOCK_LEN, DEFLATION_N_RESAMPLES,
|
||||
};
|
||||
|
||||
/// One family-producing stage's full member payload — what an emit renderer
|
||||
/// (`family_table`) prints, in ordinal order.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct StageFamily {
|
||||
pub stage: usize,
|
||||
pub block: &'static str,
|
||||
pub family_id: String,
|
||||
pub reports: Vec<RunReport>,
|
||||
}
|
||||
|
||||
/// One selection-bearing stage's payload — what an emit renderer
|
||||
/// (`selection_report`) prints.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct StageSelectionOut {
|
||||
pub stage: usize,
|
||||
pub block: &'static str,
|
||||
pub family_id: String,
|
||||
pub winner_ordinal: usize,
|
||||
pub params: Vec<(String, Scalar)>,
|
||||
pub selection: FamilySelection,
|
||||
}
|
||||
|
||||
/// Per-cell stage payloads, for the consumer's emit rendering.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct CellOutcome {
|
||||
pub families: Vec<StageFamily>,
|
||||
pub selections: Vec<StageSelectionOut>,
|
||||
}
|
||||
|
||||
/// Everything one campaign run produced: the stored realization record, the
|
||||
/// assigned run counter, and the per-cell payloads.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct CampaignOutcome {
|
||||
pub record: CampaignRunRecord,
|
||||
pub run: usize,
|
||||
pub cells: Vec<CellOutcome>,
|
||||
}
|
||||
|
||||
/// Execute a validated campaign document's process pipeline once per
|
||||
/// (strategy, instrument, window) cell, in doc order, sequentially (C1:
|
||||
/// parallelism lives inside the engine `sweep`, across members, never across
|
||||
/// cells). `campaign_id` is the campaign document's 64-hex content id (its
|
||||
/// first 8 chars prefix family names); `strategies` is index-aligned with
|
||||
/// `campaign.strategies` as (blueprint content id, canonical blueprint json).
|
||||
/// Writes one family per family-producing stage plus one campaign-run record;
|
||||
/// a `MemberFault` aborts the whole run before any write of the faulted stage.
|
||||
pub fn execute(
|
||||
campaign_id: &str,
|
||||
campaign: &CampaignDoc,
|
||||
process: &ProcessDoc,
|
||||
strategies: &[(String, String)],
|
||||
runner: &dyn MemberRunner,
|
||||
registry: &Registry,
|
||||
) -> Result<CampaignOutcome, ExecFault> {
|
||||
preflight(process, campaign)?;
|
||||
if strategies.len() != campaign.strategies.len() {
|
||||
return Err(ExecFault::PipelineShape {
|
||||
detail: format!(
|
||||
"resolved strategies ({}) are not index-aligned with campaign.strategies ({})",
|
||||
strategies.len(),
|
||||
campaign.strategies.len(),
|
||||
),
|
||||
});
|
||||
}
|
||||
let process_id = match &campaign.process.r#ref {
|
||||
DocRef::ContentId(id) | DocRef::IdentityId(id) => id.clone(),
|
||||
};
|
||||
// The store's self-keying always yields a 64-hex content id; anything else
|
||||
// is a caller bug — refuse it typed instead of panicking on the prefix
|
||||
// slice below.
|
||||
if campaign_id.len() != 64 || !campaign_id.bytes().all(|b| b.is_ascii_hexdigit()) {
|
||||
return Err(ExecFault::PipelineShape {
|
||||
detail: format!(
|
||||
"campaign_id must be a 64-hex content id (got {} chars)",
|
||||
campaign_id.len()
|
||||
),
|
||||
});
|
||||
}
|
||||
let campaign_prefix = &campaign_id[..8];
|
||||
|
||||
let mut cells_out: Vec<CellOutcome> = Vec::new();
|
||||
let mut cells_rec: Vec<CellRealization> = Vec::new();
|
||||
for (strategy_ordinal, (entry, (strategy_id, blueprint_json))) in
|
||||
campaign.strategies.iter().zip(strategies).enumerate()
|
||||
{
|
||||
for instrument in &campaign.data.instruments {
|
||||
for (window_ordinal, window) in campaign.data.windows.iter().enumerate() {
|
||||
let cell = CellSpec {
|
||||
strategy_ordinal,
|
||||
strategy_id: strategy_id.clone(),
|
||||
blueprint_json: blueprint_json.clone(),
|
||||
axes: entry.axes.clone(),
|
||||
instrument: instrument.clone(),
|
||||
window_ms: (window.from_ms, window.to_ms),
|
||||
};
|
||||
let (outcome, realization) = run_cell(
|
||||
&cell,
|
||||
process,
|
||||
campaign_prefix,
|
||||
window_ordinal,
|
||||
campaign.seed,
|
||||
runner,
|
||||
registry,
|
||||
)?;
|
||||
cells_out.push(outcome);
|
||||
cells_rec.push(realization);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut record = CampaignRunRecord {
|
||||
campaign: campaign_id.to_string(),
|
||||
process: process_id,
|
||||
run: 0,
|
||||
seed: campaign.seed,
|
||||
cells: cells_rec,
|
||||
};
|
||||
let run = registry.append_campaign_run(&record).map_err(ExecFault::Registry)?;
|
||||
record.run = run;
|
||||
Ok(CampaignOutcome { record, run, cells: cells_out })
|
||||
}
|
||||
|
||||
/// Run one cell's pipeline: stages in doc order, a realized prefix that stops
|
||||
/// at an empty gate (decision 8 — the cell truncates, the campaign continues).
|
||||
fn run_cell(
|
||||
cell: &CellSpec,
|
||||
process: &ProcessDoc,
|
||||
campaign_prefix: &str,
|
||||
window_ordinal: usize,
|
||||
seed: u64,
|
||||
runner: &dyn MemberRunner,
|
||||
registry: &Registry,
|
||||
) -> Result<(CellOutcome, CellRealization), ExecFault> {
|
||||
let grid = enumerate_grid(&cell.axes);
|
||||
let mut families: Vec<StageFamily> = Vec::new();
|
||||
let mut selections: Vec<StageSelectionOut> = Vec::new();
|
||||
let mut stages: Vec<StageRealization> = Vec::new();
|
||||
// The surviving population: (ordinal into the nearest preceding
|
||||
// family_id-bearing stage's family, param point, member report).
|
||||
let mut survivors: Vec<(usize, Vec<Scalar>, RunReport)> = Vec::new();
|
||||
|
||||
for (stage_ordinal, stage) in process.pipeline.iter().enumerate() {
|
||||
match stage {
|
||||
StageBlock::Sweep { metric, select, deflate } => {
|
||||
// Deterministic, self-describing family name (the "-{run}"
|
||||
// suffix is appended by the registry).
|
||||
let family_name = format!(
|
||||
"{campaign_prefix}-{}-{}-w{window_ordinal}-s{stage_ordinal}",
|
||||
cell.strategy_ordinal, cell.instrument,
|
||||
);
|
||||
let family = run_members(cell, &grid, runner)?;
|
||||
let family_id = registry
|
||||
.append_family(&family_name, FamilyKind::Sweep, &sweep_member_reports(&family))
|
||||
.map_err(ExecFault::Registry)?;
|
||||
let (winner, selection) =
|
||||
select_sweep_winner(&family, &grid.axis_lens, metric, *select, *deflate, seed)?;
|
||||
let winner_ordinal = family
|
||||
.points
|
||||
.iter()
|
||||
.position(|p| p == &winner)
|
||||
.expect("the winner is a member of its own family");
|
||||
let params = zip_params(&family.space, &winner.params);
|
||||
selections.push(StageSelectionOut {
|
||||
stage: stage_ordinal,
|
||||
block: "std::sweep",
|
||||
family_id: family_id.clone(),
|
||||
winner_ordinal,
|
||||
params: params.clone(),
|
||||
selection: selection.clone(),
|
||||
});
|
||||
stages.push(StageRealization {
|
||||
block: "std::sweep".to_string(),
|
||||
family_id: Some(family_id.clone()),
|
||||
survivor_ordinals: None,
|
||||
selection: Some(StageSelection { winner_ordinal, params, selection }),
|
||||
});
|
||||
// The whole family flows on (decision 3: `select` names the
|
||||
// recorded selection, never a filter).
|
||||
survivors = family
|
||||
.points
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, p)| (i, scalar_point(&family.space, &p.params), p.report.clone()))
|
||||
.collect();
|
||||
families.push(StageFamily {
|
||||
stage: stage_ordinal,
|
||||
block: "std::sweep",
|
||||
family_id,
|
||||
reports: family.points.into_iter().map(|p| p.report).collect(),
|
||||
});
|
||||
}
|
||||
StageBlock::Gate { all } => {
|
||||
// A member survives iff ALL predicates hold; the crate-root
|
||||
// `predicate_holds` compares the resolved per-member metric.
|
||||
// An unresolvable metric (an R name against `metrics.r ==
|
||||
// None`) fails the member — conservative and deterministic
|
||||
// (the metric NAME was already preflighted against
|
||||
// `PER_MEMBER_METRICS`).
|
||||
survivors.retain(|(_, _, report)| {
|
||||
all.iter().all(|p| {
|
||||
member_metric(report, &p.metric)
|
||||
.is_some_and(|value| predicate_holds(&p.cmp, value, p.value))
|
||||
})
|
||||
});
|
||||
let ordinals: Vec<usize> = survivors.iter().map(|(i, _, _)| *i).collect();
|
||||
let empty = ordinals.is_empty();
|
||||
stages.push(StageRealization {
|
||||
block: "std::gate".to_string(),
|
||||
family_id: None,
|
||||
survivor_ordinals: Some(ordinals),
|
||||
selection: None,
|
||||
});
|
||||
if empty {
|
||||
break; // decision 8: truncate this cell, keep running cells
|
||||
}
|
||||
}
|
||||
StageBlock::WalkForward { .. } => {
|
||||
// Deterministic, self-describing family name (the "-{run}"
|
||||
// suffix is appended by the registry).
|
||||
let family_name = format!(
|
||||
"{campaign_prefix}-{}-{}-w{window_ordinal}-s{stage_ordinal}",
|
||||
cell.strategy_ordinal, cell.instrument,
|
||||
);
|
||||
// The seam crossing is plain points: the walk-forward stage
|
||||
// re-sweeps the surviving param points and never needs the
|
||||
// gate bookkeeping's ordinals or reports.
|
||||
let survivor_points: Vec<Vec<Scalar>> =
|
||||
survivors.iter().map(|(_, point, _)| point.clone()).collect();
|
||||
let fam = run_walk_forward_stage(
|
||||
seed,
|
||||
cell,
|
||||
stage_ordinal,
|
||||
stage,
|
||||
&grid.specs,
|
||||
&survivor_points,
|
||||
&family_name,
|
||||
runner,
|
||||
registry,
|
||||
)?;
|
||||
stages.push(StageRealization {
|
||||
block: "std::walk_forward".to_string(),
|
||||
family_id: Some(fam.family_id.clone()),
|
||||
survivor_ordinals: None,
|
||||
selection: None,
|
||||
});
|
||||
families.push(fam);
|
||||
}
|
||||
StageBlock::MonteCarlo { .. } | StageBlock::Generalize { .. } => {
|
||||
unreachable!("preflight refuses non-v1 stages before any member runs")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok((
|
||||
CellOutcome { families, selections },
|
||||
CellRealization {
|
||||
strategy: cell.strategy_id.clone(),
|
||||
instrument: cell.instrument.clone(),
|
||||
window_ms: cell.window_ms,
|
||||
stages,
|
||||
},
|
||||
))
|
||||
}
|
||||
|
||||
/// The enumerated sweep grid of one cell: param specs (axis name = key, kind =
|
||||
/// axis kind) in BTreeMap (lexicographic) key order, the per-axis radixes, and
|
||||
/// the odometer point list — LAST axis fastest, the `GridSpace` order
|
||||
/// discipline (`optimize_plateau` reads `axis_lens` in exactly this order).
|
||||
struct SweepGrid {
|
||||
specs: Vec<ParamSpec>,
|
||||
axis_lens: Vec<usize>,
|
||||
points: Vec<Vec<Scalar>>,
|
||||
}
|
||||
|
||||
/// Odometer enumeration over the axes map, delegated to the engine's own
|
||||
/// `GridSpace` (last axis fastest) rather than a hand-synced second copy of
|
||||
/// the odometer — single-sourcing the order `optimize_plateau` reads
|
||||
/// `axis_lens` in. Zero axes yield the single empty point. Empty VALUE lists
|
||||
/// cannot reach here (`validate_campaign` refuses `EmptyAxis` upstream),
|
||||
/// hence the `expect`.
|
||||
fn enumerate_grid(axes: &BTreeMap<String, Axis>) -> SweepGrid {
|
||||
let specs: Vec<ParamSpec> = axes
|
||||
.iter()
|
||||
.map(|(name, axis)| ParamSpec { name: name.clone(), kind: axis.kind })
|
||||
.collect();
|
||||
let values: Vec<Vec<Scalar>> = axes.values().map(|axis| axis.values.clone()).collect();
|
||||
let grid = GridSpace::new(&specs, values)
|
||||
.expect("axes validated upstream (validate_campaign refuses EmptyAxis)");
|
||||
let axis_lens = grid.axis_lens();
|
||||
let points: Vec<Vec<Scalar>> = grid
|
||||
.points()
|
||||
.into_iter()
|
||||
.map(|cells| cells.iter().zip(&specs).map(|(c, ps)| Scalar::from_cell(ps.kind, *c)).collect())
|
||||
.collect();
|
||||
SweepGrid { specs, axis_lens, points }
|
||||
}
|
||||
|
||||
/// Run every grid point through the consumer's `MemberRunner` via the engine
|
||||
/// `sweep` (disjoint parallel members, C1 enumeration order). Faults are
|
||||
/// captured per slot (the closure must return a report), and the LOWEST
|
||||
/// enumeration index's fault aborts the stage after the sweep joins —
|
||||
/// deterministic regardless of thread completion order. The placeholder
|
||||
/// report for a faulted slot is never persisted or ranked: a captured fault
|
||||
/// returns before any family write.
|
||||
fn run_members(
|
||||
cell: &CellSpec,
|
||||
grid: &SweepGrid,
|
||||
runner: &dyn MemberRunner,
|
||||
) -> Result<SweepFamily, ExecFault> {
|
||||
let space = ListSpace::new(&grid.specs, grid.points.clone()).map_err(|e| {
|
||||
ExecFault::PipelineShape {
|
||||
detail: format!("axis grid does not form a valid param space: {e:?}"),
|
||||
}
|
||||
})?;
|
||||
// The engine closure receives only the point's cells; recover the
|
||||
// enumeration index by value (duplicate points — possible only when an
|
||||
// axis repeats a value — collapse to the first index, which is exactly
|
||||
// the lowest-index attribution this capture exists for).
|
||||
let cells: Vec<Vec<Cell>> =
|
||||
grid.points.iter().map(|p| p.iter().map(|s| s.cell()).collect()).collect();
|
||||
let faults: Mutex<Vec<(usize, MemberFault)>> = Mutex::new(Vec::new());
|
||||
let family = sweep(&space, |point: &[Cell]| {
|
||||
let params = zip_params(&grid.specs, point);
|
||||
match runner.run_member(cell, ¶ms, cell.window_ms) {
|
||||
Ok(report) => report,
|
||||
Err(fault) => {
|
||||
let index = cells
|
||||
.iter()
|
||||
.position(|c| c.as_slice() == point)
|
||||
.expect("the sweep only enumerates the grid's own points");
|
||||
faults.lock().expect("fault capture lock").push((index, fault));
|
||||
placeholder_report(¶ms, cell.window_ms)
|
||||
}
|
||||
}
|
||||
});
|
||||
let mut captured = faults.into_inner().expect("fault capture lock");
|
||||
captured.sort_by_key(|&(index, _)| index);
|
||||
if let Some((_, fault)) = captured.into_iter().next() {
|
||||
return Err(ExecFault::Member(fault));
|
||||
}
|
||||
Ok(family)
|
||||
}
|
||||
|
||||
/// Slot filler for a faulted member: the engine sweep contract needs one
|
||||
/// report per slot, but a captured fault aborts `execute` before any family
|
||||
/// write, so this report is never persisted and never ranked.
|
||||
fn placeholder_report(params: &[(String, Scalar)], window_ms: (i64, i64)) -> RunReport {
|
||||
RunReport {
|
||||
manifest: RunManifest {
|
||||
commit: String::new(),
|
||||
params: params.to_vec(),
|
||||
window: (Timestamp(window_ms.0), Timestamp(window_ms.1)),
|
||||
seed: 0,
|
||||
broker: "faulted-member-placeholder".to_string(),
|
||||
selection: None,
|
||||
instrument: None,
|
||||
topology_hash: None,
|
||||
project: None,
|
||||
},
|
||||
metrics: RunMetrics { total_pips: 0.0, max_drawdown: 0.0, bias_sign_flips: 0, r: None },
|
||||
}
|
||||
}
|
||||
|
||||
/// The sweep stage's winner + selection provenance. deflate=true composes only
|
||||
/// with argmax (preflighted: `DeflatePlateauConflict`), seeding the deflation
|
||||
/// nulls from the CAMPAIGN seed (C1: a campaign's realization is a pure
|
||||
/// function of the document). Bare argmax has no annotation to compute, so its
|
||||
/// `FamilySelection` is synthesized annotation-free; `raw_winner_metric`
|
||||
/// mirrors the registry's ranking read (an R metric against `r: None` ranks
|
||||
/// `NEG_INFINITY`).
|
||||
fn select_sweep_winner(
|
||||
family: &SweepFamily,
|
||||
axis_lens: &[usize],
|
||||
metric: &str,
|
||||
select: SelectRule,
|
||||
deflate: bool,
|
||||
seed: u64,
|
||||
) -> Result<(SweepPoint, FamilySelection), ExecFault> {
|
||||
match (select, deflate) {
|
||||
(SelectRule::Argmax, true) => {
|
||||
optimize_deflated(family, metric, DEFLATION_N_RESAMPLES, DEFLATION_BLOCK_LEN, seed)
|
||||
.map_err(ExecFault::Registry)
|
||||
}
|
||||
(SelectRule::Argmax, false) => {
|
||||
let winner = optimize(family, metric).map_err(ExecFault::Registry)?;
|
||||
let raw = member_metric(&winner.report, metric).unwrap_or(f64::NEG_INFINITY);
|
||||
let selection = FamilySelection {
|
||||
selection_metric: metric.to_string(),
|
||||
n_trials: family.points.len(),
|
||||
raw_winner_metric: raw,
|
||||
mode: SelectionMode::Argmax,
|
||||
deflated_score: None,
|
||||
overfit_probability: None,
|
||||
n_resamples: None,
|
||||
block_len: None,
|
||||
seed: None,
|
||||
neighbourhood_score: None,
|
||||
n_neighbours: None,
|
||||
};
|
||||
Ok((winner, selection))
|
||||
}
|
||||
(SelectRule::PlateauMean, _) => {
|
||||
optimize_plateau(family, axis_lens, metric, PlateauMode::Mean)
|
||||
.map_err(ExecFault::Registry)
|
||||
}
|
||||
(SelectRule::PlateauWorst, _) => {
|
||||
optimize_plateau(family, axis_lens, metric, PlateauMode::Worst)
|
||||
.map_err(ExecFault::Registry)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A family point's tag-free cells lifted back to self-describing scalars in
|
||||
/// `space` slot order — the survivor-point form the walk-forward stage
|
||||
/// re-sweeps through a `ListSpace`.
|
||||
fn scalar_point(space: &[ParamSpec], point: &[Cell]) -> Vec<Scalar> {
|
||||
space.iter().zip(point).map(|(ps, c)| Scalar::from_cell(ps.kind, *c)).collect()
|
||||
}
|
||||
|
||||
/// A placeholder for a faulted window slot: arity-correct chosen params (the
|
||||
/// engine asserts every window's chosen-point arity against the param-space),
|
||||
/// empty OOS equity, inert report. Never surfaces — a recorded window fault
|
||||
/// fails the stage after the roll, before any registry write.
|
||||
fn placeholder_window_run(specs: &[ParamSpec]) -> WindowRun {
|
||||
WindowRun {
|
||||
chosen_params: vec![Cell::from_i64(0); specs.len()],
|
||||
oos_equity: vec![],
|
||||
oos_report: placeholder_report(&[], (0, 0)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Execute one `std::walk_forward` stage over the cell's surviving points:
|
||||
/// roll (IS, OOS) window splits over the cell window — entirely in ms
|
||||
/// (`Timestamp` is unit-agnostic `i64`; the driver owns ms->ns at its data
|
||||
/// seam) — then per window sweep the survivor points over the IS bounds, pick
|
||||
/// the winner (argmax with trials-deflation provenance, seeded from the
|
||||
/// campaign seed — the shipped select_winner convention; plateau in
|
||||
/// walk_forward is preflight-refused), run the winner over the OOS bounds
|
||||
/// with the selection stamped on its fresh report, and append the per-window
|
||||
/// OOS reports as a `FamilyKind::WalkForward` family.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn run_walk_forward_stage(
|
||||
seed: u64,
|
||||
cell: &CellSpec,
|
||||
stage: usize,
|
||||
block: &StageBlock,
|
||||
specs: &[ParamSpec],
|
||||
survivors: &[Vec<Scalar>],
|
||||
family_name: &str,
|
||||
runner: &dyn MemberRunner,
|
||||
registry: &Registry,
|
||||
) -> Result<StageFamily, ExecFault> {
|
||||
let StageBlock::WalkForward { in_sample_ms, out_of_sample_ms, step_ms, mode, metric, .. } =
|
||||
block
|
||||
else {
|
||||
return Err(ExecFault::PipelineShape {
|
||||
detail: format!("stage {stage} is not std::walk_forward"),
|
||||
});
|
||||
};
|
||||
if survivors.is_empty() {
|
||||
// execute() truncates a cell at an empty gate before reaching here;
|
||||
// refuse (never panic) if that contract is ever violated.
|
||||
return Err(ExecFault::PipelineShape {
|
||||
detail: format!("stage {stage}: walk_forward reached with zero survivor points"),
|
||||
});
|
||||
}
|
||||
// Preflight guarantees the u64 lengths fit i64.
|
||||
let span = (Timestamp(cell.window_ms.0), Timestamp(cell.window_ms.1));
|
||||
let roll_mode = match mode {
|
||||
WfMode::Rolling => RollMode::Rolling,
|
||||
WfMode::Anchored => RollMode::Anchored,
|
||||
};
|
||||
let roller = WindowRoller::new(
|
||||
span,
|
||||
*in_sample_ms as i64,
|
||||
*out_of_sample_ms as i64,
|
||||
*step_ms as i64,
|
||||
roll_mode,
|
||||
)
|
||||
.map_err(|e| ExecFault::Window { stage, detail: format!("{e:?}") })?;
|
||||
// A second identical roller enumerates the bounds up front: the engine's
|
||||
// walk_forward consumes its roller, and the parallel closure needs each
|
||||
// window's roll index for deterministic fault attribution.
|
||||
let bounds: Vec<WindowBounds> = WindowRoller::new(
|
||||
span,
|
||||
*in_sample_ms as i64,
|
||||
*out_of_sample_ms as i64,
|
||||
*step_ms as i64,
|
||||
roll_mode,
|
||||
)
|
||||
.expect("identical roller config validated above")
|
||||
.collect();
|
||||
|
||||
let is_space = ListSpace::new(specs, survivors.to_vec()).map_err(|e| {
|
||||
ExecFault::PipelineShape {
|
||||
detail: format!("stage {stage}: survivor points do not fit the param space: {e:?}"),
|
||||
}
|
||||
})?;
|
||||
let is_points = is_space.points();
|
||||
|
||||
// Windows run in parallel and the engine closure cannot return Err:
|
||||
// faulted windows record here (keyed by roll index) and yield a
|
||||
// placeholder run; after the roll the lowest-index fault wins
|
||||
// (deterministic prose, the sweep-stage capture pattern one level up).
|
||||
let faults: Mutex<Vec<(usize, ExecFault)>> = Mutex::new(Vec::new());
|
||||
|
||||
let result = walk_forward(roller, specs.to_vec(), |w| {
|
||||
let widx =
|
||||
bounds.iter().position(|b| *b == w).expect("bounds come from an identical roller");
|
||||
// IS: engine sweep of the survivor points over the in-sample bounds,
|
||||
// with the same member fault capture the sweep stage uses.
|
||||
let is_faults: Mutex<Vec<(usize, MemberFault)>> = Mutex::new(Vec::new());
|
||||
let is_family = sweep(&is_space, |cells| {
|
||||
match runner.run_member(cell, &zip_params(specs, cells), (w.is.0 .0, w.is.1 .0)) {
|
||||
Ok(report) => report,
|
||||
Err(fault) => {
|
||||
let midx = is_points
|
||||
.iter()
|
||||
.position(|p| p.as_slice() == cells)
|
||||
.expect("sweep enumerates exactly is_points");
|
||||
is_faults.lock().unwrap().push((midx, fault));
|
||||
placeholder_report(&[], (0, 0))
|
||||
}
|
||||
}
|
||||
});
|
||||
let member_faults = is_faults.into_inner().expect("no thread panicked holding the lock");
|
||||
if let Some((_, fault)) = member_faults.into_iter().min_by_key(|(i, _)| *i) {
|
||||
faults.lock().unwrap().push((widx, ExecFault::Member(fault)));
|
||||
return placeholder_window_run(specs);
|
||||
}
|
||||
// IS winner: argmax with trials-deflation provenance, seeded from the
|
||||
// campaign seed (the shipped select_winner convention).
|
||||
let (winner, selection) = match optimize_deflated(
|
||||
&is_family,
|
||||
metric,
|
||||
DEFLATION_N_RESAMPLES,
|
||||
DEFLATION_BLOCK_LEN,
|
||||
seed,
|
||||
) {
|
||||
Ok(picked) => picked,
|
||||
Err(e) => {
|
||||
faults.lock().unwrap().push((widx, ExecFault::Registry(e)));
|
||||
return placeholder_window_run(specs);
|
||||
}
|
||||
};
|
||||
// OOS: run the winner over the out-of-sample bounds; the selection is
|
||||
// stamped on the fresh OOS report (a new record, never a mutation of
|
||||
// a stored one).
|
||||
let mut oos_report = match runner.run_member(
|
||||
cell,
|
||||
&zip_params(specs, &winner.params),
|
||||
(w.oos.0 .0, w.oos.1 .0),
|
||||
) {
|
||||
Ok(report) => report,
|
||||
Err(fault) => {
|
||||
faults.lock().unwrap().push((widx, ExecFault::Member(fault)));
|
||||
return placeholder_window_run(specs);
|
||||
}
|
||||
};
|
||||
oos_report.manifest.selection = Some(selection);
|
||||
WindowRun { chosen_params: winner.params, oos_equity: vec![], oos_report }
|
||||
});
|
||||
|
||||
let window_faults = faults.into_inner().expect("no thread panicked holding the lock");
|
||||
if let Some((_, fault)) = window_faults.into_iter().min_by_key(|(i, _)| *i) {
|
||||
return Err(fault);
|
||||
}
|
||||
|
||||
let reports = walkforward_member_reports(&result);
|
||||
let family_id = registry
|
||||
.append_family(family_name, FamilyKind::WalkForward, &reports)
|
||||
.map_err(ExecFault::Registry)?;
|
||||
Ok(StageFamily { stage, block: "std::walk_forward", family_id, reports })
|
||||
}
|
||||
@@ -11,6 +11,9 @@
|
||||
//! realizes one campaign document; it owns no topology, no data sources,
|
||||
//! and no UI.
|
||||
|
||||
mod exec;
|
||||
pub use exec::{execute, CampaignOutcome, CellOutcome, StageFamily, StageSelectionOut};
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use aura_core::Scalar;
|
||||
@@ -143,8 +146,6 @@ pub fn member_metric(report: &RunReport, name: &str) -> Option<f64> {
|
||||
}
|
||||
|
||||
/// Whether `value <cmp> threshold` holds — the gate's comparator arm.
|
||||
// Consumed by `execute`'s gate stage; the allow drops once that lands.
|
||||
#[allow(dead_code)]
|
||||
fn predicate_holds(cmp: &Cmp, value: f64, threshold: f64) -> bool {
|
||||
match cmp {
|
||||
Cmp::Gt => value > threshold,
|
||||
@@ -163,8 +164,6 @@ fn predicate_holds(cmp: &Cmp, value: f64, threshold: f64) -> bool {
|
||||
/// composes only with `argmax`; walk_forward lengths must fit `i64` (the
|
||||
/// roller's Timestamp unit). The campaign parameter is the seam for
|
||||
/// campaign-level static checks (none in v1, hence unused).
|
||||
// Consumed by `execute`; the allow drops once that lands.
|
||||
#[allow(dead_code)]
|
||||
pub(crate) fn preflight(process: &ProcessDoc, _campaign: &CampaignDoc) -> Result<(), ExecFault> {
|
||||
// v1 boundary first: an mc/generalize stage refuses wherever it sits,
|
||||
// before any shape complaint about the same stage.
|
||||
@@ -583,3 +582,445 @@ mod tests {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod wf_tests {
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use aura_analysis::SelectionMode;
|
||||
use aura_core::{Scalar, ScalarKind};
|
||||
use aura_engine::{RollMode, RunManifest, RunMetrics, RunReport, Timestamp, WindowRoller};
|
||||
use aura_registry::{FamilyKind, Registry};
|
||||
use aura_research::{
|
||||
Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation,
|
||||
ProcessDoc, ProcessRef, SelectRule, StageBlock, StrategyEntry, WfMode, Window,
|
||||
};
|
||||
|
||||
use super::{execute, CampaignOutcome, CellSpec, ExecFault, MemberFault, MemberRunner};
|
||||
|
||||
/// A per-test registry in a fresh temp dir (a Registry's family store
|
||||
/// isolates per DIRECTORY, not per filename — see Registry::open).
|
||||
fn wf_registry(name: &str) -> Registry {
|
||||
let dir = std::env::temp_dir()
|
||||
.join(format!("aura-campaign-wf-{}-{}", std::process::id(), name));
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).expect("temp dir");
|
||||
Registry::open(dir.join("runs.jsonl"))
|
||||
}
|
||||
|
||||
/// One strategy x one instrument x one window, over a single I64 axis
|
||||
/// "len" with values [1, 2, 3, 4] (the fake runner scores total_pips ==
|
||||
/// the len value, so ranking and gating are fully determined).
|
||||
fn wf_campaign(window: (i64, i64)) -> CampaignDoc {
|
||||
CampaignDoc {
|
||||
format_version: 1,
|
||||
kind: DocKind::Campaign,
|
||||
name: "wf-test".to_string(),
|
||||
description: None,
|
||||
data: DataSection {
|
||||
instruments: vec!["SYNTH".to_string()],
|
||||
windows: vec![Window { from_ms: window.0, to_ms: window.1 }],
|
||||
},
|
||||
strategies: vec![StrategyEntry {
|
||||
r#ref: DocRef::ContentId("c".repeat(64)),
|
||||
axes: BTreeMap::from([(
|
||||
"len".to_string(),
|
||||
Axis {
|
||||
kind: ScalarKind::I64,
|
||||
values: vec![
|
||||
Scalar::i64(1),
|
||||
Scalar::i64(2),
|
||||
Scalar::i64(3),
|
||||
Scalar::i64(4),
|
||||
],
|
||||
},
|
||||
)]),
|
||||
}],
|
||||
process: ProcessRef { r#ref: DocRef::ContentId("d".repeat(64)) },
|
||||
seed: 7,
|
||||
presentation: Presentation { persist_taps: vec![], emit: vec![] },
|
||||
}
|
||||
}
|
||||
|
||||
/// sweep -> (optional total_pips-gt gate) -> walk_forward, both ranked
|
||||
/// stages on total_pips / argmax, rolling mode.
|
||||
fn wf_process(gate_gt: Option<f64>, is_ms: u64, oos_ms: u64, step_ms: u64) -> ProcessDoc {
|
||||
let mut pipeline = vec![StageBlock::Sweep {
|
||||
metric: "total_pips".to_string(),
|
||||
select: SelectRule::Argmax,
|
||||
deflate: false,
|
||||
}];
|
||||
if let Some(threshold) = gate_gt {
|
||||
pipeline.push(StageBlock::Gate {
|
||||
all: vec![Predicate {
|
||||
metric: "total_pips".to_string(),
|
||||
cmp: Cmp::Gt,
|
||||
value: threshold,
|
||||
}],
|
||||
});
|
||||
}
|
||||
pipeline.push(StageBlock::WalkForward {
|
||||
in_sample_ms: is_ms,
|
||||
out_of_sample_ms: oos_ms,
|
||||
step_ms,
|
||||
mode: WfMode::Rolling,
|
||||
metric: "total_pips".to_string(),
|
||||
select: SelectRule::Argmax,
|
||||
});
|
||||
ProcessDoc {
|
||||
format_version: 1,
|
||||
kind: DocKind::Process,
|
||||
name: "wf-proc".to_string(),
|
||||
description: None,
|
||||
pipeline,
|
||||
}
|
||||
}
|
||||
|
||||
/// One (window bounds, params) -> fault mapping; a type alias rather than
|
||||
/// a struct-level `#[allow]` since `WfFakeRunner::faulty`'s own parameter
|
||||
/// re-triggers the type-complexity lint independently of the field.
|
||||
type WfFault = ((i64, i64), Vec<(String, Scalar)>, MemberFault);
|
||||
|
||||
/// Deterministic fake member runner keyed on (window bounds, params):
|
||||
/// total_pips == the summed numeric param values (here: the single "len"
|
||||
/// value), the report's manifest.window echoes the window it was run over,
|
||||
/// and every call is logged for the IS-population assertions. `faults`
|
||||
/// (empty for every plain `wf_*` fixture) lets the fault-attribution
|
||||
/// tests below make one specific (window, params) call fail instead of
|
||||
/// planting a report — the only way to reach the walk-forward stage's
|
||||
/// IS-sweep / OOS-run / per-window `min_by_key` capture paths, which a
|
||||
/// runner that always succeeds cannot exercise.
|
||||
#[allow(clippy::type_complexity)]
|
||||
struct WfFakeRunner {
|
||||
log: Mutex<Vec<((i64, i64), Vec<(String, Scalar)>)>>,
|
||||
faults: Vec<WfFault>,
|
||||
}
|
||||
|
||||
impl WfFakeRunner {
|
||||
fn new() -> Self {
|
||||
WfFakeRunner { log: Mutex::new(Vec::new()), faults: Vec::new() }
|
||||
}
|
||||
|
||||
fn faulty(faults: Vec<WfFault>) -> Self {
|
||||
WfFakeRunner { log: Mutex::new(Vec::new()), faults }
|
||||
}
|
||||
}
|
||||
|
||||
impl MemberRunner for WfFakeRunner {
|
||||
fn run_member(
|
||||
&self,
|
||||
_cell: &CellSpec,
|
||||
params: &[(String, Scalar)],
|
||||
window_ms: (i64, i64),
|
||||
) -> Result<RunReport, MemberFault> {
|
||||
self.log.lock().unwrap().push((window_ms, params.to_vec()));
|
||||
if let Some((_, _, fault)) =
|
||||
self.faults.iter().find(|(w, p, _)| *w == window_ms && p == params)
|
||||
{
|
||||
return Err(fault.clone());
|
||||
}
|
||||
let total: f64 = params
|
||||
.iter()
|
||||
.map(|(_, s)| match s {
|
||||
Scalar::I64(v) => *v as f64,
|
||||
Scalar::F64(v) => *v,
|
||||
_ => 0.0,
|
||||
})
|
||||
.sum();
|
||||
Ok(RunReport {
|
||||
manifest: RunManifest {
|
||||
commit: "wf-fake".to_string(),
|
||||
params: params.to_vec(),
|
||||
window: (Timestamp(window_ms.0), Timestamp(window_ms.1)),
|
||||
seed: 0,
|
||||
broker: "fake".to_string(),
|
||||
selection: None,
|
||||
instrument: None,
|
||||
topology_hash: None,
|
||||
project: None,
|
||||
},
|
||||
metrics: RunMetrics {
|
||||
total_pips: total,
|
||||
max_drawdown: 0.0,
|
||||
bias_sign_flips: 0,
|
||||
r: None,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn run_wf(
|
||||
campaign: &CampaignDoc,
|
||||
process: &ProcessDoc,
|
||||
runner: &WfFakeRunner,
|
||||
registry: &Registry,
|
||||
) -> Result<CampaignOutcome, ExecFault> {
|
||||
let strategies = vec![("s".repeat(64), "{}".to_string())];
|
||||
let campaign_id = "e".repeat(64);
|
||||
execute(&campaign_id, campaign, process, &strategies, runner, registry)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wf_rolls_the_declared_window_in_ms() {
|
||||
let registry = wf_registry("rolls");
|
||||
let campaign = wf_campaign((0, 99));
|
||||
let process = wf_process(None, 40, 20, 20);
|
||||
let runner = WfFakeRunner::new();
|
||||
let outcome = run_wf(&campaign, &process, &runner, ®istry).expect("wf executes");
|
||||
|
||||
// The oracle is the roller's own math over the same config.
|
||||
let expected: Vec<_> =
|
||||
WindowRoller::new((Timestamp(0), Timestamp(99)), 40, 20, 20, RollMode::Rolling)
|
||||
.expect("valid roll")
|
||||
.collect();
|
||||
assert_eq!(expected.len(), 3, "fixture sanity: 3 windows over 0..=99");
|
||||
|
||||
let fam = outcome.cells[0]
|
||||
.families
|
||||
.iter()
|
||||
.find(|f| f.block == "std::walk_forward")
|
||||
.expect("wf family present");
|
||||
assert_eq!(fam.reports.len(), expected.len());
|
||||
for (report, bounds) in fam.reports.iter().zip(&expected) {
|
||||
assert_eq!(report.manifest.window, (bounds.oos.0, bounds.oos.1));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wf_searches_only_the_survivor_points() {
|
||||
let registry = wf_registry("survivors");
|
||||
let campaign = wf_campaign((0, 99));
|
||||
// total_pips == len; gate total_pips > 2.5 keeps len 3 and len 4 of [1,2,3,4]
|
||||
let process = wf_process(Some(2.5), 40, 20, 20);
|
||||
let runner = WfFakeRunner::new();
|
||||
run_wf(&campaign, &process, &runner, ®istry).expect("wf executes");
|
||||
|
||||
let is_bounds: Vec<(i64, i64)> =
|
||||
WindowRoller::new((Timestamp(0), Timestamp(99)), 40, 20, 20, RollMode::Rolling)
|
||||
.expect("valid roll")
|
||||
.map(|w| (w.is.0 .0, w.is.1 .0))
|
||||
.collect();
|
||||
let log = runner.log.lock().unwrap();
|
||||
let mut is_total = 0;
|
||||
let mut seen3 = 0;
|
||||
let mut seen4 = 0;
|
||||
for (window, params) in log.iter() {
|
||||
if !is_bounds.contains(window) {
|
||||
continue;
|
||||
}
|
||||
is_total += 1;
|
||||
match params[0].1 {
|
||||
Scalar::I64(3) => seen3 += 1,
|
||||
Scalar::I64(4) => seen4 += 1,
|
||||
other => panic!("gated-out point ran in an IS window: {other:?}"),
|
||||
}
|
||||
}
|
||||
// 3 windows x exactly the 2 surviving points
|
||||
assert_eq!(is_total, 6);
|
||||
assert_eq!(seen3, 3);
|
||||
assert_eq!(seen4, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wf_stamps_selection_on_oos_members() {
|
||||
let registry = wf_registry("selection");
|
||||
let campaign = wf_campaign((0, 99));
|
||||
let process = wf_process(None, 40, 20, 20);
|
||||
let runner = WfFakeRunner::new();
|
||||
let outcome = run_wf(&campaign, &process, &runner, ®istry).expect("wf executes");
|
||||
|
||||
let fam = outcome.cells[0]
|
||||
.families
|
||||
.iter()
|
||||
.find(|f| f.block == "std::walk_forward")
|
||||
.expect("wf family present");
|
||||
assert!(!fam.reports.is_empty());
|
||||
for report in &fam.reports {
|
||||
let sel = report
|
||||
.manifest
|
||||
.selection
|
||||
.as_ref()
|
||||
.expect("every OOS member carries its IS selection");
|
||||
assert_eq!(sel.selection_metric, "total_pips");
|
||||
assert_eq!(sel.seed, Some(7), "deflation is seeded from campaign.seed");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wf_roller_refusal_maps_to_window_fault() {
|
||||
let registry = wf_registry("window-fault");
|
||||
// 0..=49 cannot fit is 40 + oos 20 (window 0 needs 59) -> the roller
|
||||
// refuses at runtime (zero lengths are already doc-tier faults).
|
||||
let campaign = wf_campaign((0, 49));
|
||||
let process = wf_process(None, 40, 20, 20);
|
||||
let runner = WfFakeRunner::new();
|
||||
let result = run_wf(&campaign, &process, &runner, ®istry);
|
||||
let Err(err) = result else { panic!("span too short for one window must refuse") };
|
||||
match err {
|
||||
ExecFault::Window { stage, detail } => {
|
||||
assert_eq!(stage, 1, "walk_forward is pipeline stage 1 here");
|
||||
assert!(
|
||||
detail.contains("SpanTooShort"),
|
||||
"detail carries the roller refusal: {detail}"
|
||||
);
|
||||
}
|
||||
other => panic!("expected ExecFault::Window, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// A member fault raised INSIDE a window's in-sample sweep aborts the
|
||||
/// walk_forward stage with that fault, and among several faulted windows
|
||||
/// the LOWEST roll index wins (the per-window `min_by_key` attribution
|
||||
/// mirrors the sweep stage's own lowest-enumeration-index convention) —
|
||||
/// never the faulted stage's family, since a captured fault returns
|
||||
/// before any registry write.
|
||||
#[test]
|
||||
fn wf_is_member_fault_lowest_window_wins() {
|
||||
let registry = wf_registry("is-fault");
|
||||
let campaign = wf_campaign((0, 99));
|
||||
let process = wf_process(None, 40, 20, 20);
|
||||
// IS bounds: w0 (0,39), w1 (20,59), w2 (40,79) — fault both w1's and
|
||||
// w2's IS runs; the widx-1 fault must win over widx-2's.
|
||||
let runner = WfFakeRunner::faulty(vec![
|
||||
(
|
||||
(20, 59),
|
||||
vec![("len".to_string(), Scalar::i64(3))],
|
||||
MemberFault::Run("is-w1".to_string()),
|
||||
),
|
||||
(
|
||||
(40, 79),
|
||||
vec![("len".to_string(), Scalar::i64(2))],
|
||||
MemberFault::Run("is-w2".to_string()),
|
||||
),
|
||||
]);
|
||||
let err = run_wf(&campaign, &process, &runner, ®istry)
|
||||
.expect_err("a faulted IS member aborts the walk_forward stage");
|
||||
match err {
|
||||
ExecFault::Member(fault) => assert_eq!(
|
||||
fault,
|
||||
MemberFault::Run("is-w1".to_string()),
|
||||
"the LOWEST window index's fault wins, not w2's later one",
|
||||
),
|
||||
other => panic!("expected ExecFault::Member, got {other:?}"),
|
||||
}
|
||||
let members = registry.load_family_members().expect("load members");
|
||||
assert!(members.iter().all(|m| m.kind != FamilyKind::WalkForward));
|
||||
assert!(registry.load_campaign_runs().expect("load campaign runs").is_empty());
|
||||
}
|
||||
|
||||
/// A member fault raised running the IS winner over its OUT-of-sample
|
||||
/// bounds (a call distinct from every IS-sweep call, since the OOS
|
||||
/// window's bounds differ from every window's IS bounds) also aborts the
|
||||
/// walk_forward stage with that fault, and — like the IS-sweep path —
|
||||
/// never persists the faulted stage's family.
|
||||
#[test]
|
||||
fn wf_oos_member_fault_aborts_walk_forward() {
|
||||
let registry = wf_registry("oos-fault");
|
||||
let campaign = wf_campaign((0, 99));
|
||||
let process = wf_process(None, 40, 20, 20);
|
||||
// total_pips == len, so the IS winner is always len=4 (argmax over
|
||||
// [1,2,3,4]); fault only window 0's OOS run of that winner (OOS
|
||||
// bounds (40,59) match no window's IS bounds and no other window's
|
||||
// OOS bounds).
|
||||
let runner = WfFakeRunner::faulty(vec![(
|
||||
(40, 59),
|
||||
vec![("len".to_string(), Scalar::i64(4))],
|
||||
MemberFault::Run("oos-w0".to_string()),
|
||||
)]);
|
||||
let err = run_wf(&campaign, &process, &runner, ®istry)
|
||||
.expect_err("a faulted OOS member aborts the walk_forward stage");
|
||||
match err {
|
||||
ExecFault::Member(fault) => {
|
||||
assert_eq!(fault, MemberFault::Run("oos-w0".to_string()))
|
||||
}
|
||||
other => panic!("expected ExecFault::Member, got {other:?}"),
|
||||
}
|
||||
let members = registry.load_family_members().expect("load members");
|
||||
assert!(members.iter().all(|m| m.kind != FamilyKind::WalkForward));
|
||||
}
|
||||
|
||||
/// select_sweep_winner's Plateau arms dispatch through the public
|
||||
/// `execute()` path (only Argmax is exercised by the other wf_* tests):
|
||||
/// each `SelectRule` variant lands on its matching `SelectionMode`.
|
||||
#[test]
|
||||
fn wf_sweep_stage_dispatches_plateau_select_modes() {
|
||||
for (i, (select, expected_mode)) in [
|
||||
(SelectRule::PlateauMean, SelectionMode::PlateauMean),
|
||||
(SelectRule::PlateauWorst, SelectionMode::PlateauWorst),
|
||||
]
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
{
|
||||
let registry = wf_registry(&format!("plateau-{i}"));
|
||||
let campaign = wf_campaign((0, 99));
|
||||
let process = ProcessDoc {
|
||||
format_version: 1,
|
||||
kind: DocKind::Process,
|
||||
name: "wf-plateau-proc".to_string(),
|
||||
description: None,
|
||||
pipeline: vec![StageBlock::Sweep {
|
||||
metric: "total_pips".to_string(),
|
||||
select,
|
||||
deflate: false,
|
||||
}],
|
||||
};
|
||||
let runner = WfFakeRunner::new();
|
||||
let outcome =
|
||||
run_wf(&campaign, &process, &runner, ®istry).expect("plateau sweep executes");
|
||||
let sel = &outcome.cells[0].selections[0];
|
||||
assert_eq!(
|
||||
sel.selection.mode, expected_mode,
|
||||
"SelectRule::{select:?} must dispatch to SelectionMode::{expected_mode:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// `WfMode::Anchored` maps to the engine's `RollMode::Anchored` (only
|
||||
/// Rolling is exercised by the other wf_* tests) — the oracle is the same
|
||||
/// roller construction the stage itself uses, just with `RollMode::Anchored`.
|
||||
#[test]
|
||||
fn wf_anchored_mode_maps_to_engine_rollmode_anchored() {
|
||||
let registry = wf_registry("anchored");
|
||||
let campaign = wf_campaign((0, 99));
|
||||
let process = ProcessDoc {
|
||||
format_version: 1,
|
||||
kind: DocKind::Process,
|
||||
name: "wf-anchored-proc".to_string(),
|
||||
description: None,
|
||||
pipeline: vec![
|
||||
StageBlock::Sweep {
|
||||
metric: "total_pips".to_string(),
|
||||
select: SelectRule::Argmax,
|
||||
deflate: false,
|
||||
},
|
||||
StageBlock::WalkForward {
|
||||
in_sample_ms: 40,
|
||||
out_of_sample_ms: 20,
|
||||
step_ms: 20,
|
||||
mode: WfMode::Anchored,
|
||||
metric: "total_pips".to_string(),
|
||||
select: SelectRule::Argmax,
|
||||
},
|
||||
],
|
||||
};
|
||||
let runner = WfFakeRunner::new();
|
||||
let outcome = run_wf(&campaign, &process, &runner, ®istry).expect("anchored wf executes");
|
||||
|
||||
let expected: Vec<_> =
|
||||
WindowRoller::new((Timestamp(0), Timestamp(99)), 40, 20, 20, RollMode::Anchored)
|
||||
.expect("valid roll")
|
||||
.collect();
|
||||
assert!(!expected.is_empty(), "fixture sanity: at least one anchored window");
|
||||
|
||||
let fam = outcome.cells[0]
|
||||
.families
|
||||
.iter()
|
||||
.find(|f| f.block == "std::walk_forward")
|
||||
.expect("wf family present");
|
||||
assert_eq!(fam.reports.len(), expected.len());
|
||||
for (report, bounds) in fam.reports.iter().zip(&expected) {
|
||||
assert_eq!(report.manifest.window, (bounds.oos.0, bounds.oos.1));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,360 @@
|
||||
//! Executor semantics over a fake `MemberRunner` (hermetic — no engine
|
||||
//! harness, no data): cell loop, sweep stage, gate filtering, zero-survivor
|
||||
//! truncation, determinism, fault attribution, deflation seeding.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use aura_campaign::{execute, CellSpec, ExecFault, MemberFault, MemberRunner};
|
||||
use aura_core::{Scalar, ScalarKind, Timestamp};
|
||||
use aura_engine::{RMetrics, RunManifest, RunMetrics, RunReport, SelectionMode};
|
||||
use aura_registry::{FamilyKind, Registry};
|
||||
use aura_research::{
|
||||
Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, ProcessDoc,
|
||||
ProcessRef, SelectRule, StageBlock, StrategyEntry, Window,
|
||||
};
|
||||
|
||||
const CAMPAIGN_ID: &str = "aaaabbbbccccddddeeeeffff0000111122223333444455556666777788889999";
|
||||
const PROCESS_ID: &str = "9999888877776666555544443333222211110000ffffeeeeddddccccbbbbaaaa";
|
||||
const STRATEGY_ID: &str = "1111222233334444555566667777888899990000aaaabbbbccccddddeeeeffff";
|
||||
|
||||
/// Deterministic fake: every planted metric is a pure function of
|
||||
/// (params, instrument). `fast`/`slow` are the two I64 axes;
|
||||
/// net = (fast*10 + slow)/100, negated for the instrument named "AAA".
|
||||
struct FakeRunner {
|
||||
/// (params that fault, the fault) — checked before planting a report.
|
||||
faults: Vec<(Vec<(String, Scalar)>, MemberFault)>,
|
||||
}
|
||||
|
||||
impl FakeRunner {
|
||||
fn clean() -> Self {
|
||||
FakeRunner { faults: Vec::new() }
|
||||
}
|
||||
}
|
||||
|
||||
fn param_i64(params: &[(String, Scalar)], name: &str) -> i64 {
|
||||
params
|
||||
.iter()
|
||||
.find(|(n, _)| n == name)
|
||||
.map(|(_, v)| v.as_i64())
|
||||
.expect("planted axis present")
|
||||
}
|
||||
|
||||
fn planted_report(cell: &CellSpec, params: &[(String, Scalar)], window_ms: (i64, i64)) -> RunReport {
|
||||
let fast = param_i64(params, "fast");
|
||||
let slow = param_i64(params, "slow");
|
||||
let mut net = (fast * 10 + slow) as f64 / 100.0;
|
||||
if cell.instrument == "AAA" {
|
||||
net = -net;
|
||||
}
|
||||
RunReport {
|
||||
manifest: RunManifest {
|
||||
commit: "fake".to_string(),
|
||||
params: params.to_vec(),
|
||||
window: (Timestamp(window_ms.0), Timestamp(window_ms.1)),
|
||||
seed: 0,
|
||||
broker: "fake".to_string(),
|
||||
selection: None,
|
||||
instrument: Some(cell.instrument.clone()),
|
||||
topology_hash: Some(cell.strategy_id.clone()),
|
||||
project: None,
|
||||
},
|
||||
metrics: RunMetrics {
|
||||
total_pips: net * 100.0,
|
||||
max_drawdown: 1.0,
|
||||
bias_sign_flips: 1,
|
||||
r: Some(RMetrics {
|
||||
expectancy_r: net,
|
||||
n_trades: 4,
|
||||
win_rate: 0.5,
|
||||
avg_win_r: 1.0,
|
||||
avg_loss_r: -0.5,
|
||||
profit_factor: 2.0,
|
||||
max_r_drawdown: 0.5,
|
||||
n_open_at_end: 0,
|
||||
sqn: net,
|
||||
sqn_normalized: net,
|
||||
net_expectancy_r: net,
|
||||
conviction_terciles_r: [0.0, 0.0, 0.0],
|
||||
trade_rs: Vec::new(),
|
||||
}),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
impl MemberRunner for FakeRunner {
|
||||
fn run_member(
|
||||
&self,
|
||||
cell: &CellSpec,
|
||||
params: &[(String, Scalar)],
|
||||
window_ms: (i64, i64),
|
||||
) -> Result<RunReport, MemberFault> {
|
||||
if let Some((_, fault)) = self.faults.iter().find(|(p, _)| p.as_slice() == params) {
|
||||
return Err(fault.clone());
|
||||
}
|
||||
Ok(planted_report(cell, params, window_ms))
|
||||
}
|
||||
}
|
||||
|
||||
/// fast in {2,3}, slow in {6,9} -> 4 odometer points (last axis fastest):
|
||||
/// 0:(2,6) net .26 | 1:(2,9) net .29 | 2:(3,6) net .36 | 3:(3,9) net .39.
|
||||
fn axes_2x2() -> BTreeMap<String, Axis> {
|
||||
let mut axes = BTreeMap::new();
|
||||
axes.insert(
|
||||
"fast".to_string(),
|
||||
Axis { kind: ScalarKind::I64, values: vec![Scalar::i64(2), Scalar::i64(3)] },
|
||||
);
|
||||
axes.insert(
|
||||
"slow".to_string(),
|
||||
Axis { kind: ScalarKind::I64, values: vec![Scalar::i64(6), Scalar::i64(9)] },
|
||||
);
|
||||
axes
|
||||
}
|
||||
|
||||
fn campaign(instruments: &[&str]) -> CampaignDoc {
|
||||
CampaignDoc {
|
||||
format_version: 1,
|
||||
kind: DocKind::Campaign,
|
||||
name: "exec-test".to_string(),
|
||||
description: None,
|
||||
data: DataSection {
|
||||
instruments: instruments.iter().map(|s| s.to_string()).collect(),
|
||||
windows: vec![Window { from_ms: 1_000, to_ms: 9_000 }],
|
||||
},
|
||||
strategies: vec![StrategyEntry {
|
||||
r#ref: DocRef::ContentId(STRATEGY_ID.to_string()),
|
||||
axes: axes_2x2(),
|
||||
}],
|
||||
process: ProcessRef { r#ref: DocRef::ContentId(PROCESS_ID.to_string()) },
|
||||
seed: 7,
|
||||
presentation: Presentation { persist_taps: vec![], emit: vec![] },
|
||||
}
|
||||
}
|
||||
|
||||
fn sweep_stage(deflate: bool) -> StageBlock {
|
||||
StageBlock::Sweep {
|
||||
metric: "net_expectancy_r".to_string(),
|
||||
select: SelectRule::Argmax,
|
||||
deflate,
|
||||
}
|
||||
}
|
||||
|
||||
fn gate_stage(value: f64) -> StageBlock {
|
||||
StageBlock::Gate {
|
||||
all: vec![Predicate { metric: "net_expectancy_r".to_string(), cmp: Cmp::Gt, value }],
|
||||
}
|
||||
}
|
||||
|
||||
fn process(pipeline: Vec<StageBlock>) -> ProcessDoc {
|
||||
ProcessDoc {
|
||||
format_version: 1,
|
||||
kind: DocKind::Process,
|
||||
name: "exec-test-process".to_string(),
|
||||
description: None,
|
||||
pipeline,
|
||||
}
|
||||
}
|
||||
|
||||
fn strategies() -> Vec<(String, String)> {
|
||||
vec![(STRATEGY_ID.to_string(), r#"{"format_version":1}"#.to_string())]
|
||||
}
|
||||
|
||||
/// Per-test registry DIRECTORY (family/campaign stores are per-directory
|
||||
/// siblings of the runs path — the aura-registry temp-dir idiom).
|
||||
fn temp_registry(name: &str) -> Registry {
|
||||
let dir = std::env::temp_dir()
|
||||
.join(format!("aura-campaign-exec-{}-{}", std::process::id(), name));
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).expect("create temp registry dir");
|
||||
Registry::open(dir.join("runs.jsonl"))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_sweep_only_records_family_and_selection() {
|
||||
let reg = temp_registry("sweep_only");
|
||||
let doc = campaign(&["EURUSD"]);
|
||||
let proc_doc = process(vec![sweep_stage(false)]);
|
||||
let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect("sweep-only campaign executes");
|
||||
|
||||
// outcome payloads: one cell, one family of 4 members, one selection
|
||||
assert_eq!(out.run, 0);
|
||||
assert_eq!(out.cells.len(), 1);
|
||||
assert_eq!(out.cells[0].families.len(), 1);
|
||||
assert_eq!(out.cells[0].families[0].reports.len(), 4);
|
||||
let expected_family_id = format!("{}-0-EURUSD-w0-s0-0", &CAMPAIGN_ID[..8]);
|
||||
assert_eq!(out.cells[0].families[0].family_id, expected_family_id);
|
||||
|
||||
// planted argmax: (3,9) is odometer point 3 (last axis fastest)
|
||||
assert_eq!(out.cells[0].selections.len(), 1);
|
||||
let sel = &out.cells[0].selections[0];
|
||||
assert_eq!(sel.winner_ordinal, 3);
|
||||
assert_eq!(
|
||||
sel.params,
|
||||
vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))],
|
||||
);
|
||||
|
||||
// family persisted: 4 sweep members under the derived id, ordinal-ordered
|
||||
let members = reg.load_family_members().expect("load members");
|
||||
assert_eq!(members.len(), 4);
|
||||
assert!(members.iter().all(|m| m.kind == FamilyKind::Sweep));
|
||||
assert!(members.iter().all(|m| m.family_id() == expected_family_id));
|
||||
assert_eq!(members.iter().map(|m| m.ordinal).collect::<Vec<_>>(), vec![0, 1, 2, 3]);
|
||||
|
||||
// realization persisted: one record linking the family + the selection
|
||||
let runs = reg.load_campaign_runs().expect("load campaign runs");
|
||||
assert_eq!(runs.len(), 1);
|
||||
assert_eq!(runs[0], out.record);
|
||||
assert_eq!(runs[0].campaign, CAMPAIGN_ID);
|
||||
assert_eq!(runs[0].process, PROCESS_ID);
|
||||
assert_eq!(runs[0].run, 0);
|
||||
assert_eq!(runs[0].seed, 7);
|
||||
assert_eq!(runs[0].cells.len(), 1);
|
||||
let cell = &runs[0].cells[0];
|
||||
assert_eq!(cell.strategy, STRATEGY_ID);
|
||||
assert_eq!(cell.instrument, "EURUSD");
|
||||
assert_eq!(cell.window_ms, (1_000, 9_000));
|
||||
assert_eq!(cell.stages.len(), 1);
|
||||
assert_eq!(cell.stages[0].block, "std::sweep");
|
||||
assert_eq!(cell.stages[0].family_id.as_deref(), Some(expected_family_id.as_str()));
|
||||
let stored = cell.stages[0].selection.as_ref().expect("sweep stage carries a selection");
|
||||
assert_eq!(stored.winner_ordinal, 3);
|
||||
assert_eq!(stored.selection.mode, SelectionMode::Argmax);
|
||||
assert_eq!(stored.selection.n_trials, 4);
|
||||
assert!((stored.selection.raw_winner_metric - 0.39).abs() < 1e-12);
|
||||
assert_eq!(stored.selection.seed, None, "no deflation annotation without deflate");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_gate_filters_per_member() {
|
||||
let reg = temp_registry("gate_filters");
|
||||
let doc = campaign(&["EURUSD"]);
|
||||
let proc_doc = process(vec![sweep_stage(false), gate_stage(0.3)]);
|
||||
let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect("gated campaign executes");
|
||||
|
||||
// planted nets 0.26 / 0.29 / 0.36 / 0.39 -> gt 0.3 keeps ordinals 2 and 3
|
||||
let cell = &out.record.cells[0];
|
||||
assert_eq!(cell.stages.len(), 2);
|
||||
assert_eq!(cell.stages[1].block, "std::gate");
|
||||
assert_eq!(cell.stages[1].survivor_ordinals, Some(vec![2, 3]));
|
||||
assert_eq!(cell.stages[1].family_id, None);
|
||||
assert_eq!(cell.stages[1].selection, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_zero_survivors_truncates_cell_and_continues() {
|
||||
let reg = temp_registry("zero_survivors");
|
||||
// two instruments: the fake plants NEGATIVE nets for "AAA", positive for "BBB"
|
||||
let doc = campaign(&["AAA", "BBB"]);
|
||||
// sweep -> gate(net > 0) -> gate(net > -1000): AAA dies at stage 1, BBB passes both
|
||||
let proc_doc = process(vec![sweep_stage(false), gate_stage(0.0), gate_stage(-1000.0)]);
|
||||
let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect("a zero-survivor cell is a valid result, not a fault");
|
||||
|
||||
assert_eq!(out.record.cells.len(), 2, "the second cell still runs");
|
||||
// cell 0 (AAA): realization truncated AT the empty gate — stage 2 never realized
|
||||
let aaa = &out.record.cells[0];
|
||||
assert_eq!(aaa.instrument, "AAA");
|
||||
assert_eq!(aaa.stages.len(), 2);
|
||||
assert_eq!(aaa.stages[1].survivor_ordinals, Some(vec![]));
|
||||
// cell 1 (BBB): full pipeline realized
|
||||
let bbb = &out.record.cells[1];
|
||||
assert_eq!(bbb.instrument, "BBB");
|
||||
assert_eq!(bbb.stages.len(), 3);
|
||||
assert_eq!(bbb.stages[1].survivor_ordinals, Some(vec![0, 1, 2, 3]));
|
||||
assert_eq!(bbb.stages[2].survivor_ordinals, Some(vec![0, 1, 2, 3]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_is_deterministic_twice() {
|
||||
let reg = temp_registry("deterministic_twice");
|
||||
let doc = campaign(&["EURUSD"]);
|
||||
let proc_doc = process(vec![sweep_stage(false)]);
|
||||
let first = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect("first run");
|
||||
let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect("second run");
|
||||
|
||||
// run counter advances per campaign id
|
||||
assert_eq!((first.run, second.run), (0, 1));
|
||||
assert_eq!((first.record.run, second.record.run), (0, 1));
|
||||
|
||||
// family ids differ ONLY by the "-{run}" suffix
|
||||
let base = format!("{}-0-EURUSD-w0-s0", &CAMPAIGN_ID[..8]);
|
||||
assert_eq!(first.cells[0].families[0].family_id, format!("{base}-0"));
|
||||
assert_eq!(second.cells[0].families[0].family_id, format!("{base}-1"));
|
||||
|
||||
// everything else in the record is identical (C1): normalize the two
|
||||
// divergent fields and compare whole records
|
||||
let mut normalized = second.record.clone();
|
||||
normalized.run = first.record.run;
|
||||
normalized.cells[0].stages[0].family_id = first.record.cells[0].stages[0].family_id.clone();
|
||||
assert_eq!(normalized, first.record);
|
||||
|
||||
// both records stored, in order
|
||||
let runs = reg.load_campaign_runs().expect("load campaign runs");
|
||||
assert_eq!(runs.len(), 2);
|
||||
assert_eq!(runs[0], first.record);
|
||||
assert_eq!(runs[1], second.record);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_member_fault_aborts_with_lowest_index() {
|
||||
let reg = temp_registry("member_fault");
|
||||
let doc = campaign(&["EURUSD"]);
|
||||
let proc_doc = process(vec![sweep_stage(false)]);
|
||||
// faults planted at enumeration indices 1 (2,9) and 2 (3,6)
|
||||
let point = |fast: i64, slow: i64| {
|
||||
vec![("fast".to_string(), Scalar::i64(fast)), ("slow".to_string(), Scalar::i64(slow))]
|
||||
};
|
||||
let runner = FakeRunner {
|
||||
faults: vec![
|
||||
(point(2, 9), MemberFault::Run("boom at index 1".to_string())),
|
||||
(point(3, 6), MemberFault::Run("boom at index 2".to_string())),
|
||||
],
|
||||
};
|
||||
let err = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, ®)
|
||||
.expect_err("a faulted member aborts the run");
|
||||
match err {
|
||||
ExecFault::Member(fault) => assert_eq!(
|
||||
fault,
|
||||
MemberFault::Run("boom at index 1".to_string()),
|
||||
"the LOWEST enumeration index's fault is reported",
|
||||
),
|
||||
other => panic!("expected ExecFault::Member, got {other:?}"),
|
||||
}
|
||||
// the fault aborted before any family or realization write
|
||||
assert!(reg.load_family_members().expect("load members").is_empty());
|
||||
assert!(reg.load_campaign_runs().expect("load campaign runs").is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_deflate_uses_campaign_seed() {
|
||||
let reg = temp_registry("deflate_seed");
|
||||
let doc = campaign(&["EURUSD"]); // seed: 7
|
||||
let proc_doc = process(vec![sweep_stage(true)]);
|
||||
let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect("deflated sweep executes");
|
||||
let sel = out.record.cells[0].stages[0].selection.as_ref().expect("selection recorded");
|
||||
assert_eq!(sel.selection.mode, SelectionMode::Argmax);
|
||||
assert_eq!(sel.selection.seed, Some(7), "deflation nulls seed from the campaign doc");
|
||||
assert_eq!(sel.selection.n_resamples, Some(1000));
|
||||
assert_eq!(sel.selection.block_len, Some(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_refuses_malformed_campaign_id() {
|
||||
let reg = temp_registry("bad_id");
|
||||
let doc = campaign(&["EURUSD"]);
|
||||
let proc_doc = process(vec![sweep_stage(false)]);
|
||||
let err = execute("short", &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®)
|
||||
.expect_err("a non-64-hex campaign id is refused, not sliced");
|
||||
match err {
|
||||
ExecFault::PipelineShape { detail } => assert!(
|
||||
detail.contains("campaign_id must be a 64-hex content id"),
|
||||
"detail names the id contract: {detail}",
|
||||
),
|
||||
other => panic!("expected ExecFault::PipelineShape, got {other:?}"),
|
||||
}
|
||||
assert!(reg.load_family_members().expect("load members").is_empty());
|
||||
}
|
||||
Reference in New Issue
Block a user