Files
Aura/crates/aura-ingest/tests/real_bars.rs
T
Brummel a24729e97f feat(aura-ingest): data-server M1 ingestion at the one merge boundary
aura's first real data source (closes #7, Walking-skeleton milestone). A new
aura-ingest workspace crate transposes data-server's AoS M1Parsed records into
SoA base columns (C7), normalizes Unix-ms to canonical epoch-ns at the one
ingestion boundary (C3), and feeds the existing k-way merge a real close-price
stream so a backtest runs over real bars, deterministically (C1).

Surface:
- unix_ms_to_epoch_ns(time_ms) -> Timestamp (= ms * 1_000_000), the single C3
  unit normalization.
- M1Columns: the OHLCV bar as a bundle of base columns (open/high/low/close/
  spread: f64, volume: i64, ts: epoch-ns) — SoA, C7.
- transpose_m1(&[M1Parsed]) -> M1Columns: pure AoS->SoA (C1).
- M1Columns::close_stream() -> Vec<(Timestamp, Scalar)>: the price input the
  SMA-cross sample strategy consumes.
- load_m1_window(server, symbol, from_ms, to_ms): drains data-server's
  chronological chunks, transposes once at the boundary.

Design (per spec 0011):
- Eager materialization, not a lazy/shared Source abstraction: C12's cross-sim
  Arc<[T]> sharing has no consumer until the multi-sim orchestration cycle;
  deferred, the transpose logic carries over.
- aura-ingest is the external-dependency firewall. data-server transitively
  pulls chrono + regex + zip (+ their trees) into Cargo.lock; isolating it in
  this one crate keeps aura-core/std/engine zero-external-dependency and the
  frozen deploy artifact (C6 replays recorded streams, never re-ingests) clean.
  cargo test --workspace now needs a populated cargo cache (one Gitea fetch,
  done).
- Scope: boundary + tests only, no aura run CLI arg surface (M1 first; tick and
  the CLI wiring are follow-ups).

Tests: 6 hermetic unit tests on hand-built M1Parsed (normalization, field-wise
transpose, purity, close_stream order, empty edges) + one gated integration
test (tests/real_bars.rs) that runs the cycle-0007 sample harness over real
AAPL.US 2006-08 close bars and asserts finite metrics + two-run bit-identical
JSON (C1); it skips cleanly where /mnt/tickdata is absent. On this machine it
ran the real path. Workspace gates green: test (86), clippy -D warnings, doc
-D warnings.

Minor: transient unused-import warnings in the new lib.rs across the additive
build steps resolved once load_m1_window consumed the imports; final -D warnings
gate clean.
2026-06-04 21:52:33 +02:00

108 lines
4.4 KiB
Rust

//! Gated integration test: a real data-server M1 close stream driven through
//! the cycle-0007 signal-quality sample harness (SMA-cross → Exposure →
//! SimBroker), folded into a `RunReport`. Skips with a note where the local
//! Pepperstone data directory is absent, so `cargo test --workspace` stays green
//! anywhere; it exercises the real ingestion path where data exists.
use std::sync::mpsc;
use std::sync::Arc;
use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
f64_field, summarize, Edge, Harness, RunManifest, RunReport, SourceSpec, Target,
};
use aura_ingest::load_m1_window;
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use data_server::{DataServer, DEFAULT_DATA_PATH};
/// Bootstrap the cycle-0007 two-sink signal-quality harness (mirrors
/// `aura-engine`'s `report::tests::build_two_sink_harness`, with the shipped
/// `aura_std::Recorder`), run it on `prices`, and fold the recorded equity +
/// exposure into a `RunReport` whose window is the first/last real bar ts.
fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
let (tx_eq, rx_eq) = mpsc::channel();
let (tx_ex, rx_ex) = mpsc::channel();
let mut h = Harness::bootstrap(
vec![
Box::new(Sma::new(2)), // 0
Box::new(Sma::new(4)), // 1
Box::new(Sub::new()), // 2
Box::new(Exposure::new(0.5)), // 3
Box::new(SimBroker::new(0.0001)), // 4
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
],
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
],
}],
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 },
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 signal-quality DAG");
let window = (
prices.first().map(|&(t, _)| t).unwrap_or(Timestamp(0)),
prices.last().map(|&(t, _)| t).unwrap_or(Timestamp(0)),
);
h.run(vec![prices]);
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 equity = f64_field(&eq_rows, 0);
let exposure = f64_field(&ex_rows, 0);
let metrics = summarize(&equity, &exposure);
RunReport {
manifest: RunManifest {
commit: "real-bars-test".to_string(),
params: vec![
("sma_fast".to_string(), 2.0),
("sma_slow".to_string(), 4.0),
("exposure_scale".to_string(), 0.5),
],
window,
seed: 0,
broker: "sim-optimal(pip_size=0.0001)".to_string(),
},
metrics,
}
}
#[test]
fn sample_strategy_runs_over_real_m1_bars_deterministically() {
let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
if !server.has_symbol("AAPL.US") {
eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol AAPL.US absent)");
return; // hermetic elsewhere; exercises the real path where files exist
}
// 2006-08 in inclusive Unix-ms (data-server skips files outside the window).
let (from_ms, to_ms) = (1_154_390_400_000_i64, 1_157_068_799_999_i64);
let load = || {
load_m1_window(&server, "AAPL.US", from_ms, to_ms)
.expect("AAPL.US has data in the 2006-08 window")
.close_stream()
};
let prices = load();
assert!(!prices.is_empty(), "window resolved to zero bars");
let r1 = run_sample_over(prices);
assert!(r1.metrics.total_pips.is_finite(), "a backtest ran over real bars");
// same window -> bit-identical report (C1).
let r2 = run_sample_over(load());
assert_eq!(r1.to_json(), r2.to_json());
}