1#!/usr/bin/env python32"""Tier C — realized covariance proxies for the ETF baskets (configs/study.yaml): per session, 5-minute calendar-grid3log returns (previous tick, 09:30–16:00 ET, subsampled over 5 offsets) of all basket members, RCov = mean over offsets of4sum_j r_j r_j' plus the overnight outer product r_on r_on' (percent^2). Also the daily return matrix. Needs the 1-minute5cache (run on M3U96a). Writes <cache>/rcov/<basket>.npz (dates, R (T x N) percent returns, RC (T x N x N), valid)."""6from __future__ import annotations78import glob9import sys10from pathlib import Path1112import numpy as np13import pandas as pd14import pyarrow.parquet as pq15import yaml1617ROOT = Path(__file__).resolve().parents[1]18sys.path.insert(0, str(ROOT / "src"))19from wp46.calendars import assign_sessions # noqa: E40220from wp46.findata_client.ingest import DATASET_FOR, cache_root # noqa: E40221from wp46.preprocessing import load_panel # noqa: E402222324def grid_prices(sym: str) -> dict:25 files = sorted(glob.glob(str(cache_root() / "bars" / DATASET_FOR["etf"] / sym / "*.parquet")))26 df = pd.concat([pq.read_table(f, columns=["timestamp_utc", "close"]).to_pandas() for f in files]).drop_duplicates("timestamp_utc").sort_values("timestamp_utc")27 df["timestamp_utc"] = pd.to_datetime(df["timestamp_utc"], utc=True)28 s = assign_sessions(df["timestamp_utc"], "etf")29 df = df.assign(session=s["session"].values, local=pd.DatetimeIndex(s["local"]), in_window=s["in_window"].values)30 df = df[df["in_window"]]31 out = {}32 for sess, g in df.groupby("session"):33 loc = pd.DatetimeIndex(g["local"]); start = loc[0].normalize() + pd.Timedelta(hours=9, minutes=30)34 minutes = ((loc - start).total_seconds() // 60).astype(int).values35 closes = g["close"].values.astype(float)36 P = {}37 for off in range(5):38 grid = np.arange(off, 391, 5)39 idx = np.searchsorted(minutes, grid, side="right") - 140 p = np.where(idx >= 0, closes[np.maximum(idx, 0)], np.nan)41 P[off] = p42 out[pd.Timestamp(sess)] = P43 return out444546def main() -> None:47 cfg = yaml.safe_load(open(ROOT / "configs" / "study.yaml"))48 for name, syms in cfg["tier_c_baskets"].items():49 panels = {s: load_panel(s).set_index("session") for s in syms}50 G = {s: grid_prices(s) for s in syms}51 dates = sorted(set.intersection(*[set(panels[s].index) for s in syms]) & set.intersection(*[set(G[s].keys()) for s in syms]))52 N = len(syms); T = len(dates)53 R = np.full((T, N), np.nan); RC = np.full((T, N, N), np.nan); valid = np.zeros(T, bool)54 for i, d in enumerate(dates):55 R[i] = [panels[s].loc[d, "r"] for s in syms]56 ron = np.array([panels[s].loc[d, "r_on"] for s in syms]); ron = np.where(np.isfinite(ron), ron, 0.0)57 ok = all(panels[s].loc[d, "valid_rm"] for s in syms)58 if not ok:59 continue60 acc = np.zeros((N, N)); k = 061 for off in range(5):62 P = np.column_stack([G[s][d][off] for s in syms])63 if np.any(~np.isfinite(P)):64 continue65 r = 100.0 * np.diff(np.log(P), axis=0)66 acc += r.T @ r; k += 167 if k == 0:68 continue69 RC[i] = acc / k + np.outer(ron, ron); valid[i] = True70 out = cache_root() / "rcov"; out.mkdir(parents=True, exist_ok=True)71 np.savez(out / f"{name}.npz", dates=np.array([np.datetime64(d) for d in dates]), R=R, RC=RC, valid=valid, symbols=np.array(syms))72 print(name, syms, "T", T, "valid", valid.mean().round(3), "mean diag RC / r2:", np.round(np.nanmean(np.diagonal(RC[valid], axis1=1, axis2=2), axis=0) / np.mean(R[valid] ** 2, axis=0), 3))737475if __name__ == "__main__":76 main()77