Repository Public Research preview
1"""Realized measures from intraday bar closes of ONE session (decision D-03).23Inputs: bar-open timestamps (tz-aware, local) and closes inside the session window, sorted. Returns are log returns in4PERCENT between consecutive sampled prices. Calendar-time sampling on a k-minute grid uses the last observed bar close at5or before each grid point (previous-tick); the first sampled price is the first bar of the window.67Measures (all in percent²):8 rv_1m sum of squared 1-minute returns (all consecutive bars)9 rv_5m sum of squared 5-minute calendar returns (grid anchored at the window start)10 rv_5m_ss subsampled 5-minute RV: average of the 5 RV_5m estimates with grid offsets 0..4 minutes (Zhang–Mykland–Aït-Sahalia 2005 style averaging without the bias correction)11 bv_5m bipower variation on the 5-minute grid: (pi/2) * sum |r_j||r_{j-1}| (Barndorff-Nielsen & Shephard 2004), no finite-sample scaling12 rsv_pos_5m / rsv_neg_5m realized semivariances on the 5-minute grid (Barndorff-Nielsen, Kinnebrock & Shephard 2010)13 medrv_5m median realized variance (Andersen, Dobrev & Schaumburg 2012) on the 5-minute grid14 n_bars number of 1-minute bars used; n_5m number of 5-minute returns15The caller decides whether the session is valid (coverage rule) and adds the overnight component.16"""17from __future__ import annotations1819import math2021import numpy as np2223MEASURE_COLUMNS = ["rv_1m", "rv_5m", "rv_5m_ss", "bv_5m", "rsv_pos_5m", "rsv_neg_5m", "medrv_5m", "n_bars", "n_5m"]242526def _grid_sample(minutes: np.ndarray, closes: np.ndarray, step: int, offset: int) -> np.ndarray:27 """Previous-tick sample of closes on the grid start+offset, +step, ... (minutes = minutes since window start)."""28 if len(minutes) == 0:29 return np.empty(0)30 last = minutes[-1]31 grid = np.arange(offset, last + 1, step)32 if len(grid) == 0:33 return np.empty(0)34 idx = np.searchsorted(minutes, grid, side="right") - 135 idx = idx[idx >= 0]36 p = closes[idx]37 # collapse repeated samples (no new bar between grid points -> zero return, keep as is: calendar sampling)38 return p394041def _rv(p: np.ndarray) -> float:42 if len(p) < 2:43 return math.nan44 r = 100.0 * np.diff(np.log(p))45 return float(np.sum(r * r))464748def session_measures(minutes: np.ndarray, closes: np.ndarray) -> dict:49 """minutes: integer minutes since the window start for each bar (sorted, unique); closes: positive floats."""50 out = {c: math.nan for c in MEASURE_COLUMNS}51 n = len(closes)52 out["n_bars"] = n53 if n < 2:54 return out55 r1 = 100.0 * np.diff(np.log(closes))56 out["rv_1m"] = float(np.sum(r1 * r1))57 p5 = _grid_sample(minutes, closes, 5, 0)58 r5 = 100.0 * np.diff(np.log(p5)) if len(p5) > 1 else np.empty(0)59 out["n_5m"] = len(r5)60 if len(r5) >= 2:61 out["rv_5m"] = float(np.sum(r5 * r5))62 out["bv_5m"] = float((math.pi / 2.0) * np.sum(np.abs(r5[1:]) * np.abs(r5[:-1])))63 out["rsv_pos_5m"] = float(np.sum(r5[r5 > 0] ** 2))64 out["rsv_neg_5m"] = float(np.sum(r5[r5 < 0] ** 2))65 if len(r5) >= 3:66 m = np.median(np.stack([np.abs(r5[:-2]), np.abs(r5[1:-1]), np.abs(r5[2:])]), axis=0)67 c = math.pi / (6.0 - 4.0 * math.sqrt(3.0) + math.pi)68 out["medrv_5m"] = float(c * (len(r5) / (len(r5) - 2.0)) * np.sum(m * m))69 ss = []70 for off in range(5):71 p = _grid_sample(minutes, closes, 5, off)72 if len(p) > 1:73 ss.append(_rv(p))74 out["rv_5m_ss"] = float(np.nanmean(ss)) if ss else math.nan75 return out76