"""Dense regular-hours grid per instrument and strictly causal feature construction (docs/DATA_CARD.md §grid). Grid: sessions x 391 slots (slot k = bar opening at 09:30 + k minutes ET). Slot 390 (16:00 closing print) is kept in `base` for the overnight-gap computation only; context windows and targets use slots 0..389. Missing minutes are masked (never filled). Past-only aggregates: 5-minute buckets (78/day, bucket j = slots 5j..5j+4) and 30-minute buckets (13/day, bucket j = slots 30j..30j+29), each with its own presence mask. Seasonal log-volume profile `seas[s, k]` = mean over sessions s-20..s-1 of log1p(volume) at slot k (past only, NaN when no history). """ from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path import numpy as np import polars as pl from .calendar import REGULAR_SLOTS, sessions, slot_index from .adjustments import split_factor_by_session from .quality import dedupe, qa_report N_DAY = 390 # context/target slots per day (0..389) N_B5, N_B30 = 78, 13 BASE_COLS = ["lo", "lh", "ll", "lc", "lv"] # log open/high/low/close, log1p(volume) SEAS_DAYS = 20 @dataclass class GridMeta: symbol: str asset_class: str session_dates: list[str] n_slots: list[int] early_close: list[bool] split_factors: list[float] overnight_gap: list[float | None] qa: dict counts: dict dataset_version: int | None schema_version: int = 1 def build_grid(raw: pl.DataFrame, splits: list[dict], start: str, end: str, symbol: str, asset_class: str, dataset_version: int | None = None) -> tuple[dict[str, np.ndarray], GridMeta]: sess = sessions(start, end) S = len(sess) date2idx = {s.date.date(): i for i, s in enumerate(sess)} open_utc = np.array([s.open_utc.to_datetime64() for s in sess], dtype="datetime64[s]") n_slots = np.array([s.n_slots for s in sess], dtype=np.int64) qa = qa_report(raw) bars = dedupe(raw) bars = bars.with_columns(pl.col("session_date").cast(pl.Date)) sidx = np.array([date2idx.get(d, -1) for d in bars["session_date"].to_list()], dtype=np.int64) bars = bars.with_columns(pl.Series("sidx", sidx)).filter(pl.col("sidx") >= 0) ts = bars["timestamp_utc"].dt.replace_time_zone(None).to_numpy().astype("datetime64[s]") si = bars["sidx"].to_numpy() slot = slot_index(ts, open_utc[si], n_slots[si]) bars = bars.with_columns(pl.Series("slot", slot)) n_ext = int((slot < 0).sum()) reg = bars.filter(pl.col("slot") >= 0) bad = ((reg["high"] < reg["low"]) | (reg["open"] > reg["high"]) | (reg["open"] < reg["low"]) | (reg["close"] > reg["high"]) | (reg["close"] < reg["low"]) | (reg["low"] <= 0) | (reg["volume"] < 0)) n_bad = int(bad.sum()) reg = reg.filter(~bad) # split adjustment (explicit, session based) f = split_factor_by_session(reg["session_date"].to_numpy(), splits) base = np.full((S, REGULAR_SLOTS, 5), np.nan, dtype=np.float32) mask = np.zeros((S, REGULAR_SLOTS), dtype=np.uint8) si = reg["sidx"].to_numpy(); sl = reg["slot"].to_numpy() base[si, sl, 0] = np.log(reg["open"].to_numpy() / f) base[si, sl, 1] = np.log(reg["high"].to_numpy() / f) base[si, sl, 2] = np.log(reg["low"].to_numpy() / f) base[si, sl, 3] = np.log(reg["close"].to_numpy() / f) base[si, sl, 4] = np.log1p(reg["volume"].to_numpy() * f) mask[si, sl] = 1 in_session = (np.arange(REGULAR_SLOTS)[None, :] < n_slots[:, None]).astype(np.uint8) mask &= in_session # overnight gap: first present open of day s vs last present close (incl. closing print) of day s-1 gaps: list[float | None] = [] for s in range(S): g = None if s > 0 and mask[s].any() and mask[s - 1].any(): k0 = int(np.argmax(mask[s] > 0)); k1 = int(np.max(np.flatnonzero(mask[s - 1]))) g = float(base[s, k0, 0] - base[s - 1, k1, 3]) gaps.append(g) # seasonal volume profile (past 20 sessions, present bars only) lv = np.where(mask > 0, base[:, :, 4], 0.0).astype(np.float64) cnt = (mask > 0).astype(np.float64) cs_lv = np.concatenate([np.zeros((1, REGULAR_SLOTS)), np.cumsum(lv, axis=0)], axis=0) cs_c = np.concatenate([np.zeros((1, REGULAR_SLOTS)), np.cumsum(cnt, axis=0)], axis=0) lo_i = np.clip(np.arange(S) - SEAS_DAYS, 0, None) num = cs_lv[np.arange(S)] - cs_lv[lo_i] den = cs_c[np.arange(S)] - cs_c[lo_i] with np.errstate(invalid="ignore", divide="ignore"): seas = np.where(den > 0, num / den, np.nan).astype(np.float32) b5, m5 = _buckets(base, mask, 5, N_B5) b30, m30 = _buckets(base, mask, 30, N_B30) # derived flat arrays (slots 0..389 only): log close with NaN, previous present slot, seasonal squared-return profile m390 = mask[:, :N_DAY] > 0 flat_lc = np.where(m390, base[:, :N_DAY, 3], np.nan).reshape(-1).astype(np.float32) idx = np.where(m390.reshape(-1), np.arange(S * N_DAY), -1) prev_present = np.maximum.accumulate(idx).astype(np.int32) r2 = np.zeros((S, N_DAY)); r2c = np.zeros((S, N_DAY)) lc2 = np.where(m390, base[:, :N_DAY, 3].astype(np.float64), np.nan) pr = np.full_like(lc2, np.nan) pr[:, 1:] = lc2[:, :-1] # same-session previous present close (forward carry inside the day) for k in range(1, N_DAY): pr[:, k] = np.where(np.isnan(pr[:, k]), pr[:, k - 1], pr[:, k]) ret = lc2 - pr ok = ~np.isnan(ret) r2[ok] = ret[ok] ** 2; r2c[ok] = 1.0 cs_r2 = np.concatenate([np.zeros((1, N_DAY)), np.cumsum(r2, axis=0)]); cs_c2 = np.concatenate([np.zeros((1, N_DAY)), np.cumsum(r2c, axis=0)]) num2 = cs_r2[np.arange(S)] - cs_r2[lo_i]; den2 = cs_c2[np.arange(S)] - cs_c2[lo_i] with np.errstate(invalid="ignore", divide="ignore"): seas_r2 = np.where(den2 > 0, num2 / den2, np.nan).astype(np.float32) split_by_session = split_factor_by_session(np.array([s.date.date() for s in sess], dtype="datetime64[D]"), splits) meta = GridMeta(symbol, asset_class, [s.date.date().isoformat() for s in sess], [int(x) for x in n_slots], [bool(s.early_close) for s in sess], [float(x) for x in split_by_session], gaps, qa, {"rows_raw": int(raw.height), "rows_dedup": int(bars.height), "rows_extended_hours": n_ext, "rows_regular": int(reg.height) + n_bad, "rows_dropped_integrity": n_bad, "regular_slots_expected": int(in_session.sum()), "regular_slots_present": int(mask.sum()), "coverage": float(mask.sum() / max(1, in_session.sum())), "sessions_with_any_bar": int((mask.sum(1) > 0).sum()), "sessions": S}, dataset_version) arrays = {"base": base, "mask": mask, "in_session": in_session, "seas": seas, "b5": b5, "m5": m5, "b30": b30, "m30": m30, "flat_lc": flat_lc, "prev_present": prev_present, "seas_r2": seas_r2} return arrays, meta def _buckets(base: np.ndarray, mask: np.ndarray, width: int, nb: int) -> tuple[np.ndarray, np.ndarray]: """Past-only OHLCV aggregates of `width` slots over slots 0..389 (closing print excluded).""" S = base.shape[0] x = base[:, :N_DAY].reshape(S, nb, width, 5) m = (mask[:, :N_DAY].reshape(S, nb, width) > 0) out = np.full((S, nb, 5), np.nan, dtype=np.float32) bm = m.any(axis=2) # open = first present, close = last present first = np.argmax(m, axis=2); last = width - 1 - np.argmax(m[:, :, ::-1], axis=2) ii, jj = np.indices((S, nb)) out[:, :, 0] = x[ii, jj, first, 0] out[:, :, 3] = x[ii, jj, last, 3] import warnings with np.errstate(invalid="ignore"), warnings.catch_warnings(): warnings.simplefilter("ignore", RuntimeWarning) out[:, :, 1] = np.nanmax(np.where(m, x[:, :, :, 1], np.nan), axis=2) out[:, :, 2] = np.nanmin(np.where(m, x[:, :, :, 2], np.nan), axis=2) vol = np.where(m, np.expm1(x[:, :, :, 4].astype(np.float64)), 0.0).sum(axis=2) out[:, :, 4] = np.log1p(vol) out[~bm] = np.nan return out, bm.astype(np.uint8) def save_grid(arrays: dict[str, np.ndarray], meta: GridMeta, out_dir: Path) -> None: out_dir.mkdir(parents=True, exist_ok=True) for k, v in arrays.items(): np.save(out_dir / f"{k}.npy", v) (out_dir / "meta.json").write_text(json.dumps(meta.__dict__)) def load_grid(dir_: Path, mmap: bool = True) -> tuple[dict[str, np.ndarray], dict]: arrays = {k: np.load(dir_ / f"{k}.npy", mmap_mode="r" if mmap else None) for k in ("base", "mask", "in_session", "seas", "b5", "m5", "b30", "m30", "flat_lc", "prev_present", "seas_r2")} return arrays, json.loads((dir_ / "meta.json").read_text())