Aura/fieldtests/milestone-walking-skeleton/ms_2_real_data_thread.rs

//! Milestone fieldtest — scenario 2: THE REAL-DATA THREAD.
//!
//! A downstream author wires the WHOLE walking-skeleton promise over REAL
//! recorded M1 bars, using only the public interface:
//!
//!   data-server  -> aura_ingest::load_m1_window   (ingest, C3 epoch-ns)
//!                -> M1Columns::close_stream        (SoA close column, C7)
//!                -> SMA(fast) / SMA(slow) -> Sub   (the SMA-cross signal)
//!                -> Exposure                       (intent/exposure stream, C10)
//!                -> SimBroker                      (sim-optimal pip equity, C10)
//!                -> Recorder x2                    (sink trace, C8/C22)
//!                -> f64_field + summarize          (RunMetrics, C12)
//!                -> RunReport { manifest, metrics }(structured face, C14/C18)
//!
//! Then asserts determinism (C1): two independent runs over the same window
//! fold byte-identical RunReports.
//!
//! Window: AAPL.US, 2006-08 (the symbol/window the milestone shipped its
//! integration test against). Skips cleanly if /mnt/tickdata is absent.

use std::sync::mpsc;

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

const SYMBOL: &str = "AAPL.US";
// 2006-08-01 .. 2006-09-01 UTC, inclusive Unix-ms (data-server's contract).
const FROM_MS: i64 = 1_154_390_400_000;
const TO_MS: i64 = 1_157_068_800_000;

const SMA_FAST: usize = 2;
const SMA_SLOW: usize = 4;
const EXPOSURE_SCALE: f64 = 0.5;
const PIP_SIZE: f64 = 0.0001;

/// Fold the recorded equity+exposure traces of one run into a RunReport.
fn run_once(prices: &[(Timestamp, Scalar)]) -> RunReport {
    // Node array (indices are the wiring vocabulary):
    //   0 SMA(fast)  1 SMA(slow)  2 Sub  3 Exposure
    //   4 SimBroker  5 Recorder(equity)  6 Recorder(exposure)
    let (eq_tx, eq_rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
    let (ex_tx, ex_rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();

    let nodes: Vec<Box<dyn aura_core::Node>> = vec![
        Box::new(Sma::new(SMA_FAST)),
        Box::new(Sma::new(SMA_SLOW)),
        Box::new(aura_std::Sub::new()),
        Box::new(Exposure::new(EXPOSURE_SCALE)),
        Box::new(SimBroker::new(PIP_SIZE)),
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, eq_tx)),
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, ex_tx)),
    ];

    // One source: the close price. It feeds SMA(fast) slot 0, SMA(slow) slot 0,
    // and SimBroker slot 1 (price).
    let sources = vec![SourceSpec {
        kind: ScalarKind::F64,
        targets: vec![
            Target { node: 0, slot: 0 },
            Target { node: 1, slot: 0 },
            Target { node: 4, slot: 1 },
        ],
    }];

    let edges = vec![
        // Sub = SMA_fast - SMA_slow
        Edge { from: 0, to: 2, slot: 0, from_field: 0 },
        Edge { from: 1, to: 2, slot: 1, from_field: 0 },
        // Exposure <- Sub
        Edge { from: 2, to: 3, slot: 0, from_field: 0 },
        // SimBroker slot 0 = exposure
        Edge { from: 3, to: 4, slot: 0, from_field: 0 },
        // equity sink <- SimBroker
        Edge { from: 4, to: 5, slot: 0, from_field: 0 },
        // exposure sink <- Exposure
        Edge { from: 3, to: 6, slot: 0, from_field: 0 },
    ];

    let mut harness = Harness::bootstrap(nodes, sources, edges).expect("bootstrap");
    harness.run(vec![prices.to_vec()]);
    drop(harness); // close the senders so the receivers drain to completion

    let equity_rows: Vec<(Timestamp, Vec<Scalar>)> = eq_rx.iter().collect();
    let exposure_rows: Vec<(Timestamp, Vec<Scalar>)> = ex_rx.iter().collect();

    let equity = f64_field(&equity_rows, 0);
    let exposure = f64_field(&exposure_rows, 0);
    let metrics: RunMetrics = summarize(&equity, &exposure);

    let window = (
        prices.first().map(|p| p.0).unwrap_or(Timestamp(0)),
        prices.last().map(|p| p.0).unwrap_or(Timestamp(0)),
    );
    RunReport {
        manifest: RunManifest {
            commit: "fieldtest".into(),
            params: vec![
                ("sma_fast".into(), SMA_FAST as f64),
                ("sma_slow".into(), SMA_SLOW as f64),
                ("exposure_scale".into(), EXPOSURE_SCALE),
            ],
            window,
            seed: 0,
            broker: format!("sim-optimal(pip_size={PIP_SIZE})"),
        },
        metrics,
    }
}

fn main() {
    if !std::path::Path::new(DEFAULT_DATA_PATH).exists() {
        eprintln!("SKIP: {DEFAULT_DATA_PATH} absent");
        return;
    }
    let server = std::sync::Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if !server.has_symbol(SYMBOL) {
        eprintln!("SKIP: data-server has no symbol {SYMBOL}");
        return;
    }

    let cols: M1Columns = match load_m1_window(&server, SYMBOL, FROM_MS, TO_MS) {
        Some(c) => c,
        None => {
            eprintln!("SKIP: no data in window for {SYMBOL}");
            return;
        }
    };
    let prices = cols.close_stream();
    eprintln!("ingested {} M1 close bars over [{FROM_MS},{TO_MS}]", prices.len());
    if let (Some(first), Some(last)) = (prices.first(), prices.last()) {
        eprintln!("first ts (epoch-ns): {:?}   last ts: {:?}", first.0, last.0);
    }

    let r1 = run_once(&prices);
    let r2 = run_once(&prices);

    let j1 = r1.to_json();
    let j2 = r2.to_json();
    println!("RunReport: {j1}");

    // C1: determinism.
    assert_eq!(j1, j2, "two runs over identical input must fold identical reports (C1)");
    // Validity: metrics are finite (degenerate-all-zero is valid per C2 warm-up).
    assert!(r1.metrics.total_pips.is_finite(), "total_pips finite");
    assert!(r1.metrics.max_drawdown.is_finite() && r1.metrics.max_drawdown >= 0.0, "drawdown finite, >=0");

    let m: &RunMetrics = &r1.metrics;
    let degenerate = m.total_pips == 0.0 && m.max_drawdown == 0.0 && m.exposure_sign_flips == 0;
    eprintln!(
        "metrics: total_pips={} max_drawdown={} sign_flips={}  ({})",
        m.total_pips,
        m.max_drawdown,
        m.exposure_sign_flips,
        if degenerate { "DEGENERATE (valid)" } else { "POPULATED" }
    );
    println!("SCENARIO 2: PASS (deterministic, finite{})", if degenerate { ", degenerate" } else { ", populated" });
}