// Milestone-18 fieldtest — scenario 3 (THE HEADLINE): walk-forward over a real
// M1 data window, with per-window grid-optimize and a stitched OOS equity curve.
//
// Composes: axis 3 (walk-forward = vary the data window), axis 1 (grid sweep
// in-sample), axis 2 (optimize the in-sample grid by total_pips), the Source
// ingestion seam over REAL M1 close bars (M1FieldSource, lazy chunk streaming),
// and registry lineage (persist the per-window OOS reports as a WalkForward
// family). This is the milestone's headline promise end to end.
//
// The task a downstream researcher does: "validate my SMA-cross strategy
// out-of-sample on real EURUSD M1 — roll 6-month in-sample / 1-month OOS windows,
// re-optimize the params in each in-sample slice, run the chosen params on the
// held-out month, and stitch the OOS months into one continuous pip curve."
//
// PUBLIC INTERFACE ONLY: API from the design ledger (C12 axis 3, source seam),
// glossary, specs 0041/0044/0045/0048, and `cargo doc` rustdoc. No crates/*/src
// was read. DataServer is reached for from the external `data-server` crate
// directly, because M1FieldSource::open takes &Arc<DataServer> and aura-ingest
// re-exports neither DataServer nor DEFAULT_DATA_PATH.

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

use aura_core::{Cell, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    f64_field, summarize, sweep, walk_forward, BlueprintNode, Composite, Edge, GridSpace, OutField,
    RollMode, Role, RunManifest, RunReport, Source, SweepFamily, Target, WindowBounds, WindowRoller,
    WindowRun,
};
use aura_ingest::{M1Field, M1FieldSource};
use aura_registry::{group_families, optimize, walkforward_member_reports, FamilyKind, Registry};
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use data_server::{DataServer, DEFAULT_DATA_PATH};

const SYMBOL: &str = "EURUSD";

fn sma_cross() -> Composite {
    Composite::new(
        "sma_cross",
        vec![
            Sma::builder().named("fast").into(),
            Sma::builder().named("slow").into(),
            Sub::builder().into(),
        ],
        vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 },
            Edge { from: 1, to: 2, slot: 1, from_field: 0 },
        ],
        vec![Role {
            name: "price".into(),
            targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
            source: None,
        }],
        vec![OutField { node: 2, field: 0, name: "out".into() }],
    )
}

fn harness_with_sinks() -> (
    Composite,
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
) {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let bp = Composite::new(
        "harness",
        vec![
            BlueprintNode::Composite(sma_cross()),
            Exposure::builder().into(),
            SimBroker::builder(1e-4).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
        ],
        vec![
            Edge { from: 0, to: 1, slot: 0, from_field: 0 },
            Edge { from: 1, to: 2, slot: 0, from_field: 0 },
            Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            Edge { from: 1, to: 4, slot: 0, from_field: 0 },
        ],
        vec![Role {
            name: "price".into(),
            targets: vec![Target { node: 0, slot: 0 }, Target { node: 2, slot: 1 }],
            source: Some(ScalarKind::F64),
        }],
        vec![],
    );
    (bp, rx_eq, rx_ex)
}

// epoch-ns (the roller / WindowBounds unit, == RunManifest.window) back to the
// Unix-ms M1FieldSource::open wants. aura-ingest exports unix_ms_to_epoch_ns but
// NOT its inverse, so a real-data walk-forward must hand-divide here.
fn ns_to_ms(ts: Timestamp) -> i64 {
    ts.0 / 1_000_000
}

// One run of the harness over a real [from, to] M1 close window. Returns the OOS
// equity stream and the folded RunReport. The bounds-keyed producer the #71
// firewall wants lives right here in the closure body.
fn run_window_real(
    server: &Arc<DataServer>,
    params: &[Cell],
    from: Timestamp,
    to: Timestamp,
) -> (Vec<(Timestamp, f64)>, RunReport) {
    let (bp, rx_eq, rx_ex) = harness_with_sinks();
    let mut h = bp
        .bootstrap_with_cells(params)
        .expect("chosen params kind-checked against param_space");
    let src = M1FieldSource::open(server, SYMBOL, Some(ns_to_ms(from)), Some(ns_to_ms(to)), M1Field::Close)
        .expect("real EURUSD window overlaps archived data");
    // Producer-supplied window (C18): the source knows its bounds without
    // materializing the stream.
    let window = src.bounds().unwrap_or((from, to));
    h.run(vec![Box::new(src)]);
    drop(h);
    let eq_rows: Vec<_> = rx_eq.try_iter().collect();
    let ex_rows: Vec<_> = rx_ex.try_iter().collect();
    let equity = f64_field(&eq_rows, 0);
    let exposure = f64_field(&ex_rows, 0);
    let oos_equity: Vec<(Timestamp, f64)> = eq_rows.iter().map(|(t, v)| (*t, v[0].as_f64())).collect();
    let report = RunReport {
        manifest: RunManifest {
            commit: "fieldtest".into(),
            params: vec![],
            window,
            seed: 0,
            broker: "sim-optimal(pip_size=0.0001)".into(),
        },
        metrics: summarize(&equity, &exposure),
    };
    (oos_equity, report)
}

fn main() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if !server.has_symbol(SYMBOL) {
        eprintln!("no local {SYMBOL} archive under {DEFAULT_DATA_PATH}; skipping real-data scenario");
        std::process::exit(0);
    }

    // A 9-month span of EURUSD starting 2015-01-01. is=4 months, oos=1 month,
    // step=1 month, rolling — yields several windows. Units are epoch-ns
    // (RunManifest.window / WindowBounds unit). 1 month ~= 30 days of ns.
    let day_ns: i64 = 86_400_000 * 1_000_000;
    let month_ns: i64 = 30 * day_ns;
    let origin = aura_ingest::unix_ms_to_epoch_ns(1_420_070_400_000); // 2015-01-01 UTC
    let span = (origin, Timestamp(origin.0 + 9 * month_ns));
    let roller = WindowRoller::new(span, 4 * month_ns, month_ns, month_ns, RollMode::Rolling)
        .expect("span fits at least one window");

    // The in-sample grid we re-optimize per window.
    let space = harness_with_sinks().0.param_space();
    println!("param-space slots:");
    for ps in &space {
        println!("  {} : {:?}", ps.name, ps.kind);
    }
    let grid: GridSpace = GridSpace::new(
        &space,
        vec![
            vec![Scalar::I64(3), Scalar::I64(5)],   // sma_cross.fast.length
            vec![Scalar::I64(20), Scalar::I64(50)], // sma_cross.slow.length
            vec![Scalar::F64(1.0)],                 // exposure.scale
        ],
    )
    .expect("grid well-formed for the space");

    let server_for_closure = Arc::clone(&server);
    let space_for_result = space.clone();

    // axis 3: walk_forward varies the DATA WINDOW. Inside the closure: axis 1
    // (grid sweep on the in-sample slice) + axis 2 (optimize by total_pips) +
    // an OOS run on the held-out window. All over REAL M1 close bars.
    let result = walk_forward(roller, space_for_result, move |w: WindowBounds| -> WindowRun {
        // in-sample: sweep the grid over the in-sample window, pick the best params.
        let is_family: SweepFamily = sweep(&grid, |pt: &[Cell]| {
            let (_eq, report) = run_window_real(&server_for_closure, pt, w.is.0, w.is.1);
            report
        });
        let best = optimize(&is_family, "total_pips").expect("total_pips known");
        // out-of-sample: run the chosen params on the held-out window, record equity.
        let (oos_equity, oos_report) =
            run_window_real(&server_for_closure, &best.params, w.oos.0, w.oos.1);
        WindowRun { chosen_params: best.params, oos_equity, oos_report }
    });

    println!("\naxis 3: walk-forward produced {} windows", result.windows.len());
    assert!(!result.windows.is_empty(), "the roll yields at least one window");

    // C2: no look-ahead — every OOS window starts strictly after its IS window ends.
    for (k, w) in result.windows.iter().enumerate() {
        let b = w.bounds;
        assert!(b.oos.0 > b.is.1, "window {k}: oos.0 > is.1 (no look-ahead, C2)");
        let named = result.named_params(k);
        let chosen: Vec<String> = named.iter().map(|(n, v)| format!("{n}={}", scalar_str(v))).collect();
        println!(
            "  window {k}: is=[{}..{}] oos=[{}..{}] chosen {{ {} }} oos_pips={:.4}",
            ns_to_ms(b.is.0), ns_to_ms(b.is.1), ns_to_ms(b.oos.0), ns_to_ms(b.oos.1),
            chosen.join(", "),
            w.run.oos_report.metrics.total_pips,
        );
    }

    // The stitched OOS curve — one continuous pip equity across all OOS months.
    let stitched = &result.stitched_oos_equity;
    println!(
        "\nstitched OOS curve: {} points, final cumulative = {:.4}",
        stitched.len(),
        stitched.last().map(|(_, v)| *v).unwrap_or(0.0),
    );
    assert!(!stitched.is_empty(), "real OOS windows produced equity points");
    // Stitched curve is monotone in timestamp (carried forward across windows).
    for w in stitched.windows(2) {
        assert!(w[1].0 .0 >= w[0].0 .0, "stitched timestamps are non-decreasing");
    }

    // Lineage: persist the per-window OOS reports as a WalkForward family.
    let tmp = std::env::temp_dir().join(format!("mw3-runs-{}.jsonl", std::process::id()));
    let _ = std::fs::remove_file(&tmp);
    let _ = std::fs::remove_file(tmp.with_file_name("families.jsonl"));
    let reg = Registry::open(&tmp);
    let id = reg
        .append_family("eurusd-wfo", FamilyKind::WalkForward, &walkforward_member_reports(&result))
        .expect("wfo family append");
    let families = group_families(reg.load_family_members().expect("load"));
    let f = families.iter().find(|f| f.id == id).expect("our family present");
    println!(
        "\nlineage: family {} kind={:?} members={} (one OOS report per window)",
        f.id, f.kind, f.members.len()
    );
    assert_eq!(f.kind, FamilyKind::WalkForward);
    assert_eq!(f.members.len(), result.windows.len());
    let _ = std::fs::remove_file(&tmp);

    println!("\nOK: real-data walk-forward -> per-window grid-optimize -> stitched OOS -> lineage.");
}

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(),
    }
}
