1// E6 — end-to-end streaming pipeline: FinData export (cached Parquet) → Arrow/Parquet read → session filter → returns →2// realized measures (CPU / Metal), with stage latencies, bytes, peak RSS and the number of full copies of the price3// columns. Baselines (pandas / polars / duckdb) are measured by experiments/scripts/e6_baselines.py on the same files.4// Usage: bench_e6_pipeline <e0_universe.json> <out.jsonl> [n_files] [reps]5#include <cstdio>6#include <fstream>7#include <nlohmann/json.hpp>8#include "raid/core/manifest.hpp"9#include "raid/core/hash.hpp"10#include "raid/data/io.hpp"11#include "raid/data/calendar.hpp"12#include "raid/data/returns.hpp"13#include "raid/estimators/realized.hpp"14using namespace raid; using json = nlohmann::json;15namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } }1617int main(int argc, char** argv) {18 if (argc < 3) { std::fprintf(stderr, "usage: %s e0_universe.json out.jsonl [n_files] [reps]\n", argv[0]); return 2; }19#ifndef RAID_WITH_ARROW20 std::fprintf(stderr, "E6 requires the Arrow build\n"); return 1;21#else22 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);23 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]);24 MarketCalendar cal; int fi = 0;25 for (auto& ex : e0["exports"]) { if (fi++ >= nf) break; const std::string path = ex["path"]; const usize bytes = ex["bytes"];26 for (auto backend_name : {"cpu", "metal"}) for (int rep = 0; rep < reps; ++rep) {27 const u64 rss0 = peak_rss_bytes(); const f64 t0 = now();28 auto pb = data::read_bars_parquet(path); const f64 t_read = now(); // copy 1: Arrow → BarSeries columns (+ sort/dedupe)29 std::map<std::string, data::IntradayReturns> rets; usize bars = 0, ret_n = 0;30 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: returns31 const f64 t_prep = now();32 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 concat33 const f64 t_batch = now();34 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();35 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},36 {"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},37 {"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";38 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); }39 }40 man.save(std::string(argv[2]) + ".manifest.json"); return 0;41#endif42}43