Aura/fieldtests/milestone-runway/rw_2_open_ohlc_real.rs

// Runway milestone fieldtest — scenario 2: the real GER40 M1 ingestion path via
// the CANONICAL 4-source OHLC opener (#80/#81/#92).
//
// The task a downstream researcher does: "open the real GER40 M1 archive and run
// my SMA-cross-on-close strategy against real bars, to completion — and I want to
// reach for nothing but `aura-ingest` to do it."
//
// What this exercises, top-down from the milestone promise:
//   - #81: a real-data source builds from `aura-ingest` ALONE. This file imports
//     NOTHING from `data-server`; it uses aura_ingest::{default_data_server,
//     DEFAULT_DATA_PATH, open_ohlc}. The Cargo.toml has no `data-server` dep.
//   - #92: `open_ohlc` is the single vetted home of the C4 merge order
//     (open, high, low, close). A consumer never spells the order out.
//   - #80: M1FieldSource opens a timestamp window; here via the OHLC opener over
//     a closed epoch-ns [from, to].
//
// PUBLIC INTERFACE ONLY: API from `cargo doc` rustdoc + the design ledger
// (C3/C4 ingestion, C12 source seam, C7 SoA). No crates/*/src was read.

use std::sync::mpsc;

use aura_core::{Cell, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    f64_field, summarize, BlueprintNode, Composite, Edge, OutField, Role, Source, Target,
};
use aura_ingest::{default_data_server, instrument_spec, open_ohlc, DEFAULT_DATA_PATH};
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};

const SYMBOL: &str = "GER40";

fn sma_cross() -> Composite {
    Composite::new(
        "sma_cross",
        vec![
            Sma::builder().named("fast").into(),
            Sma::builder().named("slow").into(),
            Sub::builder().into(),
        ],
        vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 },
            Edge { from: 1, to: 2, slot: 1, from_field: 0 },
        ],
        vec![Role {
            name: "price".into(),
            targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            source: None,
        }],
        vec![OutField { node: 2, field: 0, name: "out".into() }],
    )
}

// Harness with FOUR source-roles in the C4 merge order open/high/low/close, to
// mirror what open_ohlc hands back. The strategy reads CLOSE (the canonical price
// field the SMA-cross consumes). open/high/low are routed into per-field recording
// taps, so this run proves all four real OHLC streams flow to completion — not
// just close. Equity (close-driven) is recorded too.
#[allow(clippy::type_complexity)]
fn ohlc_harness() -> (
    Composite,
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>, // equity (close strategy)
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>, // open tap
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>, // high tap
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>, // low tap
) {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_o, rx_o) = mpsc::channel();
    let (tx_h, rx_h) = mpsc::channel();
    let (tx_l, rx_l) = mpsc::channel();

    // node indices:
    // 0 sma_cross (close), 1 Exposure, 2 SimBroker, 3 equity-rec,
    // 4 open-rec, 5 high-rec, 6 low-rec
    let bp = Composite::new(
        "ohlc_harness",
        vec![
            BlueprintNode::Composite(sma_cross()),
            Exposure::builder().into(),
            SimBroker::builder(1.0).into(), // GER40 index pip = 1.0
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_o).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_h).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_l).into(),
        ],
        vec![
            Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // sma_cross -> exposure
            Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker.slot0
            Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // broker equity -> rec
        ],
        // FOUR source-roles in C4 merge order: open, high, low, close. The opener
        // hands back Box<dyn Source> in exactly this order; run() maps the source
        // vec positionally onto the source-roles in declaration order.
        vec![
            Role {
                name: "open".into(),
                targets: vec![Target { node: 4, slot: 0 }],
                source: Some(ScalarKind::F64),
            },
            Role {
                name: "high".into(),
                targets: vec![Target { node: 5, slot: 0 }],
                source: Some(ScalarKind::F64),
            },
            Role {
                name: "low".into(),
                targets: vec![Target { node: 6, slot: 0 }],
                source: Some(ScalarKind::F64),
            },
            Role {
                name: "close".into(),
                // close feeds the strategy AND the broker's price slot (slot 1).
                targets: vec![
                    Target { node: 0, slot: 0 },
                    Target { node: 2, slot: 1 },
                ],
                source: Some(ScalarKind::F64),
            },
        ],
        vec![],
    );
    (bp, rx_eq, rx_o, rx_h, rx_l)
}

fn ms(t: Timestamp) -> i64 {
    t.0 / 1_000_000
}

fn main() {
    // #81: the data server builds from aura-ingest alone, over the local archive.
    let server = default_data_server();
    println!("default_data_server() over DEFAULT_DATA_PATH = {DEFAULT_DATA_PATH}");

    // The pip the milestone vouches for: a vetted index quotes in points (1.0).
    match instrument_spec(SYMBOL) {
        Some(spec) => println!("instrument_spec({SYMBOL}) = pip_size {}", spec.pip_size),
        None => {
            eprintln!("{SYMBOL} is un-specced — a real run must refuse; aborting fixture");
            std::process::exit(1);
        }
    }

    // A ~2-month window of GER40 starting 2015-01-01 (epoch-ns; open_ohlc's unit).
    let day_ns: i64 = 86_400_000 * 1_000_000;
    let from = Timestamp(1_420_070_400_000 * 1_000_000); // 2015-01-01 UTC
    let to = Timestamp(from.0 + 60 * day_ns);

    // #92: the SINGLE vetted home of the C4 merge order. We never spell out
    // open/high/low/close — the opener does, and hands back exactly four sources.
    let sources: Vec<Box<dyn Source>> = match open_ohlc(&server, SYMBOL, from, to) {
        Some(s) => s,
        None => {
            eprintln!("no local {SYMBOL} archive overlapping the window; skipping");
            std::process::exit(0);
        }
    };
    println!(
        "open_ohlc({SYMBOL}, [{}..{}]) -> {} sources (expect 4: O,H,L,C)",
        ms(from),
        ms(to),
        sources.len(),
    );
    assert_eq!(sources.len(), 4, "open_ohlc vouches for four OHLC sources");

    // Each source reports its data extent without materializing (C18/#71).
    for (i, s) in sources.iter().enumerate() {
        let field = ["open", "high", "low", "close"][i];
        let b = s.bounds();
        println!(
            "  source[{i}] {field}: bounds {:?}",
            b.map(|(f, t)| (ms(f), ms(t)))
        );
    }

    let (bp, rx_eq, rx_o, rx_h, rx_l) = ohlc_harness();
    println!("\nharness param-space slots:");
    let space = bp.param_space();
    for ps in &space {
        println!("  {} : {:?}", ps.name, ps.kind);
    }

    // Bootstrap with concrete params: fast=3, slow=20, exposure scale=1.0.
    let params: Vec<Cell> = vec![Cell::from_i64(3), Cell::from_i64(20), Cell::from_f64(1.0)];
    let mut h = bp
        .bootstrap_with_cells(&params)
        .expect("chosen params kind-check against param_space");

    // Run to completion over the four REAL OHLC streams, k-way-merged by ts (C3/C4).
    h.run(sources);
    drop(h);

    let eq_rows: Vec<_> = rx_eq.try_iter().collect();
    let o_rows: Vec<_> = rx_o.try_iter().collect();
    let hi_rows: Vec<_> = rx_h.try_iter().collect();
    let lo_rows: Vec<_> = rx_l.try_iter().collect();

    println!(
        "\nstreamed to completion: open={} high={} low={} equity={} bars recorded",
        o_rows.len(),
        hi_rows.len(),
        lo_rows.len(),
        eq_rows.len(),
    );
    assert!(!o_rows.is_empty(), "real open bars streamed");
    assert!(!hi_rows.is_empty(), "real high bars streamed");
    assert!(!lo_rows.is_empty(), "real low bars streamed");
    assert!(!eq_rows.is_empty(), "close-driven equity produced");

    // The four OHLC fields are the same bars: O/H/L taps share the timestamp set.
    assert_eq!(o_rows.len(), hi_rows.len(), "O and H are the same bar stream");
    assert_eq!(o_rows.len(), lo_rows.len(), "O and L are the same bar stream");
    for k in 0..o_rows.len() {
        assert_eq!(o_rows[k].0, hi_rows[k].0, "bar {k}: O/H share ts");
        assert_eq!(o_rows[k].0, lo_rows[k].0, "bar {k}: O/L share ts");
    }
    // The OHLC ordering invariant on real bars: low <= open,high and high >= open,low.
    let mut ohlc_ok = 0usize;
    for k in 0..o_rows.len() {
        let o = o_rows[k].1[0].as_f64();
        let hi = hi_rows[k].1[0].as_f64();
        let lo = lo_rows[k].1[0].as_f64();
        if lo <= o && o <= hi && lo <= hi {
            ohlc_ok += 1;
        }
    }
    println!(
        "OHLC ordering (low<=open<=high) holds on {}/{} real bars",
        ohlc_ok,
        o_rows.len()
    );
    assert_eq!(ohlc_ok, o_rows.len(), "every real bar respects low<=open<=high");

    let equity = f64_field(&eq_rows, 0);
    // exposure stream unused here; summarize wants both — reuse equity sign? No:
    // summarize needs an exposure series. We only recorded equity; report PnL only.
    let final_pips = equity.last().map(|(_, v)| *v).unwrap_or(0.0);
    println!(
        "\nGER40 close SMA-cross over real bars: {} equity points, final cumulative = {:.2} pips (pip_size=1.0)",
        equity.len(),
        final_pips,
    );

    // first/last bar timestamps as a sanity window.
    let first = o_rows.first().map(|(t, _)| ms(*t)).unwrap_or(0);
    let last = o_rows.last().map(|(t, _)| ms(*t)).unwrap_or(0);
    println!("real bar window actually streamed: [{first}..{last}] (unix-ms)");

    // summarize is available but needs exposure; we keep the report PnL-shaped.
    let _ = summarize; // referenced for clarity of the public surface

    println!("\nOK: real GER40 OHLC opened from aura-ingest alone -> 4 sources, C4 order -> ran to completion.");
}