1"""Build the one-minute return panel of Treasury futures aligned on 'yield days' (15:30 ET → 15:30 ET)."""2from __future__ import annotations3from pathlib import Path4import numpy as np, pandas as pd, polars as pl56ROOTS = ["ZT", "ZF", "ZN", "TN", "US", "UB"]7ET = "America/New_York"89def load_root_1m(cache: Path, root: str) -> pl.DataFrame:10 files = [f for f in sorted((cache / "futures_1m").glob(f"{root}_*.parquet")) if f.stat().st_size > 5000]11 df = pl.concat([pl.read_parquet(f, columns=["timestamp_utc", "close", "volume"]) for f in files])12 df = df.sort("timestamp_utc").unique(subset=["timestamp_utc"], keep="last", maintain_order=True)13 return df.with_columns(pl.col("timestamp_utc").dt.convert_time_zone(ET).alias("ts"))1415def roll_days(cache: Path, root: str) -> pd.DatetimeIndex:16 """Days where the raw and ratio-adjusted daily continuous series disagree on the close-to-close return (roll splices)."""17 raw = pl.concat([pl.read_parquet(f) for f in sorted((cache / "futures_1d").glob("continuous_raw_*.parquet"))]).filter(pl.col("symbol_as_reported") == root)18 adj = pl.concat([pl.read_parquet(f) for f in sorted((cache / "futures_1d").glob("continuous_ratio_*.parquet"))]).filter(pl.col("symbol_as_reported") == root)19 r = raw.select(["session_date", "close"]).sort("session_date").unique("session_date", keep="last").to_pandas().set_index("session_date")["close"]20 a = adj.select(["session_date", "close"]).sort("session_date").unique("session_date", keep="last").to_pandas().set_index("session_date")["close"]21 rr, ra = np.log(r).diff(), np.log(a).diff()22 d = (rr - ra.reindex(rr.index)).abs()23 return pd.DatetimeIndex(pd.to_datetime(d.index[d > 1e-4]))2425def minute_returns(df: pl.DataFrame, rolls: pd.DatetimeIndex, max_gap_min: int = 5) -> pl.DataFrame:26 """Close-to-close log returns in percent, within segments. Segment breaks: gap > max_gap_min, crossing 17:00 ET,27 and the first bar of the session on a roll day (session = 18:00 ET day d-1 → 17:00 ET day d, labelled d)."""28 pdf = df.select(["ts", "close", "volume"]).to_pandas()29 ts = pdf["ts"]30 gap = ts.diff().dt.total_seconds().div(60)31 # session label: date of ts shifted by +6h (so 18:00 → next day), i.e. floor((ts + 6h).date)32 sess = (ts + pd.Timedelta(hours=6)).dt.tz_localize(None).dt.normalize()33 new_sess = sess.ne(sess.shift())34 hour = ts.dt.hour + ts.dt.minute / 6035 cross_break = (hour >= 18.0) & (hour.shift() < 17.5) # first bar after the 17:00–18:00 break36 is_roll_sess = sess.isin(rolls)37 brk = (gap > max_gap_min) | new_sess | cross_break | (is_roll_sess & new_sess)38 r = 100 * np.log(pdf["close"]).diff()39 r[brk] = np.nan40 out = pd.DataFrame({"ts": ts, "session": sess, "r": r, "vol": pdf["volume"].astype(float)})41 return pl.from_pandas(out)4243def yield_day_bounds(fred_dates: pd.DatetimeIndex) -> pd.Series:44 """Boundary timestamps (15:30 ET) for each FRED yield date."""45 b = pd.DatetimeIndex(fred_dates).tz_localize(ET) + pd.Timedelta(hours=15, minutes=30)46 return pd.Series(fred_dates, index=b)4748def assign_yield_day(ts: pd.Series, bounds: pd.Series) -> pd.Series:49 """Minute with timestamp ts (bar open; the return accrues over [ts, ts+1min]) belongs to the first yield day whose 15:30 boundary >= ts + 1 min."""50 idx = bounds.index.values51 pos = np.searchsorted(idx, (ts + pd.Timedelta(minutes=1)).values, side="left")52 pos = np.clip(pos, 0, len(idx) - 1)53 return pd.Series(bounds.values[pos], index=ts.index)54