1"""Single causal pass over origins producing everything the non-neural baselines and the scorer need:2realised targets per horizon, trailing sigma, 19 tree features, HAR regressors, EWMA and seasonal variance forecasts,3GARCH(1,1)-t variance forecasts (given per-instrument parameters fitted on training data). Parallel over chunks."""4from __future__ import annotations56from concurrent.futures import ProcessPoolExecutor78import numpy as np9import pandas as pd1011from ..data.windows import OriginIndex, SymbolGrid, N_DAY, trailing_returns, make_example1213TREE_FEATURES = ["log_rv30", "log_rv390", "log_rv5d", "ret30", "ret390", "ret_day", "overnight", "relvol30", "range30",14 "minute_frac", "time_to_close", "weekday", "early_close", "log_sigma", "skew390", "kurt390", "sign_last",15 "ret5", "rv_ratio_30_390"]16HAR_FEATURES = ["log_rv30", "log_rv390", "log_rv5d"]17EWMA_LAMBDA = 0.97181920def _five_minute_hist(g: SymbolGrid, s: int, m: int, n_days: int = 3) -> np.ndarray:21 jB = (m + 1) // 5 - 122 out = []23 for d in range(max(0, s - n_days), s + 1):24 last = jB if d == s else 7725 if last < 1:26 continue27 c = np.asarray(g.flat_b5[d * 78: d * 78 + last + 1, 3]); mm = np.asarray(g.flat_m5[d * 78: d * 78 + last + 1]) > 028 ix = np.flatnonzero(mm)29 if len(ix) >= 2:30 out.append(np.diff(c[ix]))31 return np.concatenate(out) if out else np.zeros(0)323334def garch_forecast(params: dict, r5: np.ndarray, H: int) -> float:35 om, al, be = params["omega"], params["alpha"], params["beta"]36 ab = al + be37 uncond = om / max(1e-9, 1 - ab) if ab < 0.999 else float(np.var(r5)) if len(r5) else om38 h = uncond39 for r in r5[-400:]:40 h = om + al * r * r + be * h41 steps = max(1, int(np.ceil(H / 5)))42 k = np.arange(steps)43 var = float((uncond + ab ** k * (h - uncond)).sum())44 return var * (H / (5 * steps))454647def origin_row(g: SymbolGrid, s: int, m: int, horizons: list[int], garch: dict | None) -> dict:48 ex = make_example(g, s, m, type("W", (), {"L_A": 390, "L_B": 390, "L_C": 260, "H": 60})())49 sigma = float(ex["sigma"]); y, ym = ex["y"], ex["y_mask"] > 050 row = {"sym": None, "symbol": g.symbol, "s": s, "m": m, "sigma": sigma, "session_date": g.meta["session_dates"][s],51 "asset_class": g.meta["asset_class"]}52 for H in horizons:53 row[f"cum_{H}"] = float((y[:H] * ym[:H]).sum()); row[f"rv_{H}"] = float(((y[:H] * ym[:H]) ** 2).sum())54 row[f"present_{H}"] = float(ym[:H].mean())55 r = trailing_returns(g, s, m, 6 * N_DAY) # up to 6 sessions of same-session returns56 r30, r390, r5 = r[-30:], r[-390:], r[-5:]57 rv30 = float(np.mean(r30 ** 2)) if len(r30) else sigma ** 258 rv390 = float(np.mean(r390 ** 2)) if len(r390) else sigma ** 259 rv5d = float(np.mean(r ** 2)) if len(r) else sigma ** 260 lc_today = np.asarray(g.flat_lc[s * N_DAY: s * N_DAY + m + 1], dtype=np.float64); ok = ~np.isnan(lc_today)61 ret_day = float(lc_today[ok][-1] - lc_today[ok][0]) if ok.sum() >= 2 else 0.062 gap = float(g.gap[s]); gap = 0.0 if np.isnan(gap) else gap63 sl = slice(s * N_DAY + max(0, m - 29), s * N_DAY + m + 1)64 base = np.asarray(g.flat_base[sl]); mk = np.asarray(g.flat_mask[sl]) > 0; seas = np.asarray(g.flat_seas[sl])65 rel = base[mk, 4] - seas[mk]; rel = rel[~np.isnan(rel)]66 sd390 = float(r390.std()) + 1e-12 if len(r390) > 1 else 1e-1267 z390 = (r390 - r390.mean()) / sd390 if len(r390) > 1 else np.zeros(1)68 row.update({"log_rv30": np.log(rv30 + 1e-12), "log_rv390": np.log(rv390 + 1e-12), "log_rv5d": np.log(rv5d + 1e-12),69 "ret30": r30.sum() / sigma, "ret390": r390.sum() / sigma, "ret_day": ret_day / sigma, "overnight": gap / sigma,70 "relvol30": float(rel.mean()) if len(rel) else 0.0, "range30": float((base[mk, 1] - base[mk, 2]).mean()) / sigma if mk.any() else 0.0,71 "minute_frac": m / 389.0, "time_to_close": (g.n_slots[s] - 1 - m) / 389.0, "weekday": g.weekday[s] / 4.0,72 "early_close": float(g.early[s]), "log_sigma": np.log(sigma), "skew390": float(np.mean(z390 ** 3)),73 "kurt390": float(np.mean(z390 ** 4)), "sign_last": float(np.sign(r[-1])) if len(r) else 0.0,74 "ret5": r5.sum() / sigma, "rv_ratio_30_390": np.log((rv30 + 1e-12) / (rv390 + 1e-12))})75 # EWMA (minute scale) variance per minute76 rr = r[-3 * N_DAY:]77 if len(rr) >= 30:78 w = EWMA_LAMBDA ** np.arange(len(rr))[::-1]; v1 = float((w * rr * rr).sum() / w.sum())79 else:80 v1 = sigma ** 281 row["ewma_var1"] = v182 # seasonal profile (past-only squared-return profile for the future slots), level-adjusted by trailing 30-min RV83 prof = np.asarray(g.flat_seas_r2[s * N_DAY: (s + 1) * N_DAY], dtype=np.float64)84 prof = np.where(np.isnan(prof), np.nan, prof)85 ref = np.nanmean(prof[max(0, m - 29): m + 1]) if np.isfinite(prof[max(0, m - 29): m + 1]).any() else np.nan86 level = float(np.clip(rv30 / ref, 0.25, 4.0)) if (ref is not None and np.isfinite(ref) and ref > 0) else 1.087 for H in horizons:88 fut = prof[m + 1: m + 1 + H]89 if np.isfinite(fut).sum() >= H // 2:90 row[f"seas_var_{H}"] = float(np.nansum(np.where(np.isfinite(fut), fut, np.nanmean(fut))) * level)91 else:92 row[f"seas_var_{H}"] = sigma ** 2 * H93 if garch is not None:94 row[f"garch_var_{H}"] = garch_forecast(garch, _five_minute_hist(g, s, m), H)95 if garch is not None:96 row["garch_nu"] = garch["nu"]97 return row9899100def _chunk(args) -> list[dict]:101 index, ks, horizons, garch_params = args102 rows = []103 for k in ks:104 i, s, m = index.rows[k]105 r = origin_row(index.grid(int(i)), int(s), int(m), horizons, garch_params.get(int(i)) if garch_params else None)106 r["sym"] = int(i); r["k"] = int(k)107 rows.append(r)108 return rows109110111def origin_frame(index: OriginIndex, horizons: list[int], garch_params: dict | None = None, workers: int = 8,112 chunk: int = 2000) -> pd.DataFrame:113 ks = np.arange(len(index))114 chunks = [(index, ks[i:i + chunk], horizons, garch_params or {}) for i in range(0, len(ks), chunk)]115 rows = []116 if workers <= 1 or len(chunks) == 1:117 for c in chunks:118 rows.extend(_chunk(c))119 else:120 with ProcessPoolExecutor(max_workers=workers) as ex:121 for part in ex.map(_chunk, chunks):122 rows.extend(part)123 df = pd.DataFrame(rows).sort_values("k").reset_index(drop=True)124 return df125