Aura/fieldtests/research-breakout-deepdive/t1_scale.rs

// Task 1 — Scale / long horizon.
//
// The downstream researcher's task: "the shipped GER40 breakout demo runs ONE
// month. I want to know if the strategy holds up over a full year (2024) and a
// multi-year span (2018-01 .. 2024-12). Does it stay deterministic? Does memory
// stay bounded (the docs claim streaming residency is O(one chunk) — does opening
// 4 independent OHLC sources break that)? How slow is it?"
//
// PUBLIC INTERFACE ONLY. I start from the shipped example's shared wiring module
// (crates/aura-ingest/examples/shared/breakout_real.rs), included by #[path] —
// exactly as the shipped example and its test do — so I re-author NOTHING about
// the proven 13-node FlatGraph; I only change the WINDOW (a month -> a year ->
// 7 years) and add a self-RSS probe + a wall-clock timer. No crates/*/src read.

use std::sync::Arc;
use std::time::Instant;

use chrono::TimeZone;
use data_server::{DataServer, DEFAULT_DATA_PATH};

// Reuse the SHIPPED breakout wiring verbatim, the same way the shipped example
// does. The researcher does not re-type 13 nodes — the example exports them.
#[path = "../../crates/aura-ingest/examples/shared/breakout_real.rs"]
mod breakout_real;
use breakout_real::*;

/// Read this process's resident set size (KiB) from /proc/self/statm * page size.
/// A coarse but honest peak-RSS probe — no allocator hooks, just the OS view.
fn rss_kib() -> u64 {
    let statm = std::fs::read_to_string("/proc/self/statm").unwrap_or_default();
    let resident_pages: u64 = statm
        .split_whitespace()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .unwrap_or(0);
    resident_pages * 4 // page size 4 KiB on this box
}

/// Build the inclusive Unix-ms window spanning whole calendar months
/// [from (y0,m0) .. through end of (y1,m1)] in UTC.
fn utc_span_ms(y0: i32, m0: u32, y1: i32, m1: u32) -> (i64, i64) {
    let from = chrono::Utc.with_ymd_and_hms(y0, m0, 1, 0, 0, 0).unwrap().timestamp_millis();
    let (ny, nm) = if m1 == 12 { (y1 + 1, 1) } else { (y1, m1 + 1) };
    let next = chrono::Utc.with_ymd_and_hms(ny, nm, 1, 0, 0, 0).unwrap().timestamp_millis();
    (from, next - 1)
}

fn run_span(server: &Arc<DataServer>, label: &str, from_ms: i64, to_ms: i64) {
    let Some(sources) = open_ohlc_sources(server, from_ms, to_ms) else {
        println!("{label}: no overlapping {SYMBOL} files — skip");
        return;
    };

    let peak_before = rss_kib();
    let t0 = Instant::now();

    let (mut h, taps) = build_harness();
    h.run(sources);
    let trace = drain_trace(&taps);
    let report = report_from_trace(&trace, from_ms, to_ms);

    let elapsed = t0.elapsed();
    let peak_after = rss_kib();

    // count sessions (0->1 transitions)
    let mut entries = 0u32;
    let mut prev = 0.0;
    for b in &trace {
        if b.held == 1.0 && prev == 0.0 {
            entries += 1;
        }
        prev = b.held;
    }

    let m = &report.metrics;
    println!(
        "{label:<22} bars={:<7} sessions={:<4} total_pips={:>9.1} max_dd={:>8.1} flips={:<4} \
         | {:>6.2}s  RSS {}->{} MiB",
        trace.len(),
        entries,
        m.total_pips,
        m.max_drawdown,
        m.exposure_sign_flips,
        elapsed.as_secs_f64(),
        peak_before / 1024,
        peak_after / 1024,
    );
}

fn main() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if !server.has_symbol(SYMBOL) {
        println!("skip: no local {SYMBOL} data at {DEFAULT_DATA_PATH}");
        return;
    }

    println!("=== Task 1: scale the GER40 breakout over long horizons ===");
    println!("startup RSS = {} MiB\n", rss_kib() / 1024);

    // 1. The shipped one-month baseline, for reference.
    let (f, t) = utc_span_ms(2024, 9, 2024, 9);
    run_span(&server, "2024-09 (baseline)", f, t);

    // 2. A full year of GER40 (2024).
    let (f, t) = utc_span_ms(2024, 1, 2024, 12);
    run_span(&server, "2024 full year", f, t);

    // 3. A multi-year span: 2018-01 .. 2024-12 (7 years of M1 across 4 sources).
    let (f, t) = utc_span_ms(2018, 1, 2024, 12);
    run_span(&server, "2018..2024 (7y)", f, t);

    // 4. Determinism: rerun the full-year span and assert bit-identical report.
    println!("\n--- determinism (C1): rerun 2024 full year, compare report JSON ---");
    let (f, t) = utc_span_ms(2024, 1, 2024, 12);
    let rep = |srv: &Arc<DataServer>| {
        let sources = open_ohlc_sources(srv, f, t).expect("year sources");
        let (mut h, taps) = build_harness();
        h.run(sources);
        let trace = drain_trace(&taps);
        report_from_trace(&trace, f, t)
    };
    let r1 = rep(&server);
    let r2 = rep(&server);
    let j1 = r1.to_json();
    let j2 = r2.to_json();
    println!("rerun report JSON bit-identical: {}", j1 == j2);
    println!("  total_pips r1={:.1} r2={:.1}", r1.metrics.total_pips, r2.metrics.total_pips);
}