From df0574a75d902d8640ccf32968c3a4bb005d7b25 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 4 Jun 2026 21:41:58 +0200 Subject: [PATCH] spec: 0011 data-server M1 ingestion at the one merge boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First real data source (refs #7, Walking-skeleton milestone). A new `aura-ingest` crate pulls data-server as a cargo git dependency, transposes its 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 (C1). Decisions captured: - Eager materialization (not a lazy/shared Source abstraction): C12's cross-sim Arc<[T]> sharing has no consumer until the multi-sim orchestration cycle, so building it now would be speculative infra; deferred, transpose logic carries over. - aura-ingest is the external-dependency firewall: data-server transitively pulls chrono+regex+zip; isolating it in its own crate keeps core/std/engine zero-external-dep. cargo test --workspace now needs a populated cargo cache. - Scope: boundary + tests only, no aura run CLI arg surface (would force the unsettled #16 arg-parse decision); M1 first, tick deferred. grounding-check returned BLOCK on the single load-bearing external assumption (the data-server public API) — structurally unratifiable by an aura-side test because data-server is brand-new this cycle; all 9 aura-internal assumptions were ratified by named green tests. Overridden after verifying data-server's API against its actual source (lib.rs DataServer::new/has_symbol/ stream_m1_windowed/next_chunk/DEFAULT_DATA_PATH; records.rs M1Parsed fields, types, Copy), per the agent's documented override path. User-approved. --- docs/specs/0011-ingest-data-server-m1.md | 331 +++++++++++++++++++++++ 1 file changed, 331 insertions(+) create mode 100644 docs/specs/0011-ingest-data-server-m1.md diff --git a/docs/specs/0011-ingest-data-server-m1.md b/docs/specs/0011-ingest-data-server-m1.md new file mode 100644 index 0000000..b7bc00b --- /dev/null +++ b/docs/specs/0011-ingest-data-server-m1.md @@ -0,0 +1,331 @@ +# data-server M1 ingestion at the one merge boundary — Design Spec + +**Date:** 2026-06-04 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +## Goal + +Replace hand-built synthetic price streams with aura's first **real** data source. +This cycle ships the **ingestion boundary** (C3): a new `aura-ingest` crate that +pulls `data-server` as a cargo git dependency, transposes its Array-of-Structs +`M1Parsed` records into aura's Structure-of-Arrays base columns (C7), normalizes +`data-server`'s Unix-millisecond time to aura's canonical epoch-ns `Timestamp` at +that one boundary (C3), and feeds the existing k-way merge (`Harness::run`) a real +close-price stream — so a backtest runs over real bars, deterministically (C1). + +Scope is the **boundary plus its tests**, not a new CLI surface. The `aura run` +CLI keeps its argument-free built-in sample harness (cycle 0010); wiring a +`--symbol/--from/--to` invocation surface is a deliberate follow-up cycle, kept +out of this one because it would force the still-unsettled `aura run` arg-parse +strictness (issue #16) and a date-format surface before the boundary itself is +proven. M1 is the format shipped here; tick (ask/bid) is a natural extension of +the same transpose, deferred to a follow-up. + +## Architecture + +`data-server` (`~/dev/libs/data-server`, Gitea `Brummel/data-server`) is a +standalone loader/cache for Pepperstone binary M1/tick files. It yields parsed +records lock-free as `Arc<[T]>` chunks in chronological order, with `time_ms: i64` +(Unix-ms). aura's engine consumes **materialized** ascending source streams +`Vec>` and performs the one k-way merge inside +`Harness::run` (cycle 0004). The ingestion boundary bridges the two. + +**Eager materialization** (decided this cycle): the boundary drains a +`data-server` window into owned SoA columns and hands the engine the stream shape +it already consumes. No engine API change. C12's *cross-sim* zero-copy `Arc<[T]>` +sharing is **not** realized here — it has no consumer yet (the multi-sim +orchestration axes are unbuilt), so building a lazy/shared `Source` abstraction +now would be speculative infra for an absent consumer. It is deferred to the +orchestration cycle that actually shares one dataset across many harnesses; the +transpose logic written here carries over to that layer unchanged. + +**`aura-ingest` is the external-dependency firewall.** The aura workspace is +zero-external-dependency by deep design (the hand-rolled JSON in +`aura-engine::report` exists precisely to keep it so). `data-server` transitively +pulls `chrono`, `regex`, and `zip` — the one sanctioned external dependency +(INDEX.md, External components; this issue). Isolating it in its **own** crate +keeps `aura-core` / `aura-std` / `aura-engine` — the hot path and the frozen +deploy artifact (C8/C6: deploy replays *recorded* streams, never re-ingests) — +external-dependency-free. This is the substantive reason the boundary is a +dedicated crate rather than a feature-gated dep on `aura-engine`: the crate edge +*is* the firewall. + +**Offline-build consequence.** With `aura-ingest` a workspace member, +`cargo build/test --workspace` now requires `data-server` to be fetchable or +already in the cargo cache (one Gitea fetch over LAN, then cached). The previous +"workspace compiles with no network" invariant becomes "compiles from a populated +cargo cache"; this is the accepted cost of the first real data source, and the +firewall confines the transitive external tree to this one crate's build. + +## Concrete code shapes + +### Delivered code — the boundary the cycle ships + +```rust +// crates/aura-ingest/src/lib.rs +use aura_core::{Scalar, Timestamp}; +use data_server::records::M1Parsed; + +/// Normalize data-server's Unix-millisecond time to aura's canonical epoch-ns +/// `Timestamp` — the single unit normalization of C3, performed at the one +/// ingestion boundary and nowhere else. `ms * 1_000_000`; the i64 range covers +/// all market data through year ~2262 (well beyond any Pepperstone file). +pub fn unix_ms_to_epoch_ns(time_ms: i64) -> Timestamp { + Timestamp(time_ms * 1_000_000) +} + +/// One M1 window transposed Array-of-Structs -> Structure-of-Arrays (C7): the +/// OHLCV bar as a bundle of base columns, with time already normalized to +/// epoch-ns. `volume` is the one i64 column; the price/spread columns are f64. +/// All columns share an index: `ts[i]` is the timestamp of `close[i]`, etc. +pub struct M1Columns { + pub ts: Vec, // normalized epoch-ns, ascending (data-server is chronological) + pub open: Vec, + pub high: Vec, + pub low: Vec, + pub close: Vec, + pub spread: Vec, + pub volume: Vec, +} + +/// Transpose data-server's AoS M1 records into aura's SoA columns (C7), +/// normalizing time at the boundary (C3). Pure: identical bars yield identical +/// columns (C1) — it reads no clock and no external state. +pub fn transpose_m1(bars: &[M1Parsed]) -> M1Columns { + let mut c = M1Columns::with_capacity(bars.len()); + for b in bars { + c.ts.push(unix_ms_to_epoch_ns(b.time_ms)); + c.open.push(b.open); + c.high.push(b.high); + c.low.push(b.low); + c.close.push(b.close); + c.spread.push(b.spread); + c.volume.push(b.volume); + } + c +} + +impl M1Columns { + /// The close column as an engine source stream: each normalized timestamp + /// zipped with its close as a `Scalar::F64`. Ascending in timestamp iff the + /// bars were (the C3 ingestion precondition `Harness::run` relies on). This + /// is the price input the sample SMA-cross strategy consumes; a project that + /// wants other fields reads the public columns directly and zips its own. + pub fn close_stream(&self) -> Vec<(Timestamp, Scalar)> { + self.ts.iter().zip(&self.close).map(|(&t, &v)| (t, Scalar::F64(v))).collect() + } +} +``` + +```rust +// crates/aura-ingest/src/lib.rs (data-server adapter) +use std::sync::Arc; +use data_server::DataServer; + +/// Drain a data-server M1 window into transposed SoA columns. Reads the +/// chronological chunk iterator to exhaustion into one AoS buffer, then +/// transposes once at the boundary (C3 — one merge point, not per chunk). +/// `None` if the symbol has no data in `[from_ms, to_ms]` (data-server's own +/// `None`); the inclusive Unix-ms window bounds are data-server's contract. +pub fn load_m1_window( + server: &Arc, + symbol: &str, + from_ms: i64, + to_ms: i64, +) -> Option { + let mut it = server.stream_m1_windowed(symbol, Some(from_ms), Some(to_ms))?; + let mut bars: Vec = Vec::new(); + while let Some(chunk) = it.next_chunk() { + bars.extend_from_slice(&chunk); // M1Parsed: Copy; Arc<[_]> derefs to a slice + } + Some(transpose_m1(&bars)) +} +``` + +### North-star slice — a real backtest over real bars (the gated integration test) + +```rust +// crates/aura-ingest/tests/real_bars.rs (skips cleanly where local data is absent) +use std::sync::Arc; +use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_ingest::load_m1_window; + +#[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}"); + return; // hermetic elsewhere; exercises the real path where files exist + } + // 2006-08 in Unix-ms (inclusive bounds; data-server skips files outside). + let (from_ms, to_ms) = (1_154_390_400_000, 1_157_068_799_999); + + let run = || { + let cols = load_m1_window(&server, "AAPL.US", from_ms, to_ms) + .expect("symbol has data in window"); + let prices = cols.close_stream(); // real close bars as the source + let report = run_sample_over(prices); // SMA-cross -> Exposure -> SimBroker -> report + report + }; + + let r1 = run(); + assert!(r1.metrics.total_pips.is_finite()); // a backtest actually ran over real bars + assert!(!r1.manifest_window_is_empty()); // the window carried real bars + assert_eq!(run().to_json(), r1.to_json()); // same window -> bit-identical (C1) +} +``` + +`run_sample_over(prices)` is a test-local helper that bootstraps the cycle-0007 +signal-quality wiring — `Sma::new(2)` / `Sma::new(4)` → `Sub` → `Exposure::new(0.5)` +→ `SimBroker::new(0.0001)` with two shipped `aura_std::Recorder` sinks (equity on +the broker, exposure on the `Exposure` node, price tapped into the broker) — runs +it on the single real close stream, drains the receivers, and folds +`aura_engine::{f64_field, summarize}` into a `RunReport` whose `window` is the +first/last real bar timestamp. It mirrors `aura-cli`'s `sample_harness`, fed real +prices instead of synthetic. + +### Supporting shape — workspace + crate wiring (secondary) + +```toml +# Cargo.toml (workspace) — add the member +members = [ + "crates/aura-core", + "crates/aura-std", + "crates/aura-engine", + "crates/aura-cli", + "crates/aura-ingest", # NEW — the ingestion boundary / external-dep firewall +] +``` + +```toml +# crates/aura-ingest/Cargo.toml +[package] +name = "aura-ingest" +edition.workspace = true +version.workspace = true +license.workspace = true +publish.workspace = true + +[dependencies] +aura-core = { path = "../aura-core" } +data-server = { git = "http://192.168.178.103:3000/Brummel/data-server.git", branch = "main" } + +[dev-dependencies] +aura-engine = { path = "../aura-engine" } # the integration test bootstraps a harness + report +aura-std = { path = "../aura-std" } # Sma / Sub / Exposure / SimBroker / Recorder +``` + +## Components + +- **`aura-ingest` crate.** The new workspace member. Production surface: + `unix_ms_to_epoch_ns`, `M1Columns` (public SoA fields), `transpose_m1`, + `M1Columns::close_stream`, `load_m1_window`. Production deps: `aura-core` + (Scalar/Timestamp) + `data-server` (records + loader). No production dependency + on `aura-engine`/`aura-std` — the boundary produces the engine's *input shape*, + it does not wire harnesses. +- **`transpose_m1` + `unix_ms_to_epoch_ns`.** The pure C3/C7 core: AoS→SoA and + Unix-ms→epoch-ns. No I/O, no clock — hermetically testable on hand-built + records. +- **`load_m1_window`.** The thin `data-server` adapter: drains the chunk iterator + and calls `transpose_m1` once. This is the only function that touches files / + the cache; it is exercised only by the gated integration test. +- **`M1Columns::close_stream`.** The adapter from a transposed column to the + engine's `Vec<(Timestamp, Scalar)>` source-stream shape. + +## Data flow + +``` +data-server file (ZIP, Unix-ms, AoS) + │ stream_m1_windowed(symbol, from_ms, to_ms) -> SymbolChunkIter + ▼ +Arc<[M1Parsed]> chunks (chronological) + │ load_m1_window: drain to one AoS buffer + ▼ +&[M1Parsed] + │ transpose_m1 ── unix_ms_to_epoch_ns (C3 normalize) ──▶ SoA columns (C7) + ▼ +M1Columns { ts: epoch-ns, open/high/low/close/spread: f64, volume: i64 } + │ close_stream() + ▼ +Vec<(Timestamp, Scalar::F64)> ── the one real source stream ──┐ + │ │ + ▼ ▼ +Harness::run(vec![stream]) ──(k-way merge, cycle 0004; one record = one cycle, C4)──▶ + SMA(2)/SMA(4) → Sub → Exposure → SimBroker → Recorder sinks + │ drain → f64_field → summarize + ▼ +RunReport { manifest{window = first/last real bar ts}, metrics } +``` + +The merge is unchanged: the boundary delivers one ascending stream, so the k-way +merge is a 1-way pass and each bar is one cycle in global timestamp order (C3/C4). +Multi-stream merge (e.g. several symbols, or OHLCV-as-barrier) is already proven +by the engine's own tests; this cycle exercises the single-real-source path. + +## Error handling + +- **Symbol/window miss.** `load_m1_window` returns `None` when `data-server` + returns `None` (unknown symbol, or no file in the window). The integration test + branches on `has_symbol` first and skips cleanly; a real consumer decides + per-call. +- **No local data files.** The integration test prints a skip note and returns + (not a failure), so `cargo test --workspace` stays green on a machine without + `/mnt/tickdata` — correctness lives in the hermetic unit tests, the real path is + exercised where data exists. +- **Timestamp range.** `time_ms * 1_000_000` cannot overflow i64 for any real + Pepperstone timestamp (year ≤ 2262); no checked arithmetic is warranted, and a + pathological future date is out of scope (it would be a data-server-side + concern). +- **Empty window.** A window that resolves to zero bars yields empty columns; + `close_stream` yields an empty stream; `Harness::run` over an empty stream does + nothing and `summarize` returns the documented zeros. No panic. + +## Testing strategy + +Hermetic unit tests (`crates/aura-ingest/src/lib.rs`, no files, no network): + +- `unix_ms_to_epoch_ns` maps a known Unix-ms to the right epoch-ns (e.g. the + data-server epoch test value: `1_488_326_400_000` ms → `…000_000` ns). +- `transpose_m1` on a hand-built `&[M1Parsed]` produces field-correct, equal-length + SoA columns; `volume` lands as i64, prices/spread as f64; `ts` is normalized. +- `transpose_m1` is pure: two transposes of the same bars are equal (C1 at the + boundary). +- `close_stream` zips ts↔close in order and preserves ascending timestamps. +- empty input → empty columns → empty stream (the no-panic edge). + +Gated integration test (`crates/aura-ingest/tests/real_bars.rs`): the north-star +slice above — real close bars → sample strategy → sim-optimal broker → `RunReport`, +finite metrics, and two runs over the same window bit-identical (C1). Skips with a +note where `/mnt/tickdata` is absent. + +Gates (profile `commands`): `cargo test --workspace`, +`cargo clippy --workspace --all-targets -- -D warnings`, +`RUSTDOCFLAGS="-D warnings" cargo doc --workspace --no-deps`. All must be clean, +which now includes building `aura-ingest` and its `data-server` dependency tree. + +## Acceptance criteria + +Maps the issue's acceptance boxes to this spec: + +- [x] **data-server pulled as a cargo git dependency** — `aura-ingest/Cargo.toml`, + git dep on the Gitea URL, `branch = "main"`. +- [x] **M1 AoS records transposed into SoA columns at the boundary** — + `transpose_m1` → `M1Columns`, hermetically tested. +- [x] **time_ms (Unix-ms) normalized to epoch-ns at the boundary** — + `unix_ms_to_epoch_ns`, tested against a known conversion. +- [x] **The k-way merge consumes the real source; a backtest runs over real bars** + — `close_stream` → `Harness::run` → `RunReport` in the gated integration test. +- [x] **Determinism preserved (C1): same window → same cycle stream** — purity + unit test + the two-run bit-identical assertion in the integration test. + +### Non-goals (this cycle) + +- No `aura run --symbol/--from/--to` CLI surface (deferred; would force issue + #16's arg-parse decision and a date-format surface). +- No tick (ask/bid) ingestion (deferred; same transpose shape, a follow-up). +- No lazy/shared `Source` abstraction / `Arc<[T]>` cross-sim sharing (deferred to + the multi-sim orchestration cycle that has a consumer for it; C12). +- No OHLCV-barrier bundler node in the shipped path — the sample strategy consumes + the close column; the other columns are transposed and publicly available but + not wired into a bar-record node here.