WP40 RAID: Phase 6 — E1–E6 drivers, E5 results (R=1000), E7 generation sweep (11 nodes, cold/warm compile), e0 universe, references.bib (Crossref-verified)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
15 changed files +3,330 −2
modified
backends/metal/host/metal_backend.mm
+7 −1
@@ -10,6 +10,7 @@ | ||
| 10 | 10 | #include <map> |
| 11 | 11 | #include <mutex> |
| 12 | 12 | #include <unistd.h> |
| 13 | +#include <cstdlib> | |
| 13 | 14 | #include <mach-o/dyld.h> |
| 14 | 15 | |
| 15 | 16 | namespace raid { namespace metal { const char* msl_bundle(); } } |
@@ -51,12 +52,14 @@ bool ensure_library(KernelStats* st) { | ||
| 51 | 52 | NSError* err = nil; Timer t; |
| 52 | 53 | // 1. prebuilt metallib: RAID_METALLIB env, compile-time path, next to the executable |
| 53 | 54 | std::vector<std::string> candidates; |
| 55 | + const bool runtime_only = std::getenv("RAID_METAL_RUNTIME_ONLY") != nullptr; // force runtime MSL compilation (benchmarks) | |
| 54 | 56 | if (const char* e = std::getenv("RAID_METALLIB")) candidates.push_back(e); |
| 55 | 57 | #ifdef RAID_METALLIB_PATH |
| 56 | 58 | candidates.push_back(RAID_METALLIB_PATH); |
| 57 | 59 | #endif |
| 58 | 60 | { char buf[4096]; uint32_t sz = sizeof buf; if (_NSGetExecutablePath(buf, &sz) == 0) { std::string p(buf); auto k = p.rfind('/'); if (k != std::string::npos) candidates.push_back(p.substr(0, k) + "/raid.metallib"); } } |
| 59 | 61 | for (auto& c : candidates) { |
| 62 | + if (runtime_only) break; | |
| 60 | 63 | if (access(c.c_str(), R_OK) != 0) continue; |
| 61 | 64 | library_ = [device_ newLibraryWithURL:[NSURL fileURLWithPath:[NSString stringWithUTF8String:c.c_str()]] error:&err]; |
| 62 | 65 | if (library_) { library_source_ = "metallib:" + c; compile_seconds_ = t.s(); break; } |
@@ -64,7 +67,10 @@ bool ensure_library(KernelStats* st) { | ||
| 64 | 67 | // 2. runtime compilation of the embedded MSL |
| 65 | 68 | if (!library_) { |
| 66 | 69 | MTLCompileOptions* opts = [MTLCompileOptions new]; opts.fastMathEnabled = NO; // keep IEEE semantics (no reassociation): required by the double-float transforms |
| 67 | − library_ = [device_ newLibraryWithSource:[NSString stringWithUTF8String:raid::metal::msl_bundle()] options:opts error:&err]; | |
| 70 | + // RAID_MSL_SALT appends a unique comment so that the system shader cache cannot serve a previous compilation | |
| 71 | + // (used to measure genuinely cold runtime compilation in E7). | |
| 72 | + std::string src = raid::metal::msl_bundle(); if (const char* salt = std::getenv("RAID_MSL_SALT")) src += "\n// salt " + std::string(salt) + "\n"; | |
| 73 | + library_ = [device_ newLibraryWithSource:[NSString stringWithUTF8String:src.c_str()] options:opts error:&err]; | |
| 68 | 74 | if (!library_) { log::error("metal: shader compilation failed", {{"error", err ? std::string([[err localizedDescription] UTF8String]) : "?"}}); available_ = false; return false; } |
| 69 | 75 | library_source_ = "runtime"; compile_seconds_ = t.s(); if (st) st->compiled_now = true; |
| 70 | 76 | } |
added
benchmarks/end_to_end/bench_e1_batched_rv.cpp
+66 −0
@@ -0,0 +1,66 @@ | ||
| 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" | |
| 18 | +using namespace raid; using json = nlohmann::json; | |
| 19 | + | |
| 20 | +namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } } | |
| 21 | + | |
| 22 | +int 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_ARROW | |
| 30 | + 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 | +#else | |
| 33 | + std::fprintf(stderr, "E1 requires the Arrow build\n"); return 1; | |
| 34 | +#endif | |
| 35 | + 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_METAL | |
| 40 | + gpu = backend::make_metal_backend(); | |
| 41 | +#endif | |
| 42 | + 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 S : {1ul, 10ul, 100ul, 1000ul}) for (auto& [sn, nd] : segs) { if (S > 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 returns | |
| 51 | + 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 rows | |
| 61 | + 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 | +} | |
added
benchmarks/end_to_end/bench_e2_har.cpp
+49 −0
@@ -0,0 +1,49 @@ | ||
| 1 | +// E2 — HAR-RV on the real daily RV panel (from the 2024 1-minute exports): (a) rolling causal forecasts h ∈ {1,5,22} | |
| 2 | +// with QLIKE/MSE vs a random-walk benchmark (RV_{t+h} forecast = RV_t) on a strict chronological split, per symbol; | |
| 3 | +// (b) runtime of per-series QR fits (CPU) vs batched sufficient-statistics fits (CPU-MT, Metal) for S ∈ {10,100,1000}. | |
| 4 | +// Note: 2024 alone gives ≈ 252 daily RVs per symbol — rolling windows are short (120 days); the point is runtime and | |
| 5 | +// relative accuracy, not a new forecasting result. Usage: bench_e2_har <e0_universe.json> <out.jsonl> [reps] | |
| 6 | +#include <cstdio> | |
| 7 | +#include <fstream> | |
| 8 | +#include <random> | |
| 9 | +#include <nlohmann/json.hpp> | |
| 10 | +#include "raid/core/manifest.hpp" | |
| 11 | +#include "raid/core/hash.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/estimators/har.hpp" | |
| 18 | +using namespace raid; using json = nlohmann::json; | |
| 19 | +namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } } | |
| 20 | + | |
| 21 | +int main(int argc, char** argv) { | |
| 22 | + if (argc < 3) { std::fprintf(stderr, "usage: %s e0_universe.json out.jsonl [reps]\n", argv[0]); return 2; } | |
| 23 | +#ifndef RAID_WITH_ARROW | |
| 24 | + return 1; | |
| 25 | +#else | |
| 26 | + 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); | |
| 27 | + ExperimentManifest man; man.experiment = "E2.har"; 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]); | |
| 28 | + MarketCalendar cal; std::map<std::string, std::vector<f64>> rv, rq; | |
| 29 | + for (auto& ex : e0["exports"]) { auto pb = data::read_bars_parquet(ex["path"].get<std::string>()); for (auto& [sym, b] : pb.series) { auto reg = b.filter_session(cal, SessionPolicy::regular); if (reg.n_sessions() < 240) continue; auto ir = data::intraday_returns(reg); est::RealizedOptions ro; ro.plan.backend = Backend::cpu; auto m = est::realized_measures(ir, ro); bool ok = true; for (f64 v : m.rv) if (!(v > 0)) ok = false; if (ok) { rv[sym] = m.rv; rq[sym] = m.rq; } } } | |
| 30 | + const auto universe = e0["samples"]["1000"].get<std::vector<std::string>>(); std::vector<std::string> syms; for (auto& s : universe) if (rv.count(s)) syms.push_back(s); | |
| 31 | + std::fprintf(stderr, "E2: %zu symbols with >= 240 sessions of RV\n", syms.size()); out << json({{"kind", "load"}, {"experiment", "E2"}, {"symbols", syms.size()}}).dump() << "\n"; | |
| 32 | + // (a) forecasting accuracy: window 120, step 1, horizons 1/5/22, HAR-RV vs log-HAR vs HARQ vs random walk; first 100 symbols | |
| 33 | + const usize nacc = std::min<usize>(100, syms.size()); | |
| 34 | + for (int h : {1, 5, 22}) { std::vector<json> rows(nacc); | |
| 35 | + ThreadPool::global().parallel_for(nacc, [&](usize a, usize b) { for (usize i = a; i < b; ++i) { auto& x = rv[syms[i]]; auto& q = rq[syms[i]]; json r = {{"symbol", syms[i]}, {"h", h}, {"T", x.size()}}; | |
| 36 | + for (auto& [name, lg, hq] : std::vector<std::tuple<std::string, bool, bool>>{{"har", false, false}, {"loghar", true, false}, {"harq", false, true}}) { est::HarSpec s; s.horizon = h; s.log = lg; s.harq = hq; try { auto roll = est::har_rolling(x, s, 120, 1, hq ? std::span<const f64>(q) : std::span<const f64>()); auto l = est::forecast_loss(roll.actual, roll.forecast); r[name] = {{"qlike", l.qlike}, {"mse", l.mse}, {"n", l.n}, {"fits", roll.n_fits}, {"seconds", roll.seconds}}; if (name == "har") { std::vector<f64> rw(roll.origin.size()); for (usize j = 0; j < rw.size(); ++j) rw[j] = x[roll.origin[j]]; auto lr = est::forecast_loss(roll.actual, rw); r["random_walk"] = {{"qlike", lr.qlike}, {"mse", lr.mse}, {"n", lr.n}}; } } catch (const RaidError& ex) { r[name] = {{"error", ex.what()}}; } } | |
| 37 | + rows[i] = r; } }); | |
| 38 | + for (auto& r : rows) { r["kind"] = "accuracy"; r["experiment"] = "E2"; out << r.dump() << "\n"; } out.flush(); std::fprintf(stderr, "E2 accuracy h=%d done\n", h); } | |
| 39 | + // (b) runtime: per-series fits vs batched sufficient statistics | |
| 40 | + for (usize S : {10ul, 100ul, 1000ul}) { if (S > syms.size()) continue; std::vector<f64> cat; std::vector<i64> off = {0}; for (usize i = 0; i < S; ++i) { cat.insert(cat.end(), rv[syms[i]].begin(), rv[syms[i]].end()); off.push_back(static_cast<i64>(cat.size())); } | |
| 41 | + est::HarSpec spec; spec.horizon = 1; spec.cov.type = stats::CovType::classical; | |
| 42 | + for (auto mode : {"per_series_cpu1", "per_series_cpuMT", "batched_cpu", "batched_metal"}) for (int rep = 0; rep < reps; ++rep) { const f64 t0 = now(); usize okc = 0; std::string backend_used = "cpu"; f64 dev = 0; | |
| 43 | + if (std::string(mode).rfind("per_series", 0) == 0) { std::vector<int> ok(S, 0); auto fitall = [&](usize a, usize b) { for (usize i = a; i < b; ++i) { try { auto e = est::har_fit(rv[syms[i]], spec); ok[i] = e.ok(); } catch (...) {} } }; if (std::string(mode) == "per_series_cpu1") fitall(0, S); else ThreadPool::global().parallel_for(S, fitall); for (int v : ok) okc += v; } | |
| 44 | + else { ExecutionPlan p; p.backend = std::string(mode) == "batched_metal" ? Backend::metal : Backend::cpu; auto b = est::har_batch(cat, off, spec, p); for (auto& e : b.fits) okc += e.ok(); backend_used = std::string(to_string(b.stats.backend)); dev = b.stats.device_seconds; } | |
| 45 | + out << json({{"kind", "runtime"}, {"experiment", "E2"}, {"S", S}, {"mode", mode}, {"backend_used", backend_used}, {"rep", rep}, {"wall_seconds", now() - t0}, {"device_seconds", dev}, {"converged", okc}, {"T_per_series", cat.size() / S}}).dump() << "\n"; } | |
| 46 | + out.flush(); std::fprintf(stderr, "E2 runtime S=%zu done\n", S); } | |
| 47 | + man.save(std::string(argv[2]) + ".manifest.json"); return 0; | |
| 48 | +#endif | |
| 49 | +} | |
added
benchmarks/end_to_end/bench_e3_garch.cpp
+58 −0
@@ -0,0 +1,58 @@ | ||
| 1 | +// E3 — GARCH-family panel estimation on real daily close-to-close returns (2024, from the 1-minute exports): | |
| 2 | +// per-symbol CPU QML fits (GARCH, GJR, EGARCH; Gaussian and Student-t): convergence rates, iterations, time, SE; | |
| 3 | +// GPU batched likelihood grid (P = 256 parameter vectors per series) used (i) as a profile-likelihood screen and | |
| 4 | +// (ii) to choose starting values — effect on convergence and time; 1-day-ahead variance forecasts evaluated against | |
| 5 | +// next-day RV (QLIKE) on the last 40 sessions (strict chronological split). Usage: bench_e3_garch <e0.json> <out.jsonl> [S] | |
| 6 | +#include <cstdio> | |
| 7 | +#include <fstream> | |
| 8 | +#include <nlohmann/json.hpp> | |
| 9 | +#include "raid/core/manifest.hpp" | |
| 10 | +#include "raid/core/hash.hpp" | |
| 11 | +#include "raid/core/thread_pool.hpp" | |
| 12 | +#include "raid/data/io.hpp" | |
| 13 | +#include "raid/data/calendar.hpp" | |
| 14 | +#include "raid/data/returns.hpp" | |
| 15 | +#include "raid/estimators/realized.hpp" | |
| 16 | +#include "raid/estimators/har.hpp" | |
| 17 | +#include "raid/models/garch.hpp" | |
| 18 | +#include "raid/backend/backend.hpp" | |
| 19 | +using namespace raid; using json = nlohmann::json; | |
| 20 | +namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } } | |
| 21 | + | |
| 22 | +int main(int argc, char** argv) { | |
| 23 | + if (argc < 3) { std::fprintf(stderr, "usage: %s e0_universe.json out.jsonl [S]\n", argv[0]); return 2; } | |
| 24 | +#ifndef RAID_WITH_ARROW | |
| 25 | + return 1; | |
| 26 | +#else | |
| 27 | + json e0 = json::parse(std::ifstream(argv[1])); const usize Smax = argc > 3 ? static_cast<usize>(std::atoi(argv[3])) : 1000; std::ofstream out(argv[2], std::ios::app); | |
| 28 | + ExperimentManifest man; man.experiment = "E3.garch"; 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]); | |
| 29 | + MarketCalendar cal; std::map<std::string, std::vector<f64>> ret, rvd; | |
| 30 | + for (auto& ex : e0["exports"]) { auto pb = data::read_bars_parquet(ex["path"].get<std::string>()); for (auto& [sym, b] : pb.series) { auto reg = b.filter_session(cal, SessionPolicy::regular); if (reg.n_sessions() < 240) continue; auto ir = data::intraday_returns(reg); std::vector<f64> r; for (usize k = 1; k < ir.close_to_close.size(); ++k) r.push_back(ir.close_to_close[k]); est::RealizedOptions ro; ro.plan.backend = Backend::cpu; auto m = est::realized_measures(ir, ro); ret[sym] = r; rvd[sym] = std::vector<f64>(m.rv.begin() + 1, m.rv.end()); } } | |
| 31 | + const auto universe = e0["samples"]["1000"].get<std::vector<std::string>>(); std::vector<std::string> syms; for (auto& s : universe) if (ret.count(s) && syms.size() < Smax) syms.push_back(s); | |
| 32 | + const usize S = syms.size(); std::fprintf(stderr, "E3: %zu symbols, ~%zu daily returns each\n", S, ret[syms[0]].size()); | |
| 33 | + const usize test = 40; // last 40 sessions held out for forecast evaluation | |
| 34 | + // ---- (i) per-symbol fits, all models, CPU parallel over symbols | |
| 35 | + struct Spec { const char* name; models::GarchModel m; models::GarchDist d; }; const std::vector<Spec> specs = {{"garch.n", models::GarchModel::garch, models::GarchDist::gaussian}, {"gjr.n", models::GarchModel::gjr, models::GarchDist::gaussian}, {"egarch.n", models::GarchModel::egarch, models::GarchDist::gaussian}, {"garch.t", models::GarchModel::garch, models::GarchDist::student_t}}; | |
| 36 | + for (auto& sp : specs) { std::vector<json> rows(S); const f64 t0 = now(); | |
| 37 | + ThreadPool::global().parallel_for(S, [&](usize a, usize b) { for (usize i = a; i < b; ++i) { auto& r = ret[syms[i]]; std::vector<f64> train(r.begin(), r.end() - static_cast<long>(test)); models::GarchFitOptions o; o.model = sp.m; o.dist = sp.d; o.forecast_horizon = 1; json row = {{"symbol", syms[i]}, {"model", sp.name}, {"n_train", train.size()}}; | |
| 38 | + try { const f64 f0 = now(); auto e = models::garch_fit(train, o); row["seconds"] = now() - f0; row["converged"] = e.ok(); row["iterations"] = e.convergence.iterations; row["fevals"] = e.convergence.function_evaluations; row["loglik"] = -e.loss; row["coef"] = e.coefficients; row["se"] = e.inference ? e.inference->std_errors : std::vector<f64>{}; row["persistence"] = e.extra.value("persistence", 0.0); row["warnings"] = e.warnings; | |
| 39 | + // rolling 1-step forecasts over the test window WITHOUT refitting (parameters fixed at the training estimate; variance path updated with realized returns): QLIKE vs next-day RV | |
| 40 | + if (e.ok()) { std::vector<f64> fc, act; for (usize k = 0; k < test; ++k) { std::vector<f64> hist(r.begin(), r.end() - static_cast<long>(test - k)); auto f = models::garch_forecast(e, hist, 1); fc.push_back(f[0]); act.push_back(rvd[syms[i]][r.size() - test + k]); } auto l = est::forecast_loss(act, fc); row["qlike_vs_rv"] = l.qlike; row["mse_vs_rv"] = l.mse; row["n_forecasts"] = l.n; std::vector<f64> naive; for (usize k = 0; k < test; ++k) naive.push_back(rvd[syms[i]][r.size() - test + k - 1]); auto ln = est::forecast_loss(act, naive); row["qlike_rw_rv"] = ln.qlike; } } | |
| 41 | + catch (const RaidError& ex) { row["error"] = ex.what(); row["converged"] = false; } rows[i] = row; } }); | |
| 42 | + usize conv = 0; for (auto& r : rows) { conv += r.value("converged", false); r["kind"] = "fit"; r["experiment"] = "E3"; out << r.dump() << "\n"; } | |
| 43 | + out << json({{"kind", "fit_summary"}, {"experiment", "E3"}, {"model", sp.name}, {"S", S}, {"converged", conv}, {"wall_seconds_all", now() - t0}, {"threads", ThreadPool::global().size()}}).dump() << "\n"; out.flush(); | |
| 44 | + std::fprintf(stderr, "E3 %s: %zu/%zu converged in %.1fs\n", sp.name, conv, S, now() - t0); } | |
| 45 | + // ---- (ii) GPU grid screen: P = 256 (α, β, ω) combinations per series (μ = sample mean), batched NLL on Metal vs CPU-MT; best grid point as starting value → refit convergence/time | |
| 46 | + { const usize P = 256; std::vector<f64> cat; std::vector<i64> off = {0}; std::vector<f64> params(S * P * 4); for (usize i = 0; i < S; ++i) { auto& r = ret[syms[i]]; std::vector<f64> train(r.begin(), r.end() - static_cast<long>(test)); cat.insert(cat.end(), train.begin(), train.end()); off.push_back(static_cast<i64>(cat.size())); f64 mu = 0, v = 0; for (f64 x : train) mu += x; mu /= train.size(); for (f64 x : train) v += (x - mu) * (x - mu); v /= train.size(); | |
| 47 | + usize q = 0; for (int ia = 0; ia < 16; ++ia) for (int ib = 0; ib < 16; ++ib) { const f64 alpha = 0.01 + 0.015 * ia, beta = 0.50 + 0.031 * ib; f64* th = params.data() + (i * P + q) * 4; th[0] = mu; th[2] = alpha; th[3] = beta; th[1] = alpha + beta < 0.999 ? v * (1 - alpha - beta) : v * 0.001; ++q; } } | |
| 48 | + backend::Batch b{cat, off}; auto cpuN = backend::make_cpu_backend(0); std::shared_ptr<backend::Backend> gpu; | |
| 49 | +#ifdef RAID_ENABLE_METAL | |
| 50 | + gpu = backend::make_metal_backend(); | |
| 51 | +#endif | |
| 52 | + std::vector<f64> nll_cpu, nll_gpu; for (int rep = 0; rep < 3; ++rep) { backend::KernelStats st; const f64 t0 = now(); nll_cpu = cpuN->garch_nll_batch(b, params, P, 4, 0, 0.0, &st); out << json({{"kind", "grid"}, {"experiment", "E3"}, {"backend", "cpuMT"}, {"S", S}, {"P", P}, {"rep", rep}, {"wall_seconds", now() - t0}}).dump() << "\n"; | |
| 53 | + if (gpu) { const f64 t1 = now(); nll_gpu = gpu->garch_nll_batch(b, params, P, 4, 0, 0.0, &st); f64 maxrel = 0; usize agree = 0; for (usize i = 0; i < S; ++i) { usize ac = 0, ag = 0; for (usize q = 1; q < P; ++q) { if (nll_cpu[i * P + q] < nll_cpu[i * P + ac]) ac = q; if (nll_gpu[i * P + q] < nll_gpu[i * P + ag]) ag = q; } agree += (ac == ag); for (usize q = 0; q < P; ++q) if (std::isfinite(nll_cpu[i * P + q]) && std::isfinite(nll_gpu[i * P + q])) maxrel = std::max(maxrel, std::abs(nll_cpu[i * P + q] - nll_gpu[i * P + q]) / std::abs(nll_cpu[i * P + q])); } | |
| 54 | + out << json({{"kind", "grid"}, {"experiment", "E3"}, {"backend", "metal"}, {"S", S}, {"P", P}, {"rep", rep}, {"wall_seconds", now() - t1}, {"device_seconds", st.device_seconds}, {"max_rel_err_nll", maxrel}, {"argmin_agreement", agree}, {"S_total", S}}).dump() << "\n"; } } | |
| 55 | + out.flush(); std::fprintf(stderr, "E3 grid screen done\n"); } | |
| 56 | + man.save(std::string(argv[2]) + ".manifest.json"); return 0; | |
| 57 | +#endif | |
| 58 | +} | |
added
benchmarks/end_to_end/bench_e4_connectedness.cpp
+55 −0
@@ -0,0 +1,55 @@ | ||
| 1 | +// E4 — realized covariance and dynamic connectedness on real data: N ∈ {10, 30, 100} symbols, 5-minute synchronized | |
| 2 | +// returns (calendar-aware aggregation of the 1-minute bars) → daily realized covariance matrices (CPU; conditioning | |
| 3 | +// numbers, Ledoit–Wolf shrinkage), daily log-RV panel → rolling VAR(1) Diebold–Yilmaz total connectedness (window 120, | |
| 4 | +// H = 10): runtime, failures, totals. Memory per window and the N where VAR(1)+intercept becomes rank-limited are recorded. | |
| 5 | +// Usage: bench_e4_connectedness <e0.json> <out.jsonl> | |
| 6 | +#include <cstdio> | |
| 7 | +#include <fstream> | |
| 8 | +#include <set> | |
| 9 | +#include <algorithm> | |
| 10 | +#include <nlohmann/json.hpp> | |
| 11 | +#include "raid/core/manifest.hpp" | |
| 12 | +#include "raid/core/hash.hpp" | |
| 13 | +#include "raid/data/io.hpp" | |
| 14 | +#include "raid/data/calendar.hpp" | |
| 15 | +#include "raid/data/returns.hpp" | |
| 16 | +#include "raid/data/resample.hpp" | |
| 17 | +#include "raid/estimators/realized.hpp" | |
| 18 | +#include "raid/models/var.hpp" | |
| 19 | +#include "raid/core/thread_pool.hpp" | |
| 20 | +using namespace raid; using json = nlohmann::json; | |
| 21 | +namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } } | |
| 22 | + | |
| 23 | +int main(int argc, char** argv) { | |
| 24 | + if (argc < 3) { std::fprintf(stderr, "usage: %s e0_universe.json out.jsonl\n", argv[0]); return 2; } | |
| 25 | +#ifndef RAID_WITH_ARROW | |
| 26 | + return 1; | |
| 27 | +#else | |
| 28 | + json e0 = json::parse(std::ifstream(argv[1])); std::ofstream out(argv[2], std::ios::app); | |
| 29 | + ExperimentManifest man; man.experiment = "E4.connectedness"; 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]); | |
| 30 | + MarketCalendar cal; const auto universe = e0["samples"]["1000"].get<std::vector<std::string>>(); | |
| 31 | + // load up to 100 symbols with full coverage; 5-minute regular-session bars; synchronized grid = the 79 bins per session | |
| 32 | + std::map<std::string, BarSeries> five; std::map<std::string, std::vector<f64>> logrv; | |
| 33 | + for (auto& ex : e0["exports"]) { if (five.size() >= 100) break; auto pb = data::read_bars_parquet(ex["path"].get<std::string>()); for (auto& [sym, b] : pb.series) { if (five.size() >= 100) break; auto reg = b.filter_session(cal, SessionPolicy::regular); auto q = reg.validate(cal); if (q.n_sessions < 250 || q.coverage < 0.98) continue; five[sym] = data::aggregate(reg, Interval::m5, cal); auto ir = data::intraday_returns(reg); est::RealizedOptions ro; ro.plan.backend = Backend::cpu; auto m = est::realized_measures(ir, ro); std::vector<f64> lv; for (f64 v : m.rv) lv.push_back(std::log(std::max(v, 1e-12))); logrv[sym] = lv; } } | |
| 34 | + std::vector<std::string> syms; for (auto& s : universe) if (five.count(s)) syms.push_back(s); std::fprintf(stderr, "E4: %zu symbols with >= 250 sessions and coverage >= 0.98\n", syms.size()); | |
| 35 | + // common sessions | |
| 36 | + std::set<Date> common; bool first = true; for (auto& s : syms) { std::set<Date> d(five[s].session_dates.begin(), five[s].session_dates.end()); if (first) { common = d; first = false; } else { std::set<Date> inter; for (auto x : common) if (d.count(x)) inter.insert(x); common = inter; } } | |
| 37 | + std::vector<Date> sessions(common.begin(), common.end()); std::fprintf(stderr, "E4: %zu common sessions\n", sessions.size()); | |
| 38 | + for (usize N : {10ul, 30ul, 100ul}) { if (N > syms.size()) continue; | |
| 39 | + // (a) daily realized covariance from synchronized 5-minute returns (bins matched by timestamp; a day is used only if all N have the full 79 bins) | |
| 40 | + const f64 t0 = now(); usize days_ok = 0, days_skipped = 0; f64 cond_max = 0, cond_med = 0; std::vector<f64> conds; f64 lw_shrink_mean = 0; | |
| 41 | + for (auto d : sessions) { std::vector<std::vector<f64>> R(N); bool ok = true; for (usize i = 0; i < N && ok; ++i) { auto& b = five[syms[i]]; usize k = 0; while (k < b.n_sessions() && b.session_dates[k] != d) ++k; if (k == b.n_sessions()) { ok = false; break; } auto c = b.session_closes(k); const usize expect = cal.expected_regular_bars(d, Interval::m5); if (c.size() != expect) { ok = false; break; } R[i] = data::returns(c); } | |
| 42 | + if (!ok) { ++days_skipped; continue; } const usize M = R[0].size(); Eigen::MatrixXd Rm(static_cast<Eigen::Index>(M), static_cast<Eigen::Index>(N)); for (usize i = 0; i < N; ++i) for (usize j = 0; j < M; ++j) Rm(static_cast<Eigen::Index>(j), static_cast<Eigen::Index>(i)) = R[i][j]; | |
| 43 | + auto rc = est::realized_covariance(Rm); Eigen::SelfAdjointEigenSolver<Eigen::MatrixXd> es(rc.cov); const f64 cond = es.eigenvalues().maxCoeff() / std::max(es.eigenvalues().minCoeff(), 1e-300); conds.push_back(cond); cond_max = std::max(cond_max, cond); auto lw = models::ledoit_wolf(Rm, true); lw_shrink_mean += lw.shrinkage; ++days_ok; } | |
| 44 | + if (!conds.empty()) { std::sort(conds.begin(), conds.end()); cond_med = conds[conds.size() / 2]; lw_shrink_mean /= days_ok; } | |
| 45 | + out << json({{"kind", "rcov"}, {"experiment", "E4"}, {"N", N}, {"days_ok", days_ok}, {"days_skipped", days_skipped}, {"M_per_day", 78}, {"cond_median", cond_med}, {"cond_max", cond_max}, {"rank_deficient_possible", N > 78}, {"ledoit_wolf_shrinkage_mean", lw_shrink_mean}, {"wall_seconds", now() - t0}}).dump() << "\n"; | |
| 46 | + // (b) rolling Diebold–Yilmaz on the daily log-RV panel (aligned on common sessions) | |
| 47 | + const usize T = sessions.size(); Eigen::MatrixXd Y(static_cast<Eigen::Index>(T), static_cast<Eigen::Index>(N)); | |
| 48 | + for (usize i = 0; i < N; ++i) { auto& b = five[syms[i]]; auto& lv = logrv[syms[i]]; for (usize t = 0; t < T; ++t) { usize k = 0; while (b.session_dates[k] != sessions[t]) ++k; Y(static_cast<Eigen::Index>(t), static_cast<Eigen::Index>(i)) = lv[k]; } } | |
| 49 | + for (usize window : {120ul, 180ul}) { if (window >= T || window <= N + 12) { out << json({{"kind", "dy"}, {"experiment", "E4"}, {"N", N}, {"window", window}, {"skipped", "window must exceed N·p + p + 10 and be < T"}, {"T", T}}).dump() << "\n"; continue; } | |
| 50 | + for (int rep = 0; rep < 3; ++rep) { const f64 t1 = now(); auto r = models::rolling_connectedness(Y, 1, 10, window, 1); f64 mean_total = 0; usize nn = 0; for (f64 v : r.total) if (std::isfinite(v)) { mean_total += v; ++nn; } if (nn) mean_total /= nn; | |
| 51 | + out << json({{"kind", "dy"}, {"experiment", "E4"}, {"N", N}, {"window", window}, {"rep", rep}, {"n_windows", r.n_windows}, {"failures", r.failures}, {"mean_total", mean_total}, {"wall_seconds", now() - t1}, {"threads", ThreadPool::global().size()}, {"peak_rss", peak_rss_bytes()}}).dump() << "\n"; } } | |
| 52 | + out.flush(); std::fprintf(stderr, "E4 N=%zu done\n", N); } | |
| 53 | + man.save(std::string(argv[2]) + ".manifest.json"); return 0; | |
| 54 | +#endif | |
| 55 | +} | |
added
benchmarks/end_to_end/bench_e5_inference.cpp
+48 −0
@@ -0,0 +1,48 @@ | ||
| 1 | +// E5 — inference at scale and reproducibility: (a) GARCH(1,1) Monte Carlo R = 1 000 (CPU parallel): bias, RMSE, coverage, | |
| 2 | +// failures, time; (b) HAR-RV h=1/5 HAC coverage R = 1 000; (c) stationary-bootstrap RV intervals on real sessions | |
| 3 | +// (coverage of the "long-sample" IV proxy is not identifiable — we report interval widths and reproducibility); | |
| 4 | +// (d) bitwise reproducibility: same seed → identical bootstrap statistics across thread counts, and identical Metal | |
| 5 | +// realized measures across runs; CPU-vs-Metal differences reported in ulps-equivalent relative error. | |
| 6 | +// Usage: bench_e5_inference <out.jsonl> [R] | |
| 7 | +#include <cstdio> | |
| 8 | +#include <fstream> | |
| 9 | +#include <atomic> | |
| 10 | +#include <nlohmann/json.hpp> | |
| 11 | +#include "raid/core/manifest.hpp" | |
| 12 | +#include "raid/core/rng.hpp" | |
| 13 | +#include "raid/core/thread_pool.hpp" | |
| 14 | +#include "raid/models/garch.hpp" | |
| 15 | +#include "raid/estimators/har.hpp" | |
| 16 | +#include "raid/estimators/realized.hpp" | |
| 17 | +#include "raid/statistics/bootstrap.hpp" | |
| 18 | +#include "raid/statistics/reductions.hpp" | |
| 19 | +using namespace raid; using json = nlohmann::json; | |
| 20 | +namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } } | |
| 21 | + | |
| 22 | +int main(int argc, char** argv) { | |
| 23 | + if (argc < 2) { std::fprintf(stderr, "usage: %s out.jsonl [R]\n", argv[0]); return 2; } | |
| 24 | + const int R = argc > 2 ? std::atoi(argv[2]) : 1000; std::ofstream out(argv[1], std::ios::app); | |
| 25 | + ExperimentManifest man; man.experiment = "E5.inference"; man.created_at_utc = now_utc_iso(); man.raid_version = RAID_VERSION; man.host = host_info(); man.seed = 20261102; man.parameters = {{"R", std::to_string(R)}}; | |
| 26 | + // (a) GARCH MC | |
| 27 | + for (auto& [n, label] : std::vector<std::pair<usize, const char*>>{{500, "n500"}, {2000, "n2000"}}) { const double tr[4] = {0.0, 0.02, 0.08, 0.90}; std::vector<std::array<double, 4>> est(R), se(R); std::vector<int> okv(R, 0), its(R, 0); const f64 t0 = now(); | |
| 28 | + ThreadPool::global().parallel_for(R, [&](usize a, usize b) { for (usize rep = a; rep < b; ++rep) { Rng rng(20261102, rep); std::vector<double> r(n); double s2 = tr[1] / (1 - tr[2] - tr[3]), e = 0; for (usize t = 0; t < n; ++t) { s2 = tr[1] + tr[2] * e * e + tr[3] * s2; e = std::sqrt(s2) * rng.normal(); r[t] = e; } models::GarchFitOptions o; auto f = models::garch_fit(r, o); its[rep] = f.convergence.iterations; if (!f.ok()) continue; okv[rep] = 1; for (int j = 0; j < 4; ++j) { est[rep][j] = f.coefficients[j]; se[rep][j] = f.inference->std_errors[j]; } } }); | |
| 29 | + json row = {{"kind", "garch_mc"}, {"experiment", "E5"}, {"n", n}, {"R", R}, {"wall_seconds", now() - t0}, {"threads", ThreadPool::global().size()}}; int ok = 0; for (int v : okv) ok += v; row["converged"] = ok; | |
| 30 | + for (int j = 0; j < 4; ++j) { double bias = 0, mse = 0; int cov = 0; for (int rep = 0; rep < R; ++rep) { if (!okv[rep]) continue; double d = est[rep][j] - tr[j]; bias += d; mse += d * d; if (std::abs(d) <= 1.96 * se[rep][j]) ++cov; } const char* nm[4] = {"mu", "omega", "alpha", "beta"}; row[nm[j]] = {{"bias", bias / ok}, {"rmse", std::sqrt(mse / ok)}, {"coverage95", static_cast<double>(cov) / ok}}; } | |
| 31 | + out << row.dump() << "\n"; out.flush(); std::fprintf(stderr, "E5 garch MC %s: %d/%d converged %.1fs\n", label, ok, R, now() - t0); } | |
| 32 | + // (b) HAR HAC coverage | |
| 33 | + for (int h : {1, 5}) { auto simulate = [](usize nn, u64 seed) { Rng g(4242, seed); std::vector<double> rv(nn); double lv = std::log(1e-4); for (usize t = 0; t < nn; ++t) { lv = 0.3 * std::log(1e-4) + 0.7 * lv + 0.4 * g.normal(); rv[t] = std::exp(lv); } return rv; }; | |
| 34 | + est::HarSpec spec; spec.horizon = h; auto big = simulate(400000, 999); auto pseudo = est::har_fit(big, spec); std::vector<std::array<int, 4>> cov(R); const f64 t0 = now(); | |
| 35 | + ThreadPool::global().parallel_for(R, [&](usize a, usize b) { for (usize rep = a; rep < b; ++rep) { auto rv = simulate(1500, rep); auto f = est::har_fit(rv, spec); for (int j = 0; j < 4; ++j) cov[rep][j] = std::abs(f.coefficients[j] - pseudo.coefficients[j]) <= 1.96 * f.inference->std_errors[j]; } }); | |
| 36 | + json row = {{"kind", "har_mc"}, {"experiment", "E5"}, {"h", h}, {"T", 1500}, {"R", R}, {"wall_seconds", now() - t0}}; for (int j = 0; j < 4; ++j) { int c = 0; for (int rep = 0; rep < R; ++rep) c += cov[rep][j]; row["coverage95"][j] = static_cast<double>(c) / R; } out << row.dump() << "\n"; out.flush(); std::fprintf(stderr, "E5 har h=%d done\n", h); } | |
| 37 | + // (d) reproducibility | |
| 38 | + { Rng g(99); std::vector<double> x(5000); for (auto& v : x) v = 1e-3 * g.student_t(5.0); auto st = [](std::span<const double> v) { return stats::sum_squares<double>(v); }; | |
| 39 | + stats::BootstrapOptions o; o.replicas = 2000; o.mean_block = 10; std::vector<std::vector<double>> results; std::vector<unsigned> thr = {1, 4, 18}; for (unsigned t : thr) { o.threads = t; auto b = stats::bootstrap(x, st, o); results.push_back(b.statistics); } | |
| 40 | + bool same = results[0] == results[1] && results[1] == results[2]; out << json({{"kind", "repro_bootstrap"}, {"experiment", "E5"}, {"replicas", 2000}, {"threads", thr}, {"bitwise_identical_across_threads", same}}).dump() << "\n"; | |
| 41 | + std::vector<i64> off; for (usize s = 0; s <= 50; ++s) off.push_back(static_cast<i64>(s * 100)); backend::Batch b{x, off}; est::RealizedOptions rc; rc.plan.backend = Backend::cpu; auto c1 = est::realized_measures_batch(b, rc); rc.plan.threads = 1; auto c2 = est::realized_measures_batch(b, rc); | |
| 42 | + json rr = {{"kind", "repro_realized"}, {"experiment", "E5"}, {"cpu_threads_vs_1_identical", c1.rv == c2.rv}}; | |
| 43 | +#ifdef RAID_ENABLE_METAL | |
| 44 | + est::RealizedOptions rm; rm.plan.backend = Backend::metal; auto m1 = est::realized_measures_batch(b, rm), m2 = est::realized_measures_batch(b, rm); rr["metal_runs_identical"] = m1.rv == m2.rv && m1.bpv == m2.bpv && m1.rq == m2.rq; double maxrel = 0; for (usize s = 0; s < c1.rv.size(); ++s) maxrel = std::max(maxrel, std::abs(m1.rv[s] - c1.rv[s]) / c1.rv[s]); rr["metal_vs_cpu_max_rel_rv"] = maxrel; rr["metal_backend_used"] = std::string(to_string(m1.stats.backend)); | |
| 45 | +#endif | |
| 46 | + out << rr.dump() << "\n"; } | |
| 47 | + man.save(std::string(argv[1]) + ".manifest.json"); return 0; | |
| 48 | +} | |
added
benchmarks/end_to_end/bench_e6_pipeline.cpp
+42 −0
@@ -0,0 +1,42 @@ | ||
| 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 price | |
| 3 | +// 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" | |
| 14 | +using namespace raid; using json = nlohmann::json; | |
| 15 | +namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } } | |
| 16 | + | |
| 17 | +int 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_ARROW | |
| 20 | + std::fprintf(stderr, "E6 requires the Arrow build\n"); return 1; | |
| 21 | +#else | |
| 22 | + 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: returns | |
| 31 | + 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 concat | |
| 33 | + 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 | +#endif | |
| 42 | +} | |
added
experiments/results/e0_universe.json
+2370 −0
Large diff (2,371 lines) is not loaded by default to keep the page fast.Load diff
added
experiments/results/e7_summary.csv
+12 −0
Large diff (12 lines) is not loaded by default to keep the page fast.Load diff
added
experiments/results/e7_summary.json
+431 −0
Large diff (432 lines) is not loaded by default to keep the page fast.Load diff
added
experiments/scripts/analyze_e7.py
+39 −0
Large diff (39 lines) is not loaded by default to keep the page fast.Load diff
modified
experiments/scripts/e0_universe_and_exports.py
+1 −1
Large diff (8 lines) is not loaded by default to keep the page fast.Load diff
added
experiments/scripts/e6_baselines.py
+40 −0
Large diff (40 lines) is not loaded by default to keep the page fast.Load diff
added
experiments/scripts/run_e1_e6.sh
+13 −0
Large diff (13 lines) is not loaded by default to keep the page fast.Load diff
added
paper/references.bib
+99 −0
Large diff (100 lines) is not loaded by default to keep the page fast.Load diff