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

// Task 4 — Multi-instrument comparison: the SAME breakout on GER40 vs FRA40.
//
// The researcher's task: "run my breakout on another index (FRA40 — Paris, also
// CET 09:00) and compare signal quality against GER40. Is the comparison honest
// across instruments? What does the engine do about per-instrument pip metadata?"
//
// The experiment matrix here is the structural axis "which instrument" (C20): one
// strategy x N symbols, a plain Rust loop. The breakout Composite is REUSED
// verbatim across symbols; only the symbol the sources read and the pip_size the
// SimBroker bakes change.
//
// THE PIP QUESTION (the headline Task-4 finding): SimBroker::new(pip_size) takes a
// BARE f64. There is NO instrument/pip-metadata registry on the public surface
// (aura-ingest exposes only M1 streaming; Aura.toml — which the glossary says
// carries "instrument/pip metadata" — does not exist yet, it is an open thread).
// So the researcher must HARD-CODE pip_size per symbol from outside knowledge:
// GER40=1.0 (index point), FRA40=1.0 (index point), EURUSD=0.0001 (FX). Nothing in
// the engine knows or checks this. A GER40-vs-FRA40 comparison is honest ONLY
// because the researcher happens to know both are index points; the engine offers
// no help and no guard against comparing index pips to FX pips.
//
// PUBLIC SURFACE ONLY.

use std::sync::mpsc;
use std::sync::Arc;

use aura_core::{Cell, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{summarize, BlueprintNode, Composite, Edge, Role, RunManifest, RunReport, Target};
use aura_ingest::{M1Field, M1FieldSource};
use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder, Resample, Session, SimBroker};
use chrono::TimeZone;
use chrono_tz::Europe::Berlin;
use data_server::{DataServer, DEFAULT_DATA_PATH};

// What the researcher must supply BY HAND per instrument — there is no registry.
// (symbol, pip_size, human note on what a "pip" is for this asset)
const INSTRUMENTS: &[(&str, f64, &str)] = &[
    ("GER40", 1.0, "index point"),
    ("FRA40", 1.0, "index point"),
];

fn breakout(
    pip_size: f64,
) -> (Composite, mpsc::Receiver<(Timestamp, Vec<Scalar>)>, mpsc::Receiver<(Timestamp, Vec<Scalar>)>) {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_held, rx_held) = mpsc::channel();
    let nodes: Vec<BlueprintNode> = vec![
        Resample::builder().into(),
        Delay::builder().into(),
        Gt::builder().into(),
        Session::builder(9, 0, Berlin, 15).into(),
        EqConst::builder().named("entry_bar").into(),
        EqConst::builder().named("exit_bar").into(),
        And::builder().into(),
        Latch::builder().into(),
        SimBroker::builder(pip_size).into(),
        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_held).into(),
    ];
    let edges = vec![
        Edge { from: 0, to: 2, slot: 0, from_field: 3 },
        Edge { from: 0, to: 1, slot: 0, from_field: 1 },
        Edge { from: 1, to: 2, slot: 1, from_field: 0 },
        Edge { from: 0, to: 3, slot: 0, from_field: 3 },
        Edge { from: 3, to: 4, slot: 0, from_field: 0 },
        Edge { from: 3, to: 5, slot: 0, from_field: 0 },
        Edge { from: 2, to: 6, slot: 0, from_field: 0 },
        Edge { from: 4, to: 6, slot: 1, from_field: 0 },
        Edge { from: 6, to: 7, slot: 0, from_field: 0 },
        Edge { from: 5, to: 7, slot: 1, from_field: 0 },
        Edge { from: 7, to: 8, slot: 0, from_field: 0 },
        Edge { from: 0, to: 8, slot: 1, from_field: 3 },
        Edge { from: 8, to: 9, slot: 0, from_field: 0 },
        Edge { from: 7, to: 10, slot: 0, from_field: 0 },
    ];
    let roles = (0..4)
        .map(|slot| Role {
            name: ["open", "high", "low", "close"][slot as usize].into(),
            targets: vec![Target { node: 0, slot }],
            source: Some(ScalarKind::F64),
        })
        .collect();
    (Composite::new("breakout", nodes, edges, roles, vec![]), rx_eq, rx_held)
}

fn run_symbol(server: &Arc<DataServer>, symbol: &str, pip_size: f64, from_ms: i64, to_ms: i64) -> Option<(RunReport, u32)> {
    let (bp, rx_eq, rx_held) = breakout(pip_size);
    let space = bp.param_space();
    // entry=3, exit=5, period=15, lag=1 — the shipped baseline config.
    let point: Vec<Cell> = space
        .iter()
        .map(|ps| match ps.name.as_str() {
            "resample.period_minutes" => Scalar::I64(15).cell(),
            "delay.lag" => Scalar::I64(1).cell(),
            "entry_bar.target" => Scalar::I64(3).cell(),
            "exit_bar.target" => Scalar::I64(5).cell(),
            other => panic!("unexpected slot {other}"),
        })
        .collect();
    let mut h = bp.bootstrap_with_cells(&point).expect("kind-checked");
    let fields = [M1Field::Open, M1Field::High, M1Field::Low, M1Field::Close];
    let mut srcs: Vec<Box<dyn aura_engine::Source>> = Vec::new();
    for &f in &fields {
        let s = M1FieldSource::open(server, symbol, Some(from_ms), Some(to_ms), f)?;
        srcs.push(Box::new(s));
    }
    h.run(srcs);
    drop(h);
    let eq: Vec<(Timestamp, f64)> = rx_eq.try_iter().map(|(t, r)| (t, r[0].as_f64())).collect();
    let held: Vec<(Timestamp, f64)> = rx_held.try_iter().map(|(t, r)| (t, r[0].as_f64())).collect();
    // count entries (0->1 transitions)
    let mut entries = 0u32;
    let mut prev = 0.0;
    for (_, v) in &held {
        if *v == 1.0 && prev == 0.0 {
            entries += 1;
        }
        prev = *v;
    }
    let rep = RunReport {
        manifest: RunManifest { commit: "compare".into(), params: vec![], window: (Timestamp(0), Timestamp(0)), seed: 0, broker: format!("sim(pip={pip_size})") },
        metrics: summarize(&eq, &held),
    };
    Some((rep, entries))
}

fn main() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    println!("=== Task 4: multi-instrument comparison (same breakout, GER40 vs FRA40) ===\n");

    // Full year 2024 so the comparison has statistical body.
    let from = chrono::Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap().timestamp_millis();
    let to = chrono::Utc.with_ymd_and_hms(2024, 12, 31, 23, 59, 59).unwrap().timestamp_millis();

    println!("{:<8} {:<14} {:>8} {:>10} {:>10} {:>7}", "symbol", "pip-meaning", "sessions", "total_pips", "max_dd", "flips");
    println!("{}", "-".repeat(62));
    for &(symbol, pip_size, note) in INSTRUMENTS {
        if !server.has_symbol(symbol) {
            println!("{symbol:<8} (no local data — skip)");
            continue;
        }
        match run_symbol(&server, symbol, pip_size, from, to) {
            Some((rep, entries)) => {
                let m = &rep.metrics;
                println!(
                    "{symbol:<8} {note:<14} {entries:>8} {:>10.1} {:>10.1} {:>7}",
                    m.total_pips, m.max_drawdown, m.exposure_sign_flips
                );
            }
            None => println!("{symbol:<8} (no overlapping OHLC files — skip)"),
        }
    }

    println!("\nHonesty of the cross-asset comparison:");
    println!("  Both GER40 and FRA40 are index points (pip=1.0), so their pip totals");
    println!("  ARE directly comparable here — BUT only because the researcher knew");
    println!("  to pass pip=1.0 for both. The engine has NO instrument registry: pip");
    println!("  is a bare f64 to SimBroker::new, and nothing would stop (or flag) a");
    println!("  comparison of GER40 index-pips to EURUSD fx-pips with mismatched scale.");
}