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

// Task 3 — Parameter sweep of the breakout's STRUCTURAL params.
//
// The researcher's task: "sweep the breakout's entry bar index, exit bar index,
// and bar period. NOTE the coupling: the Resample period and the Session period
// must stay EQUAL, and the entry/exit bar indices are the EqConst targets. Can
// the public named-param sweep express this without a topology change?"
//
// To sweep ANYTHING I must first RE-AUTHOR the shipped breakout (a hand-wired
// FlatGraph with no param_space()) as a Composite blueprint — only a Composite
// exposes param_space() / the named-axis sweep / bootstrap_with_cells that the
// World sweep API consumes. The shipped example does NOT ship the breakout as a
// Composite, so this re-authoring is forced on the researcher (recorded as the
// headline Task-3 friction). The 13-node wiring below is the SAME topology as the
// shipped shared/breakout_real.rs, re-expressed as a Composite.
//
// PUBLIC SURFACE ONLY: Composite/axis/param_space/bootstrap_with_cells from cargo
// doc; EqConst/Resample/Session builder param declarations from their rustdoc
// (EqConst declares `target`; Resample declares `period_minutes`; Session BAKES
// open/tz/period — "NOT scalar params, the declared param list is empty").

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

use aura_core::{Cell, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    sweep, summarize, BlueprintNode, Composite, Edge, GridSpace, RunManifest, RunReport,
    Role, 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";
const PIP_SIZE: f64 = 1.0;
const BAR_MINUTES: i64 = 15;

// The breakout re-authored as a Composite. Same 9-node strategy topology +
// SimBroker + 2 recorder sinks (equity, held). Sources bind to the 4 OHLC roles.
// Returns the blueprint plus the equity/held receivers a run will drain.
fn breakout_composite() -> (
    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();

    // Node indices, parallel to the shipped FlatGraph (sans the two extra taps).
    // 0 Resample, 1 Delay, 2 Gt, 3 Session, 4 EqConst(entry), 5 EqConst(exit),
    // 6 And, 7 Latch, 8 SimBroker, 9 Rec(equity), 10 Rec(held).
    let nodes: Vec<BlueprintNode> = vec![
        Resample::builder().into(),                                  // 0
        Delay::builder().into(),                                     // 1
        Gt::builder().into(),                                        // 2
        Session::builder(9, 0, Berlin, BAR_MINUTES).into(),          // 3  (period BAKED)
        EqConst::builder().named("entry_bar").into(),                // 4  (target tunable)
        EqConst::builder().named("exit_bar").into(),                 // 5  (target tunable)
        And::builder().into(),                                       // 6
        Latch::builder().into(),                                     // 7
        SimBroker::builder(PIP_SIZE).into(),                         // 8
        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),   // 9
        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_held).into(), // 10
    ];

    const F_HIGH15: usize = 1;
    const F_CLOSE15: usize = 3;
    let edges = vec![
        Edge { from: 0, to: 2, slot: 0, from_field: F_CLOSE15 }, // close15 -> Gt.a
        Edge { from: 0, to: 1, slot: 0, from_field: F_HIGH15 },  // high15 -> Delay
        Edge { from: 1, to: 2, slot: 1, from_field: 0 },         // prevHigh15 -> Gt.b
        Edge { from: 0, to: 3, slot: 0, from_field: F_CLOSE15 }, // close15 -> Session.trigger
        Edge { from: 3, to: 4, slot: 0, from_field: 0 },         // bars -> EqConst(entry)
        Edge { from: 3, to: 5, slot: 0, from_field: 0 },         // bars -> EqConst(exit)
        Edge { from: 2, to: 6, slot: 0, from_field: 0 },         // breakout -> And.a
        Edge { from: 4, to: 6, slot: 1, from_field: 0 },         // isEntryBar -> And.b
        Edge { from: 6, to: 7, slot: 0, from_field: 0 },         // entry -> Latch.set
        Edge { from: 5, to: 7, slot: 1, from_field: 0 },         // isExitBar -> Latch.reset
        Edge { from: 7, to: 8, slot: 0, from_field: 0 },         // held -> SimBroker.exposure
        Edge { from: 0, to: 8, slot: 1, from_field: F_CLOSE15 }, // close15 -> SimBroker.price
        Edge { from: 8, to: 9, slot: 0, from_field: 0 },         // equity -> sink
        Edge { from: 7, to: 10, slot: 0, from_field: 0 },        // held -> sink
    ];

    // Four bound OHLC source roles -> the Resample's four Barrier(0) slots, in the
    // fixed C4 merge order open(0) high(1) low(2) close(3).
    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();

    let bp = Composite::new("ger40_breakout", nodes, edges, roles, vec![]);
    (bp, rx_eq, rx_held)
}

fn utc_month_window_ms(year: i32, month: u32) -> (i64, i64) {
    let from = chrono::Utc.with_ymd_and_hms(year, month, 1, 0, 0, 0).unwrap().timestamp_millis();
    let (ny, nm) = if month == 12 { (year + 1, 1) } else { (year, month + 1) };
    let next = chrono::Utc.with_ymd_and_hms(ny, nm, 1, 0, 0, 0).unwrap().timestamp_millis();
    (from, next - 1)
}

fn run_point(server: &Arc<DataServer>, point: &[Cell], from_ms: i64, to_ms: i64) -> RunReport {
    let (bp, rx_eq, rx_held) = breakout_composite();
    let mut h = bp
        .bootstrap_with_cells(point)
        .expect("sweep point kind-checked against param_space");
    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)
                    .expect("OHLC field overlaps window"),
            ) 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 ex: Vec<(Timestamp, f64)> = rx_held.try_iter().map(|(t, r)| (t, r[0].as_f64())).collect();
    RunReport {
        manifest: RunManifest {
            commit: "breakout-sweep".into(),
            params: vec![],
            window: (
                aura_ingest::unix_ms_to_epoch_ns(from_ms),
                aura_ingest::unix_ms_to_epoch_ns(to_ms),
            ),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".into(),
        },
        metrics: summarize(&eq, &ex),
    }
}

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

    println!("=== Task 3: sweep the breakout's structural params ===\n");

    // 1. What does param_space() actually expose for the re-authored breakout?
    let space = breakout_composite().0.param_space();
    println!("param_space() of the breakout Composite:");
    for ps in &space {
        println!("    {:<28} : {:?}", ps.name, ps.kind);
    }
    println!();

    // The CRUX (recorded): note what is and is NOT in the space.
    // - entry_bar.target / exit_bar.target ARE here (EqConst declares `target`).
    // - resample.period_minutes IS here (Resample declares it tunable).
    // - There is NO session.* period knob — Session BAKES its period (a structural
    //   factory arg). So "Resample period == Session period" CANNOT be a single
    //   swept named param: the Session leg of the coupling is invisible to the
    //   sweep. Sweeping resample.period_minutes alone would silently DESYNC the
    //   two periods (a correctness trap), and changing the Session period at all
    //   requires REBUILDING the Composite (a different blueprint), exactly the
    //   "topology change, not a param" line C11 draws.
    let has_resample_period = space.iter().any(|p| p.name.contains("period_minutes"));
    let has_session_period = space.iter().any(|p| p.name.to_lowercase().contains("session"));
    println!("coupling check:");
    println!("    resample.period_minutes in space?  {has_resample_period}");
    println!("    any session.* period in space?     {has_session_period}");
    println!("    => the bar-period coupling is NOT expressible as one swept param.\n");

    // 2. The sweep that DOES work: entry_bar x exit_bar (both EqConst targets),
    //    at the FIXED 15m period (no period coupling to violate). A 3x3 grid over
    //    one month of GER40.
    let (from_ms, to_ms) = utc_month_window_ms(2024, 9);
    let grid = GridSpace::new(
        &space,
        space
            .iter()
            .map(|ps| match ps.name.as_str() {
                "entry_bar.target" => vec![Scalar::I64(2), Scalar::I64(3), Scalar::I64(4)],
                "exit_bar.target" => vec![Scalar::I64(5), Scalar::I64(6), Scalar::I64(8)],
                // resample.period_minutes pinned to 15 (== the baked Session 15m).
                "resample.period_minutes" => vec![Scalar::I64(15)],
                // SURPRISE (recorded): delay.lag is a STRUCTURAL deformer (the
                // "previous bar's high" of the breakout) yet ships in param_space
                // as a tunable knob — so the grid is FORCED to enumerate it (here
                // pinned to its structural value 1) or panic. C34's bind() would
                // remove it from the space, but the node exposes it by default.
                "delay.lag" => vec![Scalar::I64(1)],
                other => panic!("unexpected param slot {other}"),
            })
            .collect(),
    )
    .expect("grid well-formed for the space");

    println!("sweeping entry_bar x exit_bar (period pinned 15m) over GER40 2024-09:");
    let server_for_closure = Arc::clone(&server);
    let family = sweep(&grid, |pt: &[Cell]| run_point(&server_for_closure, pt, from_ms, to_ms));

    // Print each point with its named params (by index — SweepFamily::named_params(i)).
    for (i, p) in family.points.iter().enumerate() {
        let named = family.named_params(i);
        let label: Vec<String> = named
            .iter()
            .filter(|(n, _)| n != "resample.period_minutes")
            .map(|(n, v)| format!("{}={}", n.replace(".target", ""), scalar_str(v)))
            .collect();
        println!(
            "    {{ {} }}  total_pips={:>8.1}  max_dd={:>7.1}",
            label.join(", "),
            p.report.metrics.total_pips,
            p.report.metrics.max_drawdown,
        );
    }
    println!("\nOK: entry/exit-bar grid sweep works; bar-period coupling does not.");
}

fn scalar_str(s: &Scalar) -> String {
    match s {
        Scalar::I64(v) => v.to_string(),
        Scalar::F64(v) => format!("{v:.2}"),
        Scalar::Bool(v) => v.to_string(),
        Scalar::Timestamp(t) => t.0.to_string(),
    }
}