feat(trace-charts): persist recorder taps to disk (iter 1)
Iteration 1 (persist half) of the visual face — spec 0056 / plan 0055:
tap data from a graph and save it to disk so it can later be charted
(the serve/chart half is iteration 2).
- ColumnarTrace (aura-engine): a drained tap <-> columnar SoA form
({tap, kinds, ts, columns}); values coerced to f64 for plotting with
the base type preserved in `kinds` (C7); to_rows rebuilds uniformly-f64
rows for the serve-side join. Pure (C1), beside join_on_ts/summarize.
- TraceStore (aura-registry): a per-run directory under
runs/traces/<name>/ — one <tap>.json per tap + an index.json (manifest
+ tap order), beside the families.jsonl sibling store. Missing run =
NotFound (no panic); corrupt file = Parse{file}.
- CLI: a shared persist_traces helper threaded into run_sample /
run_macd / run_sample_real via an orthogonal `trace: Option<&str>`;
new arms `aura run [--macd] --trace <name>` and `--trace <name>` on
`run --real`. The default (no --trace) run is byte-for-byte unchanged.
Verified by the orchestrator: `cargo test --workspace` green (incl. the
4 columnar_trace, 3 trace_store, 4 cli_run e2e, and the 4-tuple
parse_real_args tests); `cargo clippy --workspace --all-targets -D
warnings` exit 0. Implementer deviations folded in, all
compiler-prescribed: #[derive(Debug)] on RunTraces (the RED tests
panic-format a Result over it), #[allow(clippy::type_complexity)] on
the parse_real_args 4-tuple (mirrors the crate's sample_harness idiom).
Default-run byte-identity stays pinned by run_prints_json_and_exits_zero.
refs #101
This commit is contained in:
+106
-78
@@ -15,14 +15,14 @@ mod render;
|
|||||||
|
|
||||||
use aura_core::{zip_params, Cell, Firing, Scalar, ScalarKind, Timestamp};
|
use aura_core::{zip_params, Cell, Firing, Scalar, ScalarKind, Timestamp};
|
||||||
use aura_engine::{
|
use aura_engine::{
|
||||||
f64_field, monte_carlo, param_stability, summarize, walk_forward, window_of, Composite, Edge,
|
f64_field, monte_carlo, param_stability, summarize, walk_forward, window_of, ColumnarTrace,
|
||||||
FlatGraph, GraphBuilder, Harness, McAggregate, McFamily, RollMode, RunManifest, RunReport,
|
Composite, Edge, FlatGraph, GraphBuilder, Harness, McAggregate, McFamily, RollMode,
|
||||||
SourceSpec, SweepFamily, SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds,
|
RunManifest, RunReport, SourceSpec, SweepFamily, SyntheticSpec, Target, VecSource,
|
||||||
WindowRoller, WindowRun,
|
WalkForwardResult, WindowBounds, WindowRoller, WindowRun,
|
||||||
};
|
};
|
||||||
use aura_registry::{
|
use aura_registry::{
|
||||||
group_families, mc_member_reports, optimize, rank_by, sweep_member_reports,
|
group_families, mc_member_reports, optimize, rank_by, sweep_member_reports,
|
||||||
walkforward_member_reports, FamilyKind, Registry,
|
walkforward_member_reports, FamilyKind, Registry, TraceStore,
|
||||||
};
|
};
|
||||||
use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub};
|
use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub};
|
||||||
use std::sync::mpsc::{self, Receiver};
|
use std::sync::mpsc::{self, Receiver};
|
||||||
@@ -152,10 +152,31 @@ fn sim_optimal_manifest(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// 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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Run the sample harness and fold it into a `RunReport` (drain both sinks →
|
/// Run the sample harness and fold it into a `RunReport` (drain both sinks →
|
||||||
/// `f64_field` → `summarize` → pair with a `RunManifest`). Pure and deterministic
|
/// `f64_field` → `summarize` → pair with a `RunManifest`). Pure and deterministic
|
||||||
/// (C1): the same build yields the same report.
|
/// (C1): the same build yields the same report.
|
||||||
fn run_sample() -> RunReport {
|
fn run_sample(trace: Option<&str>) -> RunReport {
|
||||||
let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE);
|
let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE);
|
||||||
let sources: Vec<Box<dyn aura_engine::Source>> =
|
let sources: Vec<Box<dyn aura_engine::Source>> =
|
||||||
vec![Box::new(VecSource::new(synthetic_prices()))];
|
vec![Box::new(VecSource::new(synthetic_prices()))];
|
||||||
@@ -164,23 +185,21 @@ fn run_sample() -> RunReport {
|
|||||||
|
|
||||||
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
|
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 ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
|
||||||
let equity = f64_field(&eq_rows, 0);
|
let manifest = sim_optimal_manifest(
|
||||||
let exposure = f64_field(&ex_rows, 0);
|
vec![
|
||||||
let metrics = summarize(&equity, &exposure);
|
("sma_fast".to_string(), Scalar::i64(2)),
|
||||||
|
("sma_slow".to_string(), Scalar::i64(4)),
|
||||||
RunReport {
|
("exposure_scale".to_string(), Scalar::f64(0.5)),
|
||||||
manifest: sim_optimal_manifest(
|
],
|
||||||
vec![
|
window,
|
||||||
("sma_fast".to_string(), Scalar::i64(2)),
|
0,
|
||||||
("sma_slow".to_string(), Scalar::i64(4)),
|
SYNTHETIC_PIP_SIZE,
|
||||||
("exposure_scale".to_string(), Scalar::f64(0.5)),
|
);
|
||||||
],
|
if let Some(name) = trace {
|
||||||
window,
|
persist_traces(name, &manifest, &eq_rows, &ex_rows);
|
||||||
0,
|
|
||||||
SYNTHETIC_PIP_SIZE,
|
|
||||||
),
|
|
||||||
metrics,
|
|
||||||
}
|
}
|
||||||
|
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
|
/// `aura run --real <SYMBOL>`: run the built-in sample harness over real M1 close
|
||||||
@@ -190,7 +209,7 @@ fn run_sample() -> RunReport {
|
|||||||
/// (a Source is single-pass), so the run source streams the window untouched.
|
/// (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
|
/// 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.
|
/// 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>) -> RunReport {
|
fn run_sample_real(symbol: &str, from_ms: Option<i64>, to_ms: Option<i64>, trace: Option<&str>) -> RunReport {
|
||||||
// Per-instrument pip: look up BEFORE any data access, so an un-specced symbol
|
// Per-instrument pip: look up BEFORE any data access, so an un-specced symbol
|
||||||
// refuses without touching local data. Honest by construction.
|
// refuses without touching local data. Honest by construction.
|
||||||
let spec = match aura_ingest::instrument_spec(symbol) {
|
let spec = match aura_ingest::instrument_spec(symbol) {
|
||||||
@@ -237,23 +256,21 @@ fn run_sample_real(symbol: &str, from_ms: Option<i64>, to_ms: Option<i64>) -> Ru
|
|||||||
|
|
||||||
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
|
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 ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
|
||||||
let equity = f64_field(&eq_rows, 0);
|
let manifest = sim_optimal_manifest(
|
||||||
let exposure = f64_field(&ex_rows, 0);
|
vec![
|
||||||
let metrics = summarize(&equity, &exposure);
|
("sma_fast".to_string(), Scalar::i64(2)),
|
||||||
|
("sma_slow".to_string(), Scalar::i64(4)),
|
||||||
RunReport {
|
("exposure_scale".to_string(), Scalar::f64(0.5)),
|
||||||
manifest: sim_optimal_manifest(
|
],
|
||||||
vec![
|
window,
|
||||||
("sma_fast".to_string(), Scalar::i64(2)),
|
0,
|
||||||
("sma_slow".to_string(), Scalar::i64(4)),
|
spec.pip_size,
|
||||||
("exposure_scale".to_string(), Scalar::f64(0.5)),
|
);
|
||||||
],
|
if let Some(name) = trace {
|
||||||
window,
|
persist_traces(name, &manifest, &eq_rows, &ex_rows);
|
||||||
0,
|
|
||||||
spec.pip_size,
|
|
||||||
),
|
|
||||||
metrics,
|
|
||||||
}
|
}
|
||||||
|
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
|
/// Parse the `run --real` tail: a mandatory `<SYMBOL>` then zero-or-more
|
||||||
@@ -261,25 +278,29 @@ fn run_sample_real(symbol: &str, from_ms: Option<i64>, to_ms: Option<i64>) -> Ru
|
|||||||
/// arg grammar is unit-testable; `main` does the side effects. An unknown token, a
|
/// 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
|
/// flag without its value, a non-`i64` ms, or a repeated flag is an `Err` carrying
|
||||||
/// a short usage message.
|
/// a short usage message.
|
||||||
fn parse_real_args(rest: &[&str]) -> Result<(String, Option<i64>, Option<i64>), String> {
|
#[allow(clippy::type_complexity)]
|
||||||
let usage = || "run --real <SYMBOL> [--from <ms>] [--to <ms>]".to_string();
|
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() {
|
let (symbol, mut tail) = match rest.split_first() {
|
||||||
Some((sym, t)) if !sym.is_empty() => (*sym, t),
|
Some((sym, t)) if !sym.is_empty() => (*sym, t),
|
||||||
_ => return Err(usage()),
|
_ => return Err(usage()),
|
||||||
};
|
};
|
||||||
let mut from: Option<i64> = None;
|
let mut from: Option<i64> = None;
|
||||||
let mut to: Option<i64> = None;
|
let mut to: Option<i64> = None;
|
||||||
|
let mut trace: Option<String> = None;
|
||||||
while let Some((flag, t)) = tail.split_first() {
|
while let Some((flag, t)) = tail.split_first() {
|
||||||
let (value, t) = t.split_first().ok_or_else(usage)?;
|
let (value, t) = t.split_first().ok_or_else(usage)?;
|
||||||
let ms: i64 = value.parse().map_err(|_| usage())?;
|
|
||||||
match *flag {
|
match *flag {
|
||||||
"--from" if from.is_none() => from = Some(ms),
|
"--from" if from.is_none() => from = Some(value.parse().map_err(|_| usage())?),
|
||||||
"--to" if to.is_none() => to = Some(ms),
|
"--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()),
|
_ => return Err(usage()),
|
||||||
}
|
}
|
||||||
tail = t;
|
tail = t;
|
||||||
}
|
}
|
||||||
Ok((symbol.to_string(), from, to))
|
Ok((symbol.to_string(), from, to, trace))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The SMA-cross signal as a named composite (price -> fast/slow SMA -> spread).
|
/// The SMA-cross signal as a named composite (price -> fast/slow SMA -> spread).
|
||||||
@@ -820,7 +841,7 @@ fn macd_prices() -> Vec<(Timestamp, Scalar)> {
|
|||||||
/// (the same bootstrap path the SMA sample's compiled view uses), drive it on the
|
/// (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
|
/// synthetic stream, and fold both sinks into a `RunReport`. Pure and
|
||||||
/// deterministic (C1).
|
/// deterministic (C1).
|
||||||
fn run_macd() -> RunReport {
|
fn run_macd(trace: Option<&str>) -> RunReport {
|
||||||
let (tx_eq, rx_eq) = mpsc::channel();
|
let (tx_eq, rx_eq) = mpsc::channel();
|
||||||
let (tx_ex, rx_ex) = mpsc::channel();
|
let (tx_ex, rx_ex) = mpsc::channel();
|
||||||
let flat = macd_strategy_blueprint(tx_eq, tx_ex)
|
let flat = macd_strategy_blueprint(tx_eq, tx_ex)
|
||||||
@@ -835,28 +856,26 @@ fn run_macd() -> RunReport {
|
|||||||
|
|
||||||
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
|
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 ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
|
||||||
let equity = f64_field(&eq_rows, 0);
|
let manifest = sim_optimal_manifest(
|
||||||
let exposure = f64_field(&ex_rows, 0);
|
vec![
|
||||||
let metrics = summarize(&equity, &exposure);
|
("ema_fast".to_string(), Scalar::i64(2)),
|
||||||
|
("ema_slow".to_string(), Scalar::i64(4)),
|
||||||
RunReport {
|
("ema_signal".to_string(), Scalar::i64(3)),
|
||||||
manifest: sim_optimal_manifest(
|
("exposure_scale".to_string(), Scalar::f64(0.5)),
|
||||||
vec![
|
],
|
||||||
("ema_fast".to_string(), Scalar::i64(2)),
|
window,
|
||||||
("ema_slow".to_string(), Scalar::i64(4)),
|
0,
|
||||||
("ema_signal".to_string(), Scalar::i64(3)),
|
SYNTHETIC_PIP_SIZE,
|
||||||
("exposure_scale".to_string(), Scalar::f64(0.5)),
|
);
|
||||||
],
|
if let Some(name) = trace {
|
||||||
window,
|
persist_traces(name, &manifest, &eq_rows, &ex_rows);
|
||||||
0,
|
|
||||||
SYNTHETIC_PIP_SIZE,
|
|
||||||
),
|
|
||||||
metrics,
|
|
||||||
}
|
}
|
||||||
|
let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
|
||||||
|
RunReport { manifest, metrics }
|
||||||
}
|
}
|
||||||
|
|
||||||
const USAGE: &str =
|
const USAGE: &str =
|
||||||
"usage: aura run [--macd] | aura run --real <SYMBOL> [--from <ms>] [--to <ms>] | aura graph | aura sweep [--name <n>] | aura mc [--name <n>] | aura walkforward [--name <n>] | aura runs families | aura runs family <id> [rank <metric>]";
|
"usage: aura run [--macd] [--trace <name>] | aura run --real <SYMBOL> [--from <ms>] [--to <ms>] [--trace <name>] | aura graph | aura sweep [--name <n>] | aura mc [--name <n>] | aura walkforward [--name <n>] | aura runs families | aura runs family <id> [rank <metric>]";
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
// Collect argv and match the whole vector: every accepted form is exhaustive,
|
// Collect argv and match the whole vector: every accepted form is exhaustive,
|
||||||
@@ -864,10 +883,14 @@ fn main() {
|
|||||||
// than masquerading as a successful run (#16 strict reading).
|
// than masquerading as a successful run (#16 strict reading).
|
||||||
let args: Vec<String> = std::env::args().skip(1).collect();
|
let args: Vec<String> = std::env::args().skip(1).collect();
|
||||||
match args.iter().map(String::as_str).collect::<Vec<_>>().as_slice() {
|
match args.iter().map(String::as_str).collect::<Vec<_>>().as_slice() {
|
||||||
["run"] => println!("{}", run_sample().to_json()),
|
["run"] => println!("{}", run_sample(None).to_json()),
|
||||||
["run", "--macd"] => println!("{}", run_macd().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) {
|
["run", "--real", rest @ ..] => match parse_real_args(rest) {
|
||||||
Ok((sym, from, to)) => println!("{}", run_sample_real(&sym, from, to).to_json()),
|
Ok((sym, from, to, trace)) => {
|
||||||
|
println!("{}", run_sample_real(&sym, from, to, trace.as_deref()).to_json())
|
||||||
|
}
|
||||||
Err(msg) => {
|
Err(msg) => {
|
||||||
eprintln!("aura: {msg}");
|
eprintln!("aura: {msg}");
|
||||||
std::process::exit(2);
|
std::process::exit(2);
|
||||||
@@ -1096,8 +1119,8 @@ mod tests {
|
|||||||
// the MACD strategy authors a nested EMA-of-EMA composite, compiles it to a
|
// 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
|
// flat runnable harness (the call not panicking proves the compile+bootstrap
|
||||||
// path), and runs it. C1 determinism: two runs are bit-identical.
|
// path), and runs it. C1 determinism: two runs are bit-identical.
|
||||||
let r1 = run_macd();
|
let r1 = run_macd(None);
|
||||||
let r2 = run_macd();
|
let r2 = run_macd(None);
|
||||||
assert_eq!(r1.metrics, r2.metrics);
|
assert_eq!(r1.metrics, r2.metrics);
|
||||||
assert_eq!(r1.to_json(), r2.to_json());
|
assert_eq!(r1.to_json(), r2.to_json());
|
||||||
|
|
||||||
@@ -1177,8 +1200,8 @@ mod tests {
|
|||||||
|
|
||||||
// The headline: a real-data run over the bounded Sept-2024 window yields
|
// The headline: a real-data run over the bounded Sept-2024 window yields
|
||||||
// a finite, C1-deterministic RunReport.
|
// a finite, C1-deterministic RunReport.
|
||||||
let r1 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS));
|
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));
|
let r2 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None);
|
||||||
|
|
||||||
assert!(
|
assert!(
|
||||||
r1.metrics.total_pips.is_finite(),
|
r1.metrics.total_pips.is_finite(),
|
||||||
@@ -1212,12 +1235,12 @@ mod tests {
|
|||||||
// Bounded to the vetted Sept-2024 window so the pip-label + determinism
|
// Bounded to the vetted Sept-2024 window so the pip-label + determinism
|
||||||
// checks run fast (not the full unbounded archive).
|
// checks run fast (not the full unbounded archive).
|
||||||
let report =
|
let report =
|
||||||
run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS));
|
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.
|
// 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)");
|
assert_eq!(report.manifest.broker, "sim-optimal(pip_size=1)");
|
||||||
// Deterministic (C1): a second run yields the same report.
|
// Deterministic (C1): a second run yields the same report.
|
||||||
let again =
|
let again =
|
||||||
run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS));
|
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.manifest.broker, again.manifest.broker);
|
||||||
assert_eq!(report.metrics.total_pips, again.metrics.total_pips);
|
assert_eq!(report.metrics.total_pips, again.metrics.total_pips);
|
||||||
}
|
}
|
||||||
@@ -1227,15 +1250,20 @@ mod tests {
|
|||||||
/// value — the pure arg grammar `main` relies on for `run --real`.
|
/// value — the pure arg grammar `main` relies on for `run --real`.
|
||||||
#[test]
|
#[test]
|
||||||
fn parse_real_args_accepts_symbol_and_optional_window() {
|
fn parse_real_args_accepts_symbol_and_optional_window() {
|
||||||
assert_eq!(parse_real_args(&["EURUSD"]), Ok(("EURUSD".to_string(), None, None)));
|
assert_eq!(parse_real_args(&["EURUSD"]), Ok(("EURUSD".to_string(), None, None, None)));
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
parse_real_args(&["EURUSD", "--from", "100", "--to", "200"]),
|
parse_real_args(&["EURUSD", "--from", "100", "--to", "200"]),
|
||||||
Ok(("EURUSD".to_string(), Some(100), Some(200)))
|
Ok(("EURUSD".to_string(), Some(100), Some(200), None))
|
||||||
);
|
);
|
||||||
// flags in any order
|
// flags in any order
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
parse_real_args(&["EURUSD", "--to", "200", "--from", "100"]),
|
parse_real_args(&["EURUSD", "--to", "200", "--from", "100"]),
|
||||||
Ok(("EURUSD".to_string(), Some(100), Some(200)))
|
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
|
// 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(&["EURUSD", "--from"]).is_err());
|
||||||
@@ -1245,8 +1273,8 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn run_sample_is_deterministic_and_non_trivial() {
|
fn run_sample_is_deterministic_and_non_trivial() {
|
||||||
let r1 = run_sample();
|
let r1 = run_sample(None);
|
||||||
let r2 = run_sample();
|
let r2 = run_sample(None);
|
||||||
// C1 determinism: two runs are bit-identical (metrics + rendered JSON).
|
// C1 determinism: two runs are bit-identical (metrics + rendered JSON).
|
||||||
assert_eq!(r1.metrics, r2.metrics);
|
assert_eq!(r1.metrics, r2.metrics);
|
||||||
assert_eq!(r1.to_json(), r2.to_json());
|
assert_eq!(r1.to_json(), r2.to_json());
|
||||||
|
|||||||
@@ -248,6 +248,171 @@ fn run_real_vetted_index_pip_reaches_the_emitted_manifest() {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn run_trace_persists_taps_and_plain_run_writes_no_traces() {
|
||||||
|
let cwd = temp_cwd("run-trace");
|
||||||
|
|
||||||
|
// plain `aura run` (no --trace) persists nothing to disk.
|
||||||
|
let plain = Command::new(BIN).arg("run").current_dir(&cwd).output().expect("spawn aura run");
|
||||||
|
assert!(plain.status.success(), "plain run exit: {:?}", plain.status);
|
||||||
|
assert!(!cwd.join("runs/traces").exists(), "plain run must write no trace files");
|
||||||
|
|
||||||
|
// `aura run --trace demo` persists the two taps + index, and still prints the report.
|
||||||
|
let traced = Command::new(BIN)
|
||||||
|
.args(["run", "--trace", "demo"])
|
||||||
|
.current_dir(&cwd)
|
||||||
|
.output()
|
||||||
|
.expect("spawn aura run --trace");
|
||||||
|
assert!(traced.status.success(), "traced run exit: {:?}", traced.status);
|
||||||
|
let stdout = String::from_utf8(traced.stdout).expect("utf-8 stdout");
|
||||||
|
assert!(stdout.trim_end().starts_with("{\"manifest\":{"), "report still printed: {stdout:?}");
|
||||||
|
assert!(cwd.join("runs/traces/demo/index.json").exists(), "index.json written");
|
||||||
|
assert!(cwd.join("runs/traces/demo/equity.json").exists(), "equity tap written");
|
||||||
|
assert!(cwd.join("runs/traces/demo/exposure.json").exists(), "exposure tap written");
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(&cwd);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Property: a persisted `<tap>.json` is the **columnar SoA form** a downstream
|
||||||
|
/// chart can consume, not an opaque blob — it carries the four
|
||||||
|
/// `tap`/`kinds`/`ts`/`columns` keys, the equity tap is kind-tagged `["F64"]`
|
||||||
|
/// (C7: the base type survives on disk), and the SoA invariant holds: `ts` has one
|
||||||
|
/// entry per recorded cycle and EVERY column has exactly that many entries (parallel
|
||||||
|
/// arrays, one timestamp per row). This is acceptance criterion 2 at the built-binary
|
||||||
|
/// file boundary — the unit round-trip pins the in-memory transpose, but only an
|
||||||
|
/// end-to-end run pins that the bytes actually written to disk hold that shape.
|
||||||
|
/// Asserts on structure + array-length parity, never on the exact float values
|
||||||
|
/// (pip equity's low bits vary build-to-build), so it stays deterministic.
|
||||||
|
#[test]
|
||||||
|
fn run_trace_persists_columnar_soa_shape_on_disk() {
|
||||||
|
let cwd = temp_cwd("run-trace-shape");
|
||||||
|
let out = Command::new(BIN)
|
||||||
|
.args(["run", "--trace", "demo"])
|
||||||
|
.current_dir(&cwd)
|
||||||
|
.output()
|
||||||
|
.expect("spawn aura run --trace");
|
||||||
|
assert!(out.status.success(), "traced run exit: {:?}", out.status);
|
||||||
|
|
||||||
|
let equity = std::fs::read_to_string(cwd.join("runs/traces/demo/equity.json"))
|
||||||
|
.expect("read equity.json");
|
||||||
|
// The four SoA keys are present, equity is f64-kind-tagged, and the parallel
|
||||||
|
// arrays open as expected (the synthetic sample harness has a single column).
|
||||||
|
assert!(equity.contains("\"tap\":\"equity\""), "tap name missing: {equity}");
|
||||||
|
assert!(equity.contains("\"kinds\":[\"F64\"]"), "F64 kind tag missing: {equity}");
|
||||||
|
assert!(equity.contains("\"ts\":["), "ts axis missing: {equity}");
|
||||||
|
assert!(equity.contains("\"columns\":[["), "columns array missing: {equity}");
|
||||||
|
|
||||||
|
// SoA invariant: one timestamp per row, and every column has exactly that many
|
||||||
|
// entries. Count the ts entries (the synthetic window is 7 cycles) and the
|
||||||
|
// single column's entries by their comma-separated cardinality.
|
||||||
|
let ts_body = json_array_body(&equity, "\"ts\":[");
|
||||||
|
let col_body = json_array_body(&equity, "\"columns\":[[");
|
||||||
|
let ts_len = ts_body.split(',').count();
|
||||||
|
let col_len = col_body.split(',').count();
|
||||||
|
assert_eq!(ts_len, 7, "synthetic run records 7 cycles; ts was: {ts_body:?}");
|
||||||
|
assert_eq!(
|
||||||
|
col_len, ts_len,
|
||||||
|
"SoA parity broken: ts has {ts_len} entries but the column has {col_len}; file: {equity}"
|
||||||
|
);
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(&cwd);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Property: `index.json` is the run's **table of contents** — it carries the run's
|
||||||
|
/// `RunManifest` (so a chart page can show run context: commit, broker, window) and
|
||||||
|
/// the tap names in a **deterministic order** (`["equity","exposure"]`). The serve
|
||||||
|
/// half (`aura chart`, iteration 2) reads taps in this index order; a regression that
|
||||||
|
/// dropped the manifest, or made the tap list order-unstable, would silently corrupt
|
||||||
|
/// the chart header / the per-series mapping. Asserts on the manifest's structural
|
||||||
|
/// presence (the commit value is the volatile git HEAD, pinned elsewhere) and the
|
||||||
|
/// exact deterministic tap order.
|
||||||
|
#[test]
|
||||||
|
fn run_trace_index_carries_manifest_and_ordered_tap_list() {
|
||||||
|
let cwd = temp_cwd("run-trace-index");
|
||||||
|
let out = Command::new(BIN)
|
||||||
|
.args(["run", "--trace", "demo"])
|
||||||
|
.current_dir(&cwd)
|
||||||
|
.output()
|
||||||
|
.expect("spawn aura run --trace");
|
||||||
|
assert!(out.status.success(), "traced run exit: {:?}", out.status);
|
||||||
|
|
||||||
|
let index = std::fs::read_to_string(cwd.join("runs/traces/demo/index.json"))
|
||||||
|
.expect("read index.json");
|
||||||
|
// The manifest is embedded with its identifying context for the chart header.
|
||||||
|
assert!(index.contains("\"manifest\":{\"commit\":\""), "manifest missing: {index}");
|
||||||
|
assert!(index.contains("\"window\":[1,7]"), "window missing from manifest: {index}");
|
||||||
|
assert!(
|
||||||
|
index.contains("\"broker\":\"sim-optimal(pip_size=0.0001)\""),
|
||||||
|
"broker label missing: {index}"
|
||||||
|
);
|
||||||
|
// The tap list is the deterministic read order the serve path depends on.
|
||||||
|
assert!(
|
||||||
|
index.contains("\"taps\":[\"equity\",\"exposure\"]"),
|
||||||
|
"tap order must be deterministic [equity, exposure]: {index}"
|
||||||
|
);
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(&cwd);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Property: `--trace` is an **orthogonal modifier** — it composes with `aura run
|
||||||
|
/// --real <SYM>`, not only plain `run`, and is strictly **additive**: the looked-up
|
||||||
|
/// instrument pip still reaches stdout byte-for-byte while the same two taps land on
|
||||||
|
/// disk. The spec's headline is "one shared persist helper, every run form"; the
|
||||||
|
/// existing E2E only exercised the synthetic `run` path. Gated on local GER40 data
|
||||||
|
/// (skip cleanly when absent, the project convention) so it never fails on a
|
||||||
|
/// data-less machine; the no-data path is the distinct "no local data" refusal.
|
||||||
|
#[test]
|
||||||
|
fn run_real_with_trace_persists_taps_and_keeps_stdout_unchanged() {
|
||||||
|
const FROM_MS: &str = "1725148800000";
|
||||||
|
const TO_MS: &str = "1727740799999";
|
||||||
|
let cwd = temp_cwd("run-real-trace");
|
||||||
|
let out = Command::new(BIN)
|
||||||
|
.args(["run", "--real", "GER40", "--from", FROM_MS, "--to", TO_MS, "--trace", "ger"])
|
||||||
|
.current_dir(&cwd)
|
||||||
|
.output()
|
||||||
|
.expect("spawn aura run --real --trace");
|
||||||
|
|
||||||
|
// Skip on a data-less machine: vetted GER40 with no local data takes the
|
||||||
|
// distinct no-data path (exit 2), never the pip-refusal, and writes no traces.
|
||||||
|
if out.status.code() == Some(2) {
|
||||||
|
let stderr = String::from_utf8_lossy(&out.stderr);
|
||||||
|
assert!(
|
||||||
|
stderr.contains("no local data for symbol 'GER40'"),
|
||||||
|
"exit 2 for vetted GER40 --trace must be the no-data path, got: {stderr}"
|
||||||
|
);
|
||||||
|
eprintln!("skip: no local GER40 data");
|
||||||
|
let _ = std::fs::remove_dir_all(&cwd);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
assert_eq!(out.status.code(), Some(0), "real --trace exit: {:?}", out.status);
|
||||||
|
// Additive: the index-pip report still reaches stdout unchanged.
|
||||||
|
let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
|
||||||
|
let line = stdout.trim_end();
|
||||||
|
assert!(line.starts_with("{\"manifest\":{"), "report still printed: {line}");
|
||||||
|
assert!(
|
||||||
|
line.contains("\"broker\":\"sim-optimal(pip_size=1)\""),
|
||||||
|
"real --trace must keep the GER40 index pip on stdout: {line}"
|
||||||
|
);
|
||||||
|
// And the same two taps landed under the named run dir.
|
||||||
|
assert!(cwd.join("runs/traces/ger/index.json").exists(), "index.json written");
|
||||||
|
assert!(cwd.join("runs/traces/ger/equity.json").exists(), "equity tap written");
|
||||||
|
assert!(cwd.join("runs/traces/ger/exposure.json").exists(), "exposure tap written");
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(&cwd);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Extract the body of a JSON array opened by `marker` (e.g. `"ts":[`) up to its
|
||||||
|
/// closing `]`, by substring — no serde dependency, mirroring the sibling tests'
|
||||||
|
/// substring parsing. Used only to count comma-separated entries for the SoA
|
||||||
|
/// length-parity check.
|
||||||
|
fn json_array_body<'a>(json: &'a str, marker: &str) -> &'a str {
|
||||||
|
let start = json.find(marker).expect("array marker present") + marker.len();
|
||||||
|
let rest = &json[start..];
|
||||||
|
let end = rest.find(']').expect("array is closed");
|
||||||
|
&rest[..end]
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn no_args_prints_usage_and_exits_two() {
|
fn no_args_prints_usage_and_exits_two() {
|
||||||
let out = Command::new(BIN).output().expect("spawn aura");
|
let out = Command::new(BIN).output().expect("spawn aura");
|
||||||
|
|||||||
@@ -61,7 +61,9 @@ pub use harness::{
|
|||||||
window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SyntheticSpec, Target,
|
window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SyntheticSpec, Target,
|
||||||
VecSource,
|
VecSource,
|
||||||
};
|
};
|
||||||
pub use report::{f64_field, join_on_ts, summarize, JoinedRow, RunManifest, RunMetrics, RunReport};
|
pub use report::{
|
||||||
|
f64_field, join_on_ts, summarize, ColumnarTrace, JoinedRow, RunManifest, RunMetrics, RunReport,
|
||||||
|
};
|
||||||
pub use sweep::{
|
pub use sweep::{
|
||||||
sweep, GridSpace, ParamRange, RandomSpace, Space, SweepError, SweepFamily, SweepPoint,
|
sweep, GridSpace, ParamRange, RandomSpace, Space, SweepError, SweepFamily, SweepPoint,
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -186,6 +186,95 @@ pub fn join_on_ts(
|
|||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// One drained recorder tap in columnar (SoA) form: parallel arrays, one per
|
||||||
|
/// recorded column, plus the shared recorded-timestamp axis. Chart-ready (a column
|
||||||
|
/// is a series of numbers) and kind-tagged (C7) — values are coerced to f64 for
|
||||||
|
/// plotting; the base type survives in `kinds`. Pure: identical rows always encode
|
||||||
|
/// to the same value (C1).
|
||||||
|
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
|
||||||
|
pub struct ColumnarTrace {
|
||||||
|
pub tap: String,
|
||||||
|
pub kinds: Vec<String>,
|
||||||
|
pub ts: Vec<i64>,
|
||||||
|
pub columns: Vec<Vec<f64>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ColumnarTrace {
|
||||||
|
/// Transpose a drained tap's `(ts, row)` pairs into columns. An empty `rows`
|
||||||
|
/// yields empty `ts` and one empty column per kind. Each cell is coerced to f64:
|
||||||
|
/// f64 as-is, i64 as f64, bool 1.0/0.0, timestamp epoch as f64. Panics if any
|
||||||
|
/// row's width disagrees with `kinds.len()` — a wiring bug (a sink's column
|
||||||
|
/// count is fixed at bootstrap), surfaced as a named panic like
|
||||||
|
/// [`f64_field`] rather than a bare index-out-of-bounds (wider) or silent
|
||||||
|
/// ragged columns (narrower).
|
||||||
|
pub fn from_rows(tap: &str, kinds: &[ScalarKind], rows: &[(Timestamp, Vec<Scalar>)]) -> Self {
|
||||||
|
let ts: Vec<i64> = rows.iter().map(|(t, _)| t.0).collect();
|
||||||
|
let mut columns: Vec<Vec<f64>> = vec![Vec::with_capacity(rows.len()); kinds.len()];
|
||||||
|
for (_, row) in rows {
|
||||||
|
if row.len() != kinds.len() {
|
||||||
|
panic!(
|
||||||
|
"from_rows: row width {} disagrees with kinds.len() {} (tap {tap:?})",
|
||||||
|
row.len(),
|
||||||
|
kinds.len(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
for (c, scalar) in row.iter().enumerate() {
|
||||||
|
columns[c].push(scalar_to_f64(*scalar));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ColumnarTrace {
|
||||||
|
tap: tap.to_string(),
|
||||||
|
kinds: kinds.iter().map(kind_tag).collect(),
|
||||||
|
ts,
|
||||||
|
columns,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Inverse for the serve/align path: rebuild `(ts, row)` pairs with each cell as
|
||||||
|
/// `Scalar::f64(columns[c][r])` — uniformly f64 (the on-disk store is f64
|
||||||
|
/// columns; the base type survives only in `kinds`). A true value round-trip for
|
||||||
|
/// f64 taps; keeps the serve-side `as_f64` projection total.
|
||||||
|
pub fn to_rows(&self) -> Vec<(Timestamp, Vec<Scalar>)> {
|
||||||
|
self.ts
|
||||||
|
.iter()
|
||||||
|
.enumerate()
|
||||||
|
.map(|(r, &t)| {
|
||||||
|
let row = self.columns.iter().map(|col| Scalar::f64(col[r])).collect();
|
||||||
|
(Timestamp(t), row)
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The four-kind tag string for a column (the on-disk `kinds` form; `ScalarKind`
|
||||||
|
/// derives no serde, so tags are plain strings).
|
||||||
|
fn kind_tag(kind: &ScalarKind) -> String {
|
||||||
|
match kind {
|
||||||
|
ScalarKind::F64 => "F64",
|
||||||
|
ScalarKind::I64 => "I64",
|
||||||
|
ScalarKind::Bool => "Bool",
|
||||||
|
ScalarKind::Timestamp => "Timestamp",
|
||||||
|
}
|
||||||
|
.to_string()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Coerce a recorded scalar to the f64 a chart plots: f64 as-is, i64 as f64,
|
||||||
|
/// bool 1.0/0.0, timestamp epoch as f64.
|
||||||
|
fn scalar_to_f64(s: Scalar) -> f64 {
|
||||||
|
match s.kind() {
|
||||||
|
ScalarKind::F64 => s.as_f64(),
|
||||||
|
ScalarKind::I64 => s.as_i64() as f64,
|
||||||
|
ScalarKind::Bool => {
|
||||||
|
if s.as_bool() {
|
||||||
|
1.0
|
||||||
|
} else {
|
||||||
|
0.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ScalarKind::Timestamp => s.as_ts().0 as f64,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
@@ -510,4 +599,75 @@ mod tests {
|
|||||||
assert_eq!(joined[0].ts, Timestamp(10));
|
assert_eq!(joined[0].ts, Timestamp(10));
|
||||||
assert_eq!(joined[0].sides[0], Some(vec![Scalar::i64(7)]));
|
assert_eq!(joined[0].sides[0], Some(vec![Scalar::i64(7)]));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn columnar_trace_round_trips_f64_rows() {
|
||||||
|
let rows = vec![
|
||||||
|
(Timestamp(2), vec![Scalar::f64(10.0)]),
|
||||||
|
(Timestamp(3), vec![Scalar::f64(20.0)]),
|
||||||
|
(Timestamp(4), vec![Scalar::f64(30.0)]),
|
||||||
|
];
|
||||||
|
let ct = ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &rows);
|
||||||
|
assert_eq!(ct.tap, "equity");
|
||||||
|
assert_eq!(ct.kinds, vec!["F64".to_string()]);
|
||||||
|
assert_eq!(ct.ts, vec![2, 3, 4]);
|
||||||
|
assert_eq!(ct.columns, vec![vec![10.0, 20.0, 30.0]]);
|
||||||
|
// to_rows is the inverse for f64 taps
|
||||||
|
assert_eq!(ct.to_rows(), rows);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn columnar_trace_empty_rows_yields_empty_columns_sized_to_kinds() {
|
||||||
|
let ct = ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], &[]);
|
||||||
|
assert_eq!(ct.ts, Vec::<i64>::new());
|
||||||
|
assert_eq!(ct.columns, vec![Vec::<f64>::new()]);
|
||||||
|
assert!(ct.to_rows().is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn columnar_trace_coerces_non_f64_kinds_to_f64() {
|
||||||
|
let rows = vec![
|
||||||
|
(Timestamp(1), vec![Scalar::i64(7), Scalar::bool(true)]),
|
||||||
|
(Timestamp(2), vec![Scalar::i64(9), Scalar::bool(false)]),
|
||||||
|
];
|
||||||
|
let ct = ColumnarTrace::from_rows("mix", &[ScalarKind::I64, ScalarKind::Bool], &rows);
|
||||||
|
assert_eq!(ct.kinds, vec!["I64".to_string(), "Bool".to_string()]);
|
||||||
|
assert_eq!(ct.columns, vec![vec![7.0, 9.0], vec![1.0, 0.0]]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The fourth coercion arm: a Timestamp-kind column tags as "Timestamp" and
|
||||||
|
/// its epoch-ns survives the f64 projection (`scalar_to_f64` reads `as_ts().0`),
|
||||||
|
/// closing the kind for which `from_rows` would otherwise be untested.
|
||||||
|
#[test]
|
||||||
|
fn columnar_trace_coerces_timestamp_kind_to_epoch_f64() {
|
||||||
|
let rows = vec![
|
||||||
|
(Timestamp(1), vec![Scalar::ts(Timestamp(1_700_000_000_000_000_000))]),
|
||||||
|
(Timestamp(2), vec![Scalar::ts(Timestamp(1_700_000_000_000_000_060))]),
|
||||||
|
];
|
||||||
|
let ct = ColumnarTrace::from_rows("clock", &[ScalarKind::Timestamp], &rows);
|
||||||
|
assert_eq!(ct.kinds, vec!["Timestamp".to_string()]);
|
||||||
|
assert_eq!(
|
||||||
|
ct.columns,
|
||||||
|
vec![vec![1_700_000_000_000_000_000.0, 1_700_000_000_000_000_060.0]],
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A row wider than the declared kinds is a wiring bug (a sink's column count
|
||||||
|
/// is fixed at bootstrap), surfaced as a named panic like `f64_field` — not a
|
||||||
|
/// bare index-out-of-bounds.
|
||||||
|
#[test]
|
||||||
|
#[should_panic(expected = "row width 2 disagrees with kinds.len() 1")]
|
||||||
|
fn from_rows_panics_on_row_wider_than_kinds() {
|
||||||
|
let rows = vec![(Timestamp(1), vec![Scalar::f64(1.0), Scalar::f64(2.0)])];
|
||||||
|
let _ = ColumnarTrace::from_rows("wide", &[ScalarKind::F64], &rows);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Symmetric to the wide case: a row narrower than the declared kinds is the
|
||||||
|
/// same wiring-bug class and carries the same named panic.
|
||||||
|
#[test]
|
||||||
|
#[should_panic(expected = "row width 1 disagrees with kinds.len() 2")]
|
||||||
|
fn from_rows_panics_on_row_narrower_than_kinds() {
|
||||||
|
let rows = vec![(Timestamp(1), vec![Scalar::f64(1.0)])];
|
||||||
|
let _ = ColumnarTrace::from_rows("narrow", &[ScalarKind::F64, ScalarKind::F64], &rows);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -27,6 +27,9 @@ pub use lineage::{
|
|||||||
FamilyKind, FamilyRunRecord,
|
FamilyKind, FamilyRunRecord,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
mod trace_store;
|
||||||
|
pub use trace_store::{RunTraces, TraceStore, TraceStoreError};
|
||||||
|
|
||||||
/// An append-only run registry over a JSONL file: one serde_json line per
|
/// An append-only run registry over a JSONL file: one serde_json line per
|
||||||
/// `RunReport`.
|
/// `RunReport`.
|
||||||
pub struct Registry {
|
pub struct Registry {
|
||||||
|
|||||||
@@ -0,0 +1,210 @@
|
|||||||
|
//! The per-run trace store: a directory of columnar tap files under
|
||||||
|
//! `<runs_dir>/traces/<name>/`, beside the run registry's `runs.jsonl`. One
|
||||||
|
//! subdirectory per run name; one `<tap>.json` (a `ColumnarTrace`) per tap; plus
|
||||||
|
//! `index.json` carrying the run's `RunManifest` and the tap order. Mirrors the
|
||||||
|
//! registry's directory-co-located sibling-store discipline; a missing run reads as
|
||||||
|
//! `NotFound` (treat-as-absent), never a panic.
|
||||||
|
|
||||||
|
use std::fmt;
|
||||||
|
use std::fs;
|
||||||
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
|
use aura_engine::{ColumnarTrace, RunManifest};
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
/// A per-run trace directory rooted at `<runs_dir>/traces`. One subdir per run name.
|
||||||
|
pub struct TraceStore {
|
||||||
|
dir: PathBuf,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A run's read-back traces: its manifest plus its taps in index (write) order.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub struct RunTraces {
|
||||||
|
pub manifest: RunManifest,
|
||||||
|
pub taps: Vec<ColumnarTrace>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `index.json`: the run manifest + the tap names in write order (= read order).
|
||||||
|
#[derive(Serialize, Deserialize)]
|
||||||
|
struct Index {
|
||||||
|
manifest: RunManifest,
|
||||||
|
taps: Vec<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TraceStore {
|
||||||
|
/// Bind to `<runs_dir>/traces`. No I/O — directories are created on write.
|
||||||
|
pub fn open(runs_dir: impl AsRef<Path>) -> TraceStore {
|
||||||
|
TraceStore { dir: runs_dir.as_ref().join("traces") }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Write one run's taps + index under `traces/<name>/`, creating directories as
|
||||||
|
/// needed. Each tap lands in `<tap>.json`; the manifest + tap order in
|
||||||
|
/// `index.json`.
|
||||||
|
pub fn write(
|
||||||
|
&self,
|
||||||
|
name: &str,
|
||||||
|
manifest: &RunManifest,
|
||||||
|
taps: &[ColumnarTrace],
|
||||||
|
) -> Result<(), TraceStoreError> {
|
||||||
|
let run_dir = self.dir.join(name);
|
||||||
|
fs::create_dir_all(&run_dir)?;
|
||||||
|
for tap in taps {
|
||||||
|
let path = run_dir.join(format!("{}.json", tap.tap));
|
||||||
|
let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse {
|
||||||
|
file: path.display().to_string(),
|
||||||
|
source,
|
||||||
|
})?;
|
||||||
|
fs::write(&path, body)?;
|
||||||
|
}
|
||||||
|
let index = Index {
|
||||||
|
manifest: manifest.clone(),
|
||||||
|
taps: taps.iter().map(|t| t.tap.clone()).collect(),
|
||||||
|
};
|
||||||
|
let index_path = run_dir.join("index.json");
|
||||||
|
let body = serde_json::to_string(&index).map_err(|source| TraceStoreError::Parse {
|
||||||
|
file: index_path.display().to_string(),
|
||||||
|
source,
|
||||||
|
})?;
|
||||||
|
fs::write(&index_path, body)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read a run's index + taps back, in index order. A missing run directory (no
|
||||||
|
/// `index.json`) is `NotFound`, not a panic; a malformed file is `Parse{file,..}`.
|
||||||
|
pub fn read(&self, name: &str) -> Result<RunTraces, TraceStoreError> {
|
||||||
|
let run_dir = self.dir.join(name);
|
||||||
|
let index_path = run_dir.join("index.json");
|
||||||
|
let index_text = match fs::read_to_string(&index_path) {
|
||||||
|
Ok(t) => t,
|
||||||
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
|
||||||
|
return Err(TraceStoreError::NotFound(name.to_string()))
|
||||||
|
}
|
||||||
|
Err(e) => return Err(TraceStoreError::Io(e)),
|
||||||
|
};
|
||||||
|
let index: Index = serde_json::from_str(&index_text).map_err(|source| {
|
||||||
|
TraceStoreError::Parse { file: index_path.display().to_string(), source }
|
||||||
|
})?;
|
||||||
|
let mut taps = Vec::with_capacity(index.taps.len());
|
||||||
|
for tap_name in &index.taps {
|
||||||
|
let path = run_dir.join(format!("{tap_name}.json"));
|
||||||
|
let text = fs::read_to_string(&path)?;
|
||||||
|
let tap: ColumnarTrace = serde_json::from_str(&text).map_err(|source| {
|
||||||
|
TraceStoreError::Parse { file: path.display().to_string(), source }
|
||||||
|
})?;
|
||||||
|
taps.push(tap);
|
||||||
|
}
|
||||||
|
Ok(RunTraces { manifest: index.manifest, taps })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// What can go wrong reading or writing the trace store.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum TraceStoreError {
|
||||||
|
/// An I/O error reading or writing a trace file.
|
||||||
|
Io(std::io::Error),
|
||||||
|
/// A trace/index file did not parse (the offending file named).
|
||||||
|
Parse { file: String, source: serde_json::Error },
|
||||||
|
/// No recorded run of this name (no `index.json` under `traces/<name>/`).
|
||||||
|
NotFound(String),
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Display for TraceStoreError {
|
||||||
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
|
match self {
|
||||||
|
TraceStoreError::Io(e) => write!(f, "trace store i/o: {e}"),
|
||||||
|
TraceStoreError::Parse { file, source } => {
|
||||||
|
write!(f, "trace store parse error in {file}: {source}")
|
||||||
|
}
|
||||||
|
TraceStoreError::NotFound(name) => {
|
||||||
|
write!(f, "no recorded run '{name}' under runs/traces")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::error::Error for TraceStoreError {}
|
||||||
|
|
||||||
|
impl From<std::io::Error> for TraceStoreError {
|
||||||
|
fn from(e: std::io::Error) -> Self {
|
||||||
|
TraceStoreError::Io(e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use aura_core::{Scalar, ScalarKind, Timestamp};
|
||||||
|
|
||||||
|
fn temp_traces_root(name: &str) -> PathBuf {
|
||||||
|
let dir =
|
||||||
|
std::env::temp_dir().join(format!("aura-tracestore-{}-{}", std::process::id(), name));
|
||||||
|
let _ = fs::remove_dir_all(&dir);
|
||||||
|
fs::create_dir_all(&dir).expect("create temp traces root");
|
||||||
|
dir
|
||||||
|
}
|
||||||
|
|
||||||
|
fn sample_manifest() -> RunManifest {
|
||||||
|
RunManifest {
|
||||||
|
commit: "c".to_string(),
|
||||||
|
params: vec![("p".to_string(), Scalar::f64(1.0))],
|
||||||
|
window: (Timestamp(1), Timestamp(5)),
|
||||||
|
seed: 0,
|
||||||
|
broker: "sim-optimal(pip_size=0.0001)".to_string(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn sample_taps() -> Vec<ColumnarTrace> {
|
||||||
|
vec![
|
||||||
|
ColumnarTrace::from_rows(
|
||||||
|
"equity",
|
||||||
|
&[ScalarKind::F64],
|
||||||
|
&[(Timestamp(1), vec![Scalar::f64(0.0)]), (Timestamp(2), vec![Scalar::f64(0.5)])],
|
||||||
|
),
|
||||||
|
ColumnarTrace::from_rows(
|
||||||
|
"exposure",
|
||||||
|
&[ScalarKind::F64],
|
||||||
|
&[(Timestamp(1), vec![Scalar::f64(1.0)]), (Timestamp(2), vec![Scalar::f64(-1.0)])],
|
||||||
|
),
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn write_then_read_round_trips() {
|
||||||
|
let root = temp_traces_root("roundtrip");
|
||||||
|
let store = TraceStore::open(&root);
|
||||||
|
let manifest = sample_manifest();
|
||||||
|
let taps = sample_taps();
|
||||||
|
store.write("demo", &manifest, &taps).expect("write");
|
||||||
|
let back = store.read("demo").expect("read");
|
||||||
|
assert_eq!(back.manifest, manifest);
|
||||||
|
assert_eq!(back.taps, taps);
|
||||||
|
assert!(root.join("traces/demo/index.json").exists());
|
||||||
|
assert!(root.join("traces/demo/equity.json").exists());
|
||||||
|
assert!(root.join("traces/demo/exposure.json").exists());
|
||||||
|
let _ = fs::remove_dir_all(&root);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn read_missing_run_is_not_found() {
|
||||||
|
let root = temp_traces_root("missing");
|
||||||
|
let store = TraceStore::open(&root);
|
||||||
|
match store.read("nope") {
|
||||||
|
Err(TraceStoreError::NotFound(name)) => assert_eq!(name, "nope"),
|
||||||
|
other => panic!("expected NotFound, got {other:?}"),
|
||||||
|
}
|
||||||
|
let _ = fs::remove_dir_all(&root);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn corrupt_tap_file_is_a_parse_error_naming_the_file() {
|
||||||
|
let root = temp_traces_root("corrupt");
|
||||||
|
let store = TraceStore::open(&root);
|
||||||
|
store.write("demo", &sample_manifest(), &sample_taps()).expect("write");
|
||||||
|
fs::write(root.join("traces/demo/equity.json"), "not json").expect("clobber");
|
||||||
|
match store.read("demo") {
|
||||||
|
Err(TraceStoreError::Parse { file, .. }) => assert!(file.ends_with("equity.json")),
|
||||||
|
other => panic!("expected Parse, got {other:?}"),
|
||||||
|
}
|
||||||
|
let _ = fs::remove_dir_all(&root);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user