diff --git a/crates/aura-ingest/examples/ger40_breakout_compare.rs b/crates/aura-ingest/examples/ger40_breakout_compare.rs index dd7688c..ff0067d 100644 --- a/crates/aura-ingest/examples/ger40_breakout_compare.rs +++ b/crates/aura-ingest/examples/ger40_breakout_compare.rs @@ -23,12 +23,10 @@ use std::sync::Arc; use aura_core::{Cell, Scalar, Timestamp}; -use aura_engine::{summarize, RunMetrics, Source}; -use chrono::TimeZone; +use aura_engine::{summarize, RunMetrics}; use chrono_tz::Europe::Berlin; -use data_server::{DataServer, DEFAULT_DATA_PATH}; -use aura_ingest::{M1Field, M1FieldSource}; +use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH}; #[path = "shared/breakout_real.rs"] mod breakout_real; @@ -48,7 +46,7 @@ const INSTRUMENTS: &[(&str, f64, &str)] = &[ ]; /// Bootstrap the blueprint (at `bar_period`) at `{entry=3, exit=5}`, run it over -/// `[from_ms, to_ms]` of real `symbol` OHLC, and summarize — or `None` if the +/// `[from, to]` (epoch-ns) of real `symbol` OHLC, and summarize — or `None` if the /// symbol has no file overlapping the window. The `pip_size` rides on the /// blueprint's baked SimBroker (always 1.0 here; both indices). Returns the /// report plus the entry count (0→1 transitions of held). @@ -56,8 +54,8 @@ fn run_cell( server: &Arc, symbol: &str, bar_period: i64, - from_ms: i64, - to_ms: i64, + from: Timestamp, + to: Timestamp, ) -> Option<(RunMetrics, u32)> { let (bp, taps) = ger40_breakout_blueprint(bar_period, SESSION_HOUR, SESSION_MINUTE, Berlin); let point: Vec = bp @@ -71,12 +69,7 @@ fn run_cell( .collect(); let mut h = bp.bootstrap_with_cells(&point).expect("point kind-checked"); - let fields = [M1Field::Open, M1Field::High, M1Field::Low, M1Field::Close]; - let mut srcs: Vec> = Vec::with_capacity(4); - for &f in &fields { - let s = M1FieldSource::open(server, symbol, Some(from_ms), Some(to_ms), f)?; - srcs.push(Box::new(s)); - } + let srcs = open_ohlc(server, symbol, from, to)?; h.run(srcs); drop(h); @@ -104,21 +97,15 @@ fn run_cell( } fn main() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { println!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)."); return; } - // Full year 2024 for body. - let from_ms = chrono::Utc - .with_ymd_and_hms(2024, 1, 1, 0, 0, 0) - .unwrap() - .timestamp_millis(); - let to_ms = chrono::Utc - .with_ymd_and_hms(2024, 12, 31, 23, 59, 59) - .unwrap() - .timestamp_millis(); + // Full year 2024 (the same UTC span, via the shared Timestamp helper). + let (from, _) = utc_month_window(2024, 1); + let (_, to) = utc_month_window(2024, 12); println!("=== GER40 session-breakout — World `compare` (structural matrix) ==="); println!("window=2024 (UTC, full year), fixed tuning point entry={ENTRY_BAR}, exit={EXIT_BAR}\n"); @@ -135,7 +122,7 @@ fn main() { println!("{symbol:<8} (no local data — skip)"); continue; } - match run_cell(&server, symbol, BAR_MINUTES, from_ms, to_ms) { + match run_cell(&server, symbol, BAR_MINUTES, from, to) { Some((m, entries)) => println!( "{symbol:<8} {note:<14} {entries:>9} {:>11.1} {:>10.1} {:>7}", m.total_pips, m.max_drawdown, m.exposure_sign_flips @@ -156,7 +143,7 @@ fn main() { ); println!("{}", "-".repeat(52)); for &period in &[15_i64, 30] { - match run_cell(&server, SYMBOL, period, from_ms, to_ms) { + match run_cell(&server, SYMBOL, period, from, to) { Some((m, entries)) => println!( "{:<10} {entries:>9} {:>11.1} {:>10.1} {:>7}", format!("{period}m"), diff --git a/crates/aura-ingest/examples/ger40_breakout_real.rs b/crates/aura-ingest/examples/ger40_breakout_real.rs index 87f2755..0d2ab04 100644 --- a/crates/aura-ingest/examples/ger40_breakout_real.rs +++ b/crates/aura-ingest/examples/ger40_breakout_real.rs @@ -20,11 +20,9 @@ //! so the strategy's behaviour over the ~20 sessions in the month is visibly //! inspectable. Skips cleanly (no panic) where the archive is absent. -use std::sync::Arc; - use chrono::TimeZone; use chrono_tz::Europe::Berlin; -use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_ingest::{default_data_server, open_ohlc, DEFAULT_DATA_PATH}; // The breakout wiring is defined once in the shared module and included by both // this example and the gated determinism test — never duplicated. @@ -37,7 +35,7 @@ const YEAR: i32 = 2024; const MONTH: u32 = 9; fn main() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { println!( "skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent) — \ @@ -46,9 +44,9 @@ fn main() { return; } - let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH); + let (from, to) = utc_month_window(YEAR, MONTH); - let Some(sources) = open_ohlc_sources(&server, from_ms, to_ms) else { + let Some(sources) = open_ohlc(&server, SYMBOL, from, to) else { println!( "skip: {SYMBOL} has no M1 file overlapping {YEAR}-{MONTH:02} — \ nothing to demonstrate here." @@ -72,7 +70,7 @@ fn main() { // channels), then fold the report from that same trace so the mpsc channels // are drained exactly once (shared with the determinism test). let trace = drain_trace(&taps); - let report = report_from_trace(&trace, from_ms, to_ms); + let report = report_from_trace(&trace, from, to); let metrics = &report.metrics; println!("=== GER40 15m session-breakout — REAL bars ==="); diff --git a/crates/aura-ingest/examples/ger40_breakout_sweep.rs b/crates/aura-ingest/examples/ger40_breakout_sweep.rs index cc7e819..5d6ee0c 100644 --- a/crates/aura-ingest/examples/ger40_breakout_sweep.rs +++ b/crates/aura-ingest/examples/ger40_breakout_sweep.rs @@ -18,7 +18,7 @@ use std::sync::Arc; use aura_core::{Cell, Scalar, Timestamp}; use aura_engine::{sweep, summarize, GridSpace, RunManifest, RunReport}; use chrono_tz::Europe::Berlin; -use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH}; #[path = "shared/breakout_real.rs"] mod breakout_real; @@ -29,17 +29,18 @@ use breakout_real::*; const ENTRY_BARS: &[i64] = &[2, 3, 4]; const EXIT_BARS: &[i64] = &[4, 5, 6]; -/// Bootstrap the blueprint at one grid point, run it over `[from_ms, to_ms]` of -/// real GER40 OHLC, and fold the recorded equity/held taps into a `RunReport` -/// (`summarize` over the drained channels). A fresh blueprint per call (fresh -/// nodes + channels) keeps the disjoint sweep points independent (C1). -fn run_point(server: &Arc, point: &[Cell], from_ms: i64, to_ms: i64) -> RunReport { +/// Bootstrap the blueprint at one grid point, run it over the epoch-ns +/// [`Timestamp`] window `[from, to]` of real GER40 OHLC, and fold the recorded +/// equity/held taps into a `RunReport` (`summarize` over the drained channels). +/// A fresh blueprint per call (fresh nodes + channels) keeps the disjoint sweep +/// points independent (C1). +fn run_point(server: &Arc, point: &[Cell], from: Timestamp, to: Timestamp) -> RunReport { let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin); let mut h = bp .bootstrap_with_cells(point) .expect("sweep point kind-checked against param_space"); let sources = - open_ohlc_sources(server, from_ms, to_ms).expect("window overlaps GER40 data"); + open_ohlc(server, SYMBOL, from, to).expect("window overlaps GER40 data"); h.run(sources); drop(h); let equity: Vec<(Timestamp, f64)> = taps @@ -56,10 +57,7 @@ fn run_point(server: &Arc, point: &[Cell], from_ms: i64, to_ms: i64) manifest: RunManifest { commit: "ger40-breakout-sweep".to_string(), params: vec![], - window: ( - aura_ingest::unix_ms_to_epoch_ns(from_ms), - aura_ingest::unix_ms_to_epoch_ns(to_ms), - ), + window: (from, to), seed: 0, broker: "sim-optimal(pip_size=1)".to_string(), }, @@ -68,15 +66,15 @@ fn run_point(server: &Arc, point: &[Cell], from_ms: i64, to_ms: i64) } fn main() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { println!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)."); return; } // Full-year 2024. - let (from_ms, _) = utc_month_window_ms(2024, 1); - let (_, to_ms) = utc_month_window_ms(2024, 12); + let (from, _) = utc_month_window(2024, 1); + let (_, to) = utc_month_window(2024, 12); // The grid is built from the blueprint's param_space() directly — the two // EqConst targets are the swept axes (no re-authoring). @@ -109,7 +107,7 @@ fn main() { let server_for_closure = Arc::clone(&server); let family = sweep(&grid, |pt: &[Cell]| { - run_point(&server_for_closure, pt, from_ms, to_ms) + run_point(&server_for_closure, pt, from, to) }); // Rank the family by total_pips, descending. diff --git a/crates/aura-ingest/examples/ger40_breakout_walkforward.rs b/crates/aura-ingest/examples/ger40_breakout_walkforward.rs index ecb8e3f..6dfae38 100644 --- a/crates/aura-ingest/examples/ger40_breakout_walkforward.rs +++ b/crates/aura-ingest/examples/ger40_breakout_walkforward.rs @@ -24,7 +24,7 @@ use aura_engine::{ }; use chrono::TimeZone; use chrono_tz::Europe::Berlin; -use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH}; #[path = "shared/breakout_real.rs"] mod breakout_real; @@ -35,28 +35,22 @@ use breakout_real::*; const ENTRY_BARS: &[i64] = &[2, 3, 4]; const EXIT_BARS: &[i64] = &[4, 5, 6]; -/// ns → ms: `open_ohlc_sources` opens by Unix-ms, but `WindowBounds` are -/// epoch-ns. `aura-ingest` exports only the forward direction, so the inverse is -/// a local hand-divide (#80, filed). -fn ns_to_ms(ts: Timestamp) -> i64 { - ts.0 / 1_000_000 -} - -/// Bootstrap the blueprint at one grid point, run it over `[from_ms, to_ms]` of -/// real GER40 OHLC, and fold the recorded equity/held taps into `(RunReport, -/// equity-segment)`. Fresh blueprint per call (C1). +/// Bootstrap the blueprint at one grid point, run it over the epoch-ns +/// [`Timestamp`] window `[from, to]` of real GER40 OHLC, and fold the recorded +/// equity/held taps into `(RunReport, equity-segment)`. Fresh blueprint per call +/// (C1). fn run_point( server: &Arc, point: &[Cell], - from_ms: i64, - to_ms: i64, + from: Timestamp, + to: Timestamp, ) -> (RunReport, Vec<(Timestamp, f64)>) { let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin); let mut h = bp .bootstrap_with_cells(point) .expect("point kind-checked against param_space"); let sources = - open_ohlc_sources(server, from_ms, to_ms).expect("window overlaps GER40 data"); + open_ohlc(server, SYMBOL, from, to).expect("window overlaps GER40 data"); h.run(sources); drop(h); let equity: Vec<(Timestamp, f64)> = taps @@ -73,10 +67,7 @@ fn run_point( manifest: RunManifest { commit: "ger40-breakout-wfo".to_string(), params: vec![], - window: ( - aura_ingest::unix_ms_to_epoch_ns(from_ms), - aura_ingest::unix_ms_to_epoch_ns(to_ms), - ), + window: (from, to), seed: 0, broker: "sim-optimal(pip_size=1)".to_string(), }, @@ -86,7 +77,7 @@ fn run_point( } fn main() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { println!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)."); return; @@ -143,9 +134,8 @@ fn main() { .collect(), ) .expect("IS grid well-formed"); - let (is_from, is_to) = (ns_to_ms(w.is.0), ns_to_ms(w.is.1)); let is_family = sweep(&is_grid, |pt: &[Cell]| { - run_point(&server_for_closure, pt, is_from, is_to).0 + run_point(&server_for_closure, pt, w.is.0, w.is.1).0 }); let best = is_family .points @@ -161,9 +151,8 @@ fn main() { let chosen = best.params.clone(); // Apply the chosen params on the OOS window — the honest out-of-sample run. - let (oos_from, oos_to) = (ns_to_ms(w.oos.0), ns_to_ms(w.oos.1)); let (oos_report, oos_equity) = - run_point(&server_for_closure, &chosen, oos_from, oos_to); + run_point(&server_for_closure, &chosen, w.oos.0, w.oos.1); WindowRun { chosen_params: chosen, oos_equity, oos_report } }); diff --git a/crates/aura-ingest/examples/shared/breakout_real.rs b/crates/aura-ingest/examples/shared/breakout_real.rs index bd9d8f5..3303e96 100644 --- a/crates/aura-ingest/examples/shared/breakout_real.rs +++ b/crates/aura-ingest/examples/shared/breakout_real.rs @@ -37,19 +37,15 @@ #![allow(dead_code)] // each consumer (example / test) uses a subset of this API use std::sync::mpsc; -use std::sync::Arc; use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp}; use aura_engine::{ - summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport, Source, + summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport, SourceSpec, Target, }; use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder, Resample, Session, SimBroker}; use chrono::TimeZone; use chrono_tz::Europe::Berlin; -use data_server::DataServer; - -use aura_ingest::{M1Field, M1FieldSource}; // --------------------------------------------------------------------------- // Node indices in the FlatGraph (one fixed layout, the same nodes 0..=8 as @@ -306,32 +302,14 @@ pub fn ger40_breakout_blueprint( (bp, Taps { equity: rx_equity, held: rx_held, bars: rx_bars, breakout: rx_breakout }) } -/// Open the four real OHLC `M1FieldSource`s over `[from_ms, to_ms]` (inclusive -/// Unix-ms) in the FIXED order open, high, low, close — the C4 merge tie-break -/// order Resample's `Barrier(0)` group expects. Returns `None` if any field has -/// no file overlapping the window (so the caller can skip cleanly). -/// -/// This is exactly the multi-field OHLC open that `M1FieldSource` makes the -/// caller spell out four times by hand — see the friction note in the report. -pub fn open_ohlc_sources( - server: &Arc, - from_ms: i64, - to_ms: i64, -) -> Option>> { - let open = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Open)?; - let high = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::High)?; - let low = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Low)?; - let close = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Close)?; - let srcs: Vec> = - vec![Box::new(open), Box::new(high), Box::new(low), Box::new(close)]; - Some(srcs) -} - -/// Build the inclusive Unix-ms window for the whole calendar month `(year, -/// month)` in UTC: `[first instant of the 1st, last ms of the last day]`. UTC -/// keeps the window boundary trivial to reason about; the Session node still -/// indexes bars in Berlin wall-clock inside the graph. -pub fn utc_month_window_ms(year: i32, month: u32) -> (i64, i64) { +/// Build the inclusive window for the whole calendar month `(year, month)` in +/// UTC: `[first instant of the 1st, last ms of the last day]`, in aura's native +/// epoch-ns [`Timestamp`] currency. UTC keeps the boundary trivial to reason +/// about; the Session node still indexes bars in Berlin wall-clock inside the +/// graph. The SAME UTC instants as before, typed as `Timestamp` (each bound +/// wrapped through the one `unix_ms_to_epoch_ns` seam), so the consumer layer is +/// Timestamp-native and no caller round-trips ms→Timestamp→ms. +pub fn utc_month_window(year: i32, month: u32) -> (Timestamp, Timestamp) { let from = chrono::Utc .with_ymd_and_hms(year, month, 1, 0, 0, 0) .unwrap() @@ -342,7 +320,10 @@ pub fn utc_month_window_ms(year: i32, month: u32) -> (i64, i64) { .with_ymd_and_hms(ny, nm, 1, 0, 0, 0) .unwrap() .timestamp_millis(); - (from, next - 1) + ( + aura_ingest::unix_ms_to_epoch_ns(from), + aura_ingest::unix_ms_to_epoch_ns(next - 1), + ) } /// One bar's recorded trace row, fused across the four taps by **timestamp**. @@ -403,8 +384,8 @@ pub fn drain_trace(taps: &Taps) -> Vec { /// window, exactly as `tests/real_bars.rs` does for the SMA sample. The trace is /// the single drain of the recording channels (`drain_trace`); this reduction is /// pure over it, so the example and the test share one report definition. The -/// window is the requested `[from_ms, to_ms]` normalized to epoch-ns. -pub fn report_from_trace(trace: &[BarTrace], from_ms: i64, to_ms: i64) -> RunReport { +/// window is the requested `[from, to]` epoch-ns window. +pub fn report_from_trace(trace: &[BarTrace], from: Timestamp, to: Timestamp) -> RunReport { let equity: Vec<(Timestamp, f64)> = trace.iter().map(|b| (b.ts, b.equity)).collect(); let exposure: Vec<(Timestamp, f64)> = trace.iter().map(|b| (b.ts, b.held)).collect(); let metrics = summarize(&equity, &exposure); @@ -419,10 +400,7 @@ pub fn report_from_trace(trace: &[BarTrace], from_ms: i64, to_ms: i64) -> RunRep ("entry_bar".to_string(), Scalar::i64(3)), ("exit_bar".to_string(), Scalar::i64(5)), ], - window: ( - aura_ingest::unix_ms_to_epoch_ns(from_ms), - aura_ingest::unix_ms_to_epoch_ns(to_ms), - ), + window: (from, to), seed: 0, broker: "sim-optimal(pip_size=1)".to_string(), }, diff --git a/crates/aura-ingest/src/lib.rs b/crates/aura-ingest/src/lib.rs index 407615c..10475fb 100644 --- a/crates/aura-ingest/src/lib.rs +++ b/crates/aura-ingest/src/lib.rs @@ -22,7 +22,12 @@ use aura_core::{Scalar, Timestamp}; use data_server::records::M1Parsed; -use data_server::{DataServer, SymbolChunkIter}; +use data_server::SymbolChunkIter; + +/// Re-export of the data-server archive entry points (#81): a real-data source +/// builds from `aura-ingest` alone — a consumer never names the external +/// `data_server` crate directly. +pub use data_server::{DataServer, DEFAULT_DATA_PATH}; use std::sync::Arc; /// Normalize data-server's Unix-millisecond time to aura's canonical epoch-ns @@ -33,6 +38,14 @@ pub fn unix_ms_to_epoch_ns(time_ms: i64) -> Timestamp { Timestamp(time_ms * 1_000_000) } +/// Inverse of [`unix_ms_to_epoch_ns`]: project aura's canonical epoch-ns +/// [`Timestamp`] back to data-server's Unix-millisecond time. **Private** — the +/// seam owns the ms↔ns convention (C3); a consumer threads `Timestamp`s and never +/// converts. Floor division by 1e6 ns/ms (archived M1 data has no sub-ms instant). +fn epoch_ns_to_unix_ms(ts: Timestamp) -> i64 { + ts.0 / 1_000_000 +} + /// One M1 window transposed Array-of-Structs → Structure-of-Arrays (C7): the /// OHLCV bar as a bundle of base columns, time already normalized to epoch-ns. /// `volume` is the one `i64` column; the price/spread columns are `f64`. All @@ -216,6 +229,28 @@ impl M1FieldSource { Some(s) } + /// Open over the `[from, to]` window in aura's native epoch-ns [`Timestamp`] + /// currency — the engine-side mirror of [`open`](Self::open), which takes + /// data-server's Unix-ms. Each bound is mapped through the seam-private + /// `epoch_ns_to_unix_ms` and delegated to `open`, so the ms↔ns crossing + /// happens once, here, and a consumer never divides. `None` bounds preserve + /// open-ended windows exactly as `open` does. + pub fn open_window( + server: &Arc, + symbol: &str, + from: Option, + to: Option, + field: M1Field, + ) -> Option { + Self::open( + server, + symbol, + from.map(epoch_ns_to_unix_ms), + to.map(epoch_ns_to_unix_ms), + field, + ) + } + /// Decode the head at the cursor, refilling chunks as needed. Sets /// `self.head = None` at exhaustion. fn advance(&mut self) { @@ -271,6 +306,39 @@ impl aura_engine::Source for M1FieldSource { } } +/// Open the four real OHLC [`M1FieldSource`]s for `symbol` over the closed +/// epoch-ns window `[from, to]`, in the FIXED order open, high, low, close — the +/// C4 merge tie-break order a resampler's `Barrier(0)` group depends on (#92). +/// This is the single vetted home of that order; consumers never spell it out. +/// All four sources share the one `Arc` (one `FileCache`), so a +/// window's bars are parsed once and reused across the four field decodes, and +/// the same `Arc` flows across the disjoint sims of a sweep / walk-forward (C12). +/// +/// Returns `None` if any field has no archived file overlapping the window (the +/// caller skips cleanly), propagating each [`open_window`](M1FieldSource::open_window)'s +/// file-level `Option`. A window that overlaps a file but holds zero bars yields +/// four sources whose first `peek` is `None`, not a `None` here. +pub fn open_ohlc( + server: &Arc, + symbol: &str, + from: Timestamp, + to: Timestamp, +) -> Option>> { + let open = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::Open)?; + let high = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::High)?; + let low = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::Low)?; + let close = M1FieldSource::open_window(server, symbol, Some(from), Some(to), M1Field::Close)?; + Some(vec![Box::new(open), Box::new(high), Box::new(low), Box::new(close)]) +} + +/// Construct the default data-server over the local archive at +/// [`DEFAULT_DATA_PATH`], wrapped in the `Arc` the streaming sources share (#81). +/// Build it once and clone the `Arc` across a family's sims — one cache, never +/// one server per field (C12). +pub fn default_data_server() -> Arc { + Arc::new(DataServer::new(DEFAULT_DATA_PATH)) +} + #[cfg(test)] mod tests { use super::*; @@ -286,6 +354,16 @@ mod tests { assert_eq!(unix_ms_to_epoch_ns(0), Timestamp(0)); } + #[test] + fn epoch_ns_to_unix_ms_inverts_unix_ms_to_epoch_ns() { + // The seam-owned inverse round-trips the forward normalization for every + // ms instant, so a Timestamp window bound fed back to data-server's ms + // contract recovers the exact ms (C3: one currency crossing, owned here). + for ms in [0_i64, 1, 1_488_326_400_000, 1_727_000_000_000] { + assert_eq!(epoch_ns_to_unix_ms(unix_ms_to_epoch_ns(ms)), ms); + } + } + /// A hand-built M1 bar with distinct per-field values so a transpose test /// can tell the columns apart. fn full_bar(time_ms: i64) -> M1Parsed { diff --git a/crates/aura-ingest/tests/ger40_breakout_blueprint.rs b/crates/aura-ingest/tests/ger40_breakout_blueprint.rs index 516148c..6c6f96c 100644 --- a/crates/aura-ingest/tests/ger40_breakout_blueprint.rs +++ b/crates/aura-ingest/tests/ger40_breakout_blueprint.rs @@ -16,9 +16,9 @@ use std::sync::Arc; -use aura_core::Scalar; +use aura_core::{Scalar, Timestamp}; use chrono_tz::Europe::Berlin; -use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH}; #[path = "../examples/shared/breakout_real.rs"] mod breakout_real; @@ -69,10 +69,11 @@ fn blueprint_param_space_is_construction_deterministic() { assert_eq!(a, b, "blueprint param_space() is a deterministic function of its args"); } -/// Run the blueprint-bootstrapped harness over `[from_ms, to_ms]` and drain the -/// recorded per-bar `(held, equity)` series. Fresh build per call (fresh nodes + -/// channels) keeps two runs disjoint for the C1 determinism assertion. -fn run_blueprint(server: &Arc, from_ms: i64, to_ms: i64) -> Vec<(f64, f64)> { +/// Run the blueprint-bootstrapped harness over the epoch-ns [`Timestamp`] window +/// `[from, to]` and drain the recorded per-bar `(held, equity)` series. Fresh +/// build per call (fresh nodes + channels) keeps two runs disjoint for the C1 +/// determinism assertion. +fn run_blueprint(server: &Arc, from: Timestamp, to: Timestamp) -> Vec<(f64, f64)> { let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin); let mut h = bp .with("entry_bar.target", Scalar::i64(3)) @@ -80,7 +81,7 @@ fn run_blueprint(server: &Arc, from_ms: i64, to_ms: i64) -> Vec<(f64 .bootstrap() .expect("blueprint bootstraps with entry=3, exit=5"); let sources = - open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window"); + open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window"); h.run(sources); let trace = drain_trace(&taps); trace.iter().map(|b| (b.held, b.equity)).collect() @@ -88,10 +89,10 @@ fn run_blueprint(server: &Arc, from_ms: i64, to_ms: i64) -> Vec<(f64 /// Run the shipped hand-wired `FlatGraph` over the same window — the C23 /// reference series. -fn run_flatgraph(server: &Arc, from_ms: i64, to_ms: i64) -> Vec<(f64, f64)> { +fn run_flatgraph(server: &Arc, from: Timestamp, to: Timestamp) -> Vec<(f64, f64)> { let (mut h, taps) = build_harness(); let sources = - open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window"); + open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window"); h.run(sources); let trace = drain_trace(&taps); trace.iter().map(|b| (b.held, b.equity)).collect() @@ -108,26 +109,26 @@ fn run_flatgraph(server: &Arc, from_ms: i64, to_ms: i64) -> Vec<(f64 /// disjoint rerun (C1). #[test] fn composite_matches_flatgraph_bit_identical() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)"); return; // hermetic elsewhere; exercises the real path where files exist } - let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH); + let (from, to) = utc_month_window(YEAR, MONTH); - let blueprint_series = run_blueprint(&server, from_ms, to_ms); + let blueprint_series = run_blueprint(&server, from, to); assert!(!blueprint_series.is_empty(), "window resolved to zero bars"); // C23: the Composite blueprint reproduces the hand-wired FlatGraph EXACTLY. - let flatgraph_series = run_flatgraph(&server, from_ms, to_ms); + let flatgraph_series = run_flatgraph(&server, from, to); assert_eq!( blueprint_series, flatgraph_series, "blueprint must reproduce the hand-wired FlatGraph bit-identically (C23)", ); // C1: a fresh disjoint blueprint run over the identical window is bit-identical. - let blueprint_rerun = run_blueprint(&server, from_ms, to_ms); + let blueprint_rerun = run_blueprint(&server, from, to); assert_eq!( blueprint_series, blueprint_rerun, "blueprint recorded (held, equity) series bit-identical across runs (C1)", diff --git a/crates/aura-ingest/tests/ger40_breakout_real.rs b/crates/aura-ingest/tests/ger40_breakout_real.rs index daf3209..5698988 100644 --- a/crates/aura-ingest/tests/ger40_breakout_real.rs +++ b/crates/aura-ingest/tests/ger40_breakout_real.rs @@ -12,7 +12,8 @@ use std::sync::Arc; -use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_core::Timestamp; +use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH}; #[path = "../examples/shared/breakout_real.rs"] mod breakout_real; @@ -22,16 +23,17 @@ use breakout_real::*; const YEAR: i32 = 2024; const MONTH: u32 = 9; -/// Build the real breakout harness, run it over `[from_ms, to_ms]`, and return -/// the folded `RunReport` plus the drained per-bar `(held, equity)` series. A -/// fresh build per call keeps the two determinism runs disjoint (C1). +/// Build the real breakout harness, run it over the epoch-ns [`Timestamp`] +/// window `[from, to]`, and return the folded `RunReport` plus the drained +/// per-bar `(held, equity)` series. A fresh build per call keeps the two +/// determinism runs disjoint (C1). fn run_real( server: &Arc, - from_ms: i64, - to_ms: i64, + from: Timestamp, + to: Timestamp, ) -> (aura_engine::RunReport, Vec<(f64, f64)>) { let sources = - open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window"); + open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window"); let (mut h, taps) = build_harness(); h.run(sources); @@ -39,7 +41,7 @@ fn run_real( // helper so the mpsc channels are consumed exactly once and the report shape // matches the example's byte-for-byte. let trace = drain_trace(&taps); - let report = report_from_trace(&trace, from_ms, to_ms); + let report = report_from_trace(&trace, from, to); let series: Vec<(f64, f64)> = trace.iter().map(|b| (b.held, b.equity)).collect(); (report, series) } @@ -56,15 +58,15 @@ fn run_real( /// data, which the synthetic single-session capstone cannot reach. #[test] fn ger40_breakout_real_bars_run_is_deterministic() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)"); return; // hermetic elsewhere; exercises the real path where files exist } - let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH); + let (from, to) = utc_month_window(YEAR, MONTH); - let (r1, s1) = run_real(&server, from_ms, to_ms); + let (r1, s1) = run_real(&server, from, to); assert!(!s1.is_empty(), "window resolved to zero bars"); // Metrics are finite — a backtest actually ran over real bars. @@ -82,7 +84,7 @@ fn ger40_breakout_real_bars_run_is_deterministic() { // C1 determinism: a fresh disjoint run over the identical window is // bit-identical in both the folded report and the raw recorded series. - let (r2, s2) = run_real(&server, from_ms, to_ms); + let (r2, s2) = run_real(&server, from, to); assert_eq!(r1.to_json(), r2.to_json(), "report bit-identical across runs (C1)"); assert_eq!(s1, s2, "recorded (held, equity) series bit-identical across runs (C1)"); } diff --git a/crates/aura-ingest/tests/ger40_breakout_world.rs b/crates/aura-ingest/tests/ger40_breakout_world.rs index 55f8c16..9325c2c 100644 --- a/crates/aura-ingest/tests/ger40_breakout_world.rs +++ b/crates/aura-ingest/tests/ger40_breakout_world.rs @@ -27,31 +27,24 @@ use aura_engine::{ }; use chrono::TimeZone; use chrono_tz::Europe::Berlin; -use data_server::{DataServer, DEFAULT_DATA_PATH}; +use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH}; #[path = "../examples/shared/breakout_real.rs"] mod breakout_real; use breakout_real::*; -/// ns → ms: the shipped `open_ohlc_sources` opens by Unix-ms, but the World -/// `WindowBounds` are epoch-ns. `aura-ingest` exports only the forward direction -/// (`unix_ms_to_epoch_ns`), so the inverse is a local hand-divide (#80, filed). -fn ns_to_ms(ts: Timestamp) -> i64 { - ts.0 / 1_000_000 -} - /// Bootstrap the shipped blueprint at one `{entry_bar, exit_bar}` point, run it -/// over `[from_ms, to_ms]` of real GER40 OHLC, and fold the recorded equity/held -/// taps into a `RunReport` (`summarize` over the drained channels — exactly as -/// the SMA-cross sweep fixture and the fieldtest's `run_point` do). A fresh -/// blueprint per call (fresh nodes + channels) keeps the disjoint sweep points -/// independent (C1). -fn run_point(server: &Arc, point: &[Cell], from_ms: i64, to_ms: i64) -> RunReport { +/// over the epoch-ns [`Timestamp`] window `[from, to]` of real GER40 OHLC, and +/// fold the recorded equity/held taps into a `RunReport` (`summarize` over the +/// drained channels — exactly as the SMA-cross sweep fixture and the fieldtest's +/// `run_point` do). A fresh blueprint per call (fresh nodes + channels) keeps the +/// disjoint sweep points independent (C1). +fn run_point(server: &Arc, point: &[Cell], from: Timestamp, to: Timestamp) -> RunReport { let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin); let mut h = bp .bootstrap_with_cells(point) .expect("sweep point kind-checked against param_space"); - let sources = open_ohlc_sources(server, from_ms, to_ms).expect("window overlaps GER40 data"); + let sources = open_ohlc(server, SYMBOL, from, to).expect("window overlaps GER40 data"); h.run(sources); drop(h); let equity: Vec<(Timestamp, f64)> = taps.equity.try_iter().collect::>() @@ -66,10 +59,7 @@ fn run_point(server: &Arc, point: &[Cell], from_ms: i64, to_ms: i64) manifest: RunManifest { commit: "ger40-breakout-world".to_string(), params: vec![], - window: ( - aura_ingest::unix_ms_to_epoch_ns(from_ms), - aura_ingest::unix_ms_to_epoch_ns(to_ms), - ), + window: (from, to), seed: 0, broker: "sim-optimal(pip_size=1)".to_string(), }, @@ -91,13 +81,13 @@ fn scalar_f64(s: &Scalar) -> f64 { /// `param_space()` directly, so the two `EqConst` targets are the swept axes. #[test] fn sweep_consumes_blueprint_over_named_params() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)"); return; } - let (from_ms, to_ms) = utc_month_window_ms(2024, 9); + let (from, to) = utc_month_window(2024, 9); let space = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin) .0 .param_space(); @@ -117,7 +107,7 @@ fn sweep_consumes_blueprint_over_named_params() { let server_for_closure = Arc::clone(&server); let family = sweep(&grid, |pt: &[Cell]| { - run_point(&server_for_closure, pt, from_ms, to_ms) + run_point(&server_for_closure, pt, from, to) }); // A full 3×3 grid was enumerated over the two named params (no re-authoring). @@ -152,7 +142,7 @@ fn sweep_consumes_blueprint_over_named_params() { /// over the raw FlatGraph, where `chosen_params == vec![]` everywhere). #[test] fn walk_forward_consumes_blueprint_non_degenerate() { - let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + let server = default_data_server(); if !server.has_symbol(SYMBOL) { eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)"); return; @@ -201,9 +191,8 @@ fn walk_forward_consumes_blueprint_non_degenerate() { .collect(), ) .expect("IS grid well-formed"); - let (is_from, is_to) = (ns_to_ms(w.is.0), ns_to_ms(w.is.1)); let is_family = sweep(&is_grid, |pt: &[Cell]| { - run_point(&server_for_closure, pt, is_from, is_to) + run_point(&server_for_closure, pt, w.is.0, w.is.1) }); let best = is_family .points @@ -219,13 +208,12 @@ fn walk_forward_consumes_blueprint_non_degenerate() { let chosen = best.params.clone(); // Apply the chosen params on the OOS window. - let (oos_from, oos_to) = (ns_to_ms(w.oos.0), ns_to_ms(w.oos.1)); - let oos_report = run_point(&server_for_closure, &chosen, oos_from, oos_to); + let oos_report = run_point(&server_for_closure, &chosen, w.oos.0, w.oos.1); // Re-run once on OOS to capture the recorded equity segment for stitching. let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin); let mut h = bp.bootstrap_with_cells(&chosen).expect("chosen point kind-checked"); - let sources = open_ohlc_sources(&server_for_closure, oos_from, oos_to) + let sources = open_ohlc(&server_for_closure, SYMBOL, w.oos.0, w.oos.1) .expect("OOS window overlaps data"); h.run(sources); drop(h); diff --git a/crates/aura-ingest/tests/open_ohlc_absent_archive.rs b/crates/aura-ingest/tests/open_ohlc_absent_archive.rs new file mode 100644 index 0000000..17a58d5 --- /dev/null +++ b/crates/aura-ingest/tests/open_ohlc_absent_archive.rs @@ -0,0 +1,102 @@ +//! Hermetic coverage for the Task-1 OHLC source-open surface (#80/#81/#92): the +//! documented *file-level* `None` contract of the canonical openers, and the #81 +//! re-export firewall (a real-data source builds from `aura-ingest` alone). Every +//! item here is reached through `aura_ingest` ONLY — this test never names the +//! external `data_server` crate, which is the firewall it pins. +//! +//! Unlike the gated GER40 tests (which skip without a local Pepperstone archive), +//! these run everywhere: they drive a `DataServer` over a guaranteed-EMPTY base +//! path, so no archived file overlaps any window and the openers take their +//! documented "no data source present" branch deterministically — same input, +//! same `None`, every run, on any machine. + +use std::sync::Arc; + +// Firewall (#81): the ENTIRE source-open stack is imported from `aura_ingest`. +// `DataServer` / `DEFAULT_DATA_PATH` are re-exports; if a future change drops +// the re-export, this `use` stops compiling — that is the test. +use aura_core::Timestamp; +use aura_ingest::{ + default_data_server, open_ohlc, DataServer, M1Field, M1FieldSource, DEFAULT_DATA_PATH, +}; + +/// A `DataServer` over a base path that is guaranteed not to exist, so its symbol +/// index is empty on every machine (data-server's directory scan returns `Err` +/// for an absent path and yields no symbols). Gives a deterministic "no archive" +/// server with no filesystem writes and no dependence on local data. +fn empty_server() -> Arc { + // A path component that cannot exist as a real archive directory; uniqueness + // is irrelevant (an absent path scans empty regardless), but keep it obviously + // synthetic so an accidental match is impossible. + let absent = std::env::temp_dir().join("aura-ingest-no-such-archive-DO-NOT-CREATE"); + Arc::new(DataServer::new(absent)) +} + +/// Property: `open_ohlc` returns the *file-level* `None` (the "no data source to +/// read from" branch) when no archived file overlaps the window — distinct from +/// the "overlaps a file but holds zero bars" case, which would return +/// `Some(empty)`. Pins the documented Option semantics of the canonical OHLC +/// bundle opener (#80/#92) so a future refactor cannot silently turn "no source" +/// into a panic or an empty `Some`. +#[test] +fn open_ohlc_yields_file_level_none_when_no_archive_overlaps() { + let server = empty_server(); + let from = Timestamp(0); + let to = Timestamp(1_000_000_000_000_000_000); // ~2001, well-formed epoch-ns window + assert!( + open_ohlc(&server, "GER40", from, to).is_none(), + "open_ohlc over an empty archive must return the file-level None (no source present)", + ); +} + +/// Property: `M1FieldSource::open_window` — the engine-side epoch-ns `Timestamp` +/// mirror of the ms-path `open` — threads its `Timestamp` bounds through the +/// seam-private ms<->ns crossing and propagates data-server's file-level `None` +/// for an absent symbol, without panicking on the ns->ms divide. Pins that the +/// Timestamp opener is a faithful, total mirror of `open` on the "no source" +/// branch (the C3 one-crossing seam, #80). +#[test] +fn open_window_propagates_file_level_none_for_absent_symbol() { + let server = empty_server(); + let from = Timestamp(0); + let to = Timestamp(1_000_000_000_000_000_000); + assert!( + M1FieldSource::open_window(&server, "GER40", Some(from), Some(to), M1Field::Close) + .is_none(), + "open_window over an empty archive must propagate the file-level None", + ); + // An open-ended window (None bounds) takes the same no-source branch — the + // ns->ms mapping is a no-op on None, so the openness is preserved. + assert!( + M1FieldSource::open_window(&server, "GER40", None, None, M1Field::Close).is_none(), + "open-ended open_window over an empty archive must also propagate None", + ); +} + +/// Property: the #81 source-open surface is reachable from `aura_ingest` alone — +/// `default_data_server()` constructs the canonical-path server and the +/// re-exported `DEFAULT_DATA_PATH` is the path it points at, with the external +/// `data_server` crate never named by a consumer. Observable: the convenience +/// constructor agrees with constructing `DataServer::new(DEFAULT_DATA_PATH)` by +/// hand on the symbol-presence query (both report the same absence of a synthetic +/// symbol that no real archive can hold), so the convenience is a faithful alias. +#[test] +fn default_data_server_builds_from_ingest_alone_over_canonical_path() { + let convenience = default_data_server(); + // The re-export is reachable and is what the convenience constructor uses. + let by_hand = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + // A symbol no real Pepperstone archive holds: both servers agree it is absent, + // independent of whether the local archive exists — so the convenience + // constructor is a faithful alias for `DataServer::new(DEFAULT_DATA_PATH)`, + // and the whole stack was built without naming `data_server`. + const SYNTHETIC_ABSENT: &str = "AURA-INGEST-SYNTHETIC-NONEXISTENT-SYMBOL"; + assert_eq!( + convenience.has_symbol(SYNTHETIC_ABSENT), + by_hand.has_symbol(SYNTHETIC_ABSENT), + "default_data_server() must agree with DataServer::new(DEFAULT_DATA_PATH)", + ); + assert!( + !convenience.has_symbol(SYNTHETIC_ABSENT), + "a synthetic symbol no archive holds must be absent", + ); +} diff --git a/crates/aura-ingest/tests/open_ohlc_seam.rs b/crates/aura-ingest/tests/open_ohlc_seam.rs new file mode 100644 index 0000000..110c52c --- /dev/null +++ b/crates/aura-ingest/tests/open_ohlc_seam.rs @@ -0,0 +1,71 @@ +//! Gated behaviour-preservation proof for the consolidated OHLC opener (#80/#92): +//! the library `open_ohlc` over an epoch-ns `Timestamp` window produces a recorded +//! `(held, equity)` series BIT-IDENTICAL to opening the four `M1FieldSource`s by +//! hand over the equivalent Unix-ms window via the untouched ms-path `open`. Same +//! instants, same bars, same numbers — the migration changes the surface, not the +//! behaviour. +//! +//! Mirrors the other gated GER40 tests: skips with a note where the local +//! Pepperstone archive is absent, so `cargo test --workspace` stays green +//! anywhere; exercises the real path where the files exist. + +use std::sync::Arc; + +use aura_engine::{Harness, Source}; +use chrono::TimeZone; +use data_server::{DataServer, DEFAULT_DATA_PATH}; + +use aura_ingest::{open_ohlc, M1Field, M1FieldSource}; + +#[path = "../examples/shared/breakout_real.rs"] +mod breakout_real; +use breakout_real::*; + +/// Run a fresh harness over `sources` and drain the recorded `(held, equity)` +/// series (the same fold the determinism tests use). +fn series_from(mut h: Harness, taps: &Taps, sources: Vec>) -> Vec<(f64, f64)> { + h.run(sources); + drop(h); + let trace = drain_trace(taps); + trace.iter().map(|b| (b.held, b.equity)).collect() +} + +/// Property: the Timestamp `open_ohlc` path and the hand-rolled ms-path `open` +/// (×4, in O/H/L/C order) over the equivalent window yield bit-identical recorded +/// series — behaviour-preserving consolidation (AC5). +#[test] +fn open_ohlc_timestamp_path_matches_ms_path_open() { + let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + if !server.has_symbol(SYMBOL) { + eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)"); + return; + } + + // September 2024 (inclusive UTC), the window the other gated tests pin. + let from_ms = chrono::Utc.with_ymd_and_hms(2024, 9, 1, 0, 0, 0).unwrap().timestamp_millis(); + let to_ms = chrono::Utc.with_ymd_and_hms(2024, 9, 30, 23, 59, 59).unwrap().timestamp_millis(); + let from_ts = aura_ingest::unix_ms_to_epoch_ns(from_ms); + let to_ts = aura_ingest::unix_ms_to_epoch_ns(to_ms); + + // Path A: the consolidated Timestamp opener. + let (h_a, taps_a) = build_harness(); + let sources_a = open_ohlc(&server, SYMBOL, from_ts, to_ts).expect("window overlaps GER40 data"); + let series_a = series_from(h_a, &taps_a, sources_a); + assert!(!series_a.is_empty(), "window resolved to zero bars"); + + // Path B: the untouched ms-path open, four fields by hand in O/H/L/C order. + let fields = [M1Field::Open, M1Field::High, M1Field::Low, M1Field::Close]; + let mut sources_b: Vec> = Vec::with_capacity(4); + for &f in &fields { + let s = M1FieldSource::open(&server, SYMBOL, Some(from_ms), Some(to_ms), f) + .expect("window overlaps GER40 data"); + sources_b.push(Box::new(s)); + } + let (h_b, taps_b) = build_harness(); + let series_b = series_from(h_b, &taps_b, sources_b); + + assert_eq!( + series_a, series_b, + "open_ohlc (Timestamp) must reproduce the ms-path open bit-identically (behaviour-preserving, #80/#92)", + ); +}