Aura/fieldtests/milestone-runway/rw_3_multitap_ts_join.rs

// Runway milestone fieldtest — scenario 3: multi-tap trace honesty (#93).
//
// The task a downstream researcher does: "I record two taps off the same real
// GER40 run — the exposure stream and the cumulative-equity stream — and I want a
// single per-bar trace that lines each tap up against the other. The taps fire at
// DIFFERENT cadences (exposure warms up later than the raw bar clock; equity only
// fires once the broker has a prev price), so a positional zip-by-index would
// misalign. I want them joined on the recorded TIMESTAMP."
//
// What this exercises, top-down from the milestone promise: #93's `join_on_ts` /
// `JoinedRow` — exactly one joined row per spine entry, each side Some(row) where
// it fired at that ts and None where it did not. The honest fusion the milestone
// promises: traces join on timestamp, never by index.
//
// It also re-uses the #81/#92 real-data seam (open_ohlc from aura-ingest alone) so
// the join runs over REAL bars, not a synthetic toy.
//
// PUBLIC INTERFACE ONLY: API from `cargo doc` rustdoc + the design ledger
// (C8/C18 post-run reduction, C3 no in-graph join, C1 one row per ts). No
// crates/*/src was read.

use std::sync::mpsc;

use aura_core::{Cell, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{join_on_ts, BlueprintNode, Composite, Edge, OutField, Role, Source, Target};
use aura_ingest::{default_data_server, open_ohlc};
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() }],
    )
}

// A single-price-role harness with TWO taps off different nodes:
//   - exposure tap (off the Exposure node) — fires once SMA-cross warms up
//   - equity   tap (off the SimBroker)     — fires once the broker has a prev price
// Their cadences differ, so their recorded streams have different lengths and
// firing instants — the exact #93 shape.
#[allow(clippy::type_complexity)]
fn harness_two_taps() -> (
    Composite,
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>, // exposure tap
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>, // equity tap
) {
    let (tx_ex, rx_ex) = mpsc::channel();
    let (tx_eq, rx_eq) = mpsc::channel();
    // 0 sma_cross, 1 Exposure, 2 SimBroker, 3 exposure-rec, 4 equity-rec
    let bp = Composite::new(
        "two_tap_harness",
        vec![
            BlueprintNode::Composite(sma_cross()),
            Exposure::builder().into(),
            SimBroker::builder(1.0).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).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: 1, to: 3, slot: 0, from_field: 0 }, // exposure -> exposure tap
            Edge { from: 2, to: 4, slot: 0, from_field: 0 }, // equity   -> equity tap
        ],
        vec![Role {
            name: "price".into(),
            targets: vec![
                Target { node: 0, slot: 0 }, // strategy
                Target { node: 2, slot: 1 }, // broker price slot
            ],
            source: Some(ScalarKind::F64),
        }],
        vec![],
    );
    (bp, rx_ex, rx_eq)
}

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

fn main() {
    let server = default_data_server();

    // ~1 month of GER40 from 2015-01-01 (epoch-ns), close-driven price.
    let day_ns: i64 = 86_400_000 * 1_000_000;
    let from = Timestamp(1_420_070_400_000 * 1_000_000);
    let to = Timestamp(from.0 + 31 * day_ns);

    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);
        }
    };
    // The strategy here uses ONE price role (close). open_ohlc gives four; we take
    // the close source (index 3 in the fixed O/H/L/C order #92 vouches for).
    let close: Box<dyn Source> = sources.into_iter().nth(3).expect("close is source 3");

    let (bp, rx_ex, rx_eq) = harness_two_taps();
    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("params kind-check");
    h.run(vec![close]);
    drop(h);

    let exposure_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
    let equity_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();

    println!(
        "two taps, DIFFERENT cadences: exposure fired {} times, equity fired {} times",
        exposure_rows.len(),
        equity_rows.len(),
    );
    // The whole point of #93: a positional zip-by-index is WRONG when the lengths
    // differ. Show the mismatch the index approach would hit:
    if exposure_rows.len() != equity_rows.len() {
        println!(
            "  -> lengths differ by {} rows; zip-by-index would silently misalign every row \
             past the first divergence",
            (exposure_rows.len() as i64 - equity_rows.len() as i64).abs(),
        );
    }

    // The honest fusion: join on the recorded timestamp. Use the equity stream as
    // the spine (the trace we anchor on), exposure as a side stream.
    let exp_slice: &[(Timestamp, Vec<Scalar>)] = &exposure_rows;
    let joined = join_on_ts(&equity_rows, &[exp_slice]);

    println!(
        "\njoin_on_ts: spine(equity)={} rows -> {} JoinedRows (exactly one per spine entry)",
        equity_rows.len(),
        joined.len(),
    );
    assert_eq!(
        joined.len(),
        equity_rows.len(),
        "exactly one JoinedRow per spine entry (C1/C8)"
    );

    // Every joined row's ts is the spine ts, in spine order.
    for (k, jr) in joined.iter().enumerate() {
        assert_eq!(jr.ts, equity_rows[k].0, "row {k}: joined ts == spine ts");
    }

    // Where the exposure side fired at the spine ts -> Some; where it did not -> None.
    // Cross-check Some-rows against the exposure stream by ts (not by index!).
    use std::collections::HashMap;
    let exp_by_ts: HashMap<i64, f64> =
        exposure_rows.iter().map(|(t, r)| (t.0, r[0].as_f64())).collect();

    let mut present = 0usize;
    let mut absent = 0usize;
    let mut mismatches = 0usize;
    for jr in &joined {
        match &jr.sides[0] {
            Some(side_row) => {
                present += 1;
                let joined_val = side_row[0].as_f64();
                match exp_by_ts.get(&jr.ts.0) {
                    Some(truth) if (*truth - joined_val).abs() < 1e-12 => {}
                    _ => mismatches += 1,
                }
            }
            None => {
                absent += 1;
                // a None side MUST mean exposure genuinely did not fire at this ts.
                if exp_by_ts.contains_key(&jr.ts.0) {
                    mismatches += 1;
                }
            }
        }
    }
    println!(
        "exposure side: present(Some)={present}, absent(None)={absent}, by-ts cross-check mismatches={mismatches}",
    );
    assert_eq!(mismatches, 0, "every Some/None matches the exposure stream BY TIMESTAMP");

    // Show the first few divergent rows: spine bars where exposure had NOT yet
    // warmed up (None) vs where both fired (Some) — proving the alignment is by ts.
    println!("\nfirst 6 joined rows (equity spine | exposure side):");
    for jr in joined.iter().take(6) {
        let eq = jr.spine[0].as_f64();
        let ex = jr.sides[0].as_ref().map(|r| r[0].as_f64());
        println!(
            "  ts={} equity={:.4} exposure={}",
            ms(jr.ts),
            eq,
            match ex {
                Some(v) => format!("Some({v:.4})"),
                None => "None (had not fired at this ts)".into(),
            }
        );
    }

    println!(
        "\nOK: two real-data taps at different cadences fused HONESTLY on timestamp \
         ({present} aligned, {absent} side-absent), never by index."
    );
}