1"""Build the one-minute return panel from INDIVIDUAL CONTRACTS: for each root and each CME session (18:00 ET → 17:00 ET next day) the2contract with the largest session volume is used; within-session close-to-close log returns in percent; no return across sessions, so3there is no roll splice. Outputs data/cache/returns_1m.parquet (wide minute panel, with per-root 'active' flags), returns_daily.parquet4(per yield day) and diagnostics. Yield day = 15:30 ET → 15:30 ET (H.15 reading time)."""5import sys, json6from pathlib import Path7import numpy as np, pandas as pd, polars as pl8ROOT = Path(__file__).resolve().parents[1]; sys.path.insert(0, str(ROOT / "src"))9from wp47.returns import ROOTS, yield_day_bounds, assign_yield_day10CACHE = ROOT / "data/cache"; OUT = ROOT / "results/metrics"; OUT.mkdir(parents=True, exist_ok=True)11ET = "America/New_York"12F = pd.read_csv(OUT / "daily_factors_cmt.csv", index_col=0, parse_dates=True).dropna(); bounds = yield_day_bounds(F.index)13diag = {}; wide = None14for root in ROOTS:15 files = sorted((CACHE / "contracts_1m").glob(f"{root}_*.parquet"))16 df = pl.concat([pl.read_parquet(f, columns=["symbol_as_reported", "timestamp_utc", "close", "volume"]) for f in files])17 df = df.with_columns(pl.col("timestamp_utc").dt.convert_time_zone(ET).alias("ts"))18 # session label: ET date of (ts + 6h) so that 18:00 belongs to the next calendar day's session19 df = df.with_columns(((pl.col("ts") + pl.duration(hours=6)).dt.date()).alias("session"))20 vol = df.group_by(["session", "symbol_as_reported"]).agg(pl.col("volume").sum().alias("v"), pl.len().alias("n"))21 best = vol.sort(["session", "v", "n"], descending=[False, True, True]).unique("session", keep="first", maintain_order=True).select(["session", "symbol_as_reported"]).rename({"symbol_as_reported": "best"})22 d = df.join(best, on="session").filter(pl.col("symbol_as_reported") == pl.col("best")).sort("ts").unique("ts", keep="last", maintain_order=True)23 pdf = d.select(["ts", "session", "close", "volume", "symbol_as_reported"]).to_pandas()24 ts = pdf["ts"]; gap = ts.diff().dt.total_seconds().div(60); hour = ts.dt.hour + ts.dt.minute / 6025 new_sess = pdf["session"].ne(pdf["session"].shift()); cross_break = (hour >= 18.0) & (hour.shift() < 17.5)26 r = 100 * np.log(pdf["close"]).diff(); r[new_sess | cross_break | (gap > 5)] = np.nan27 big = r.abs() > 0.25; nxt = r.shift(-1); bounce = big & (nxt.abs() > 0.25) & ((r + nxt).abs() < 0.2 * r.abs()); n_bounce = int(bounce.sum())28 r = r.mask(bounce | bounce.shift(1, fill_value=False)); pdf["r"] = r29 n_contracts_used = pdf["symbol_as_reported"].nunique()30 switches = int((pdf.groupby("session")["symbol_as_reported"].first().ne(pdf.groupby("session")["symbol_as_reported"].first().shift())).sum())31 jumps = pdf.loc[r.abs() > 0.4, "ts"]32 diag[root] = {"bars": int(len(pdf)), "contracts_used": int(n_contracts_used), "contract_switches": switches, "segment_breaks": int(r.isna().sum()), "bounce_pairs_removed": n_bounce,33 "n_abs_r_gt_40bp": int(len(jumps)), "jumps_by_hour_et": jumps.dt.hour.value_counts().sort_index().to_dict(), "sd_1m_bp": float(r.std() * 100), "first": str(ts.min()), "last": str(ts.max())}34 print(root, {k: v for k, v in diag[root].items() if k != "jumps_by_hour_et"}, flush=True)35 s = pdf.set_index("ts")[["r", "volume"]].rename(columns={"r": root, "volume": f"vol_{root}"})36 wide = s if wide is None else wide.join(s, how="outer")37wide = wide.sort_index(); wide = wide[~wide.index.duplicated(keep="last")]38# active flag: a bar within the previous 15 minutes (so a minute without a trade is a zero return while active)39idx = np.arange(len(wide))40for root in ROOTS:41 has = wide[root].notna().values42 last = pd.Series(np.where(has, idx, np.nan)).ffill().values43 wide[f"act_{root}"] = (idx - last) <= 1544wide["yday"] = assign_yield_day(pd.Series(wide.index, index=wide.index), bounds).values45wide.index.name = "ts"; wide.reset_index().to_parquet(CACHE / "returns_1m.parquet", index=False)46g = wide.groupby("yday"); daily = g[ROOTS].sum(min_count=1); cov = g[ROOTS].count(); cov.columns = [f"n_{c}" for c in ROOTS]47act = g[[f"act_{c}" for c in ROOTS]].sum(); act.columns = [f"nact_{c}" for c in ROOTS]48daily = daily.join(cov).join(act); daily.index.name = "yday"; daily.to_parquet(CACHE / "returns_daily.parquet")49diag["panel"] = {"minutes": int(len(wide)), "yield_days": int(daily.shape[0]), "first": str(wide.index.min()), "last": str(wide.index.max()),50 "median_minutes_per_yday": {c: float(cov[f"n_{c}"].median()) for c in ROOTS}, "median_active_minutes_per_yday": {c: float(act[f"nact_{c}"].median()) for c in ROOTS},51 "daily_sd_pct": daily[ROOTS].std().round(3).to_dict(), "corr_daily": daily[ROOTS].corr().round(3).to_dict()}52json.dump(diag, open(OUT / "returns_diagnostics.json", "w"), indent=1, default=str)53print(json.dumps({k: v for k, v in diag["panel"].items() if k != "corr_daily"}, indent=1, default=str))54