1"""GARCH(1,1) / EGARCH(1,1) with Student-t innovations on 5-minute training returns per instrument (arch package),2forecasting the variance over the next H minutes by aggregating the term structure of the fitted process on the35-minute grid (H/5 steps; H < 5 scaled proportionally). Mean fixed at zero. Fitted once per instrument on the4training period; used out of sample with the last `n_days` days of 5-minute returns as filtering history."""5from __future__ import annotations67import warnings89import numpy as np1011from ..data.windows import SymbolGrid121314def five_minute_returns(g: SymbolGrid, s0: int, s1: int) -> np.ndarray:15 out = []16 for s in range(s0, s1 + 1):17 c = g.flat_b5[s * 78:(s + 1) * 78, 3]; m = g.flat_m5[s * 78:(s + 1) * 78]18 idx = np.flatnonzero(m)19 if len(idx) >= 2:20 out.append(np.diff(c[idx]))21 return np.concatenate(out) if out else np.zeros(0)222324class GARCHBaseline:25 def __init__(self, kind: str = "garch", scale: float = 1e4):26 self.kind, self.scale = kind, scale27 self.models: dict[str, object] = {}28 self.name = kind2930 def fit(self, symbol: str, r5: np.ndarray) -> bool:31 from arch import arch_model32 if len(r5) < 500:33 return False34 y = r5 * self.scale35 with warnings.catch_warnings():36 warnings.simplefilter("ignore")37 if self.kind == "garch":38 am = arch_model(y, mean="Zero", vol="GARCH", p=1, q=1, dist="t")39 else:40 am = arch_model(y, mean="Zero", vol="EGARCH", p=1, o=1, q=1, dist="t")41 res = am.fit(disp="off", show_warning=False)42 self.models[symbol] = res43 return True4445 def predict(self, symbol: str, r5_hist: np.ndarray, H: int) -> dict | None:46 res = self.models.get(symbol)47 if res is None or len(r5_hist) < 50:48 return None49 from arch import arch_model50 y = r5_hist * self.scale51 with warnings.catch_warnings():52 warnings.simplefilter("ignore")53 if self.kind == "garch":54 am = arch_model(y, mean="Zero", vol="GARCH", p=1, q=1, dist="t")55 else:56 am = arch_model(y, mean="Zero", vol="EGARCH", p=1, o=1, q=1, dist="t")57 steps = max(1, int(np.ceil(H / 5)))58 fc = am.fix(res.params.values).forecast(horizon=steps, method="simulation" if self.kind != "garch" else "analytic",59 simulations=500 if self.kind != "garch" else None, reindex=False)60 v5 = fc.variance.values[-1, :steps] / self.scale ** 261 var = v5.sum() * (H / (5 * steps))62 nu = float(res.params.get("nu", 6.0)); nu = float(np.clip(nu, 2.5, 50))63 return {"mu": 0.0, "scale": np.sqrt(var * (nu - 2) / nu), "nu": nu, "var": var}64