1// E4 — realized covariance and dynamic connectedness on real data: N ∈ {10, 30, 100} symbols, 5-minute synchronized2// returns (calendar-aware aggregation of the 1-minute bars) → daily realized covariance matrices (CPU; conditioning3// 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"20using namespace raid; using json = nlohmann::json;21namespace { f64 now() { return std::chrono::duration<f64>(std::chrono::steady_clock::now().time_since_epoch()).count(); } }2223int 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_ARROW26 return 1;27#else28 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 session32 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.95) 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 sessions36 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 N0 : {10ul, 30ul, 60ul, 100ul}) { const usize N = std::min(N0, syms.size()); if (N0 > syms.size() && N0 / 2 >= 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#endif55}56