//! 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)>(); let (ex_tx, ex_rx) = mpsc::channel::<(Timestamp, Vec)>(); let nodes: Vec> = 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)> = eq_rx.iter().collect(); let exposure_rows: Vec<(Timestamp, Vec)> = 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" }); }