// 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, 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| { 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); }