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

//! Fieldtest c0011 #2 — close_stream -> a real engine run (axis b).
//!
//! Question under test: a project author transposes recorded M1 bars to
//! `M1Columns`, takes `close_stream()`, and feeds it straight into a Harness
//! (SMA-cross -> Exposure -> SimBroker -> Recorder, the cycle-0007/0009 pattern
//! over the shipped aura_std::Recorder), runs it, drains the recorded equity
//! curve, and the recorded stream is what a downstream consumer expects — the
//! ingestion boundary connects to the engine without a hand-authored stream.
//!
//! This is the C3 -> C1 hand-off: the (Timestamp, Scalar) shape `close_stream`
//! returns is exactly `Harness::run`'s source-stream input shape.
//!
//! Public-surface facts used (rustdoc + design ledger):
//!  - aura_ingest::{M1Columns via transpose_m1, close_stream}.
//!  - aura_std::{Sma, Sub, Exposure, SimBroker, Recorder}.
//!  - aura_engine::{Harness, Edge, SourceSpec, Target, RunManifest, RunReport,
//!    f64_field, summarize}.
//!  - SimBroker slots: 0 = exposure, 1 = price (ledger C10 Realization 0007).

use std::sync::mpsc;

use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{Edge, Harness, RunManifest, RunReport, SourceSpec, Target, f64_field, summarize};
use aura_ingest::transpose_m1;
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use data_server::records::M1Parsed;

/// Build a deterministic synthetic set of M1 bars (close rises then reverses),
/// the way a project author would after recording a real window. We only use
/// the close column downstream, but transpose_m1 carries the full OHLCV bundle.
fn synthetic_bars() -> Vec<M1Parsed> {
    // close: 100,102,104,106,108,106,104 — a rise then a reversal, so the
    // SMA-cross flips sign once (exercises exposure_sign_flips like the CLI demo).
    let closes = [100.0, 102.0, 104.0, 106.0, 108.0, 106.0, 104.0];
    closes
        .iter()
        .enumerate()
        .map(|(i, &c)| M1Parsed {
            time_ms: (i as i64 + 1) * 60_000, // one M1 bar per minute, Unix-ms
            open: c,
            high: c + 0.5,
            low: c - 0.5,
            close: c,
            spread: 0.1,
            volume: 1_000,
        })
        .collect()
}

fn main() {
    // --- Ingestion boundary: AoS bars -> SoA columns -> close source stream ---
    let bars = synthetic_bars();
    let cols = transpose_m1(&bars);
    let price_stream: Vec<(Timestamp, Scalar)> = cols.close_stream();
    println!("close source stream from ingestion = {price_stream:?}");

    // The C3 precondition close_stream documents: ascending in timestamp.
    for w in price_stream.windows(2) {
        assert!(w[0].0 < w[1].0, "close_stream ascending in timestamp (C3 precondition)");
    }

    // --- Wire the signal-quality harness over the ingested close stream. ---
    let (eq_tx, eq_rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
    let (exp_tx, exp_rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();

    // nodes: 0=SMA(2) fast, 1=SMA(4) slow, 2=Sub, 3=Exposure(4), 4=SimBroker(1.0),
    //        5=equity sink (taps broker), 6=exposure sink (taps Exposure).
    let mut h = Harness::bootstrap(
        vec![
            Box::new(Sma::new(2)),
            Box::new(Sma::new(4)),
            Box::new(Sub::new()),
            Box::new(Exposure::new(4.0)),
            Box::new(SimBroker::new(1.0)),
            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 }, // -> SMA fast
                Target { node: 1, slot: 0 }, // -> SMA slow
                Target { node: 4, slot: 1 }, // -> SimBroker price (slot 1)
            ],
        }],
        vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA2 -> Sub.in0
            Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA4 -> Sub.in1
            Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // Sub -> Exposure
            Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // Exposure -> SimBroker.in0
            Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // Exposure -> exposure sink
            Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // SimBroker -> equity sink
        ],
    )
    .expect("valid signal-quality DAG over the ingested close stream");

    // Run #1 over the ingested stream.
    h.run(vec![price_stream.clone()]);
    let equity_rows: Vec<(Timestamp, Vec<Scalar>)> = eq_rx.try_iter().collect();
    let exposure_rows: Vec<(Timestamp, Vec<Scalar>)> = exp_rx.try_iter().collect();
    println!("equity rows   = {equity_rows:?}");
    println!("exposure rows = {exposure_rows:?}");

    let equity = f64_field(&equity_rows, 0);
    let exposure = f64_field(&exposure_rows, 0);
    let metrics = summarize(&equity, &exposure);
    println!("metrics over ingested bars = {metrics:?}");

    // The recorded equity curve must be non-empty (the source stream warmed the
    // chain) and the metrics finite — the downstream "did the ingested data
    // actually drive a sim?" check.
    assert!(!equity_rows.is_empty(), "ingested close stream produced a recorded equity curve");
    assert!(metrics.total_pips.is_finite(), "total_pips finite");
    assert!(metrics.max_drawdown.is_finite() && metrics.max_drawdown >= 0.0, "drawdown finite, >= 0");
    // The reversal in the close series must register at least one exposure flip.
    assert!(metrics.exposure_sign_flips >= 1, "the rise-then-reverse close flips exposure at least once");

    // The recorded equity timestamps are exactly the ingested (normalized) bar
    // timestamps — confirming epoch-ns flowed end to end, unchanged, through the
    // run loop (the broker fires on every price-fresh cycle, C10 Realization).
    let recorded_ts: Vec<Timestamp> = equity_rows.iter().map(|(t, _)| *t).collect();
    let ingested_ts: Vec<Timestamp> = price_stream.iter().map(|(t, _)| *t).collect();
    assert_eq!(recorded_ts, ingested_ts, "recorded equity ts == ingested normalized bar ts (epoch-ns end to end)");

    // --- Determinism (C1): a second disjoint run over the same ingested stream
    //     yields a bit-identical report. ---
    let (eq_tx2, eq_rx2) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
    let (exp_tx2, exp_rx2) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
    let mut h2 = Harness::bootstrap(
        vec![
            Box::new(Sma::new(2)),
            Box::new(Sma::new(4)),
            Box::new(Sub::new()),
            Box::new(Exposure::new(4.0)),
            Box::new(SimBroker::new(1.0)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, eq_tx2)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, exp_tx2)),
        ],
        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 second DAG");
    h2.run(vec![price_stream]);
    let equity2 = f64_field(&eq_rx2.try_iter().collect::<Vec<_>>(), 0);
    let exposure2 = f64_field(&exp_rx2.try_iter().collect::<Vec<_>>(), 0);
    let metrics2 = summarize(&equity2, &exposure2);

    let report1 = RunReport {
        manifest: RunManifest {
            commit: "c0011-ingested-synthetic".to_string(),
            params: vec![("sma_fast".to_string(), 2.0), ("sma_slow".to_string(), 4.0)],
            window: (ingested_ts[0], *ingested_ts.last().unwrap()),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".to_string(),
        },
        metrics,
    };
    let report2 = RunReport {
        manifest: report1.manifest.clone(),
        metrics: metrics2,
    };
    assert_eq!(report1.to_json(), report2.to_json(), "two runs over the same ingested stream are bit-identical (C1)");
    println!("c0011_2 OK: ingested close_stream drove a deterministic end-to-end signal-quality run");
}