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

// Task 3 (deep dive) — the silent period DESYNC.
//
// The coupling claim from t3_sweep: Resample's period IS a swept param, but
// Session's period is BAKED, so they cannot move together. This probe shows the
// SHARP edge of that: a researcher who sweeps `resample.period_minutes` to 30
// (expecting "30-minute bars") gets a graph where the BARS are 30m but the
// SESSION still indexes 15m slots. `entry_bar.target == 3` then no longer means
// "the 3rd 30m bar"; the two clocks silently disagree and the strategy computes a
// DIFFERENT, unintended thing — with NO error, NO warning. That is the danger of a
// coupling the param system cannot enforce.
//
// We run the SAME entry/exit-bar config at period_minutes = 15 (coupled, correct)
// vs period_minutes = 30 (desynced) and show both bootstrap+run happily, but the
// 30m run is semantically broken (Session still ticks 15m).
//
// PUBLIC SURFACE ONLY. The Composite + builders are the public re-authoring; the
// desync is a property of Session BAKING its period (its rustdoc) while Resample
// exposes its own.

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};

const SYMBOL: &str = "GER40";
// The SESSION period is baked here — it never moves no matter what the sweep does
// to resample.period_minutes. THIS is the coupling the param system cannot keep.
const SESSION_PERIOD_BAKED: i64 = 15;

fn breakout(
    session_period: i64,
) -> (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, session_period).into(),
        EqConst::builder().named("entry_bar").into(),
        EqConst::builder().named("exit_bar").into(),
        And::builder().into(),
        Latch::builder().into(),
        SimBroker::builder(1.0).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(server: &Arc<DataServer>, resample_period: i64, session_period: i64, from_ms: i64, to_ms: i64) -> (RunReport, usize) {
    let (bp, rx_eq, rx_held) = breakout(session_period);
    let space = bp.param_space();
    // Build the param vector in param_space() order.
    let point: Vec<Cell> = space
        .iter()
        .map(|ps| match ps.name.as_str() {
            "resample.period_minutes" => Scalar::I64(resample_period).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 srcs: Vec<Box<dyn aura_engine::Source>> = fields
        .iter()
        .map(|&f| Box::new(M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), f).unwrap()) as Box<dyn aura_engine::Source>)
        .collect();
    h.run(srcs);
    drop(h);
    let eq: Vec<(Timestamp, f64)> = rx_eq.try_iter().map(|(t, r)| (t, r[0].as_f64())).collect();
    let n_bars = eq.len();
    let ex: Vec<(Timestamp, f64)> = rx_held.try_iter().map(|(t, r)| (t, r[0].as_f64())).collect();
    let rep = RunReport {
        manifest: RunManifest { commit: "desync".into(), params: vec![], window: (Timestamp(0), Timestamp(0)), seed: 0, broker: "sim".into() },
        metrics: summarize(&eq, &ex),
    };
    (rep, n_bars)
}

fn main() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if !server.has_symbol(SYMBOL) {
        println!("skip: no local {SYMBOL} data");
        return;
    }
    let from = chrono::Utc.with_ymd_and_hms(2024, 9, 1, 0, 0, 0).unwrap().timestamp_millis();
    let to = chrono::Utc.with_ymd_and_hms(2024, 10, 1, 0, 0, 0).unwrap().timestamp_millis() - 1;

    println!("=== Task 3 deep dive: the silent period desync ===");
    println!("entry_bar=3, exit_bar=5 fixed; Session period BAKED at {SESSION_PERIOD_BAKED}m.\n");

    // Coupled (correct): resample 15m == session 15m. This is the shipped baseline.
    let (coupled, n15) = run(&server, 15, SESSION_PERIOD_BAKED, from, to);
    println!("resample=15m, session=15m (COUPLED, correct): bars={n15:<6} total_pips={:.1}", coupled.metrics.total_pips);

    // Desynced: a researcher sweeps resample to 30m, EXPECTING 30m bars. Session
    // still ticks 15m (baked). entry_bar==3 now indexes a 15m slot on a 30m bar
    // stream — a different, unintended signal. NO error is raised.
    let (desync, n30) = run(&server, 30, SESSION_PERIOD_BAKED, from, to);
    println!("resample=30m, session=15m (DESYNCED, silent): bars={n30:<6} total_pips={:.1}", desync.metrics.total_pips);

    // Contrast: to ACTUALLY get a coherent 30m strategy the researcher must REBUILD
    // the Composite with session_period=30 (a different blueprint — topology, not a
    // swept param). That run is coherent but is NOT reachable from the sweep.
    let (rebuilt, n30b) = run(&server, 30, 30, from, to);
    println!("resample=30m, session=30m (REBUILT blueprint): bars={n30b:<6} total_pips={:.1}", rebuilt.metrics.total_pips);

    println!("\nObservation: all three bootstrap+run with NO error. The desynced");
    println!("middle run is silently wrong — the param system cannot couple the two");
    println!("periods, and only a blueprint rebuild (not a sweep) gives a coherent 30m.");
}