Aura/crates/aura-ingest/src/lib.rs

//! aura's first real data source: the C3/C7 ingestion boundary.
//!
//! Transposes [`data_server`]'s Array-of-Structs `M1Parsed` records into aura's
//! Structure-of-Arrays base columns (C7) and normalizes data-server's
//! Unix-millisecond time to aura's canonical epoch-nanosecond [`Timestamp`] at
//! this one boundary (C3). A transposed [`M1Columns`] exposes its close column
//! as the engine source-stream shape (`Vec<(Timestamp, Scalar)>`) the SMA-cross
//! sample strategy consumes.
//!
//! Two ingestion shapes coexist: the **eager** [`load_m1_window`] → [`M1Columns`]
//! transpose (a bounded window materialized to owned SoA columns), and the **lazy
//! streaming** [`M1FieldSource`] — a `Source` that pulls one data-server
//! `Arc<[M1Parsed]>` chunk at a time and decodes each `Scalar` on demand
//! (resident O(one chunk), not O(window length)) for backtests over long
//! horizons. Both normalize ms→epoch-ns through the one [`unix_ms_to_epoch_ns`]
//! seam (C3).
//!
//! This crate is the **data-source ingestion edge**: it links `data-server`
//! (and its transitive `chrono`/`regex`/`zip`) and normalizes it into aura's
//! columns here. Dependencies follow the amended C16 per-case policy (INDEX.md),
//! not a blanket zero-dependency wall.

use aura_core::{Scalar, Timestamp};
use data_server::records::M1Parsed;
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
/// [`Timestamp`] — the single unit normalization of C3, performed at the one
/// ingestion boundary and nowhere else. The `i64` epoch-ns range covers all
/// market data through year ~2262, well beyond any real file.
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
/// columns share an index: `ts[i]` is the timestamp of `close[i]`, etc.
#[derive(Clone, Debug, PartialEq)]
pub struct M1Columns {
    pub ts: Vec<Timestamp>,
    pub open: Vec<f64>,
    pub high: Vec<f64>,
    pub low: Vec<f64>,
    pub close: Vec<f64>,
    pub spread: Vec<f64>,
    pub volume: Vec<i64>,
}

impl M1Columns {
    /// Pre-size all columns for `n` bars.
    fn with_capacity(n: usize) -> Self {
        Self {
            ts: Vec::with_capacity(n),
            open: Vec::with_capacity(n),
            high: Vec::with_capacity(n),
            low: Vec::with_capacity(n),
            close: Vec::with_capacity(n),
            spread: Vec::with_capacity(n),
            volume: Vec::with_capacity(n),
        }
    }

    /// Append one chunk of AoS bars, transposing into these SoA columns and
    /// normalizing time ms->epoch-ns at the boundary (C3). Reserves per chunk so
    /// repeated appends grow amortized; the push order is preserved (the
    /// chronological merge order C3 relies on). Lets `load_m1_window` transpose
    /// chunk-by-chunk without an intermediate AoS buffer.
    fn extend_from_bars(&mut self, bars: &[M1Parsed]) {
        let n = bars.len();
        self.ts.reserve(n);
        self.open.reserve(n);
        self.high.reserve(n);
        self.low.reserve(n);
        self.close.reserve(n);
        self.spread.reserve(n);
        self.volume.reserve(n);
        for b in bars {
            self.ts.push(unix_ms_to_epoch_ns(b.time_ms));
            self.open.push(b.open);
            self.high.push(b.high);
            self.low.push(b.low);
            self.close.push(b.close);
            self.spread.push(b.spread);
            self.volume.push(b.volume);
        }
    }

    /// 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 SMA-cross sample strategy consumes; a project
    /// that wants another field reads the public columns 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()
    }
}

/// 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());
    c.extend_from_bars(bars);
    c
}

/// Drain a data-server M1 window into transposed SoA columns. Reads the
/// chronological chunk iterator to exhaustion, transposing each chunk directly
/// into the SoA columns at the boundary (C3 — one merge point; no intermediate
/// AoS buffer, so peak memory is the columns alone, not columns + an AoS copy).
///
/// Each bound is an `Option<i64>` of inclusive Unix-ms (data-server's window
/// contract): `None` means unbounded on that side (data-server skips the
/// coarse `unix_ms_to_year_month` file filter rather than reaching for a
/// sentinel). `from_ms = None` reads from the start of history; `to_ms = None`
/// reads to the end. This threads data-server's own per-bound `Option`
/// contract through unchanged, so "to the end of history" is expressible
/// without an `i64::MAX` sentinel (which panics upstream in chrono's
/// `from_timestamp_millis`).
///
/// The returned `Option` is **file-level**, propagating data-server's own
/// `None`: it is `None` only when no archived file overlaps the window (e.g. a
/// far-future window, or an unknown symbol) — i.e. there is no data source to
/// read from. A window that *does* overlap a loaded file but happens to hold
/// zero bars (a narrow gap inside coverage) returns `Some(M1Columns)` with
/// empty columns, **not** `None`. So `Some`/`None` distinguishes "data source
/// present" from "nothing to read from", not "has bars" from "has none": a
/// consumer testing for "no bars in this window" must check
/// `cols.close.is_empty()`, not the `Option` alone.
pub fn load_m1_window(
    server: &Arc<DataServer>,
    symbol: &str,
    from_ms: Option<i64>,
    to_ms: Option<i64>,
) -> Option<M1Columns> {
    let mut it = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
    // Transpose each chunk straight into the SoA columns — no intermediate AoS
    // buffer (halves peak load memory, drops one full copy). The per-chunk push
    // order is the chronological merge order (C3).
    let mut cols = M1Columns::with_capacity(0);
    // M1Parsed is Copy; &Arc<[M1Parsed]> derefs to &[M1Parsed] in arg position.
    while let Some(chunk) = it.next_chunk() {
        cols.extend_from_bars(&chunk);
    }
    Some(cols)
}

/// Which base column of an M1 bar a source streams (C7: a composite window is a
/// bundle of base columns; one `Source` per consumed field).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum M1Field {
    Open,
    High,
    Low,
    Close,
    Spread,
    Volume,
}

/// Project one M1 bar into `(normalized ts, scalar)` for `field`. Pure: a
/// function of `(field, bar)` only — no iterator, no clock — so it is testable
/// in isolation. Time is normalized ms→epoch-ns at this one seam (C3).
fn decode(field: M1Field, bar: &M1Parsed) -> (Timestamp, Scalar) {
    let value = match field {
        M1Field::Open => Scalar::f64(bar.open),
        M1Field::High => Scalar::f64(bar.high),
        M1Field::Low => Scalar::f64(bar.low),
        M1Field::Close => Scalar::f64(bar.close),
        M1Field::Spread => Scalar::f64(bar.spread),
        M1Field::Volume => Scalar::i64(bar.volume),
    };
    (unix_ms_to_epoch_ns(bar.time_ms), value)
}

/// A streaming [`Source`](aura_engine::Source) over a data-server M1 window. Holds
/// at most one `Arc` chunk (a zero-copy clone of the cache's chunk) and a cursor;
/// constructs each `Scalar` per-pull; refills via `next_chunk()` when the chunk
/// drains. The **source**'s resident footprint is O(one chunk), independent of
/// window length (see [`resident_records`](aura_engine::Source::resident_records)) — a per-source
/// bound, not whole-process RSS: the data-server cache below retains the loaded
/// window's parsed chunks for the pass (the replay-many sharing model, #95).
pub struct M1FieldSource {
    iter: SymbolChunkIter<M1Parsed>,
    chunk: Option<Arc<[M1Parsed]>>,
    pos: usize,
    field: M1Field,
    head: Option<(Timestamp, Scalar)>,
    /// The requested window (inclusive Unix-ms), kept so `bounds()` can report a
    /// producer-supplied extent without streaming (#71). `None` = open-ended.
    from_ms: Option<i64>,
    to_ms: Option<i64>,
}

impl M1FieldSource {
    /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract).
    /// `None` when no archived file overlaps the window (unknown symbol /
    /// far-future window) — propagating `stream_m1_windowed`'s file-level Option.
    /// A window that overlaps a file but holds zero matching bars yields a source
    /// whose first `peek` is `None` (immediately exhausted), not `None` here.
    pub fn open(
        server: &Arc<DataServer>,
        symbol: &str,
        from_ms: Option<i64>,
        to_ms: Option<i64>,
        field: M1Field,
    ) -> Option<Self> {
        let iter = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
        let mut s = Self { iter, chunk: None, pos: 0, field, head: None, from_ms, to_ms };
        s.advance();
        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<DataServer>,
        symbol: &str,
        from: Option<Timestamp>,
        to: Option<Timestamp>,
        field: M1Field,
    ) -> Option<Self> {
        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) {
        loop {
            if self.chunk.is_none() {
                self.chunk = self.iter.next_chunk();
                self.pos = 0;
                if self.chunk.is_none() {
                    self.head = None;
                    return;
                }
            }
            let chunk = self.chunk.as_ref().unwrap();
            if self.pos < chunk.len() {
                self.head = Some(decode(self.field, &chunk[self.pos]));
                return;
            }
            self.chunk = None; // current chunk drained — loop to refill
        }
    }

}

impl aura_engine::Source for M1FieldSource {
    /// Records resident in this source right now: the current chunk's length (or
    /// 0 at exhaustion). The ring-residency probe (#95) — bounded by one chunk
    /// length by construction (the source holds at most one `Arc` chunk and no
    /// accumulating field), so it can never grow with window length; always `Some`
    /// (this source reports). Probes the **source ring** only; the data-server
    /// `FileCache` below retains the loaded window's parsed chunks for the pass and
    /// is not counted here.
    fn resident_records(&self) -> Option<usize> {
        Some(self.chunk.as_ref().map_or(0, |c| c.len()))
    }

    fn peek(&self) -> Option<Timestamp> {
        self.head.map(|(t, _)| t)
    }
    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
        let item = self.head?;
        self.pos += 1;
        self.advance();
        Some(item)
    }
    fn bounds(&self) -> Option<(Timestamp, Timestamp)> {
        // the producer-supplied window: the requested [from_ms, to_ms] normalized
        // to epoch-ns, known without streaming a bar (#71). Open-ended (either
        // bound absent) -> no known extent (Non-goals: archive-extent query).
        match (self.from_ms, self.to_ms) {
            (Some(from), Some(to)) => Some((unix_ms_to_epoch_ns(from), unix_ms_to_epoch_ns(to))),
            _ => None,
        }
    }
}

/// 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<DataServer>` (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<DataServer>,
    symbol: &str,
    from: Timestamp,
    to: Timestamp,
) -> Option<Vec<Box<dyn aura_engine::Source>>> {
    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<DataServer> {
    Arc::new(DataServer::new(DEFAULT_DATA_PATH))
}

/// Per-instrument reference metadata held beside the hot path (C7/C15): non-scalar,
/// never streamed. Minimal today — extend with tick size / digits / quote currency
/// as later cycles need, without breaking this signature.
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct InstrumentSpec {
    /// Price units per pip / point (> 0). The sim-optimal broker's divisor.
    pub pip_size: f64,
}

/// Look up per-instrument reference metadata by `symbol` at the ingestion edge.
/// Rust-authored vetted table (NOT `Aura.toml`); `None` for an instrument with no
/// vetted spec — the caller must refuse rather than guess a pip.
///
/// Seeded with the instruments the engine's own paths exercise. Indices quote in
/// points (pip = 1.0); 5-decimal FX majors = 0.0001. NB: JPY pairs are 0.01, NOT
/// 0.0001 — only vetted values are seeded.
pub fn instrument_spec(symbol: &str) -> Option<InstrumentSpec> {
    let pip_size = match symbol {
        "GER40" | "FRA40" => 1.0,
        "EURUSD" | "GBPUSD" | "USDCAD" => 0.0001,
        _ => return None,
    };
    Some(InstrumentSpec { pip_size })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn instrument_spec_lookup_by_symbol() {
        // index points → pip 1.0
        assert_eq!(instrument_spec("GER40"), Some(InstrumentSpec { pip_size: 1.0 }));
        assert_eq!(instrument_spec("FRA40"), Some(InstrumentSpec { pip_size: 1.0 }));
        // 5-decimal FX major → pip 0.0001
        assert_eq!(instrument_spec("EURUSD"), Some(InstrumentSpec { pip_size: 0.0001 }));
        // un-specced → None (the honesty lever)
        assert_eq!(instrument_spec("NONEXISTENT"), None);
    }

    #[test]
    fn unix_ms_to_epoch_ns_scales_ms_to_ns() {
        // data-server's own epoch fixture: 2017-03-01 00:00 UTC.
        assert_eq!(
            unix_ms_to_epoch_ns(1_488_326_400_000),
            Timestamp(1_488_326_400_000_000_000)
        );
        // epoch maps to epoch.
        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 {
        M1Parsed {
            time_ms,
            open: 1.0,
            high: 2.0,
            low: 0.5,
            close: 1.5,
            spread: 0.1,
            volume: 100,
        }
    }

    #[test]
    fn transpose_m1_maps_every_field_to_its_column() {
        let bars = [full_bar(1_000), full_bar(2_000)];
        let c = transpose_m1(&bars);
        // ts normalized ms -> ns; each column equal length to the input.
        assert_eq!(c.ts, vec![Timestamp(1_000_000_000), Timestamp(2_000_000_000)]);
        assert_eq!(c.open, vec![1.0, 1.0]);
        assert_eq!(c.high, vec![2.0, 2.0]);
        assert_eq!(c.low, vec![0.5, 0.5]);
        assert_eq!(c.close, vec![1.5, 1.5]);
        assert_eq!(c.spread, vec![0.1, 0.1]);
        // volume is the one i64 column.
        assert_eq!(c.volume, vec![100_i64, 100_i64]);
    }

    #[test]
    fn transpose_m1_is_pure() {
        let bars = [full_bar(1_000), full_bar(2_000), full_bar(3_000)];
        assert_eq!(transpose_m1(&bars), transpose_m1(&bars));
    }

    #[test]
    fn transpose_m1_empty_input_yields_empty_columns() {
        let c = transpose_m1(&[]);
        assert!(c.ts.is_empty());
        assert!(c.close.is_empty());
        assert!(c.volume.is_empty());
    }

    #[test]
    fn chunked_accumulation_equals_single_transpose() {
        // `load_m1_window` now transposes chunk-by-chunk into the SoA columns
        // (no intermediate AoS buffer). That must be byte-identical to one
        // transpose over the concatenation — the path it replaced. This pins the
        // shared `extend_from_bars` logic without needing a live DataServer.
        let all = [full_bar(1_000), full_bar(2_000), full_bar(3_000)];
        let single = transpose_m1(&all);
        let mut chunked = M1Columns::with_capacity(0);
        chunked.extend_from_bars(&all[0..2]); // first chunk
        chunked.extend_from_bars(&all[2..3]); // second chunk
        assert_eq!(chunked, single);
    }

    #[test]
    fn close_stream_zips_ts_with_close_in_order() {
        // distinct close per bar so order is observable; ts ascending.
        let bars = [
            M1Parsed { time_ms: 1, open: 0.0, high: 0.0, low: 0.0, close: 10.0, spread: 0.0, volume: 0 },
            M1Parsed { time_ms: 2, open: 0.0, high: 0.0, low: 0.0, close: 20.0, spread: 0.0, volume: 0 },
            M1Parsed { time_ms: 3, open: 0.0, high: 0.0, low: 0.0, close: 30.0, spread: 0.0, volume: 0 },
        ];
        let stream = transpose_m1(&bars).close_stream();
        assert_eq!(
            stream,
            vec![
                (Timestamp(1_000_000), Scalar::f64(10.0)),
                (Timestamp(2_000_000), Scalar::f64(20.0)),
                (Timestamp(3_000_000), Scalar::f64(30.0)),
            ]
        );
        // ascending in timestamp (the merge precondition).
        assert!(stream.windows(2).all(|w| w[0].0 < w[1].0));
    }

    #[test]
    fn close_stream_empty_on_empty_columns() {
        assert!(transpose_m1(&[]).close_stream().is_empty());
    }

    #[test]
    fn decode_projects_each_field_to_its_scalar_kind() {
        // full_bar: open 1.0, high 2.0, low 0.5, close 1.5, spread 0.1, volume 100.
        let bar = full_bar(1_000);
        assert_eq!(decode(M1Field::Open, &bar),   (Timestamp(1_000_000_000), Scalar::f64(1.0)));
        assert_eq!(decode(M1Field::High, &bar),   (Timestamp(1_000_000_000), Scalar::f64(2.0)));
        assert_eq!(decode(M1Field::Low, &bar),    (Timestamp(1_000_000_000), Scalar::f64(0.5)));
        assert_eq!(decode(M1Field::Close, &bar),  (Timestamp(1_000_000_000), Scalar::f64(1.5)));
        assert_eq!(decode(M1Field::Spread, &bar), (Timestamp(1_000_000_000), Scalar::f64(0.1)));
        // volume is the one i64 column.
        assert_eq!(decode(M1Field::Volume, &bar), (Timestamp(1_000_000_000), Scalar::i64(100)));
    }
}