Files
Aura/crates/aura-cli/src/main.rs
T
Brummel 2957561c30 fix(chart): mean-decimate the bounded exposure series instead of a min/max band
The serve-time decimation (#108) reduced every series by per-bucket min/max. That is
the right envelope for a cumulative equity curve but wrong for the bounded exposure
stream (C10, f64 ∈ [-1,+1]): over a multi-year window each ~hundreds-of-bars bucket
straddles sign flips, so min/max is ±1 in nearly every bucket and the exposure
collapses to a solid -1..+1 band that reads as per-point oscillation — even for a
calm strategy (a 50/200 member flips only 0.81% of bars yet still bands out).

Make decimation tap-aware (RED-first, #111): a new `ReduceKind {MinMax, Mean}` on
each Series, set by the chart builders via `reduce_for_tap` (exposure -> Mean, else
MinMax). `decimate` honours it — a Mean series emits the per-bucket mean (its
net/duty-cycle level) in both spine slots, a MinMax series keeps min/max. The shared
spine, the equity rendering, and the page payload are unchanged (reduce is
server-side only, `#[serde(skip)]`). Verified on real GER40 1y M1: the served
exposure series goes from a ±1 band to a smooth net-level line in [-0.38, +0.34].

Rendering the min/max envelope honestly (range bars / OHLC, vs today's polyline) is
the deferred other half — filed as #112.

Verified: cargo test --workspace = 447 passed / 0 failed (incl. the RED-then-GREEN
decimate_mean_reduces_a_bipolar_series_to_its_bucket_level); clippy -D warnings clean.

closes #111
refs #112
2026-06-22 14:34:21 +02:00

2694 lines
124 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! `aura` — the programmatic / CLI face of the engine (the surface the LLM and
//! automation drive: author a node, run a sim/sweep, emit structured metrics).
//!
//! The walking skeleton's closing seam: `aura run` bootstraps a built-in sample
//! signal-quality harness (synthetic source → SMA-cross → Exposure → SimBroker →
//! recording sinks), runs it deterministically (C1), and prints the run's
//! metrics + manifest (#6) as canonical JSON to stdout (the headline C14 move).
//!
//! `aura run --real <SYMBOL> [--from <ms>] [--to <ms>]` feeds that same harness
//! real M1 close bars, streamed lazily from the local data-server archive through
//! the #71 Source seam (`M1FieldSource`) instead of the synthetic stream — the
//! first real-data backtest from the CLI.
mod render;
use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series};
use aura_core::{zip_params, Cell, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
f64_field, join_on_ts, monte_carlo, param_stability, summarize, walk_forward, window_of,
ColumnarTrace, Composite, Edge, FlatGraph, GraphBuilder, Harness, JoinedRow, McAggregate,
McFamily, RollMode, RunManifest, RunReport, SourceSpec, SweepFamily, SyntheticSpec, Target,
VecSource, WalkForwardResult, WindowBounds, WindowRoller, WindowRun,
};
use aura_registry::{
group_families, mc_member_reports, optimize, rank_by, sweep_member_reports,
walkforward_member_reports, FamilyKind, FamilyMember, NameKind, Registry, RunTraces, TraceStore,
WriteKind,
};
use aura_std::{Ema, Exposure, LinComb, LongOnly, Recorder, SimBroker, Sma, Sub};
use std::sync::mpsc::{self, Receiver};
use std::collections::HashSet;
/// The pip size the built-in *synthetic* harnesses run at: a 5-decimal FX major
/// (`EURUSD`-shaped). The synthetic streams carry no instrument, so there is no
/// `instrument_spec` to thread; a single named source keeps the broker's divisor
/// (`sample_harness` / `SimBroker::builder`) and the recorded broker label
/// (`sim_optimal_manifest`) in lockstep, so they cannot silently drift apart.
/// The real path threads the looked-up `spec.pip_size` instead.
const SYNTHETIC_PIP_SIZE: f64 = 0.0001;
/// Real walk-forward roller sizes (Fork D/F). `WindowRoller` takes sizes in the
/// stream's epoch-unit; for real M1 that is nanoseconds. A classic 3-month
/// in-sample / 1-month out-of-sample / 1-month step (contiguous OOS tiling).
const WF_DAY_NS: i64 = 86_400_000_000_000;
const WF_REAL_IS_NS: i64 = 90 * WF_DAY_NS;
const WF_REAL_OOS_NS: i64 = 30 * WF_DAY_NS;
const WF_REAL_STEP_NS: i64 = 30 * WF_DAY_NS;
/// The built-in synthetic price stream: rises through t=4 then reverses, so the
/// demo trace carries one exposure sign flip and a real drawdown (C22 populated
/// trace). Deterministic and fixed (C1).
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()
}
/// A warm-up-adequate synthetic stream for the enriched sample/sweep: ~18 ticks
/// rising, falling, then rising again so the trend SMA spread and the MACD
/// EMA-of-EMA histogram both warm up and flip sign. It shares the proven warm-up
/// profile of `macd_prices` (the enriched sample embeds the same `macd` composite,
/// so it needs the same warm-up length); the flat `run_sample` keeps the shorter
/// `synthetic_prices`. Deterministic and fixed (C1).
fn showcase_prices() -> Vec<(Timestamp, Scalar)> {
[
1.0000_f64, 1.0008, 1.0021, 1.0039, 1.0062, 1.0090, 1.0083, 1.0061, 1.0034,
1.0012, 0.9998, 1.0006, 1.0024, 1.0047, 1.0069, 1.0086, 1.0097, 1.0092,
]
.iter()
.enumerate()
.map(|(i, &p)| (Timestamp(i as i64 + 1), Scalar::f64(p)))
.collect()
}
/// Bootstrap the sample signal-quality harness with two recording sinks (equity
/// tapped on the SimBroker, exposure tapped on the Exposure node). Rust-authored
/// wiring (C17/C20) over the raw bootstrap API — no builder DSL this cycle. The
/// price taps both SMAs and the broker's price slot (slot 1); exposure feeds the
/// broker's slot 0 (slot order is load-bearing — both are f64).
// The harness-plus-two-drained-sink-receivers tuple has exactly one call site
// (`run_sample`); a named type would be speculative abstraction this cycle.
#[allow(clippy::type_complexity)]
fn sample_harness(pip_size: f64) -> (
Harness,
Receiver<(Timestamp, Vec<Scalar>)>,
Receiver<(Timestamp, Vec<Scalar>)>,
) {
let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel();
let f64_recorder_sig = || aura_engine::NodeSchema {
inputs: vec![aura_engine::PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
output: vec![],
params: vec![],
};
let h = Harness::bootstrap(FlatGraph {
nodes: vec![
Box::new(Sma::new(2)), // 0 fast SMA
Box::new(Sma::new(4)), // 1 slow SMA
Box::new(Sub::new()), // 2 spread
Box::new(Exposure::new(0.5)), // 3 exposure
Box::new(SimBroker::new(pip_size)), // 4 sim-optimal broker
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
],
signatures: vec![
Sma::builder().schema().clone(),
Sma::builder().schema().clone(),
Sub::builder().schema().clone(),
Exposure::builder().schema().clone(),
SimBroker::builder(pip_size).schema().clone(),
f64_recorder_sig(),
f64_recorder_sig(),
],
sources: vec![SourceSpec {
kind: ScalarKind::F64,
targets: vec![
Target { node: 0, slot: 0 },
Target { node: 1, slot: 0 },
Target { node: 4, slot: 1 }, // price into the broker's price slot
],
}],
edges: vec![
Edge { from: 0, to: 2, slot: 0, from_field: 0 },
Edge { from: 1, to: 2, slot: 1, from_field: 0 },
Edge { from: 2, to: 3, slot: 0, from_field: 0 },
Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // exposure into broker slot 0
Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
],
})
.expect("valid sample signal-quality DAG");
(h, rx_eq, rx_ex)
}
/// Build the sim-optimal `RunManifest`: the engine-external descriptor fields
/// (commit, seed-free synthetic run, broker label) are constant across the CLI's
/// built-in harnesses — only `params` and `window` vary. Centralizing the broker
/// label keeps it a single source (and a single edit point for per-asset pip
/// work): the `pip_size` it renders is the one its caller ran the broker at.
fn sim_optimal_manifest(
params: Vec<(String, Scalar)>,
window: (Timestamp, Timestamp),
seed: u64,
pip_size: f64,
) -> RunManifest {
// Typed params pass straight through: the manifest carries self-describing
// Scalars, so a length stays `i64` and a scale `f64` in the record.
RunManifest {
commit: option_env!("AURA_COMMIT").unwrap_or("unknown").to_string(),
params,
window,
seed,
broker: format!("sim-optimal(pip_size={pip_size})"),
}
}
/// Persist a run's drained taps to the on-disk trace store under `runs/traces/<name>/`
/// (beside the run registry's `runs/`). Shared by every run form — all drain the same
/// two f64 taps (equity off the broker, exposure off the strategy) and carry a
/// `RunManifest`. Pure wiring over `ColumnarTrace` + `TraceStore`; the engine is
/// untouched. An I/O failure is a usage-level error (stderr + exit 2), per the spec.
fn persist_traces(
name: &str,
manifest: &RunManifest,
eq_rows: &[(Timestamp, Vec<Scalar>)],
ex_rows: &[(Timestamp, Vec<Scalar>)],
) {
let taps = vec![
ColumnarTrace::from_rows("equity", &[ScalarKind::F64], eq_rows),
ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], ex_rows),
];
if let Err(e) = TraceStore::open("runs").write(name, manifest, &taps) {
eprintln!("aura: trace persist failed: {e}");
std::process::exit(2);
}
}
/// The maximum length (bytes) of an on-disk member-key path component. Comfortably
/// under the 255-byte POSIX `NAME_MAX` and inside Windows' 260-char `MAX_PATH`
/// once the `runs/traces/<name>/` prefix and `/<tap>.json` suffix are added.
const MAX_KEY: usize = 200;
/// Map any byte outside the portable directory-name charset `[A-Za-z0-9._-]` to
/// `_`. The single source of filesystem-portability for an on-disk path component
/// (valid on Linux / Windows / macOS, also URL-path- and cloud-sync-safe).
fn sanitize_component(s: &str) -> String {
s.chars()
.map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c } else { '_' })
.collect()
}
/// Render a scalar value case-lessly: integers/timestamps as decimal digits, bool
/// as `true`/`false`, f64 via Rust's `Display` (decimal, shortest round-trip, NO
/// scientific notation). Case-less rendering is what keeps two members of one
/// family from ever differing only by letter case (case-insensitive-FS safety).
fn render_value(v: &Scalar) -> String {
match v {
Scalar::I64(n) => n.to_string(),
Scalar::F64(x) => x.to_string(),
Scalar::Bool(b) => b.to_string(),
Scalar::Timestamp(t) => t.0.to_string(),
}
}
/// FNV-1a-64 over `bytes` — a fixed, version-stable, non-cryptographic hash used
/// only as the over-cap member-key disambiguator (`std`'s `DefaultHasher` is
/// explicitly unstable across releases, so it cannot name an on-disk artefact).
fn fnv1a64(bytes: &[u8]) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for &b in bytes {
h ^= b as u64;
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
/// The portable, collision-free member key for a swept grid point: one `name-value`
/// token per *varying* axis (in `named`'s param-space slot order), joined by `_`,
/// every token sanitised to the portable charset. Pinned (singleton) axes carry no
/// information and are omitted. A 1-point grid (no varying axis) keys as `"m"`. A
/// key over `MAX_KEY` degrades to a conformant `h-<16-hex>` FNV fallback so the
/// key is one valid path component for ANY grid (the #105 generalisation).
fn member_key(named: &[(String, Scalar)], varying: &HashSet<String>) -> String {
let key: String = named
.iter()
.filter(|(n, _)| varying.contains(n))
.map(|(n, v)| format!("{}-{}", sanitize_component(n), sanitize_component(&render_value(v))))
.collect::<Vec<_>>()
.join("_");
let key = if key.is_empty() { "m".to_string() } else { key };
if key.len() <= MAX_KEY { key } else { format!("h-{:016x}", fnv1a64(key.as_bytes())) }
}
/// Default decimation budget: target horizontal buckets. ~2000 buckets ⇒ ≤ ~4000
/// spine slots (min+max per bucket) — a few-thousand-point page regardless of the
/// underlying multi-year M1 point count.
const CHART_DECIMATE_BUCKETS: usize = 2000;
/// Per-tap decimation kind (#111): the bounded exposure stream (C10, f64 ∈ [-1,+1])
/// reduces by per-bucket mean, so its net/duty-cycle level survives decimation
/// instead of collapsing to a -1..+1 band (every bucket of a multi-year exposure
/// straddles many sign flips, so min/max would be ±1 everywhere). An unbounded
/// cumulative curve (equity) keeps the min/max envelope so drawdowns survive. Keyed
/// on the tap name — `exposure` is the only bounded level tap today.
fn reduce_for_tap(tap: &str) -> ReduceKind {
if tap == "exposure" {
ReduceKind::Mean
} else {
ReduceKind::MinMax
}
}
/// Serve-time decimation on the aligned `ChartData` (#108). Partition the shared
/// `xs` into at most `buckets` contiguous index ranges; per non-empty bucket emit the
/// bucket's first (and, if it spans >1 index, last) timestamp as shared spine slots,
/// and reduce each series per its [`ReduceKind`] (#111): a `MinMax` series emits min
/// then max (the envelope — equity drawdowns survive), a `Mean` series emits the
/// per-bucket mean in both slots (the net level — a bounded exposure shows its
/// duty-cycle instead of a -1..+1 band). An all-null bucket emits null. `meta` passes
/// through unchanged. Deterministic (C1). Full data stays on disk; only the served
/// page is thinned. No-op when `xs.len() <= 2 * buckets`.
fn decimate(data: ChartData, buckets: usize) -> ChartData {
let buckets = buckets.max(1);
let n = data.xs.len();
if n <= 2 * buckets {
return data;
}
let ChartData { xs, series, meta } = data;
// Bucket index bounds (lo, hi_exclusive, two_slots) + the decimated shared spine.
// xs is sorted+deduped (strictly increasing) -> boundary timestamps are strictly
// increasing across and within buckets, so the spine stays monotonic for uPlot.
let mut bounds: Vec<(usize, usize, bool)> = Vec::with_capacity(buckets);
let mut out_xs: Vec<i64> = Vec::with_capacity(2 * buckets);
for b in 0..buckets {
let lo = b * n / buckets;
let hi = (b + 1) * n / buckets;
if lo >= hi {
continue;
}
let two = hi - lo > 1;
out_xs.push(xs[lo]);
if two {
out_xs.push(xs[hi - 1]);
}
bounds.push((lo, hi, two));
}
let out_series: Vec<Series> = series
.into_iter()
.map(|s| {
let mut points: Vec<Option<f64>> = Vec::with_capacity(out_xs.len());
for &(lo, hi, two) in &bounds {
let (first, second) = match s.reduce {
ReduceKind::MinMax => {
// envelope: min at the first slot, max at the second.
let mut mn = f64::INFINITY;
let mut mx = f64::NEG_INFINITY;
let mut any = false;
for v in s.points[lo..hi].iter().flatten() {
any = true;
if *v < mn {
mn = *v;
}
if *v > mx {
mx = *v;
}
}
if any { (Some(mn), Some(mx)) } else { (None, None) }
}
ReduceKind::Mean => {
// net level: the per-bucket mean written to both slots (a flat
// step), so a bounded high-flip series shows its duty-cycle
// instead of a -1..+1 band (#111).
let mut sum = 0.0;
let mut cnt = 0u32;
for v in s.points[lo..hi].iter().flatten() {
sum += *v;
cnt += 1;
}
let m = if cnt > 0 { Some(sum / cnt as f64) } else { None };
(m, m)
}
};
points.push(first);
if two {
points.push(second);
}
}
Series { name: s.name, y_scale_id: s.y_scale_id, points, reduce: s.reduce }
})
.collect();
ChartData { xs: out_xs, series: out_series, meta }
}
/// Build the serve-ready `ChartData` from a run's read-back traces by the spec-§6
/// 3-step union-spine alignment — no tap privileged, no point dropped:
/// (1) xs = the sorted, deduped union of every tap's timestamps;
/// (2) synthesize an empty-payload spine over xs and pass ALL taps (via
/// `ColumnarTrace::to_rows`, which yields uniformly-f64 rows) as symmetric sides
/// of `join_on_ts`, so no side row is dropped and none occupies the privileged
/// `JoinedRow.spine`;
/// (3) flatten each (tap, column) to a `Series` of `Option<f64>` over xs.
fn build_chart_data(name: &str, traces: RunTraces) -> ChartData {
let mut xs: Vec<i64> = traces.taps.iter().flat_map(|t| t.ts.iter().copied()).collect();
xs.sort_unstable();
xs.dedup();
let spine: Vec<(Timestamp, Vec<Scalar>)> = xs.iter().map(|&t| (Timestamp(t), Vec::new())).collect();
let tap_rows: Vec<Vec<(Timestamp, Vec<Scalar>)>> = traces.taps.iter().map(|t| t.to_rows()).collect();
let sides: Vec<&[(Timestamp, Vec<Scalar>)]> = tap_rows.iter().map(|r| r.as_slice()).collect();
let joined: Vec<JoinedRow> = join_on_ts(&spine, &sides);
let mut series: Vec<Series> = Vec::new();
for (i, tap) in traces.taps.iter().enumerate() {
for c in 0..tap.columns.len() {
let name = if tap.columns.len() == 1 { tap.tap.clone() } else { format!("{}[{c}]", tap.tap) };
let y_scale_id = format!("y_{}", series.len());
let points: Vec<Option<f64>> =
joined.iter().map(|r| r.sides[i].as_ref().map(|row| row[c].as_f64())).collect();
series.push(Series { name, y_scale_id, points, reduce: reduce_for_tap(&tap.tap) });
}
}
let m = &traces.manifest;
let meta = ChartMeta {
kind: "run".to_string(),
name: name.to_string(),
commit: m.commit.clone(),
window: (m.window.0.0, m.window.1.0),
broker: m.broker.clone(),
seed: m.seed,
taps: traces.taps.iter().map(|t| t.tap.clone()).collect(),
members: None,
params: m.params.iter().map(|(k, v)| (k.clone(), render_value(v))).collect(),
};
ChartData { xs, series, meta }
}
/// One member's contribution to the comparison build: its key (the future series
/// name) paired with the chosen tap's drained `(ts, row)` pairs.
type MemberRows = (String, Vec<(Timestamp, Vec<Scalar>)>);
/// Build the comparison `ChartData` for a family: one `Series` per member (the
/// chosen `tap`'s column), labelled by `member.key`, ALL sharing ONE `y_scale_id`
/// (the members measure one identical quantity, so a shared scale is what makes
/// them comparable — unlike the single-run overlay, whose series are different
/// taps). Aligned on the union-ts spine via the same `join_on_ts` build_chart_data
/// uses. `Err` if NO member carries `tap` (refuse-don't-guess).
fn build_comparison_chart_data(
name: &str,
members: &[FamilyMember],
tap: &str,
) -> Result<ChartData, String> {
let mut member_rows: Vec<MemberRows> = Vec::new();
for m in members {
if let Some(t) = m.traces.taps.iter().find(|t| t.tap == tap) {
member_rows.push((m.key.clone(), t.to_rows()));
}
}
if member_rows.is_empty() {
return Err(format!("no family member has a tap named '{tap}'"));
}
let mut xs: Vec<i64> =
member_rows.iter().flat_map(|(_, r)| r.iter().map(|(t, _)| t.0)).collect();
xs.sort_unstable();
xs.dedup();
let spine: Vec<(Timestamp, Vec<Scalar>)> =
xs.iter().map(|&t| (Timestamp(t), Vec::new())).collect();
let sides: Vec<&[(Timestamp, Vec<Scalar>)]> =
member_rows.iter().map(|(_, r)| r.as_slice()).collect();
let joined: Vec<JoinedRow> = join_on_ts(&spine, &sides);
// One shared y-scale across all member series (same quantity).
let y_scale_id = format!("y_cmp_{tap}");
let mut series: Vec<Series> = Vec::new();
for (i, (key, _)) in member_rows.iter().enumerate() {
// Project column 0 — the doc's "chosen tap's column" (singular). The
// comparison taps in scope (equity / exposure) are single-column `f64`.
// A future multi-column tap selection would need a column index here.
let points: Vec<Option<f64>> =
joined.iter().map(|r| r.sides[i].as_ref().map(|row| row[0].as_f64())).collect();
series.push(Series { name: key.clone(), y_scale_id: y_scale_id.clone(), points, reduce: reduce_for_tap(tap) });
}
// member_rows is non-empty here (checked above) => members is non-empty, so
// members[0] is safe. commit/broker ARE shared across a family (one frozen
// artifact, one broker profile), but the window is NOT: a walk-forward family's
// members are disjoint OOS windows (commit 4c64feb), so the family window is the
// SPAN across all members — (min from, max to). For sweep/MC, whose members
// share one window, the span collapses to that shared window, so this is the one
// correct reading for all three kinds.
let m = &members[0].traces.manifest;
let window = (
members.iter().map(|fm| fm.traces.manifest.window.0.0).min().unwrap(),
members.iter().map(|fm| fm.traces.manifest.window.1.0).max().unwrap(),
);
let meta = ChartMeta {
kind: "family".to_string(),
name: name.to_string(),
commit: m.commit.clone(),
window,
broker: m.broker.clone(),
seed: m.seed,
taps: vec![tap.to_string()],
members: Some(members.len()),
params: Vec::new(),
};
Ok(ChartData { xs, series, meta })
}
/// Restrict a single-run `ChartData` to the one series named `tap`. `Err` if the
/// run has no such tap (refuse-don't-guess). Used by the `--tap` flag on the
/// single-run chart path; without `--tap` the single-run page is unchanged.
fn filter_to_tap(data: ChartData, tap: &str) -> Result<ChartData, String> {
let series: Vec<Series> = data.series.into_iter().filter(|s| s.name == tap).collect();
if series.is_empty() {
return Err(format!("run has no tap named '{tap}'"));
}
let mut meta = data.meta;
meta.taps = vec![tap.to_string()];
Ok(ChartData { xs: data.xs, series, meta })
}
/// `aura chart <name> [--tap <t>] [--panels]`: classify the name and render. A
/// single run charts all its taps (or the one `--tap` selects); a family overlays
/// one tap (default `equity`) across its members; an unknown name is a usage error
/// (stderr + exit 2), never a panic.
fn emit_chart(name: &str, tap: Option<&str>, mode: ChartMode) {
let store = TraceStore::open("runs");
match store.name_kind(name) {
NameKind::Run => {
let traces = match store.read(name) {
Ok(t) => t,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
let mut data = build_chart_data(name, traces);
if let Some(t) = tap {
data = match filter_to_tap(data, t) {
Ok(d) => d,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
}
let data = decimate(data, CHART_DECIMATE_BUCKETS);
print!("{}", render::render_chart_html(&data, mode));
}
NameKind::Family => {
let members = match store.read_family(name) {
Ok(m) => m,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
let data = match build_comparison_chart_data(name, &members, tap.unwrap_or("equity")) {
Ok(d) => d,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
let data = decimate(data, CHART_DECIMATE_BUCKETS);
print!("{}", render::render_chart_html(&data, mode));
}
NameKind::NotFound => {
eprintln!(
"aura: no recorded run or family '{name}' under runs/traces \
(run `aura run --trace {name}` or `aura sweep --trace {name}` first)"
);
std::process::exit(2);
}
}
}
/// Run the sample harness and fold it into a `RunReport` (drain both sinks →
/// `f64_field` → `summarize` → pair with a `RunManifest`). Pure and deterministic
/// (C1): the same build yields the same report.
fn run_sample(trace: Option<&str>) -> RunReport {
if let Some(n) = trace
&& let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run)
{
eprintln!("aura: {e}");
std::process::exit(2);
}
let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE);
let sources: Vec<Box<dyn aura_engine::Source>> =
vec![Box::new(VecSource::new(synthetic_prices()))];
let window = window_of(&sources).expect("non-empty synthetic stream");
h.run(sources);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let manifest = sim_optimal_manifest(
vec![
("sma_fast".to_string(), Scalar::i64(2)),
("sma_slow".to_string(), Scalar::i64(4)),
("exposure_scale".to_string(), Scalar::f64(0.5)),
],
window,
0,
SYNTHETIC_PIP_SIZE,
);
if let Some(name) = trace {
persist_traces(name, &manifest, &eq_rows, &ex_rows);
}
let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
RunReport { manifest, metrics }
}
/// `aura run --real <SYMBOL>`: run the built-in sample harness over real M1 close
/// bars streamed lazily from the local data-server archive through the #71 Source
/// seam (`M1FieldSource`, a `Box<dyn Source>`), not the synthetic `VecSource`. Same
/// fold as `run_sample`. The manifest window is read from a *separate* probe source
/// (a Source is single-pass), so the run source streams the window untouched.
/// A no-local-data condition (unknown symbol, or a window overlapping no file / no
/// bars) is a usage error: stderr + exit(2), not a panic.
fn run_sample_real(symbol: &str, from_ms: Option<i64>, to_ms: Option<i64>, trace: Option<&str>) -> RunReport {
if let Some(n) = trace
&& let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run)
{
eprintln!("aura: {e}");
std::process::exit(2);
}
// Per-instrument pip: look up BEFORE any data access, so an un-specced symbol
// refuses without touching local data. Honest by construction.
let spec = instrument_spec_or_refuse(symbol);
let (mut h, rx_eq, rx_ex) = sample_harness(spec.pip_size);
let server = std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH));
if !server.has_symbol(symbol) {
no_real_data(symbol);
}
// Manifest window: drain a separate probe (single-pass Source) for first/last ts.
let window = probe_window(&server, symbol, from_ms, to_ms);
let source: Box<dyn aura_engine::Source> =
match aura_ingest::M1FieldSource::open(&server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close) {
Some(s) => Box::new(s),
None => no_real_data(symbol),
};
h.run(vec![source]);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let manifest = sim_optimal_manifest(
vec![
("sma_fast".to_string(), Scalar::i64(2)),
("sma_slow".to_string(), Scalar::i64(4)),
("exposure_scale".to_string(), Scalar::f64(0.5)),
],
window,
0,
spec.pip_size,
);
if let Some(name) = trace {
persist_traces(name, &manifest, &eq_rows, &ex_rows);
}
let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
RunReport { manifest, metrics }
}
/// Parse the `run --real` tail: a mandatory `<SYMBOL>` then zero-or-more
/// `--from <ms>` / `--to <ms>` pairs in any order. Pure (no I/O, no exit) so the
/// arg grammar is unit-testable; `main` does the side effects. An unknown token, a
/// flag without its value, a non-`i64` ms, or a repeated flag is an `Err` carrying
/// a short usage message.
#[allow(clippy::type_complexity)]
fn parse_real_args(
rest: &[&str],
) -> Result<(String, Option<i64>, Option<i64>, Option<String>), String> {
let usage = || "run --real <SYMBOL> [--from <ms>] [--to <ms>] [--trace <name>]".to_string();
let (symbol, mut tail) = match rest.split_first() {
Some((sym, t)) if !sym.is_empty() => (*sym, t),
_ => return Err(usage()),
};
let mut from: Option<i64> = None;
let mut to: Option<i64> = None;
let mut trace: Option<String> = None;
while let Some((flag, t)) = tail.split_first() {
let (value, t) = t.split_first().ok_or_else(usage)?;
match *flag {
"--from" if from.is_none() => from = Some(value.parse().map_err(|_| usage())?),
"--to" if to.is_none() => to = Some(value.parse().map_err(|_| usage())?),
"--trace" if trace.is_none() => trace = Some((*value).to_string()),
_ => return Err(usage()),
}
tail = t;
}
Ok((symbol.to_string(), from, to, trace))
}
/// Parse `aura chart` args: `<name> [--tap <t>] [--panels]` in any order.
fn parse_chart_args(rest: &[&str]) -> Result<(String, Option<String>, ChartMode), String> {
let mut name: Option<String> = None;
let mut tap: Option<String> = None;
let mut mode = ChartMode::Overlay;
let mut i = 0;
while i < rest.len() {
match rest[i] {
"--panels" => {
mode = ChartMode::Panels;
i += 1;
}
"--tap" => {
let t = rest.get(i + 1).ok_or("--tap needs a value")?;
tap = Some((*t).to_string());
i += 2;
}
other if !other.starts_with("--") && name.is_none() => {
name = Some(other.to_string());
i += 1;
}
other => return Err(format!("unexpected chart argument '{other}'")),
}
}
let name = name.ok_or("chart needs a <name>")?;
Ok((name, tap, mode))
}
/// The shared `--real <SYMBOL> [--from <ms>] [--to <ms>]` accumulator for the
/// family parsers (`parse_sweep_args` / `parse_walkforward_args`). It holds the
/// one home of the real/window grammar — flag-repeat strictness, the empty-symbol
/// rejection, and the "window flags require `--real`" rule — so the two siblings
/// agree with each other and with `parse_real_args`. Each parser owns its other
/// flags (strategy / name / trace); on a `--real`/`--from`/`--to` token it calls
/// `accept`, and at the end calls `finish` to validate and build the `DataChoice`.
#[derive(Default)]
struct RealWindowGrammar {
symbol: Option<String>,
from_ms: Option<i64>,
to_ms: Option<i64>,
}
impl RealWindowGrammar {
/// Consume one `--real`/`--from`/`--to` flag and its already-split value.
/// Returns `Ok(true)` if the flag was a real/window flag (consumed),
/// `Ok(false)` if it is not ours (the caller handles it), `Err` on a malformed
/// real/window flag: a repeated flag (mirroring `parse_real_args`'s
/// `if from.is_none()` strictness) or an empty `--real` symbol.
fn accept(&mut self, flag: &str, value: &str, usage: &impl Fn() -> String) -> Result<bool, String> {
match flag {
"--real" if self.symbol.is_none() && !value.is_empty() => {
self.symbol = Some(value.to_string());
}
"--from" if self.from_ms.is_none() => {
self.from_ms = Some(value.parse().map_err(|_| usage())?);
}
"--to" if self.to_ms.is_none() => {
self.to_ms = Some(value.parse().map_err(|_| usage())?);
}
"--real" | "--from" | "--to" => return Err(usage()), // repeated / empty symbol
_ => return Ok(false),
}
Ok(true)
}
/// Validate (window flags require `--real`) and assemble the `DataChoice`.
fn finish(self, usage: &impl Fn() -> String) -> Result<DataChoice, String> {
match self.symbol {
Some(symbol) => Ok(DataChoice::Real { symbol, from_ms: self.from_ms, to_ms: self.to_ms }),
None if self.from_ms.is_some() || self.to_ms.is_some() => Err(usage()),
None => Ok(DataChoice::Synthetic),
}
}
}
/// What `--real` parsing yields: the synthetic default, or a real symbol + an
/// optional window (parsed, not yet opened). Pure, so the grammar is unit-testable.
#[derive(Debug, Clone, PartialEq)]
enum DataChoice {
Synthetic,
Real { symbol: String, from_ms: Option<i64>, to_ms: Option<i64> },
}
/// The source provider threaded into the family builders: synthetic built-in
/// streams, or real M1 close bars from the data-server archive. Replaces the
/// hardcoded `VecSource` so a member's source, pip, window, and roller sizes come
/// from one place (Fork B/D/F).
///
/// `Synthetic` denotes a *consumer-dependent* built-in stream, not one fixed
/// series: the full-window consumers (`full_window` / `run_sources`, used by
/// sweep / MC) draw the 18-bar `showcase_prices()`, while the windowed consumers
/// (`windowed_sources` / `wf_window_sizes`, used by walk-forward) draw the 60-bar
/// `walkforward_prices()` so the `(24,12,12)`-bar roller fits its span. The two
/// faces never reach one consumer (a family is either full-window or windowed), so
/// the split is invisible per call site but real across the type — read both
/// family builders to see it whole.
enum DataSource {
Synthetic,
Real {
server: std::sync::Arc<data_server::DataServer>,
symbol: String,
from_ms: Option<i64>,
to_ms: Option<i64>,
pip: f64,
},
}
/// No-local-data refusal — stderr + exit(2), mirroring `run_sample_real`.
fn no_real_data(symbol: &str) -> ! {
eprintln!("aura: no local data for symbol '{symbol}' at {}", data_server::DEFAULT_DATA_PATH);
std::process::exit(2)
}
/// Look up the vetted per-instrument pip/spec, or refuse (stderr + exit 2) on an
/// un-specced symbol — the single home of the guessed-pip refusal message, shared
/// by `run_sample_real` and `DataSource::from_choice`. Refusing BEFORE any data
/// access keeps the pip honest by construction (never guessed).
fn instrument_spec_or_refuse(symbol: &str) -> aura_ingest::InstrumentSpec {
match aura_ingest::instrument_spec(symbol) {
Some(s) => s,
None => {
eprintln!("aura: no vetted pip/instrument spec for symbol '{symbol}' — refusing to run a real instrument with a guessed pip (add it to the instrument table)");
std::process::exit(2);
}
}
}
/// Probe the full data window: open a single-pass probe `M1FieldSource`, drain it
/// for the first/last timestamp, and return `(first, last)`. Refuses (via
/// `no_real_data`) when the symbol/window yields no source or no bars. Shared by
/// `run_sample_real` (which needs the manifest window from a probe separate from
/// the run source) and `DataSource::full_window`.
fn probe_window(
server: &std::sync::Arc<data_server::DataServer>,
symbol: &str,
from_ms: Option<i64>,
to_ms: Option<i64>,
) -> (Timestamp, Timestamp) {
let mut probe = aura_ingest::M1FieldSource::open(server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close)
.unwrap_or_else(|| no_real_data(symbol));
let first = aura_engine::Source::peek(&probe).unwrap_or_else(|| no_real_data(symbol));
let mut last = first;
while let Some((t, _)) = aura_engine::Source::next(&mut probe) {
last = t;
}
(first, last)
}
impl DataSource {
/// Build a provider from a parsed choice, or refuse (stderr + exit 2) on an
/// un-vetted symbol / absent data — both BEFORE any member runs (Fork C/G),
/// via the same `instrument_spec_or_refuse` / `no_real_data` helpers as
/// `run_sample_real`.
fn from_choice(choice: DataChoice) -> DataSource {
match choice {
DataChoice::Synthetic => DataSource::Synthetic,
DataChoice::Real { symbol, from_ms, to_ms } => {
let spec = instrument_spec_or_refuse(&symbol);
let server = std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH));
if !server.has_symbol(&symbol) {
no_real_data(&symbol);
}
DataSource::Real { server, symbol, from_ms, to_ms, pip: spec.pip_size }
}
}
}
fn pip_size(&self) -> f64 {
match self {
DataSource::Synthetic => SYNTHETIC_PIP_SIZE,
DataSource::Real { pip, .. } => *pip,
}
}
/// The full run window, probed once. Synthetic: the showcase span. Real:
/// `probe_window` drains a separate single-pass probe source for first/last ts
/// (the same helper `run_sample_real` uses for its manifest window).
fn full_window(&self) -> (Timestamp, Timestamp) {
match self {
DataSource::Synthetic => {
let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(showcase_prices()))];
window_of(&s).expect("non-empty showcase stream")
}
DataSource::Real { server, symbol, from_ms, to_ms, .. } => {
probe_window(server, symbol, *from_ms, *to_ms)
}
}
}
/// The full walk-forward span. Synthetic draws the 60-bar `walkforward_prices`
/// span — NOT `showcase_prices` (which `full_window` uses): walk-forward is a
/// *windowed* consumer whose roller `(24,12,12)` needs 36 bars, so it uses the
/// longer built-in stream (byte-unchanged from the pre-`DataSource`
/// `walkforward_family`, which derived its span the same way). Real: the same
/// probed `--from..--to` window as `full_window`.
fn wf_full_span(&self) -> (Timestamp, Timestamp) {
match self {
DataSource::Synthetic => {
let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(walkforward_prices()))];
window_of(&s).expect("non-empty walkforward stream")
}
DataSource::Real { server, symbol, from_ms, to_ms, .. } => {
probe_window(server, symbol, *from_ms, *to_ms)
}
}
}
/// A fresh full-window source per member (single-pass). Synthetic: showcase.
fn run_sources(&self) -> Vec<Box<dyn aura_engine::Source>> {
match self {
DataSource::Synthetic => vec![Box::new(VecSource::new(showcase_prices()))],
DataSource::Real { server, symbol, from_ms, to_ms, .. } => vec![Box::new(
aura_ingest::M1FieldSource::open(server, symbol, *from_ms, *to_ms, aura_ingest::M1Field::Close)
.unwrap_or_else(|| no_real_data(symbol)),
)],
}
}
/// A fresh windowed source for an IS/OOS sub-window (walk-forward). Synthetic:
/// `walkforward_window_source`. Real: `open_window` (ns-native `Timestamp`).
fn windowed_sources(&self, from: Timestamp, to: Timestamp) -> Vec<Box<dyn aura_engine::Source>> {
match self {
DataSource::Synthetic => vec![Box::new(walkforward_window_source(from, to))],
DataSource::Real { server, symbol, .. } => vec![Box::new(
aura_ingest::M1FieldSource::open_window(server, symbol, Some(from), Some(to), aura_ingest::M1Field::Close)
.unwrap_or_else(|| no_real_data(symbol)),
)],
}
}
/// WindowRoller sizes per data kind (Fork F): bar-index for synthetic (24/12/12
/// over the 60-bar span), calendar-ns for real.
fn wf_window_sizes(&self) -> (i64, i64, i64) {
match self {
DataSource::Synthetic => (24, 12, 12),
DataSource::Real { .. } => (WF_REAL_IS_NS, WF_REAL_OOS_NS, WF_REAL_STEP_NS),
}
}
/// The built-in strategy's length grid, **per data kind**. Synthetic keeps the
/// short lengths that fit the 18/60-bar demo streams (a 200-bar MA would never
/// warm on a 60-bar stream); real uses realistic M1 lengths so the SMA-cross +
/// MACD signal is a real trend cross over tens of thousands of bars, not noise.
/// Returns `(trend_fast grid, trend_slow grid, (macd_fast, macd_slow, macd_signal))`;
/// only the two trend axes vary (a 2×2 sweep), the MACD lengths are pinned.
fn strategy_lengths(&self) -> ([i64; 2], [i64; 2], (i64, i64, i64)) {
match self {
DataSource::Synthetic => ([2, 3], [4, 5], (2, 4, 3)),
// 50/200-style intraday crosses on M1 (minutes), standard 12/26/9 MACD.
DataSource::Real { .. } => ([50, 100], [200, 400], (12, 26, 9)),
}
}
}
/// The SMA-cross signal as a named composite (price -> fast/slow SMA -> spread).
/// CLI-local sample builder; the engine ships no sample (the duplication with
/// `blueprint.rs`'s test helper is the dedup tracked in #14). Value-empty: the SMA
/// lengths are injected at compile, not baked here.
fn sma_cross(name: &str) -> Composite {
let mut g = GraphBuilder::new(name);
let fast = g.add(Sma::builder().named("fast")); // fast SMA leg
let slow = g.add(Sma::builder().named("slow")); // slow SMA leg
let sub = g.add(Sub::builder());
let price = g.input_role("price");
g.feed(price, [fast.input("series"), slow.input("series")]);
g.connect(fast.output("value"), sub.input("lhs"));
g.connect(slow.output("value"), sub.input("rhs"));
g.expose(sub.output("value"), "cross");
g.build().expect("sample sma_cross wiring resolves")
}
/// The blended signal: a trend leg (SMA-cross) and a momentum leg (MACD), combined
/// by a weighted sum. A multiply-nested composite (root → signals → {trend,
/// momentum}); the blend is a multi-param node living inside it, with one weight
/// bound as a structural constant (so it drops out of the sweepable surface).
fn signals(name: &str) -> Composite {
let mut g = GraphBuilder::new(name);
let trend = g.add(sma_cross("trend")); // trend leg (one f64 "cross")
let momentum = g.add(macd("momentum")); // momentum leg (3 outputs)
// blend: Σ wᵢ·termᵢ over [trend.cross, momentum.histogram, momentum.signal].
// weights[2] is bound (a fixed signal-line weight) → removed from param_space;
// weights[0]/[1] stay tunable. `.named("blend")` makes the path signals.blend.*
// (and renders the `blend:` prefix).
let blend = g.add(
LinComb::builder(3)
.named("blend")
.bind("weights[2]", Scalar::f64(0.5)),
);
let price = g.input_role("price");
g.feed(price, [trend.input("price"), momentum.input("price")]);
g.connect(trend.output("cross"), blend.input("term[0]")); // trend.cross → blend.term[0]
g.connect(momentum.output("histogram"), blend.input("term[1]")); // momentum.histogram → blend.term[1]
g.connect(momentum.output("signal"), blend.input("term[2]")); // momentum.signal → blend.term[2]
g.expose(blend.output("value"), "signal");
g.build().expect("sample signals wiring resolves")
}
/// The sample signal-quality blueprint (value-empty) **with its two recording
/// sinks reachable**: returns the equity + exposure receivers a per-point sweep
/// run drains (the eight free params — the trend SMA lengths, the momentum EMA
/// lengths, the two open blend weights, and the exposure scale — are injected at
/// compile via the point vector). The root harness of the sample topology, whose
/// signal is built by the nested `signals` composite; `build_sample` (the `aura
/// graph` entry) is expressed on top of it.
#[allow(clippy::type_complexity)]
fn sample_blueprint_with_sinks(pip_size: f64) -> (
Composite,
Receiver<(Timestamp, Vec<Scalar>)>,
Receiver<(Timestamp, Vec<Scalar>)>,
) {
let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel();
let mut g = GraphBuilder::new("sample");
let sig = g.add(signals("signals"));
let exposure = g.add(Exposure::builder());
let broker = g.add(SimBroker::builder(pip_size));
let eq = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq));
let ex = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex));
let price = g.source_role("price", ScalarKind::F64);
g.feed(price, [sig.input("price"), broker.input("price")]);
g.connect(sig.output("signal"), exposure.input("signal")); // blended signal -> Exposure
g.connect(exposure.output("exposure"), broker.input("exposure")); // exposure -> broker slot 0
g.connect(broker.output("equity"), eq.input("col[0]")); // equity -> sink
g.connect(exposure.output("exposure"), ex.input("col[0]")); // exposure -> sink
let bp = g.build().expect("sample blueprint wiring resolves");
(bp, rx_eq, rx_ex)
}
/// The sample blueprint without its sink receivers — the `aura graph` render
/// entry, which never runs the graph (so the receivers are dropped).
fn build_sample() -> Composite {
sample_blueprint_with_sinks(SYNTHETIC_PIP_SIZE).0
}
/// The built-in sample rendered by `aura graph`.
fn sample_blueprint() -> Composite {
build_sample()
}
/// Run the built-in sample over a small built-in grid (fast ∈ {2,3},
/// slow ∈ {4,5}, scale ∈ {0.5} — 4 points) and render one JSON line per point in
/// enumeration (odometer) order. Pure + deterministic (C1): the same build yields
/// the same report. Each point builds a fresh blueprint (fresh sink channels),
/// bootstraps it under the point vector, runs it, and folds the drained sinks to
/// metrics — the per-point closure the engine `sweep` drives disjointly.
fn sweep_family(trace: Option<&str>, data: &DataSource) -> SweepFamily {
let pip = data.pip_size();
let window = data.full_window();
let bp = sample_blueprint_with_sinks(pip).0;
let space = bp.param_space();
let (tf, ts, (mf, ms, msig)) = data.strategy_lengths();
let binder = bp
.axis("signals.trend.fast.length", tf)
.axis("signals.trend.slow.length", ts)
.axis("signals.momentum.fast.length", [mf])
.axis("signals.momentum.slow.length", [ms])
.axis("signals.momentum.signal.length", [msig])
.axis("signals.blend.weights[0]", [1.0])
.axis("signals.blend.weights[1]", [1.0])
.axis("exposure.scale", [0.5]);
let varying: HashSet<String> = binder.varying_axes().into_iter().collect();
binder
.sweep(|point| {
let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(pip);
let mut h = bp
.bootstrap_with_cells(point)
.expect("grid points are kind-checked against param_space");
let sources = data.run_sources();
h.run(sources);
let eq_rows = rx_eq.try_iter().collect::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
let named = zip_params(&space, point);
let key = member_key(&named, &varying);
let manifest = sim_optimal_manifest(named, window, 0, pip);
if let Some(name) = trace {
persist_traces(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows);
}
let equity = f64_field(&eq_rows, 0);
let exposure = f64_field(&ex_rows, 0);
RunReport { manifest, metrics: summarize(&equity, &exposure) }
})
.expect("the built-in named grid matches the sample param-space")
}
/// The EMA-distance momentum demo strategy with its two recording sinks reachable.
/// The signal is how far price sits above/below its own EMA (`price - ema`), sized
/// by `Exposure`, then passed through the long-only `LongOnly` gate. Three swept
/// knobs of three kinds — `ema.length` (i64), `exposure.scale` (f64),
/// `longonly.enabled` (bool) — with nothing in common with the SMA-cross demo: the
/// generic-sweep proof. The exposure sink taps the FINAL (gated) exposure so the
/// bool's effect is visible in the trace.
#[allow(clippy::type_complexity)]
fn momentum_blueprint_with_sinks(pip_size: f64) -> (
Composite,
Receiver<(Timestamp, Vec<Scalar>)>,
Receiver<(Timestamp, Vec<Scalar>)>,
) {
let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel();
let mut g = GraphBuilder::new("momentum");
// Name the param-bearing nodes explicitly so the swept paths are guaranteed
// ema.length / exposure.scale / longonly.enabled regardless of default-name
// derivation (the member-key examples + the param-space test depend on these).
let ema = g.add(Ema::builder().named("ema")); // ema.length
let dist = g.add(Sub::builder()); // momentum = price - ema
let expo = g.add(Exposure::builder().named("exposure")); // exposure.scale
let gate = g.add(LongOnly::builder().named("longonly")); // longonly.enabled (the bool param)
let broker = g.add(SimBroker::builder(pip_size));
let eq = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq));
let ex = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex));
let price = g.source_role("price", ScalarKind::F64);
g.feed(price, [ema.input("series"), dist.input("lhs"), broker.input("price")]);
g.connect(ema.output("value"), dist.input("rhs")); // momentum = price - ema
g.connect(dist.output("value"), expo.input("signal"));
g.connect(expo.output("exposure"), gate.input("exposure"));
g.connect(gate.output("exposure"), broker.input("exposure")); // gated exposure -> broker
g.connect(broker.output("equity"), eq.input("col[0]")); // equity -> sink
g.connect(gate.output("exposure"), ex.input("col[0]")); // gated exposure -> sink
let bp = g.build().expect("momentum blueprint wiring resolves");
(bp, rx_eq, rx_ex)
}
/// Run the momentum strategy over its built-in grid — `ema.length ∈ {5,10}` ×
/// `exposure.scale ∈ {0.5,1.0}` × `longonly.enabled ∈ {true,false}` = 8 points,
/// all three axes varying. Mirrors `sweep_family`: capture the binder's varying
/// axes, key each member via the generic portable `member_key`. With `--trace`,
/// persist each member under `runs/traces/<name>/<member_key>/`.
fn momentum_sweep_family(trace: Option<&str>, data: &DataSource) -> SweepFamily {
let pip = data.pip_size();
let window = data.full_window();
let bp = momentum_blueprint_with_sinks(pip).0;
let space = bp.param_space();
let binder = bp
.axis("ema.length", [5, 10])
.axis("exposure.scale", [0.5, 1.0])
.axis("longonly.enabled", [true, false]);
let varying: HashSet<String> = binder.varying_axes().into_iter().collect();
binder
.sweep(|point| {
let (bp, rx_eq, rx_ex) = momentum_blueprint_with_sinks(pip);
let mut h = bp
.bootstrap_with_cells(point)
.expect("grid points are kind-checked against param_space");
let sources = data.run_sources();
h.run(sources);
let eq_rows = rx_eq.try_iter().collect::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
let named = zip_params(&space, point);
let key = member_key(&named, &varying);
let manifest = sim_optimal_manifest(named, window, 0, pip);
if let Some(name) = trace {
persist_traces(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows);
}
let equity = f64_field(&eq_rows, 0);
let exposure = f64_field(&ex_rows, 0);
RunReport { manifest, metrics: summarize(&equity, &exposure) }
})
.expect("the momentum named grid matches the momentum param-space")
}
/// Render a sweep family as one `RunReport` JSON line per point. Test helper:
/// production (`run_sweep`) renders *and* persists per point.
#[cfg(test)]
fn sweep_report() -> String {
let mut out = String::new();
for pt in &sweep_family(None, &DataSource::Synthetic).points {
out.push_str(&pt.report.to_json());
out.push('\n');
}
out
}
/// The default run registry: an append-only JSONL store under the current
/// working directory. (A project-configured runs-dir via `Aura.toml` is a later
/// refinement.)
fn default_registry() -> Registry {
Registry::open("runs/runs.jsonl")
}
/// Which built-in strategy `aura sweep` runs. Default (today's behaviour) is the
/// SMA-cross sample; `momentum` is the bool-param demo.
#[derive(Clone, Copy, PartialEq, Debug)]
enum Strategy {
SmaCross,
Momentum,
}
/// Parse the `sweep` tail:
/// `[--strategy <sma|momentum>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>]`.
/// Defaults: SMA-cross, synthetic, name "sweep", no persist (today's bare `aura
/// sweep`). `--name` and `--trace` are mutually exclusive; `--from`/`--to` (ms,
/// `i64`) require `--real` (there is no synthetic window knob). Pure (no I/O /
/// exit) so the grammar is unit-testable; `main` does the side effects. An unknown
/// token, a flag without its value, an unknown strategy, both name flags, or a
/// window flag without `--real` rejects.
fn parse_sweep_args(rest: &[&str]) -> Result<(Strategy, String, bool, DataChoice), String> {
let usage = || "sweep [--strategy <sma|momentum>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>]".to_string();
let mut strategy = Strategy::SmaCross;
let mut name: Option<(String, bool)> = None; // (name, persist)
let mut real = RealWindowGrammar::default();
let mut tail = rest;
while let Some((flag, t)) = tail.split_first() {
let (value, t) = t.split_first().ok_or_else(usage)?;
if real.accept(flag, value, &usage)? {
tail = t;
continue;
}
match *flag {
"--strategy" => {
strategy = match *value {
"sma" => Strategy::SmaCross,
"momentum" => Strategy::Momentum,
_ => return Err(usage()),
};
}
"--name" if name.is_none() => name = Some(((*value).to_string(), false)),
"--trace" if name.is_none() => name = Some(((*value).to_string(), true)),
_ => return Err(usage()),
}
tail = t;
}
let (name, persist) = name.unwrap_or_else(|| ("sweep".to_string(), false));
Ok((strategy, name, persist, real.finish(&usage)?))
}
/// Parse the `walkforward` tail: `[--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>]`.
/// Defaults: synthetic, name "walkforward", no persist. `--name`/`--trace` are
/// mutually exclusive; `--from`/`--to` require `--real`. Pure (no I/O / exit).
fn parse_walkforward_args(rest: &[&str]) -> Result<(String, bool, DataChoice), String> {
let usage = || "walkforward [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>]".to_string();
let mut name: Option<(String, bool)> = None; // (name, persist)
let mut real = RealWindowGrammar::default();
let mut tail = rest;
while let Some((flag, t)) = tail.split_first() {
let (value, t) = t.split_first().ok_or_else(usage)?;
if real.accept(flag, value, &usage)? {
tail = t;
continue;
}
match *flag {
"--name" if name.is_none() => name = Some(((*value).to_string(), false)),
"--trace" if name.is_none() => name = Some(((*value).to_string(), true)),
_ => return Err(usage()),
}
tail = t;
}
let (name, persist) = name.unwrap_or_else(|| ("walkforward".to_string(), false));
Ok((name, persist, real.finish(&usage)?))
}
/// Render a family-member stdout line: the assigned `family_id` plus the embedded
/// `RunReport`. The report is emitted in its own declaration key order (manifest
/// leads with `commit`, C18) so the line is byte-identical to the stored
/// `families.jsonl`. `serde_json::json!` would route the report through a
/// `serde_json::Value` and re-alphabetize the manifest keys (broker-first),
/// diverging from the store — hence the report is spliced in pre-serialized (#99).
fn family_member_line(id: &str, report: &RunReport) -> String {
format!(
r#"{{"family_id":{},"report":{}}}"#,
serde_json::to_string(id).expect("a string id always serializes"),
report.to_json()
)
}
/// Monte-Carlo variant of [`family_member_line`]: the per-draw line also carries the
/// realization `seed` (between `family_id` and `report`), matching `run_mc`'s shape.
fn mc_member_line(id: &str, seed: u64, report: &RunReport) -> String {
format!(
r#"{{"family_id":{},"seed":{},"report":{}}}"#,
serde_json::to_string(id).expect("a string id always serializes"),
seed,
report.to_json()
)
}
/// `aura sweep [--strategy <sma|momentum>] [--name <n>|--trace <n>]`: run the
/// selected built-in sweep, persist it as a *family* (related records sharing one
/// `family_id`, C18/C21) via `append_family`, and print each point's record line
/// carrying the assigned id. With `--trace`, also persist each member's streams
/// under `runs/traces/<n>/<member_key>/` (opt-in).
fn run_sweep(strategy: Strategy, name: &str, persist: bool, data: DataSource) {
if persist
&& let Err(e) = TraceStore::open("runs").ensure_name_free(name, WriteKind::Family)
{
eprintln!("aura: {e}");
std::process::exit(2);
}
let reg = default_registry();
let family = match strategy {
Strategy::SmaCross => sweep_family(persist.then_some(name), &data),
Strategy::Momentum => momentum_sweep_family(persist.then_some(name), &data),
};
let id = match reg.append_family(name, FamilyKind::Sweep, &sweep_member_reports(&family)) {
Ok(id) => id,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
for pt in &family.points {
println!("{}", family_member_line(&id, &pt.report));
}
}
/// `aura walkforward [--name <n>|--trace <n>]`: run a built-in rolling walk-forward
/// over the sample blueprint + a synthetic windowed source. Per window: sweep the
/// built-in grid on the in-sample slice, optimize by total_pips (axis 2 inside axis 3,
/// where aura-cli bridges engine + registry), run the chosen params out-of-sample.
/// Persist the per-window OOS reports as a *family* (C18/C21) via `append_family`,
/// print each carrying the assigned id, then the stitched summary line. With
/// `--trace`, also persist each OOS member's streams under
/// `runs/traces/<n>/oos<ns>/` (opt-in). Deterministic (C1).
fn run_walkforward(name: &str, persist: bool, data: DataSource) {
if persist
&& let Err(e) = TraceStore::open("runs").ensure_name_free(name, WriteKind::Family)
{
eprintln!("aura: {e}");
std::process::exit(2);
}
let reg = default_registry();
let result = walkforward_family(persist.then_some(name), &data);
let id =
match reg.append_family(name, FamilyKind::WalkForward, &walkforward_member_reports(&result))
{
Ok(id) => id,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
for w in &result.windows {
println!("{}", family_member_line(&id, &w.run.oos_report));
}
println!("{}", walkforward_summary_json(&result));
}
/// The built-in rolling walk-forward: 24-bar in-sample, 12-bar out-of-sample,
/// stepping 12 (contiguous OOS tiling), over the 60-bar synthetic span -> 3
/// windows. Each window sweeps the built-in grid in-sample, optimizes by
/// total_pips (axis 2), and runs the chosen params out-of-sample.
fn walkforward_family(trace: Option<&str>, data: &DataSource) -> WalkForwardResult {
let span = data.wf_full_span();
let (is_len, oos_len, step) = data.wf_window_sizes();
let roller = match WindowRoller::new(span, is_len, oos_len, step, RollMode::Rolling) {
Ok(r) => r,
Err(e) => {
eprintln!("aura: walk-forward window too short for one IS+OOS span: {e:?}");
std::process::exit(2);
}
};
let space = sample_blueprint_with_sinks(data.pip_size()).0.param_space();
walk_forward(roller, space, |w: WindowBounds| {
let is_family = sweep_over(w.is.0, w.is.1, data);
let best = optimize(&is_family, "total_pips").expect("total_pips is a known metric");
let (oos_equity, oos_report) = run_oos(&best.params, w.oos.0, w.oos.1, trace, data);
WindowRun {
// The tag-free sweep winner is the chosen point; its kinds live on
// WalkForwardResult.space (computed once above from the same blueprint).
chosen_params: best.params,
oos_equity,
oos_report,
}
})
}
/// Sweep the built-in named grid over an in-sample window, sourcing the in-memory
/// windowed stream. Mirrors `sweep_family`, but windowed by `[from, to]`.
fn sweep_over(from: Timestamp, to: Timestamp, data: &DataSource) -> SweepFamily {
let pip = data.pip_size();
let bp = sample_blueprint_with_sinks(pip).0;
let space = bp.param_space();
let (tf, ts, (mf, ms, msig)) = data.strategy_lengths();
bp.axis("signals.trend.fast.length", tf)
.axis("signals.trend.slow.length", ts)
.axis("signals.momentum.fast.length", [mf])
.axis("signals.momentum.slow.length", [ms])
.axis("signals.momentum.signal.length", [msig])
.axis("signals.blend.weights[0]", [1.0])
.axis("signals.blend.weights[1]", [1.0])
.axis("exposure.scale", [0.5])
.sweep(|point| {
let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(pip);
let mut h = bp
.bootstrap_with_cells(point)
.expect("grid points are kind-checked against param_space");
let sources = data.windowed_sources(from, to);
let window = window_of(&sources).expect("non-empty in-sample window");
h.run(sources);
let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
RunReport {
manifest: sim_optimal_manifest(zip_params(&space, point), window, 0, pip),
metrics: summarize(&equity, &exposure),
}
})
.expect("the built-in named grid matches the sample param-space")
}
/// Run the chosen params over an out-of-sample window; return the recorded
/// pip-equity segment (for stitching) and the OOS RunReport (the C18 record).
fn run_oos(
params: &[Cell],
from: Timestamp,
to: Timestamp,
trace: Option<&str>,
data: &DataSource,
) -> (Vec<(Timestamp, f64)>, RunReport) {
let pip = data.pip_size();
let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(pip);
let space = bp.param_space();
let mut h = bp
.bootstrap_with_cells(params)
.expect("chosen params pre-validated by the in-sample GridSpace::new");
let sources = data.windowed_sources(from, to);
let window = window_of(&sources).expect("non-empty out-of-sample window");
h.run(sources);
let eq_rows = rx_eq.try_iter().collect::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
let manifest = sim_optimal_manifest(zip_params(&space, params), window, 0, pip);
if let Some(name) = trace {
persist_traces(&format!("{name}/oos{}", from.0), &manifest, &eq_rows, &ex_rows);
}
let equity = f64_field(&eq_rows, 0);
let exposure = f64_field(&ex_rows, 0);
let report = RunReport { manifest, metrics: summarize(&equity, &exposure) };
(equity, report)
}
/// The walk-forward summary line: window count, stitched OOS total pips (the last
/// stitched-curve value), and the on-demand per-param stability. Canonical JSON
/// (C14).
fn walkforward_summary_json(result: &WalkForwardResult) -> String {
let total = result.stitched_oos_equity.last().map(|&(_, v)| v).unwrap_or(0.0);
serde_json::json!({
"walkforward": {
"windows": result.windows.len(),
"stitched_total_pips": total,
"param_stability": param_stability(result),
}
})
.to_string()
}
/// A longer deterministic stream than `showcase_prices` — enough for several
/// IS/OOS windows with SMA warm-up. Seed-determined via `SyntheticSpec` (C1).
fn walkforward_prices() -> Vec<(Timestamp, Scalar)> {
let spec = SyntheticSpec { start: 1.0, len: 60, step: 1 };
let mut src = spec.source(7);
let mut out = Vec::new();
while let Some(item) = aura_engine::Source::next(&mut src) {
out.push(item);
}
out
}
/// The in-memory windowed source the built-in demo uses (the firewall mapping to
/// `DataServer::stream_m1_windowed` is the real-data path; the demo stays in-memory,
/// mirroring `run_sweep`'s `showcase_prices`). Inclusive `[from, to]`.
fn walkforward_window_source(from: Timestamp, to: Timestamp) -> VecSource {
VecSource::new(
walkforward_prices()
.into_iter()
.filter(|&(t, _)| t >= from && t <= to)
.collect(),
)
}
/// Render the built-in walk-forward as the per-window OOS RunReport lines plus the
/// summary line — the `run_walkforward` shape minus registry persistence. Test
/// helper (mirrors `sweep_report`).
#[cfg(test)]
fn walkforward_report() -> String {
let result = walkforward_family(None, &DataSource::Synthetic);
let mut out = String::new();
for w in &result.windows {
out.push_str(&w.run.oos_report.to_json());
out.push('\n');
}
out.push_str(&walkforward_summary_json(&result));
out.push('\n');
out
}
/// The built-in Monte-Carlo family: the sample harness over a fixed (empty) base
/// point, re-seeded across a built-in seed set — each seed a disjoint C1
/// realization of a synthetic price walk (C12 axis 4). Mirrors `sweep_family`,
/// varying the *seed* rather than a tuning param. The seed -> `Source`
/// construction lives inside the per-draw closure (eager-agnostic, #71).
fn mc_family(trace: Option<&str>) -> McFamily {
let base_point: Vec<Scalar> = Vec::new();
monte_carlo(&base_point, &[1, 2, 3], |seed, _base| {
let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE);
let spec = SyntheticSpec { start: 1.0, len: 32, step: 1 };
let sources: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(spec.source(seed))];
let window = window_of(&sources).expect("non-empty synthetic stream");
h.run(sources);
let eq_rows = rx_eq.try_iter().collect::<Vec<_>>();
let ex_rows = rx_ex.try_iter().collect::<Vec<_>>();
let manifest = sim_optimal_manifest(
vec![
("sma_fast".to_string(), Scalar::i64(2)),
("sma_slow".to_string(), Scalar::i64(4)),
("exposure_scale".to_string(), Scalar::f64(0.5)),
],
window,
seed,
SYNTHETIC_PIP_SIZE,
);
if let Some(name) = trace {
persist_traces(&format!("{name}/seed{seed}"), &manifest, &eq_rows, &ex_rows);
}
let equity = f64_field(&eq_rows, 0);
let exposure = f64_field(&ex_rows, 0);
RunReport { manifest, metrics: summarize(&equity, &exposure) }
})
}
/// Render an `McAggregate` as one canonical JSON line. `McAggregate` itself is not
/// `Serialize` (only its `MetricStats` fields are), so the line is built from the
/// three per-metric stat blocks.
fn mc_aggregate_json(agg: &McAggregate) -> String {
serde_json::json!({
"mc_aggregate": {
"total_pips": agg.total_pips,
"max_drawdown": agg.max_drawdown,
"exposure_sign_flips": agg.exposure_sign_flips,
}
})
.to_string()
}
/// `aura mc [--name <n>|--trace <n>]`: run the built-in Monte-Carlo family, persist
/// it to the family store via `append_family` (C18/C21), print each draw's record
/// line (carrying the assigned `family_id`) plus the aggregate line. With `--trace`,
/// also persist each draw's streams under `runs/traces/<n>/seed<seed>/` (opt-in).
fn run_mc(name: &str, persist: bool) {
if persist
&& let Err(e) = TraceStore::open("runs").ensure_name_free(name, WriteKind::Family)
{
eprintln!("aura: {e}");
std::process::exit(2);
}
let reg = default_registry();
let family = mc_family(persist.then_some(name));
let id = match reg.append_family(name, FamilyKind::MonteCarlo, &mc_member_reports(&family)) {
Ok(id) => id,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
for draw in &family.draws {
println!("{}", mc_member_line(&id, draw.seed, &draw.report));
}
println!("{}", mc_aggregate_json(&family.aggregate));
}
/// Render the built-in Monte-Carlo family as the per-draw `RunReport` lines plus
/// the aggregate line — the `run_mc` shape minus registry persistence (no
/// `family_id`, which is store-assigned). Test helper, mirroring `sweep_report` /
/// `walkforward_report`: it carries the C1-determinism test of the family
/// computation, separate from the store-dependent id.
#[cfg(test)]
fn mc_report() -> String {
let family = mc_family(None);
let mut out = String::new();
for draw in &family.draws {
out.push_str(&draw.report.to_json());
out.push('\n');
}
out.push_str(&mc_aggregate_json(&family.aggregate));
out.push('\n');
out
}
/// `aura runs families`: one header line per stored family (id, kind, member
/// count), in first-seen store order.
fn runs_families() {
let reg = default_registry();
let members = match reg.load_family_members() {
Ok(m) => m,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
for fam in group_families(members) {
println!(
"{}",
serde_json::json!({ "family_id": fam.id, "kind": fam.kind, "members": fam.members.len() })
);
}
}
/// `aura runs family <id> [rank <metric>]`: list one family's member reports in
/// ordinal order, or best-first by `metric`. An unknown id is an empty family
/// (prints nothing, exit 0); an unknown metric is a usage error (stderr + exit 2).
fn runs_family(id: &str, rank: Option<&str>) {
let reg = default_registry();
let members = match reg.load_family_members() {
Ok(m) => m,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
};
let Some(family) = group_families(members).into_iter().find(|f| f.id == id) else {
return; // unknown family id: empty, exit 0
};
let reports: Vec<RunReport> = family.members.iter().map(|m| m.report.clone()).collect();
let ordered = match rank {
Some(metric) => match rank_by(reports, metric) {
Ok(r) => r,
Err(e) => {
eprintln!("aura: {e}");
std::process::exit(2);
}
},
None => reports,
};
for report in &ordered {
println!("{}", report.to_json());
}
}
// --- MACD proof-of-concept (a richer, nested indicator + strategy) -----------
/// The MACD signal as a named composite: price → fast/slow `Ema` → the MACD line
/// (their spread) → a signal `Ema` of that line → the histogram (line signal).
/// The composite exposes all **three MACD lines** as a named output record
/// (`macd`, `signal`, `histogram`); the strategy trades the histogram by reading
/// `from_field: 2`. A richer fixture than `sma_cross`: a nested EMA-of-EMA chain
/// with interior fan-out (the MACD line feeds *both* the signal EMA and the
/// histogram). Three `length` knobs (fast, slow, signal) are injected at compile
/// in node order; value-empty here.
fn macd(name: &str) -> Composite {
let mut g = GraphBuilder::new(name);
let fast = g.add(Ema::builder().named("fast")); // fast EMA
let slow = g.add(Ema::builder().named("slow")); // slow EMA
let line = g.add(Sub::builder()); // MACD line = fast slow
let signal = g.add(Ema::builder().named("signal")); // signal EMA of the MACD line
let hist = g.add(Sub::builder()); // histogram = MACD line signal
let price = g.input_role("price");
g.feed(price, [fast.input("series"), slow.input("series")]);
g.connect(fast.output("value"), line.input("lhs")); // fast → line
g.connect(slow.output("value"), line.input("rhs")); // slow → line
g.connect(line.output("value"), signal.input("series")); // line → signal EMA
g.connect(line.output("value"), hist.input("lhs")); // line → histogram
g.connect(signal.output("value"), hist.input("rhs")); // signal → histogram
g.expose(line.output("value"), "macd"); // the MACD line
g.expose(signal.output("value"), "signal"); // the signal line
g.expose(hist.output("value"), "histogram"); // the histogram
g.build().expect("sample macd wiring resolves")
}
/// The MACD strategy blueprint (value-empty): the `macd` histogram → `Exposure` →
/// `SimBroker` → recording sinks. Channels are threaded so a run can drain the
/// sinks; `macd_blueprint` drops the receivers for the structural render.
fn macd_strategy_blueprint(
tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
) -> Composite {
let mut g = GraphBuilder::new("macd_strategy");
let macd_node = g.add(macd("macd"));
let exposure = g.add(Exposure::builder());
let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE));
let eq = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq));
let ex = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex));
let price = g.source_role("price", ScalarKind::F64);
g.feed(price, [macd_node.input("price"), broker.input("price")]);
g.connect(macd_node.output("histogram"), exposure.input("signal")); // histogram → Exposure
g.connect(exposure.output("exposure"), broker.input("exposure")); // exposure → broker slot 0
g.connect(broker.output("equity"), eq.input("col[0]")); // equity → sink
g.connect(exposure.output("exposure"), ex.input("col[0]")); // exposure → sink
g.build().expect("macd_strategy wiring resolves")
}
/// The MACD strategy blueprint as a param-space fixture (receivers dropped, since
/// `param_space()` reads structure only, never runs the graph).
#[cfg(test)]
fn macd_blueprint() -> Composite {
let (tx_eq, _rx_eq) = mpsc::channel();
let (tx_ex, _rx_ex) = mpsc::channel();
macd_strategy_blueprint(tx_eq, tx_ex)
}
/// The point vector for the MACD strategy, in `param_space()` slot order:
/// `[fast EMA length, slow EMA length, signal EMA length, exposure scale]`. Short
/// windows so the 7-tick synthetic stream still produces a non-trivial trace
/// (conventional MACD is 12/26/9, meaningless on 7 points).
fn macd_point() -> Vec<Scalar> {
vec![Scalar::i64(2), Scalar::i64(4), Scalar::i64(3), Scalar::f64(0.5)]
}
/// A longer synthetic stream than the SMA sample's 7 ticks: MACD's EMAs each warm
/// up over their `length`, so the stream rises, falls, then rises again to give the
/// histogram room to flip sign more than once *after* warm-up. Deterministic (C1).
fn macd_prices() -> Vec<(Timestamp, Scalar)> {
[
1.0000_f64, 1.0008, 1.0021, 1.0039, 1.0062, 1.0090, 1.0083, 1.0061, 1.0034,
1.0012, 0.9998, 1.0006, 1.0024, 1.0047, 1.0069, 1.0086, 1.0097, 1.0092,
]
.iter()
.enumerate()
.map(|(i, &p)| (Timestamp(i as i64 + 1), Scalar::f64(p)))
.collect()
}
/// Run the MACD strategy: compile the nested composite blueprint to a flat harness
/// (the same bootstrap path the SMA sample's compiled view uses), drive it on the
/// synthetic stream, and fold both sinks into a `RunReport`. Pure and
/// deterministic (C1).
fn run_macd(trace: Option<&str>) -> RunReport {
if let Some(n) = trace
&& let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run)
{
eprintln!("aura: {e}");
std::process::exit(2);
}
let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel();
let flat = macd_strategy_blueprint(tx_eq, tx_ex)
.compile_with_params(&macd_point())
.expect("valid macd blueprint");
let mut h = Harness::bootstrap(flat).expect("valid macd harness");
let sources: Vec<Box<dyn aura_engine::Source>> =
vec![Box::new(VecSource::new(macd_prices()))];
let window = window_of(&sources).expect("non-empty macd stream");
h.run(sources);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let manifest = sim_optimal_manifest(
vec![
("ema_fast".to_string(), Scalar::i64(2)),
("ema_slow".to_string(), Scalar::i64(4)),
("ema_signal".to_string(), Scalar::i64(3)),
("exposure_scale".to_string(), Scalar::f64(0.5)),
],
window,
0,
SYNTHETIC_PIP_SIZE,
);
if let Some(name) = trace {
persist_traces(name, &manifest, &eq_rows, &ex_rows);
}
let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
RunReport { manifest, metrics }
}
const USAGE: &str =
"usage: aura run [--macd] [--trace <name>] | aura run --real <SYMBOL> [--from <ms>] [--to <ms>] [--trace <name>] | aura chart <name> [--tap <t>] [--panels] | aura graph | aura sweep [--strategy <sma|momentum>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura mc [--name <n>|--trace <n>] | aura walkforward [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura runs families | aura runs family <id> [rank <metric>]";
fn main() {
// Collect argv and match the whole vector: every accepted form is exhaustive,
// so an unexpected trailing token falls through to the usage-error path rather
// than masquerading as a successful run (#16 strict reading).
let args: Vec<String> = std::env::args().skip(1).collect();
match args.iter().map(String::as_str).collect::<Vec<_>>().as_slice() {
["run"] => println!("{}", run_sample(None).to_json()),
["run", "--macd"] => println!("{}", run_macd(None).to_json()),
["run", "--trace", name] => println!("{}", run_sample(Some(name)).to_json()),
["run", "--macd", "--trace", name] => println!("{}", run_macd(Some(name)).to_json()),
["run", "--real", rest @ ..] => match parse_real_args(rest) {
Ok((sym, from, to, trace)) => {
println!("{}", run_sample_real(&sym, from, to, trace.as_deref()).to_json())
}
Err(msg) => {
eprintln!("aura: {msg}");
std::process::exit(2);
}
},
["chart", rest @ ..] => match parse_chart_args(rest) {
Ok((name, tap, mode)) => emit_chart(&name, tap.as_deref(), mode),
Err(msg) => {
eprintln!("aura: {msg}");
std::process::exit(2);
}
},
["graph"] => print!("{}", render::render_html(&sample_blueprint())),
["sweep", rest @ ..] => match parse_sweep_args(rest) {
Ok((strategy, name, persist, choice)) => {
run_sweep(strategy, &name, persist, DataSource::from_choice(choice))
}
Err(msg) => {
eprintln!("aura: {msg}");
std::process::exit(2);
}
},
["walkforward", rest @ ..] => match parse_walkforward_args(rest) {
Ok((name, persist, choice)) => {
run_walkforward(&name, persist, DataSource::from_choice(choice))
}
Err(msg) => {
eprintln!("aura: {msg}");
std::process::exit(2);
}
},
["mc"] => run_mc("mc", false),
["mc", "--name", n] => run_mc(n, false),
["mc", "--trace", n] => run_mc(n, true),
["runs", "families"] => runs_families(),
["runs", "family", id] => runs_family(id, None),
["runs", "family", id, "rank", metric] => runs_family(id, Some(metric)),
["--help"] | ["-h"] => println!("{USAGE}"),
_ => {
eprintln!("aura: {USAGE}");
std::process::exit(2);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn cmp_member(key: &str, ts: &[i64], vals: &[f64]) -> FamilyMember {
cmp_member_win(key, ts, vals, (0, 0))
}
/// Like [`cmp_member`] but with an explicit manifest `window` so a test can
/// model walk-forward members (disjoint per-member OOS windows) and assert the
/// family window spans them.
fn cmp_member_win(key: &str, ts: &[i64], vals: &[f64], window: (i64, i64)) -> FamilyMember {
let rows: Vec<(Timestamp, Vec<Scalar>)> =
ts.iter().zip(vals).map(|(&t, &v)| (Timestamp(t), vec![Scalar::f64(v)])).collect();
let tap = ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &rows);
FamilyMember {
key: key.to_string(),
traces: RunTraces {
manifest: sim_optimal_manifest(
vec![],
(Timestamp(window.0), Timestamp(window.1)),
0,
1.0,
),
taps: vec![tap],
},
}
}
#[test]
fn comparison_overlays_one_shared_scale_series_per_member() {
let members = vec![
cmp_member("a", &[1, 2, 3], &[10.0, 11.0, 12.0]),
cmp_member("b", &[1, 2, 3], &[20.0, 21.0, 22.0]),
];
let data = build_comparison_chart_data("fam", &members, "equity").expect("builds");
assert_eq!(data.xs, vec![1, 2, 3]);
assert_eq!(data.series.len(), 2);
assert_eq!(data.series[0].name, "a");
assert_eq!(data.series[1].name, "b");
// ONE shared y-scale across members (same quantity).
assert_eq!(data.series[0].y_scale_id, data.series[1].y_scale_id);
// shared ts -> dense, no nulls.
assert!(data.series[0].points.iter().all(Option::is_some));
// #102 meta wiring: a family carries kind/name/member-count + the one
// compared tap, and never the per-member params (those are the labels).
assert_eq!(data.meta.kind, "family");
assert_eq!(data.meta.name, "fam");
assert_eq!(data.meta.members, Some(2));
assert_eq!(data.meta.taps, vec!["equity".to_string()]);
assert!(data.meta.params.is_empty(), "family meta must not repeat per-member params");
}
#[test]
fn comparison_disjoint_members_are_null_complementary() {
let members = vec![
cmp_member("oos1", &[1, 2], &[10.0, 11.0]),
cmp_member("oos2", &[3, 4], &[20.0, 21.0]),
];
let data = build_comparison_chart_data("fam", &members, "equity").expect("builds");
assert_eq!(data.xs, vec![1, 2, 3, 4]);
assert_eq!(data.series[0].points, vec![Some(10.0), Some(11.0), None, None]);
assert_eq!(data.series[1].points, vec![None, None, Some(20.0), Some(21.0)]);
}
/// #102 family-window semantics: the header's `window` for a family is the
/// SPAN across all members — `(min member.from, max member.to)` — not the first
/// member's window. The distinction is load-bearing for a walk-forward family,
/// whose members are DISJOINT OOS windows (commit 4c64feb): labelling such a
/// family with `members[0]`'s window mislabels the family's true coverage. The
/// span reading is correct for all three kinds (sweep/MC members share a window,
/// so their span collapses to that shared window).
#[test]
fn comparison_window_spans_disjoint_walk_forward_members() {
let members = vec![
cmp_member_win("oos1", &[10, 20], &[1.0, 2.0], (10, 20)),
cmp_member_win("oos2", &[30, 40], &[3.0, 4.0], (30, 40)),
cmp_member_win("oos3", &[50, 60], &[5.0, 6.0], (50, 60)),
];
let data = build_comparison_chart_data("wf", &members, "equity").expect("builds");
// SPAN of all OOS windows (10..60), NOT members[0]'s window (10..20).
assert_eq!(data.meta.window, (10, 60));
}
#[test]
fn comparison_errors_when_no_member_has_the_tap() {
let members = vec![cmp_member("a", &[1], &[1.0])];
assert!(build_comparison_chart_data("fam", &members, "nosuch").is_err());
}
#[test]
fn decimate_bounds_the_spine_to_twice_the_bucket_count() {
let n = 10_000usize;
let xs: Vec<i64> = (0..n as i64).collect();
let points: Vec<Option<f64>> = (0..n).map(|i| Some(i as f64)).collect();
let data = ChartData {
xs,
series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }],
meta: ChartMeta::default(),
};
let out = decimate(data, 2000);
assert!(out.xs.len() <= 4000, "spine not bounded: {}", out.xs.len());
assert_eq!(out.xs.len(), out.series[0].points.len(), "xs and points must stay aligned");
}
#[test]
fn decimate_preserves_per_bucket_min_and_max() {
// 10 points, 2 buckets -> bucket 0 = idx 0..5 (a spike), bucket 1 = idx 5..10 (a trough).
let xs: Vec<i64> = (0..10).collect();
let mut pv = vec![1.0_f64; 10];
pv[3] = 999.0;
pv[7] = -50.0;
let points: Vec<Option<f64>> = pv.into_iter().map(Some).collect();
let data = ChartData {
xs,
series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }],
meta: ChartMeta::default(),
};
let out = decimate(data, 2);
let got = out.series[0].points.clone();
assert!(got.contains(&Some(999.0)), "bucket max (spike) dropped: {got:?}");
assert!(got.contains(&Some(-50.0)), "bucket min (trough) dropped: {got:?}");
}
#[test]
fn decimate_keeps_an_all_null_bucket_null() {
let xs: Vec<i64> = (0..10).collect();
let mut points: Vec<Option<f64>> = (0..5).map(|i| Some(i as f64)).collect();
points.extend(std::iter::repeat_n(None, 5));
let data = ChartData {
xs,
series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }],
meta: ChartMeta::default(),
};
let out = decimate(data, 2);
assert_eq!(*out.series[0].points.last().unwrap(), None, "all-null bucket must stay null");
}
#[test]
fn decimate_is_a_noop_within_budget() {
let data = ChartData {
xs: vec![1, 2, 3],
series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points: vec![Some(1.0), Some(2.0), Some(3.0)], reduce: ReduceKind::MinMax }],
meta: ChartMeta::default(),
};
let out = decimate(data, 2000);
assert_eq!(out.xs, vec![1, 2, 3], "within-budget data must pass through unchanged");
assert_eq!(out.series[0].points, vec![Some(1.0), Some(2.0), Some(3.0)]);
}
#[test]
fn decimate_passes_meta_through_and_keeps_xs_monotonic() {
let n = 10_000usize;
let xs: Vec<i64> = (0..n as i64).collect();
let points: Vec<Option<f64>> = (0..n).map(|i| Some(i as f64)).collect();
let meta = ChartMeta { name: "keep-me".into(), ..Default::default() };
let data = ChartData { xs, series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }], meta };
let out = decimate(data, 2000);
assert_eq!(out.meta.name, "keep-me", "meta must pass through decimation");
assert!(out.xs.windows(2).all(|w| w[0] < w[1]), "decimated spine must stay strictly increasing");
}
/// #111: a bounded *level* series with `reduce = Mean` decimates to each bucket's
/// MEAN, not its min/max envelope — so a high-flip bipolar exposure shows its
/// net/duty-cycle level instead of collapsing to a -1..+1 band. RED under the
/// shipped min/max-only decimation (any bucket holding a +1 emits +1); GREEN once
/// `decimate` honours `ReduceKind::Mean`.
#[test]
fn decimate_mean_reduces_a_bipolar_series_to_its_bucket_level() {
// 10 points, 2 buckets. Bucket 0 (idx 0..5) = [+1,+1,-1,+1,+1] -> mean +0.6;
// bucket 1 (idx 5..10) = all -1 -> mean -1.0.
let xs: Vec<i64> = (0..10).collect();
let pv = vec![1.0, 1.0, -1.0, 1.0, 1.0, -1.0, -1.0, -1.0, -1.0, -1.0];
let points: Vec<Option<f64>> = pv.into_iter().map(Some).collect();
let data = ChartData {
xs,
series: vec![Series { name: "exposure".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::Mean }],
meta: ChartMeta::default(),
};
let out = decimate(data, 2);
let got = out.series[0].points.clone();
// No -1..+1 envelope: bucket 0 is its mean (+0.6), not a min/max pair.
assert!(!got.contains(&Some(1.0)), "mean reduce must not emit a +1 envelope point: {got:?}");
assert!(got.contains(&Some(0.6)), "bucket-0 duty-cycle mean (+0.6) missing: {got:?}");
// bucket 0 spans two slots, both = the mean (a flat step, not a -1->+1 ramp).
assert_eq!(got[0], Some(0.6), "first slot must be the bucket mean");
assert_eq!(got[1], Some(0.6), "second slot must also be the bucket mean");
}
/// #102 single-run meta wiring: `build_chart_data` maps the `RunManifest` into
/// `ChartData.meta` — kind "run", the name arg, the manifest window/broker, the
/// charted taps, and the bound params stringified (each typed `Scalar` rendered
/// via `render_value`, preserving its lexical form: `i64` decimal, `f64`
/// shortest round-trip). A single run carries no member count.
#[test]
fn build_chart_data_threads_run_manifest_into_meta() {
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> =
[1i64, 2, 3].iter().map(|&t| (Timestamp(t), vec![Scalar::f64(t as f64)])).collect();
let traces = RunTraces {
manifest: sim_optimal_manifest(
vec![("len".into(), Scalar::i64(10)), ("scale".into(), Scalar::f64(0.5))],
(Timestamp(1), Timestamp(3)),
7,
1.0,
),
taps: vec![ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &eq_rows)],
};
let data = build_chart_data("demo", traces);
let meta = &data.meta;
assert_eq!(meta.kind, "run");
assert_eq!(meta.name, "demo");
assert_eq!(meta.window, (1, 3));
assert_eq!(meta.broker, "sim-optimal(pip_size=1)");
assert_eq!(meta.seed, 7);
assert_eq!(meta.taps, vec!["equity".to_string()]);
assert_eq!(meta.members, None);
// params stringified via render_value: typed Scalars keep their lexical form.
assert_eq!(
meta.params,
vec![("len".to_string(), "10".to_string()), ("scale".to_string(), "0.5".to_string())]
);
}
/// #99: a sweep/walk-forward family-member stdout line embeds the `RunReport` in
/// its own declaration key order (manifest leads with `commit`), byte-matching the
/// stored `families.jsonl` — never `serde_json::Value`'s alphabetical order (which
/// would lead the manifest with `broker`).
#[test]
fn family_member_line_keeps_report_in_store_key_order() {
let report = RunReport {
manifest: sim_optimal_manifest(vec![], (Timestamp(0), Timestamp(0)), 0, 1.0),
metrics: summarize(&[], &[]),
};
let line = family_member_line("demo-1", &report);
assert!(
line.starts_with(r#"{"family_id":"demo-1","report":{"manifest":{"commit":"#),
"got: {line}"
);
assert!(
!line.contains(r#""manifest":{"broker":"#),
"manifest re-alphabetized (broker-first), should be commit-first: {line}"
);
}
/// #99: the Monte-Carlo per-draw line carries the `seed` between `family_id` and
/// `report`, and the embedded report stays in store (commit-first) key order.
#[test]
fn mc_member_line_keeps_report_in_store_key_order_with_seed() {
let report = RunReport {
manifest: sim_optimal_manifest(vec![], (Timestamp(0), Timestamp(0)), 7, 1.0),
metrics: summarize(&[], &[]),
};
let line = mc_member_line("mc-1", 7, &report);
assert!(
line.starts_with(r#"{"family_id":"mc-1","seed":7,"report":{"manifest":{"commit":"#),
"got: {line}"
);
assert!(
!line.contains(r#""manifest":{"broker":"#),
"manifest re-alphabetized (broker-first), should be commit-first: {line}"
);
}
// The vetted GER40 real-data window: the whole of September 2024 (UTC,
// inclusive), the same calendar month the gated ingest `ger40_breakout_real`
// test drives. Expressed in Unix-ms (`run_sample_real`'s window currency):
// `[2024-09-01T00:00:00Z, 2024-10-01T00:00:00Z - 1ms]`. Both gated GER40
// tests bound their runs to this window so the C1-determinism check stays
// fast — an unbounded `None, None` run drains the full archive every call.
const GER40_SEP2024_FROM_MS: i64 = 1_725_148_800_000;
const GER40_SEP2024_TO_MS: i64 = 1_727_740_799_999;
#[test]
fn data_source_synthetic_pip_and_window_match_the_built_ins() {
let d = DataSource::Synthetic;
assert_eq!(d.pip_size(), SYNTHETIC_PIP_SIZE);
assert!(!d.run_sources().is_empty());
assert_eq!(d.wf_window_sizes(), (24, 12, 12));
// full_window equals window_of over the showcase stream (byte-unchanged source)
let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(showcase_prices()))];
assert_eq!(d.full_window(), window_of(&s).unwrap());
}
#[test]
fn wf_real_roller_sizes_are_90_30_30_days_in_ns() {
// Independent expected value: a day reconstructed from its time units
// (24 h * 60 min * 60 s * 1e9 ns), not the constant's own `86_400_000_000_000`
// literal — so the test fails if either the literal or the day-count is wrong.
let day_ns: i64 = 24 * 60 * 60 * 1_000_000_000;
assert_eq!(WF_REAL_IS_NS, 90 * day_ns);
assert_eq!(WF_REAL_OOS_NS, 30 * day_ns);
assert_eq!(WF_REAL_STEP_NS, 30 * day_ns);
}
#[test]
fn strategy_lengths_are_short_for_synthetic_realistic_for_real() {
// Synthetic keeps the demo-stream lengths (byte-unchanged: the 18/60-bar
// built-in streams cannot warm a long MA).
assert_eq!(DataSource::Synthetic.strategy_lengths(), ([2, 3], [4, 5], (2, 4, 3)));
// Real uses realistic M1 lengths — no 2-5-bar noise over tens of thousands
// of bars. Constructing Real needs a server, but strategy_lengths matches on
// the variant only (no data access).
let real = DataSource::Real {
server: std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH)),
symbol: "EURUSD".into(),
from_ms: None,
to_ms: None,
pip: 0.0001,
};
let (tf, ts, macd) = real.strategy_lengths();
assert_eq!((tf, ts, macd), ([50, 100], [200, 400], (12, 26, 9)));
// every trend-fast < every trend-slow (a valid SMA cross, both variants).
assert!(tf.iter().max().unwrap() < ts.iter().min().unwrap());
let (stf, sts, _) = DataSource::Synthetic.strategy_lengths();
assert!(stf.iter().max().unwrap() < sts.iter().min().unwrap());
}
#[test]
fn walkforward_report_is_deterministic() {
// The built-in WFO render is byte-identical across two
// calls (C1).
assert_eq!(walkforward_report(), walkforward_report());
}
#[test]
fn walkforward_report_has_one_oos_line_per_window_plus_summary() {
// N per-window OOS RunReport lines + one summary line.
let out = walkforward_report();
let lines: Vec<&str> = out.lines().collect();
assert_eq!(lines.len(), 4); // built-in roll = 3 windows + 1 summary
assert!(lines[3].contains(r#""walkforward""#), "summary line: {}", lines[3]);
for line in &lines[..3] {
assert!(
line.contains(r#""manifest""#) && line.contains(r#""metrics""#),
"expected an OOS RunReport line, got: {line}",
);
}
}
/// The drained sink trace of a seeded run — the recorded rows of the equity
/// and exposure sinks. Compared row-for-row so the C1 seed-determinism
/// property is tested at the trace level (strictly stronger than the folded
/// 3-field metrics). `PartialEq` not `Eq`: `Scalar` carries `f64`.
#[derive(Debug, PartialEq)]
struct SeededTrace {
equity: Vec<(Timestamp, Vec<Scalar>)>,
exposure: Vec<(Timestamp, Vec<Scalar>)>,
}
/// A seeded run of the sample harness: the synthetic stream is generated
/// from `seed`, that same seed is recorded into the manifest, and the
/// drained sink trace is returned alongside the report. Every byte of both
/// is a function of `seed`.
fn run_sample_seeded(seed: u64) -> (RunReport, SeededTrace) {
let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE);
let spec = SyntheticSpec { start: 1.0, len: 64, step: 1 };
let window = (Timestamp(1), Timestamp((spec.len as i64 - 1) * spec.step + 1));
h.run(vec![Box::new(spec.source(seed))]);
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
let report = RunReport {
manifest: sim_optimal_manifest(
vec![
("sma_fast".to_string(), Scalar::i64(2)),
("sma_slow".to_string(), Scalar::i64(4)),
("exposure_scale".to_string(), Scalar::f64(0.5)),
],
window,
seed,
SYNTHETIC_PIP_SIZE,
),
metrics,
};
(report, SeededTrace { equity: eq_rows, exposure: ex_rows })
}
#[test]
fn same_seed_bit_identical_trace() {
// Bit-identical sink trace for a fixed seed (acceptance bullet 1, C1).
let (_, trace_a) = run_sample_seeded(42);
let (_, trace_b) = run_sample_seeded(42);
assert_eq!(trace_a, trace_b);
}
#[test]
fn different_seed_different_trace() {
// Different seeds perturb the trace (acceptance bullet 2).
let (a, _) = run_sample_seeded(1);
let (b, _) = run_sample_seeded(2);
assert_ne!(a.metrics, b.metrics);
}
#[test]
fn seed_recorded_in_manifest() {
// The seed that drove the run is recorded (acceptance bullet 3).
let (report, _) = run_sample_seeded(7);
assert_eq!(report.manifest.seed, 7);
}
#[test]
fn sample_blueprint_with_sinks_bootstraps_runs_and_drains() {
// the factory returns the two Recorder receivers (build_sample drops them),
// so a caller can bootstrap one point, run it, and drain both sinks.
let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks(SYNTHETIC_PIP_SIZE);
let mut h = bp
.with("signals.trend.fast.length", 2)
.with("signals.trend.slow.length", 4)
.with("signals.momentum.fast.length", 2)
.with("signals.momentum.slow.length", 4)
.with("signals.momentum.signal.length", 3)
.with("signals.blend.weights[0]", 1.0)
.with("signals.blend.weights[1]", 1.0)
.with("exposure.scale", 0.5)
.bootstrap()
.expect("sample blueprint compiles under a valid point");
h.run(vec![Box::new(VecSource::new(showcase_prices()))]);
assert!(!rx_eq.try_iter().collect::<Vec<_>>().is_empty(), "equity sink drained empty");
assert!(!rx_ex.try_iter().collect::<Vec<_>>().is_empty(), "exposure sink drained empty");
}
#[test]
fn sweep_report_renders_four_points_in_odometer_order() {
let out = sweep_report();
let lines: Vec<&str> = out.lines().collect();
assert_eq!(lines.len(), 4, "one JSON line per grid point; got: {out:?}");
// each line is a full RunReport; the commit is the real git HEAD
// (volatile), so pin the per-point manifest params (odometer order, last
// axis fastest) + the metric keys, not the commit value.
for line in &lines {
assert!(line.starts_with(r#"{"manifest":{"commit":""#), "not a RunReport: {line}");
}
assert!(lines[0].contains(r#""params":[["signals.trend.fast.length",{"I64":2}],["signals.trend.slow.length",{"I64":4}],["signals.momentum.fast.length",{"I64":2}],["signals.momentum.slow.length",{"I64":4}],["signals.momentum.signal.length",{"I64":3}],["signals.blend.weights[0]",{"F64":1.0}],["signals.blend.weights[1]",{"F64":1.0}],["exposure.scale",{"F64":0.5}]]"#), "line0: {}", lines[0]);
assert!(lines[1].contains(r#""params":[["signals.trend.fast.length",{"I64":2}],["signals.trend.slow.length",{"I64":5}],["signals.momentum.fast.length",{"I64":2}],["signals.momentum.slow.length",{"I64":4}],["signals.momentum.signal.length",{"I64":3}],["signals.blend.weights[0]",{"F64":1.0}],["signals.blend.weights[1]",{"F64":1.0}],["exposure.scale",{"F64":0.5}]]"#), "line1: {}", lines[1]);
assert!(lines[2].contains(r#""params":[["signals.trend.fast.length",{"I64":3}],["signals.trend.slow.length",{"I64":4}],["signals.momentum.fast.length",{"I64":2}],["signals.momentum.slow.length",{"I64":4}],["signals.momentum.signal.length",{"I64":3}],["signals.blend.weights[0]",{"F64":1.0}],["signals.blend.weights[1]",{"F64":1.0}],["exposure.scale",{"F64":0.5}]]"#), "line2: {}", lines[2]);
assert!(lines[3].contains(r#""params":[["signals.trend.fast.length",{"I64":3}],["signals.trend.slow.length",{"I64":5}],["signals.momentum.fast.length",{"I64":2}],["signals.momentum.slow.length",{"I64":4}],["signals.momentum.signal.length",{"I64":3}],["signals.blend.weights[0]",{"F64":1.0}],["signals.blend.weights[1]",{"F64":1.0}],["exposure.scale",{"F64":0.5}]]"#), "line3: {}", lines[3]);
for line in &lines {
assert!(line.contains(r#""total_pips":"#), "missing total_pips: {line}");
assert!(line.contains(r#""max_drawdown":"#), "missing max_drawdown: {line}");
assert!(line.contains(r#""exposure_sign_flips":"#), "missing flips: {line}");
assert!(line.ends_with('}'), "line not closed: {line}");
}
}
#[test]
fn sweep_report_is_deterministic() {
// C1 at the CLI edge: the same build yields a bit-identical report.
assert_eq!(sweep_report(), sweep_report());
}
#[test]
fn mc_report_is_deterministic_and_one_line_per_seed() {
// C1 at the CLI edge: the family computation renders bit-identically.
assert_eq!(mc_report(), mc_report());
let out = mc_report();
let lines: Vec<&str> = out.lines().collect();
// three seeds -> three member lines + one aggregate line
assert_eq!(lines.len(), 4, "expected 3 members + 1 aggregate: {out}");
for line in &lines[..3] {
assert!(line.contains(r#""total_pips":"#), "member line missing metrics: {line}");
}
assert!(lines[3].contains(r#""mc_aggregate":"#), "missing aggregate line: {}", lines[3]);
}
#[test]
fn cli_families_persist_and_round_trip_per_kind() {
use aura_registry::{
group_families, mc_member_reports, sweep_member_reports, walkforward_member_reports,
FamilyKind, Registry,
};
let dir = std::env::temp_dir().join(format!("aura-cli-fam-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("temp dir");
let reg = Registry::open(dir.join("runs.jsonl"));
// the exact persist chain `run_sweep`/`run_mc`/`run_walkforward` use, against
// a fresh temp store (the run_* fns themselves bind `default_registry()`):
// engine family -> per-kind extractor -> append_family.
let sid = reg
.append_family(
"sweep",
FamilyKind::Sweep,
&sweep_member_reports(&sweep_family(None, &DataSource::Synthetic)),
)
.expect("sweep family");
let mid = reg
.append_family("mc", FamilyKind::MonteCarlo, &mc_member_reports(&mc_family(None)))
.expect("mc family");
let wid = reg
.append_family(
"walkforward",
FamilyKind::WalkForward,
&walkforward_member_reports(&walkforward_family(None, &DataSource::Synthetic)),
)
.expect("walkforward family");
assert_eq!((sid.as_str(), mid.as_str(), wid.as_str()), ("sweep-0", "mc-0", "walkforward-0"));
let families = group_families(reg.load_family_members().expect("load"));
assert_eq!(families.len(), 3);
let by_id = |id: &str| families.iter().find(|f| f.id == id).expect("family present");
assert_eq!(by_id("sweep-0").kind, FamilyKind::Sweep);
assert_eq!(by_id("mc-0").kind, FamilyKind::MonteCarlo);
assert_eq!(by_id("mc-0").members.len(), 3); // 3 seeds
assert_eq!(by_id("walkforward-0").kind, FamilyKind::WalkForward);
assert_eq!(by_id("walkforward-0").members.len(), 3); // 3 windows
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn run_macd_compiles_from_nested_composite_and_is_deterministic() {
// the MACD strategy authors a nested EMA-of-EMA composite, compiles it to a
// flat runnable harness (the call not panicking proves the compile+bootstrap
// path), and runs it. C1 determinism: two runs are bit-identical.
let r1 = run_macd(None);
let r2 = run_macd(None);
assert_eq!(r1.metrics, r2.metrics);
assert_eq!(r1.to_json(), r2.to_json());
// the synthetic stream is carried end-to-end and the trace is well-formed.
let (from, to) = r1.manifest.window;
assert_eq!((from.0, to.0), (1, 18));
assert!(r1.metrics.total_pips.is_finite(), "macd pips must be finite: {:?}", r1.metrics);
assert!(r1.metrics.max_drawdown >= 0.0, "drawdown is non-negative: {:?}", r1.metrics);
// after warm-up the EMA-of-EMA histogram crosses zero, so the strategy
// reverses exposure at least once — a genuinely non-trivial trace.
assert!(
r1.metrics.exposure_sign_flips >= 1,
"macd trace should flip exposure: {:?}",
r1.metrics
);
}
/// E2E acceptance (#41, the worked example): the real MACD strategy
/// blueprint's swept param surface qualifies the three otherwise-indistinguishable
/// EMA `length` slots by node name to `macd.fast.length` / `macd.slow.length` /
/// `macd.signal.length` — the named composite boundary visible end-to-end through
/// `param_space()`, with the slot count and order unchanged (C23 — node names are
/// non-load-bearing: every interior slot stays sweepable, the `exposure.scale`
/// knob is unaffected).
#[test]
fn macd_param_space_surfaces_the_three_named_legs() {
let names: Vec<String> =
macd_blueprint().param_space().into_iter().map(|p| p.name).collect();
// three named composite-interior slots, in declared (fast, slow, signal)
// order, then the strategy-level Exposure `scale` (a root-level leaf).
assert_eq!(
names,
vec![
"macd.fast.length".to_string(),
"macd.slow.length".to_string(),
"macd.signal.length".to_string(),
"exposure.scale".to_string(),
],
"MACD param surface must expose the three named EMA lengths + scale",
);
}
/// `aura run --real <SYMBOL>` dogfoods the #71 streaming Source seam: the same
/// built-in sample signal-quality harness, but fed real M1 **close** bars
/// streamed lazily through `aura_ingest::M1FieldSource` (a `Box<dyn Source>`),
/// not synthetic `VecSource` prices. The property: over the verified bounded
/// Sept-2024 GER40 window, `run_sample_real` yields a `RunReport` whose
/// `total_pips` is finite and is C1-deterministic — two runs of the same
/// window are bit-identical JSON. Bounding the window (vs the full unbounded
/// archive) keeps the determinism check fast.
///
/// Gated like the ingest `streaming_seam` test: skip (early return) when the
/// local Pepperstone archive is absent, so the test never fails on a machine
/// without the data. Uses `GER40` — a *vetted* symbol (pip 1.0): the
/// per-instrument-pip refusal makes `run_sample_real` reject an un-specced
/// symbol before any data access (`std::process::exit(2)`), so this CLI-level
/// test must drive a symbol in the instrument table. The bounded-window AAPL.US
/// streaming property still lives in the ingest `streaming_seam` test, which
/// builds its source literally without the spec lookup.
#[test]
fn run_sample_real_streams_real_close_bars_deterministically() {
// GER40 is in the vetted instrument table (index pip 1.0); the un-specced
// AAPL.US would now refuse at the spec lookup before any data access.
const SYMBOL: &str = "GER40";
// Mirror skip_if_no_data: never fail where the local archive is absent.
let server = std::sync::Arc::new(data_server::DataServer::new(
data_server::DEFAULT_DATA_PATH,
));
if !server.has_symbol(SYMBOL) {
eprintln!(
"skip: no local data at {} (symbol {SYMBOL} absent)",
data_server::DEFAULT_DATA_PATH
);
return;
}
// The headline: a real-data run over the bounded Sept-2024 window yields
// a finite, C1-deterministic RunReport.
let r1 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None);
let r2 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None);
assert!(
r1.metrics.total_pips.is_finite(),
"real-data run must yield finite pips: {:?}",
r1.metrics
);
// C1 at the CLI edge: the same real window streamed twice is bit-identical.
assert_eq!(
r1.to_json(),
r2.to_json(),
"two real-data runs of the same window must be bit-identical (C1)"
);
}
#[test]
fn sim_optimal_manifest_renders_per_instrument_pip() {
let m = sim_optimal_manifest(vec![], (Timestamp(1), Timestamp(2)), 0, 1.0);
assert_eq!(m.broker, "sim-optimal(pip_size=1)");
let m2 = sim_optimal_manifest(vec![], (Timestamp(1), Timestamp(2)), 0, 0.0001);
assert_eq!(m2.broker, "sim-optimal(pip_size=0.0001)");
}
#[test]
fn run_real_ger40_uses_index_pip() {
// Gated: needs local GER40 data. Mirrors the existing real-path test's skip.
let server = data_server::DataServer::new(data_server::DEFAULT_DATA_PATH);
if !server.has_symbol("GER40") {
eprintln!("skip: no local GER40 data at {}", data_server::DEFAULT_DATA_PATH);
return;
}
// Bounded to the vetted Sept-2024 window so the pip-label + determinism
// checks run fast (not the full unbounded archive).
let report =
run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None);
// The looked-up index pip (1.0) reaches the manifest — not the FX 0.0001.
assert_eq!(report.manifest.broker, "sim-optimal(pip_size=1)");
// Deterministic (C1): a second run yields the same report.
let again =
run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None);
assert_eq!(report.manifest.broker, again.manifest.broker);
assert_eq!(report.metrics.total_pips, again.metrics.total_pips);
}
/// `parse_real_args` accepts a bare symbol (no window) and a full
/// symbol + `--from`/`--to` pair in any order, and rejects a flag missing its
/// value — the pure arg grammar `main` relies on for `run --real`.
#[test]
fn parse_real_args_accepts_symbol_and_optional_window() {
assert_eq!(parse_real_args(&["EURUSD"]), Ok(("EURUSD".to_string(), None, None, None)));
assert_eq!(
parse_real_args(&["EURUSD", "--from", "100", "--to", "200"]),
Ok(("EURUSD".to_string(), Some(100), Some(200), None))
);
// flags in any order
assert_eq!(
parse_real_args(&["EURUSD", "--to", "200", "--from", "100"]),
Ok(("EURUSD".to_string(), Some(100), Some(200), None))
);
// the --trace tail flag carries a string name, parsed alongside the window
assert_eq!(
parse_real_args(&["EURUSD", "--from", "100", "--trace", "demo"]),
Ok(("EURUSD".to_string(), Some(100), None, Some("demo".to_string())))
);
// a flag without its value, an empty symbol, and a non-numeric ms all reject
assert!(parse_real_args(&["EURUSD", "--from"]).is_err());
assert!(parse_real_args(&[]).is_err());
assert!(parse_real_args(&["EURUSD", "--from", "notanumber"]).is_err());
}
/// `parse_chart_args` parses `<name>`, `--tap <t>`, and `--panels` in any order
/// (the loop grammar's reason for being — the name need not lead), and surfaces
/// each of its three distinct errors: `--tap` missing its value, an unexpected
/// extra argument, and a missing `<name>`. The any-order contract is the only
/// reason `main` runs a loop here instead of an exhaustive argv match, so it is
/// asserted as a property, not just exercised.
#[test]
fn parse_chart_args_accepts_any_order_and_rejects_three_ways() {
// bare name -> default tap (None) + Overlay mode
let (name, tap, mode) = parse_chart_args(&["fam"]).expect("bare name parses");
assert_eq!(name, "fam");
assert_eq!(tap, None);
assert!(matches!(mode, ChartMode::Overlay));
// name + --tap + --panels in canonical order
let (name, tap, mode) =
parse_chart_args(&["fam", "--tap", "equity", "--panels"]).expect("full parses");
assert_eq!(name, "fam");
assert_eq!(tap.as_deref(), Some("equity"));
assert!(matches!(mode, ChartMode::Panels));
// any order: flags before the name, and --panels between name and --tap
let (name, tap, mode) =
parse_chart_args(&["--tap", "equity", "fam"]).expect("flag-first parses");
assert_eq!(name, "fam");
assert_eq!(tap.as_deref(), Some("equity"));
assert!(matches!(mode, ChartMode::Overlay));
let (name, tap, mode) =
parse_chart_args(&["--panels", "fam", "--tap", "exposure"]).expect("interleaved parses");
assert_eq!(name, "fam");
assert_eq!(tap.as_deref(), Some("exposure"));
assert!(matches!(mode, ChartMode::Panels));
// three distinct error branches, by message (the Ok tuple holds a
// non-Debug ChartMode, so match the Err arm instead of unwrapping).
let err = |args: &[&str]| match parse_chart_args(args) {
Err(e) => e,
Ok(_) => panic!("expected an error for {args:?}"),
};
assert_eq!(err(&["fam", "--tap"]), "--tap needs a value");
assert_eq!(err(&["fam", "extra"]), "unexpected chart argument 'extra'");
assert_eq!(err(&["--panels"]), "chart needs a <name>");
}
#[test]
fn run_sample_is_deterministic_and_non_trivial() {
let r1 = run_sample(None);
let r2 = run_sample(None);
// C1 determinism: two runs are bit-identical (metrics + rendered JSON).
assert_eq!(r1.metrics, r2.metrics);
assert_eq!(r1.to_json(), r2.to_json());
let m = &r1.metrics;
// exactly one exposure sign flip in the demo trace (rises then reverses).
assert_eq!(m.exposure_sign_flips, 1);
// a non-trivial, populated trace: a real drawdown.
assert!(m.max_drawdown > 0.0);
// hand-computed magnitudes for the chosen stream (float tolerance; the
// computation's dust is ~1e-15).
assert!(
(m.max_drawdown - 0.17).abs() < 1e-9,
"max_drawdown = {}",
m.max_drawdown
);
assert!(
(m.total_pips - (-0.13)).abs() < 1e-9,
"total_pips = {}",
m.total_pips
);
// manifest carries the sample's known configuration.
let (from, to) = r1.manifest.window;
assert_eq!((from.0, to.0), (1, 7));
// commit is the build's git identity (or the no-git "unknown" fallback);
// either way it is non-empty and fixed at compile time, so it is stable
// across runs of the same build (C1 determinism, already asserted above
// via `to_json()`).
assert!(!r1.manifest.commit.is_empty());
assert_eq!(r1.manifest.commit, r2.manifest.commit);
}
fn pair(name: &str, v: Scalar) -> (String, Scalar) {
(name.to_string(), v)
}
#[test]
fn member_key_renders_varying_axes_portably() {
let named = vec![
pair("ema.length", Scalar::i64(5)),
pair("exposure.scale", Scalar::f64(0.5)),
pair("longonly.enabled", Scalar::bool(true)),
];
let varying: std::collections::HashSet<String> =
named.iter().map(|(n, _)| n.clone()).collect();
let key = member_key(&named, &varying);
assert_eq!(key, "ema.length-5_exposure.scale-0.5_longonly.enabled-true");
assert!(key.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')));
}
#[test]
fn member_key_omits_pinned_axes() {
let named = vec![
pair("ema.length", Scalar::i64(5)),
pair("exposure.scale", Scalar::f64(0.5)),
pair("longonly.enabled", Scalar::bool(false)),
];
let mut varying = std::collections::HashSet::new();
varying.insert("longonly.enabled".to_string());
assert_eq!(member_key(&named, &varying), "longonly.enabled-false");
}
#[test]
fn member_key_handles_negative_float_and_sanitises_names() {
let named = vec![pair("exposure.scale", Scalar::f64(-0.5))];
let varying: std::collections::HashSet<String> =
["exposure.scale".to_string()].into_iter().collect();
assert_eq!(member_key(&named, &varying), "exposure.scale--0.5");
let named2 = vec![pair("weird key!", Scalar::f64(1.0))];
let varying2: std::collections::HashSet<String> =
["weird key!".to_string()].into_iter().collect();
assert_eq!(member_key(&named2, &varying2), "weird_key_-1");
}
#[test]
fn member_key_is_m_when_no_axis_varies() {
let named = vec![pair("exposure.scale", Scalar::f64(0.5))];
let varying = std::collections::HashSet::new();
assert_eq!(member_key(&named, &varying), "m");
}
#[test]
fn member_key_caps_length_with_conformant_hash_fallback() {
let named: Vec<(String, Scalar)> = (0..40)
.map(|i| pair(&format!("some.long.axis.path.number.{i}"), Scalar::i64(i)))
.collect();
let varying: std::collections::HashSet<String> =
named.iter().map(|(n, _)| n.clone()).collect();
let key = member_key(&named, &varying);
assert!(key.len() <= MAX_KEY, "over-cap key not bounded: {} bytes", key.len());
assert!(key.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')));
let mut named2 = named.clone();
named2[0].1 = Scalar::i64(999);
assert_ne!(key, member_key(&named2, &varying), "distinct over-cap inputs must differ");
}
#[test]
fn member_key_collision_free_over_a_non_trend_axis_set() {
// #105 regression: vary exposure.scale + a bool (NOT the old hardcoded
// trend.fast/slow). Distinct points -> distinct keys; the old f…s… key
// would have collapsed them all to one dir.
let varying: std::collections::HashSet<String> =
["exposure.scale".to_string(), "longonly.enabled".to_string()].into_iter().collect();
let p = |s: f64, b: bool| {
vec![
pair("ema.length", Scalar::i64(5)), // pinned -> omitted from key
pair("exposure.scale", Scalar::f64(s)),
pair("longonly.enabled", Scalar::bool(b)),
]
};
let keys: Vec<String> = [(0.5, true), (0.5, false), (1.0, true), (1.0, false)]
.iter()
.map(|&(s, b)| member_key(&p(s, b), &varying))
.collect();
let unique: std::collections::HashSet<&String> = keys.iter().collect();
assert_eq!(unique.len(), 4, "distinct points must yield distinct keys: {keys:?}");
}
#[test]
fn momentum_param_space_is_ema_exposure_longonly() {
// pins the default node-name path segments (ema / exposure / longonly) and
// the param order/kinds the member key + sweep depend on.
let names: Vec<String> =
momentum_blueprint_with_sinks(SYNTHETIC_PIP_SIZE).0.param_space().into_iter().map(|p| p.name).collect();
assert_eq!(
names,
vec![
"ema.length".to_string(),
"exposure.scale".to_string(),
"longonly.enabled".to_string(),
],
);
}
#[test]
fn momentum_sweep_is_deterministic_and_has_eight_points() {
let a = momentum_sweep_family(None, &DataSource::Synthetic);
let b = momentum_sweep_family(None, &DataSource::Synthetic);
assert_eq!(a.points.len(), 8, "2x2x2 grid = 8 points");
assert_eq!(a, b, "C1: the momentum family is a pure function of the build");
}
#[test]
fn parse_sweep_args_defaults_selects_and_rejects() {
assert_eq!(
parse_sweep_args(&[]),
Ok((Strategy::SmaCross, "sweep".to_string(), false, DataChoice::Synthetic))
);
assert_eq!(
parse_sweep_args(&["--trace", "swp"]),
Ok((Strategy::SmaCross, "swp".to_string(), true, DataChoice::Synthetic))
);
assert_eq!(
parse_sweep_args(&["--name", "s"]),
Ok((Strategy::SmaCross, "s".to_string(), false, DataChoice::Synthetic))
);
assert_eq!(
parse_sweep_args(&["--strategy", "momentum", "--trace", "mom"]),
Ok((Strategy::Momentum, "mom".to_string(), true, DataChoice::Synthetic))
);
assert!(parse_sweep_args(&["--strategy", "bogus"]).is_err());
assert!(parse_sweep_args(&["--name", "a", "--trace", "b"]).is_err()); // mutually exclusive
assert!(parse_sweep_args(&["--trace"]).is_err()); // flag missing its value
}
/// `parse_sweep_args` admits a real symbol with an optional `--from`/`--to`
/// window (yielding `DataChoice::Real`), still honouring `--strategy`/`--trace`;
/// a `--real` without its symbol, or a window flag without `--real`, is rejected.
#[test]
fn parse_sweep_args_accepts_real_symbol_and_window() {
assert_eq!(
parse_sweep_args(&["--real", "EURUSD", "--from", "100", "--to", "200", "--trace", "s"]),
Ok((Strategy::SmaCross, "s".to_string(), true,
DataChoice::Real { symbol: "EURUSD".to_string(), from_ms: Some(100), to_ms: Some(200) }))
);
assert!(parse_sweep_args(&["--real"]).is_err()); // --real needs a symbol
assert!(parse_sweep_args(&["--from", "100"]).is_err()); // --from without --real
}
/// `parse_sweep_args` mirrors `parse_real_args`'s real/window strictness: an
/// empty `--real` symbol and a repeated `--real`/`--from`/`--to` are usage
/// errors (not last-write-wins, not a downstream refusal) — so sibling commands
/// reject the same malformed real-grammar the same way.
#[test]
fn parse_sweep_args_rejects_empty_symbol_and_repeated_real_window() {
assert!(parse_sweep_args(&["--real", ""]).is_err()); // empty symbol
assert!(parse_sweep_args(&["--real", "EURUSD", "--real", "GBPUSD"]).is_err()); // repeated --real
assert!(parse_sweep_args(&["--real", "EURUSD", "--from", "1", "--from", "2"]).is_err()); // repeated --from
assert!(parse_sweep_args(&["--real", "EURUSD", "--to", "1", "--to", "2"]).is_err()); // repeated --to
}
/// `parse_walkforward_args` defaults to synthetic / name "walkforward" / no
/// persist, admits `--real <SYMBOL>` (window-less here) yielding `DataChoice::Real`,
/// and rejects two name flags or a `--real` missing its symbol.
#[test]
fn parse_walkforward_args_defaults_and_accepts_real() {
assert_eq!(parse_walkforward_args(&[]), Ok(("walkforward".to_string(), false, DataChoice::Synthetic)));
assert_eq!(
parse_walkforward_args(&["--real", "EURUSD", "--trace", "w"]),
Ok(("w".to_string(), true, DataChoice::Real { symbol: "EURUSD".to_string(), from_ms: None, to_ms: None }))
);
assert!(parse_walkforward_args(&["--name", "a", "--trace", "b"]).is_err());
assert!(parse_walkforward_args(&["--real"]).is_err());
}
/// `parse_walkforward_args` mirrors `parse_real_args`'s real/window strictness:
/// an empty `--real` symbol and a repeated `--real`/`--from`/`--to` are usage
/// errors — the same real-grammar rejection its sibling `parse_sweep_args` gives.
#[test]
fn parse_walkforward_args_rejects_empty_symbol_and_repeated_real_window() {
assert!(parse_walkforward_args(&["--real", ""]).is_err()); // empty symbol
assert!(parse_walkforward_args(&["--real", "EURUSD", "--real", "GBPUSD"]).is_err());
assert!(parse_walkforward_args(&["--real", "EURUSD", "--from", "1", "--from", "2"]).is_err());
assert!(parse_walkforward_args(&["--real", "EURUSD", "--to", "1", "--to", "2"]).is_err());
}
}