Aura/fieldtests/cycle-0011-ingest/c0011_3_load_window_real.rs

//! Fieldtest c0011 #3 — load_m1_window against real data (axis c).
//!
//! Question under test: a downstream researcher opens the local data-server
//! archive, asks `load_m1_window` for a real symbol/window, gets back SoA
//! `M1Columns`, takes `close_stream()`, and runs an end-to-end signal-quality
//! backtest over REAL recorded AAPL.US 2006 M1 close bars — deterministically.
//! And: the documented `None` path (symbol/window miss) behaves as the rustdoc
//! says.
//!
//! Public-surface facts used (rustdoc + data-server README):
//!  - aura_ingest::load_m1_window(&Arc<DataServer>, symbol, from_ms, to_ms)
//!      -> Option<M1Columns>; "None if the symbol has no data in [from_ms,
//!      to_ms]"; inclusive Unix-ms bounds (data-server's contract).
//!  - data_server::{DataServer, DEFAULT_DATA_PATH}; DataServer::new(base_path);
//!    has_symbol(symbol) -> bool; stream is over Arc<DataServer>.
//!  - close_stream() ascending in timestamp -> Harness::run source shape.
//!
//! Skips cleanly (exit 0, prints SKIP) where the local archive is absent, so it
//! is hermetic on machines without /mnt/tickdata — mirroring the gated
//! integration test's documented behaviour.

use std::path::Path;
use std::sync::{Arc, mpsc};

use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{Edge, Harness, SourceSpec, Target, f64_field, summarize};
use aura_ingest::load_m1_window;
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use data_server::{DEFAULT_DATA_PATH, DataServer};

// AAPL.US 2006-08 is present per the carrier. Window the whole month, inclusive.
const SYMBOL: &str = "AAPL.US";
// 2006-08-01T00:00:00Z .. 2006-09-01T00:00:00Z in Unix-ms.
const FROM_MS: i64 = 1_154_390_400_000;
const TO_MS: i64 = 1_157_068_800_000;

fn main() {
    if !Path::new(DEFAULT_DATA_PATH).exists() {
        println!("c0011_3 SKIP: {DEFAULT_DATA_PATH} absent (no local archive)");
        return;
    }

    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if !server.has_symbol(SYMBOL) {
        println!("c0011_3 SKIP: symbol {SYMBOL} not in local archive");
        return;
    }

    // ---- The happy path: load a real M1 window. ----------------------------
    let cols = match load_m1_window(&server, SYMBOL, FROM_MS, TO_MS) {
        Some(c) => c,
        None => {
            println!("c0011_3 SKIP: {SYMBOL} present but no bars in the 2006-08 window");
            return;
        }
    };
    let n = cols.close.len();
    println!("loaded {n} real M1 bars for {SYMBOL} in 2006-08");
    assert!(n > 0, "a present symbol+window yields at least one bar");

    // SoA column lengths are all aligned (C7: one index across columns).
    assert_eq!(cols.ts.len(), n, "ts column length matches close");
    assert_eq!(cols.open.len(), n, "open length");
    assert_eq!(cols.high.len(), n, "high length");
    assert_eq!(cols.low.len(), n, "low length");
    assert_eq!(cols.spread.len(), n, "spread length");
    assert_eq!(cols.volume.len(), n, "volume (i64) length");

    // Timestamps normalized to epoch-ns and inside the requested window
    // (the boundary normalizes Unix-ms -> ns; the window bounds are Unix-ms).
    let from_ns = Timestamp(FROM_MS * 1_000_000);
    let to_ns = Timestamp(TO_MS * 1_000_000);
    for (i, t) in cols.ts.iter().enumerate() {
        assert!(*t >= from_ns && *t <= to_ns, "bar {i} ts {t:?} inside the normalized window");
    }

    // ---- close_stream is ascending (the C3 precondition Harness::run needs). -
    let price_stream: Vec<(Timestamp, Scalar)> = cols.close_stream();
    assert_eq!(price_stream.len(), n, "one close-stream entry per bar");
    for w in price_stream.windows(2) {
        assert!(w[0].0 <= w[1].0, "real close_stream non-decreasing in timestamp (C3)");
    }
    println!(
        "first bar = (ts {:?}, close {:?}), last = (ts {:?}, close {:?})",
        price_stream[0].0, price_stream[0].1,
        price_stream[n - 1].0, price_stream[n - 1].1,
    );

    // ---- End-to-end signal-quality run over REAL close bars. ---------------
    let report1 = run_signal_quality(&price_stream);
    let report2 = run_signal_quality(&price_stream); // disjoint second run
    println!("metrics over real AAPL.US 2006-08 close bars = {report1:?}");

    assert!(report1.0.is_finite(), "total_pips finite over real bars");
    assert!(report1.1.is_finite() && report1.1 >= 0.0, "max_drawdown finite, >= 0 over real bars");
    // Determinism (C1): same recorded input -> bit-identical metrics.
    assert_eq!(report1, report2, "two runs over the same real window are bit-identical (C1)");

    // ---- The documented None path: a symbol with no data in the window. ----
    // A far-future window for a present symbol -> data-server's own None.
    let future_from = 30_000_000_000_000_i64; // ~year 2920 in Unix-ms
    let future_to = 30_000_086_400_000_i64;
    let miss = load_m1_window(&server, SYMBOL, future_from, future_to);
    assert!(miss.is_none(), "present symbol, empty window -> None (rustdoc)");
    println!("None-path: {SYMBOL} with a far-future window -> {miss:?}");

    // An unknown symbol -> None as well.
    let miss_sym = load_m1_window(&server, "NO.SUCH.SYMBOL_XYZ", FROM_MS, TO_MS);
    assert!(miss_sym.is_none(), "unknown symbol -> None (rustdoc / data-server contract)");
    println!("None-path: unknown symbol -> {miss_sym:?}");

    // ---- The grey zone: a window INSIDE the file's date coverage but with no
    //      bars in it. The rustdoc says "None if the symbol has no data in
    //      [from_ms, to_ms]". A narrow single-day window inside 2006-08 (whose
    //      file IS loaded) but containing zero bars returns Some(EMPTY columns),
    //      NOT None — data-server's `None` is FILE-level (no file overlaps the
    //      window, skipped without I/O), while a loaded file that yields no bar
    //      in the exact sub-window transposes to Some(empty). Recorded as a
    //      spec_gap: the Some(empty) vs None boundary is not crisp on the public
    //      surface. We assert the OBSERVED reading.
    let one_day_from = FROM_MS; // 2006-08-01T00:00Z
    let one_day_to = FROM_MS + 86_400_000; // +1 day
    let narrow = load_m1_window(&server, SYMBOL, one_day_from, one_day_to);
    match &narrow {
        Some(c) if c.close.is_empty() => {
            println!("grey-zone: in-coverage empty sub-window -> Some(EMPTY), not None");
        }
        Some(c) => println!("grey-zone: in-coverage sub-window -> Some({} bars)", c.close.len()),
        None => println!("grey-zone: in-coverage empty sub-window -> None"),
    }
    // Observed on this archive: Some(empty). Asserting the observed reading so
    // the fixture is a faithful record (a future tightening would flip this).
    assert!(
        matches!(&narrow, Some(c) if c.close.is_empty()),
        "in-coverage empty sub-window returns Some(empty), not None (observed reading)"
    );

    println!("c0011_3 OK: load_m1_window drove a deterministic end-to-end run over real bars; None paths hold");
}

/// SMA-cross -> Exposure -> SimBroker -> Recorder; returns (total_pips,
/// max_drawdown, exposure_sign_flips) as the summarized metrics tuple.
fn run_signal_quality(price_stream: &[(Timestamp, Scalar)]) -> (f64, f64, u64) {
    let (eq_tx, eq_rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
    let (exp_tx, exp_rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();

    let mut h = Harness::bootstrap(
        vec![
            Box::new(Sma::new(10)),
            Box::new(Sma::new(30)),
            Box::new(Sub::new()),
            Box::new(Exposure::new(1.0)),
            Box::new(SimBroker::new(0.0001)), // AAPL.US pip_size; reference metadata
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, eq_tx)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, exp_tx)),
        ],
        vec![SourceSpec {
            kind: ScalarKind::F64,
            targets: vec![
                Target { node: 0, slot: 0 },
                Target { node: 1, slot: 0 },
                Target { node: 4, slot: 1 },
            ],
        }],
        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: 3, to: 6, slot: 0, from_field: 0 },
            Edge { from: 4, to: 5, slot: 0, from_field: 0 },
        ],
    )
    .expect("valid real-bars signal-quality DAG");

    h.run(vec![price_stream.to_vec()]);
    let equity = f64_field(&eq_rx.try_iter().collect::<Vec<_>>(), 0);
    let exposure = f64_field(&exp_rx.try_iter().collect::<Vec<_>>(), 0);
    let m = summarize(&equity, &exposure);
    (m.total_pips, m.max_drawdown, m.exposure_sign_flips)
}