1// E1 — batched realized measures on a real equity panel (FinData 2024 exports, cache) and synthetic panels:2// scaling versus panel width S ∈ {1,10,100,1000,10000(synthetic)} and segment length (1 day = ~390, 1 week ≈ 1950,3// 1 month ≈ 8 580 returns, 1 year ≈ 98 000), RAID CPU 1 thread / CPU all threads / Metal (fused and unfused).4// Usage: bench_e1_batched_rv <e0_universe.json> <out.jsonl> [reps]5#include <cstdio>6#include <fstream>7#include <random>8#include <nlohmann/json.hpp>9#include "raid/core/manifest.hpp"10#include "raid/core/hash.hpp"11#include "raid/core/rng.hpp"12#include "raid/core/thread_pool.hpp"13#include "raid/data/io.hpp"14#include "raid/data/calendar.hpp"15#include "raid/data/returns.hpp"16#include "raid/estimators/realized.hpp"17#include "raid/backend/backend.hpp"18using namespace raid; using json = nlohmann::json;1920namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } }2122int main(int argc, char** argv) {23 if (argc < 3) { std::fprintf(stderr, "usage: %s e0_universe.json out.jsonl [reps]\n", argv[0]); return 2; }24 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);25 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)}};26 man.inputs["e0_universe.json"] = Sha256::of_file(argv[1]); man.parameters["dataset_versions"] = e0["dataset_versions"].dump();27 // ---- load the real panel: per symbol, regular-session intraday log returns for 2024 (CSR by symbol, and by symbol×session)28 MarketCalendar cal; std::map<std::string, data::IntradayReturns> rets; f64 load_s = 0, prep_s = 0; usize bars_total = 0;29#ifdef RAID_WITH_ARROW30 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();31 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; }32#else33 std::fprintf(stderr, "E1 requires the Arrow build\n"); return 1;34#endif35 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);36 std::fprintf(stderr, "loaded %zu symbols (%zu bars, load %.1fs, prep %.1fs)\n", syms.size(), bars_total, load_s, prep_s);37 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";38 auto cpu1 = backend::make_cpu_backend(1), cpuN = backend::make_cpu_backend(0); std::shared_ptr<backend::Backend> gpu;39#ifdef RAID_ENABLE_METAL40 gpu = backend::make_metal_backend();41#endif42 struct Case { std::string source; usize S; std::string seg; usize seg_len; std::string backend; bool fused; };43 std::vector<Case> cases; const std::vector<std::pair<std::string, usize>> segs = {{"day", 1}, {"week", 5}, {"month", 22}, {"year", 252}};44 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"}); }45 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}); }46 std::mt19937_64 order(20261102); std::shuffle(cases.begin(), cases.end(), order);47 usize done = 0;48 for (auto& c : cases) {49 if ((c.backend == "metal" || c.backend == "metal_unfused") && !gpu) continue;50 // Build the batch: segments = consecutive groups of `seg_len` sessions per symbol (real) or synthetic Student-t returns51 std::vector<f64> v; std::vector<i64> off = {0}; const f64 tb0 = now();52 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())); } } }53 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); }54 const f64 build_s = now() - tb0; backend::Batch b{v, off};55 auto be = c.backend == "cpu1" ? cpu1 : (c.backend == "cpuN" ? cpuN : gpu);56 est::RealizedOptions ro; ro.fuse_kernels = c.fused; ro.plan.backend = be->kind();57 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); };58 for (int w = 0; w < 2; ++w) run();59 for (int rep = 0; rep < reps; ++rep) { const f64 t0 = now(); auto m = run(); const f64 wall = now() - t0;60 // accuracy vs CPU FP64 reference (first segment summary) — recorded for the GPU rows61 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]); }62 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"; }63 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());64 }65 man.save(std::string(argv[2]) + ".manifest.json"); return 0;66}67