// E6 — end-to-end streaming pipeline: FinData export (cached Parquet) → Arrow/Parquet read → session filter → returns →
// realized measures (CPU / Metal), with stage latencies, bytes, peak RSS and the number of full copies of the price
// columns. Baselines (pandas / polars / duckdb) are measured by experiments/scripts/e6_baselines.py on the same files.
// Usage: bench_e6_pipeline <e0_universe.json> <out.jsonl> [n_files] [reps]
#include <cstdio>
#include <fstream>
#include <nlohmann/json.hpp>
#include "raid/core/manifest.hpp"
#include "raid/core/hash.hpp"
#include "raid/data/io.hpp"
#include "raid/data/calendar.hpp"
#include "raid/data/returns.hpp"
#include "raid/estimators/realized.hpp"
using namespace raid; using json = nlohmann::json;
namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } }

int main(int argc, char** argv) {
  if (argc < 3) { std::fprintf(stderr, "usage: %s e0_universe.json out.jsonl [n_files] [reps]\n", argv[0]); return 2; }
#ifndef RAID_WITH_ARROW
  std::fprintf(stderr, "E6 requires the Arrow build\n"); return 1;
#else
  json e0 = json::parse(std::ifstream(argv[1])); const int nf = argc > 3 ? std::atoi(argv[3]) : 4; const int reps = argc > 4 ? std::atoi(argv[4]) : 3; std::ofstream out(argv[2], std::ios::app);
  ExperimentManifest man; man.experiment = "E6.pipeline"; man.created_at_utc = now_utc_iso(); man.raid_version = RAID_VERSION; man.host = host_info(); man.inputs["e0_universe.json"] = Sha256::of_file(argv[1]);
  MarketCalendar cal; int fi = 0;
  for (auto& ex : e0["exports"]) { if (fi++ >= nf) break; const std::string path = ex["path"]; const usize bytes = ex["bytes"];
    for (auto backend_name : {"cpu", "metal"}) for (int rep = 0; rep < reps; ++rep) {
      const u64 rss0 = peak_rss_bytes(); const f64 t0 = now();
      auto pb = data::read_bars_parquet(path); const f64 t_read = now();                                  // copy 1: Arrow → BarSeries columns (+ sort/dedupe)
      std::map<std::string, data::IntradayReturns> rets; usize bars = 0, ret_n = 0;
      for (auto& [sym, b] : pb.series) { bars += b.size(); auto reg = b.filter_session(cal, SessionPolicy::regular); rets[sym] = data::intraday_returns(reg); ret_n += rets[sym].r.size(); }   // copy 2: regular filter; copy 3: returns
      const f64 t_prep = now();
      std::vector<f64> v; std::vector<i64> off = {0}; for (auto& [sym, ir] : rets) { for (usize k = 0; k + 1 < ir.session_offsets.size(); ++k) { v.insert(v.end(), ir.r.begin() + ir.session_offsets[k], ir.r.begin() + ir.session_offsets[k + 1]); off.push_back(static_cast<i64>(v.size())); } }   // copy 4: batch concat
      const f64 t_batch = now();
      est::RealizedOptions ro; ro.plan.backend = backend_from_string(backend_name); ro.plan.allow_fallback = true; auto m = est::realized_measures_batch(backend::Batch{v, off}, ro); const f64 t_end = now();
      out << json({{"experiment", "E6"}, {"file", path}, {"file_bytes", bytes}, {"backend_requested", backend_name}, {"backend_used", std::string(to_string(m.stats.backend))}, {"rep", rep}, {"symbols", pb.series.size()}, {"bars", bars}, {"returns", ret_n}, {"segments", off.size() - 1},
                   {"read_seconds", t_read - t0}, {"prep_seconds", t_prep - t_read}, {"batch_seconds", t_batch - t_prep}, {"estimate_seconds", t_end - t_batch}, {"total_seconds", t_end - t0}, {"device_seconds", m.stats.device_seconds},
                   {"bars_per_second", bars / (t_end - t0)}, {"peak_rss_before", rss0}, {"peak_rss_after", peak_rss_bytes()}, {"zero_copy_possible", pb.metadata.count("zero_copy_possible") ? pb.metadata["zero_copy_possible"] : "?"}, {"column_copies", 4}, {"duplicates_dropped", pb.duplicates_dropped}}).dump() << "\n";
      std::fprintf(stderr, "%s %s rep %d: %zu bars in %.2fs (read %.2f prep %.2f batch %.2f est %.3f)\n", path.c_str(), backend_name, rep, bars, t_end - t0, t_read - t0, t_prep - t_read, t_batch - t_prep, t_end - t_batch); }
  }
  man.save(std::string(argv[2]) + ".manifest.json"); return 0;
#endif
}