//! `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 [--from ] [--to ]` 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, ChartMode, 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, Registry, RunTraces, TraceStore, TraceStoreError, }; 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)>, Receiver<(Timestamp, Vec)>, ) { 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//` /// (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)], ex_rows: &[(Timestamp, Vec)], ) { 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//` prefix and `/.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 { let key: String = named .iter() .filter(|(n, _)| varying.contains(n)) .map(|(n, v)| format!("{}-{}", sanitize_component(n), sanitize_component(&render_value(v)))) .collect::>() .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())) } } /// 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` over xs. fn build_chart_data(traces: RunTraces) -> ChartData { let mut xs: Vec = traces.taps.iter().flat_map(|t| t.ts.iter().copied()).collect(); xs.sort_unstable(); xs.dedup(); let spine: Vec<(Timestamp, Vec)> = xs.iter().map(|&t| (Timestamp(t), Vec::new())).collect(); let tap_rows: Vec)>> = traces.taps.iter().map(|t| t.to_rows()).collect(); let sides: Vec<&[(Timestamp, Vec)]> = tap_rows.iter().map(|r| r.as_slice()).collect(); let joined: Vec = join_on_ts(&spine, &sides); let mut series: Vec = 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> = joined.iter().map(|r| r.sides[i].as_ref().map(|row| row[c].as_f64())).collect(); series.push(Series { name, y_scale_id, points }); } } ChartData { xs, series } } /// `aura chart [--panels]`: read the persisted run's traces, align them, and /// print the self-contained uPlot page. A missing run is a usage error (stderr + /// exit 2), per the CLI's convention — never a panic. fn emit_chart(name: &str, mode: ChartMode) { let traces = match TraceStore::open("runs").read(name) { Ok(t) => t, Err(TraceStoreError::NotFound(n)) => { eprintln!("aura: no recorded run '{n}' under runs/traces (run `aura run --trace {n}` first)"); std::process::exit(2); } Err(e) => { eprintln!("aura: {e}"); std::process::exit(2); } }; print!("{}", render::render_chart_html(&build_chart_data(traces), mode)); } /// 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 { let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE); let sources: Vec> = 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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = 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 `: 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`), 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, to_ms: Option, trace: Option<&str>) -> RunReport { // 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 = 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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = 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 `` then zero-or-more /// `--from ` / `--to ` 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, Option, Option), String> { let usage = || "run --real [--from ] [--to ] [--trace ]".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 = None; let mut to: Option = None; let mut trace: Option = 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)) } /// The shared `--real [--from ] [--to ]` 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, from_ms: Option, to_ms: Option, } 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 { 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 { 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, to_ms: Option }, } /// 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, symbol: String, from_ms: Option, to_ms: Option, 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, symbol: &str, from_ms: Option, to_ms: Option, ) -> (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> = 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> = 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> { 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> { 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)>, Receiver<(Timestamp, Vec)>, ) { 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 = 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::>(); let ex_rows = rx_ex.try_iter().collect::>(); 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)>, Receiver<(Timestamp, Vec)>, ) { 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///`. 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 = 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::>(); let ex_rows = rx_ex.try_iter().collect::>(); 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 ] [--real [--from ] [--to ]] [--name | --trace ]`. /// 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 ] [--real [--from ] [--to ]] [--name | --trace ]".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 [--from ] [--to ]] [--name | --trace ]`. /// 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 [--from ] [--to ]] [--name | --trace ]".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 ] [--name |--trace ]`: 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///` (opt-in). fn run_sweep(strategy: Strategy, name: &str, persist: bool, data: DataSource) { 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 |--trace ]`: 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//oos/` (opt-in). Deterministic (C1). fn run_walkforward(name: &str, persist: bool, data: DataSource) { 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::>(), 0); let exposure = f64_field(&rx_ex.try_iter().collect::>(), 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::>(); let ex_rows = rx_ex.try_iter().collect::>(); 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 = 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> = 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::>(); let ex_rows = 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, 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 |--trace ]`: 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//seed/` (opt-in). fn run_mc(name: &str, persist: bool) { 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 [rank ]`: 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 = 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)>, tx_ex: mpsc::Sender<(Timestamp, Vec)>, ) -> 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 { 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 { 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> = 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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = 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 ] | aura run --real [--from ] [--to ] [--trace ] | aura chart [--panels] | aura graph | aura sweep [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] | aura mc [--name |--trace ] | aura walkforward [--real [--from ] [--to ]] [--name |--trace ] | aura runs families | aura runs family [rank ]"; 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 = std::env::args().skip(1).collect(); match args.iter().map(String::as_str).collect::>().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", name] => emit_chart(name, ChartMode::Overlay), ["chart", name, "--panels"] => emit_chart(name, ChartMode::Panels), ["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::*; /// #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> = 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)>, exposure: Vec<(Timestamp, Vec)>, } /// 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)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = 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::>().is_empty(), "equity sink drained empty"); assert!(!rx_ex.try_iter().collect::>().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 = 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 ` 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`), /// 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()); } #[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 = 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 = ["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 = ["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 = 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 = ["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 = [(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 = 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 ` (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()); } }