Aura/crates/aura-ingest/tests/real_bars.rs

//! Gated integration test: a real data-server M1 close stream driven through
//! the cycle-0007 signal-quality sample harness (SMA-cross → Bias →
//! 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, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target,
    VecSource,
};
use aura_ingest::load_m1_window;
use aura_std::{Bias, 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 f64_recorder_sig = || NodeSchema {
        inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
        output: vec![],
        params: vec![],
    };
    let mut h = Harness::bootstrap(FlatGraph {
        nodes: vec![
            Box::new(Sma::new(2)),                                            // 0
            Box::new(Sma::new(4)),                                            // 1
            Box::new(Sub::new()),                                             // 2
            Box::new(Bias::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
        ],
        signatures: vec![
            Sma::builder().schema().clone(),
            Sma::builder().schema().clone(),
            Sub::builder().schema().clone(),
            Bias::builder().schema().clone(),
            SimBroker::builder(0.0001).schema().clone(),
            f64_recorder_sig(),
            f64_recorder_sig(),
        ],
        sources: 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
            ],
        }],
        edges: 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![Box::new(VecSource::new(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(), Scalar::i64(2)),
                ("sma_slow".to_string(), Scalar::i64(4)),
                ("bias_scale".to_string(), Scalar::f64(0.5)),
            ],
            window,
            seed: 0,
            broker: "sim-optimal(pip_size=0.0001)".to_string(),
            selection: None,
            instrument: None,
            topology_hash: None,
        },
        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", Some(from_ms), Some(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());
}