// E1 — batched realized measures on a real equity panel (FinData 2024 exports, cache) and synthetic panels:
// scaling versus panel width S ∈ {1,10,100,1000,10000(synthetic)} and segment length (1 day = ~390, 1 week ≈ 1950,
// 1 month ≈ 8 580 returns, 1 year ≈ 98 000), RAID CPU 1 thread / CPU all threads / Metal (fused and unfused).
// Usage: bench_e1_batched_rv <e0_universe.json> <out.jsonl> [reps]
#include <cstdio>
#include <fstream>
#include <random>
#include <nlohmann/json.hpp>
#include "raid/core/manifest.hpp"
#include "raid/core/hash.hpp"
#include "raid/core/rng.hpp"
#include "raid/core/thread_pool.hpp"
#include "raid/data/io.hpp"
#include "raid/data/calendar.hpp"
#include "raid/data/returns.hpp"
#include "raid/estimators/realized.hpp"
#include "raid/backend/backend.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 [reps]\n", argv[0]); return 2; }
  json e0 = json::parse(std::ifstream(argv[1])); const int reps = argc > 3 ? std::atoi(argv[3]) : 5; std::ofstream out(argv[2], std::ios::app);
  ExperimentManifest man; man.experiment = "E1.batched_rv"; man.created_at_utc = now_utc_iso(); man.raid_version = RAID_VERSION; man.host = host_info(); man.seed = 20261102; man.parameters = {{"reps", std::to_string(reps)}};
  man.inputs["e0_universe.json"] = Sha256::of_file(argv[1]); man.parameters["dataset_versions"] = e0["dataset_versions"].dump();
  // ---- load the real panel: per symbol, regular-session intraday log returns for 2024 (CSR by symbol, and by symbol×session)
  MarketCalendar cal; std::map<std::string, data::IntradayReturns> rets; f64 load_s = 0, prep_s = 0; usize bars_total = 0;
#ifdef RAID_WITH_ARROW
  for (auto& ex : e0["exports"]) { const f64 t0 = now(); auto pb = data::read_bars_parquet(ex["path"].get<std::string>()); load_s += now() - t0; const f64 t1 = now();
    for (auto& [sym, b] : pb.series) { bars_total += b.size(); auto reg = b.filter_session(cal, SessionPolicy::regular); if (reg.n_sessions() < 200) continue; rets[sym] = data::intraday_returns(reg); } prep_s += now() - t1; }
#else
  std::fprintf(stderr, "E1 requires the Arrow build\n"); return 1;
#endif
  const auto universe = e0["samples"]["1000"].get<std::vector<std::string>>(); std::vector<std::string> syms; for (auto& s : universe) if (rets.count(s)) syms.push_back(s);
  std::fprintf(stderr, "loaded %zu symbols (%zu bars, load %.1fs, prep %.1fs)\n", syms.size(), bars_total, load_s, prep_s);
  out << json({{"kind", "load"}, {"symbols", syms.size()}, {"bars", bars_total}, {"load_seconds", load_s}, {"prep_seconds", prep_s}, {"peak_rss", peak_rss_bytes()}}).dump() << "\n";
  auto cpu1 = backend::make_cpu_backend(1), cpuN = backend::make_cpu_backend(0); std::shared_ptr<backend::Backend> gpu;
#ifdef RAID_ENABLE_METAL
  gpu = backend::make_metal_backend();
#endif
  struct Case { std::string source; usize S; std::string seg; usize seg_len; std::string backend; bool fused; };
  std::vector<Case> cases; const std::vector<std::pair<std::string, usize>> segs = {{"day", 1}, {"week", 5}, {"month", 22}, {"year", 252}};
  for (usize S0 : {1ul, 10ul, 100ul, 1000ul}) for (auto& [sn, nd] : segs) { const usize S = std::min(S0, syms.size()); if (S0 > syms.size() && S0 / 10 >= syms.size()) continue; for (auto bk : {"cpu1", "cpuN", "metal", "metal_unfused"}) cases.push_back({"real", S, sn, nd, bk, std::string(bk) != "metal_unfused"}); }
  for (usize S : {10000ul}) for (auto& [sn, nd] : segs) { if (nd > 22) continue; for (auto bk : {"cpu1", "cpuN", "metal"}) cases.push_back({"synthetic", S, sn, nd, bk, true}); }
  std::mt19937_64 order(20261102); std::shuffle(cases.begin(), cases.end(), order);
  usize done = 0;
  for (auto& c : cases) {
    if ((c.backend == "metal" || c.backend == "metal_unfused") && !gpu) continue;
    // Build the batch: segments = consecutive groups of `seg_len` sessions per symbol (real) or synthetic Student-t returns
    std::vector<f64> v; std::vector<i64> off = {0}; const f64 tb0 = now();
    if (c.source == "real") { for (usize s = 0; s < c.S; ++s) { auto& ir = rets[syms[s]]; const usize ns = ir.session_dates.size(); for (usize k = 0; k + c.seg_len <= ns; k += c.seg_len) { const usize a = static_cast<usize>(ir.session_offsets[k]), b = static_cast<usize>(ir.session_offsets[k + c.seg_len]); v.insert(v.end(), ir.r.begin() + a, ir.r.begin() + b); off.push_back(static_cast<i64>(v.size())); } } }
    else { Rng r(7, c.S); const usize n = 390 * c.seg_len; v.resize(c.S * n); for (auto& x : v) x = 1e-3 * r.student_t(5.0); off.resize(c.S + 1); for (usize s = 0; s <= c.S; ++s) off[s] = static_cast<i64>(s * n); }
    const f64 build_s = now() - tb0; backend::Batch b{v, off};
    auto be = c.backend == "cpu1" ? cpu1 : (c.backend == "cpuN" ? cpuN : gpu);
    est::RealizedOptions ro; ro.fuse_kernels = c.fused; ro.plan.backend = be->kind();
    auto run = [&]() { if (c.backend == "cpu1") { ExecutionPlan p; p.backend = Backend::cpu; p.threads = 1; est::RealizedOptions r1 = ro; r1.plan = p; return est::realized_measures_batch(b, r1); } return est::realized_measures_batch(b, ro); };
    for (int w = 0; w < 2; ++w) run();
    for (int rep = 0; rep < reps; ++rep) { const f64 t0 = now(); auto m = run(); const f64 wall = now() - t0;
      // accuracy vs CPU FP64 reference (first segment summary) — recorded for the GPU rows
      f64 max_rel = 0; if (be->kind() == Backend::metal) { est::RealizedOptions rc = ro; rc.plan.backend = Backend::cpu; auto ref = est::realized_measures_batch(b, rc); for (usize s = 0; s < m.rv.size(); ++s) if (ref.rv[s] > 0) max_rel = std::max(max_rel, std::abs(m.rv[s] - ref.rv[s]) / ref.rv[s]); }
      out << json({{"kind", "bench"}, {"experiment", "E1"}, {"source", c.source}, {"S", c.S}, {"segments", off.size() - 1}, {"segment", c.seg}, {"segment_sessions", c.seg_len}, {"elements", v.size()}, {"backend", c.backend}, {"fused", c.fused}, {"rep", rep}, {"wall_seconds", wall}, {"kernel_wall_seconds", m.stats.wall_seconds}, {"device_seconds", m.stats.device_seconds}, {"build_seconds", build_s}, {"max_rel_err_rv_vs_fp64", max_rel}, {"precision", std::string(to_string(m.stats.precision))}, {"threads", c.backend == "cpu1" ? 1u : ThreadPool::global().size()}, {"peak_rss", peak_rss_bytes()}}).dump() << "\n"; }
    out.flush(); ++done; std::fprintf(stderr, "[%zu/%zu] %s S=%zu seg=%s %s elements=%zu\n", done, cases.size(), c.source.c_str(), c.S, c.seg.c_str(), c.backend.c_str(), v.size());
  }
  man.save(std::string(argv[2]) + ".manifest.json"); return 0;
}