Repository Public Research preview
1"""Daily panel per instrument: close-to-close log return (percent), overnight/intraday split, realized measures.23Conventions (docs/data_dictionary.md): returns in PERCENT (100 * log price ratio); variances in percent².4Equities (etf/stock): daily close-to-close return r = 100*log(C_t/C_{t-1}) from the vendor split+dividend-adjusted5DAILY bars (price-return convention with dividends reinvested at the ex-date); overnight r_on = 100*log(O_t/C_{t-1})6from the same series; intraday realized measures from the 1-minute SPLIT-adjusted bars in the 09:30–16:00 ET window7(closing print included). Futures/FX/crypto: everything from the 1-minute series using the session windows of8wp46.calendars (CME trading day 18:00→17:00 ET, FX day 17:00→17:00 ET, crypto UTC day); the session close is the last9bar close in the window and r_on = 0 by construction (FX/crypto) or the 17:00→18:00 break return (futures).10Session validity (`valid_rm`): number of 1-minute bars in the window >= MIN_FRACTION of the expected count (equities:11391 on a full day / 211 on an early close; futures 1380; fx 1440; crypto 1440). Invalid sessions keep the return12but have NaN realized measures. Full-day variance proxies are assembled in wp46.scoring.proxies.13"""14from __future__ import annotations1516import glob17import json18import os19from pathlib import Path2021import numpy as np22import pandas as pd23import pyarrow.parquet as pq2425from wp46.calendars import ASSET_SESSION, assign_sessions26from wp46.calendars.sessions import MIN_FRACTION, xnys_early_closes27from wp46.findata_client.ingest import DAILY_FOR, DATASET_FOR, cache_root28from wp46.realized_measures import MEASURE_COLUMNS, session_measures2930PANEL_COLUMNS = ["session", "close", "open", "r", "r_on", "r_oc", "valid_rm", "expected_bars"] + MEASURE_COLUMNS31EXPECTED = {"futures": 1380, "fx": 1440, "crypto": 1440} # nominal; replaced by the instrument's modal bar count (D-05)32RM_VALID_FROM = {"crypto": "2022-01-01"} # D-06: crypto 1-minute bars before 2022 are contaminated by cross-venue bounce33 # (audit: 1-min RV 10-100x the squared daily return in 2013-2019) -> no realized measure343536def _read_cache(dataset_id: str, symbol: str, cols: list[str]) -> pd.DataFrame:37 files = sorted(glob.glob(str(cache_root() / "bars" / dataset_id / symbol / "*.parquet")))38 if not files:39 raise FileNotFoundError(f"no cached bars for {dataset_id}/{symbol}")40 cols = list(dict.fromkeys(["timestamp_utc"] + cols))41 df = pd.concat([pq.read_table(f, columns=cols).to_pandas() for f in files], ignore_index=True)42 df["timestamp_utc"] = pd.to_datetime(df["timestamp_utc"], utc=True)43 df = df.drop_duplicates("timestamp_utc").sort_values("timestamp_utc").reset_index(drop=True)44 return df454647def _intraday(asset_class: str, symbol: str) -> pd.DataFrame:48 bars = _read_cache(DATASET_FOR[asset_class], symbol, ["timestamp_utc", "open", "close", "volume"])49 bars = bars[(bars["close"] > 0) & np.isfinite(bars["close"])]50 s = assign_sessions(bars["timestamp_utc"], asset_class)51 bars = bars.assign(session=s["session"].values, local=pd.DatetimeIndex(s["local"]), in_window=s["in_window"].values)52 bars = bars[bars["in_window"]]53 spec = ASSET_SESSION[asset_class]54 rows = []55 ec = {}56 if asset_class in ("etf", "stock"):57 ec = xnys_early_closes(str(bars["session"].min().date()), str(bars["session"].max().date()))58 for sess, g in bars.groupby("session", sort=True):59 loc = pd.DatetimeIndex(g["local"])60 if asset_class in ("etf", "stock"):61 start = loc[0].normalize() + pd.Timedelta(hours=9, minutes=30)62 exp = 211 if str(sess.date()) in ec else 39163 elif asset_class == "crypto":64 start = loc[0].normalize()65 exp = EXPECTED[asset_class]66 else:67 # window for session D starts on D-1 at 18:00 ET (futures) / 17:00 ET (fx) and ends on D at 17:00 ET68 start = (pd.Timestamp(sess) - pd.Timedelta(days=1)).tz_localize(spec.tz) + pd.Timedelta(hours=spec.start.hour)69 exp = EXPECTED[asset_class]70 minutes = ((loc - start).total_seconds() // 60).astype(int).values71 closes = g["close"].values.astype(float)72 m = session_measures(minutes, closes)73 m["session"] = sess74 m["first_open"] = float(g["open"].values[0])75 m["last_close"] = float(closes[-1])76 m["expected_bars"] = exp77 m["valid_rm"] = bool(len(closes) >= MIN_FRACTION * exp)78 rows.append(m)79 return pd.DataFrame(rows).set_index("session")808182def build_daily_panel(asset_class: str, symbol: str) -> pd.DataFrame:83 ir = _intraday(asset_class, symbol)84 if asset_class in ("etf", "stock"):85 d = _read_cache(DAILY_FOR[asset_class], symbol, ["session_date", "open", "close"])86 d["session"] = pd.to_datetime(d["session_date"])87 d = d.drop_duplicates("session").set_index("session").sort_index()88 d = d[(d["close"] > 0) & (d["open"] > 0)]89 out = pd.DataFrame(index=d.index)90 out["close"], out["open"] = d["close"], d["open"]91 out["r"] = 100.0 * np.log(d["close"] / d["close"].shift(1))92 out["r_on"] = 100.0 * np.log(d["open"] / d["close"].shift(1))93 out["r_oc"] = 100.0 * np.log(d["close"] / d["open"])94 out = out.join(ir[MEASURE_COLUMNS + ["expected_bars", "valid_rm"]], how="left")95 out["valid_rm"] = out["valid_rm"].fillna(False).astype(bool)96 else:97 out = pd.DataFrame(index=ir.index)98 out["close"], out["open"] = ir["last_close"], ir["first_open"]99 out["r"] = 100.0 * np.log(out["close"] / out["close"].shift(1))100 out["r_on"] = 100.0 * np.log(out["open"] / out["close"].shift(1))101 out["r_oc"] = 100.0 * np.log(out["close"] / out["open"])102 for c in MEASURE_COLUMNS + ["expected_bars", "valid_rm"]:103 out[c] = ir[c]104 # drop stub sessions (Sunday evenings, holiday fragments): they are not trading days for the daily model105 modal = int(out["n_bars"].round(-1).mode().iloc[0])106 out = out[out["n_bars"] >= 0.25 * modal]107 out["r"] = 100.0 * np.log(out["close"] / out["close"].shift(1))108 out["r_on"] = 100.0 * np.log(out["open"] / out["close"].shift(1))109 out.index.name = "session"110 out = out.iloc[1:] # first return undefined111 if asset_class not in ("etf", "stock"):112 # D-05: expected bars = modal session length of THIS instrument (grains trade ~12.5 h, metals ~22 h, FX 24 h)113 modal = int(out["n_bars"].round(-1).mode().iloc[0])114 out["expected_bars"] = modal115 out["valid_rm"] = out["n_bars"] >= MIN_FRACTION * modal116 out["valid_rm"] = out["valid_rm"] & np.isfinite(out["rv_5m_ss"]) & (out["rv_5m_ss"] > 0)117 if asset_class in RM_VALID_FROM:118 out.loc[out.index < pd.Timestamp(RM_VALID_FROM[asset_class]), "valid_rm"] = False119 return out[PANEL_COLUMNS[1:]].reset_index()120121122def panel_path(symbol: str) -> Path:123 return Path(os.environ.get("WP46_PANELS", str(cache_root() / "panels"))).expanduser() / f"{symbol}.parquet"124125126def load_panel(symbol: str) -> pd.DataFrame:127 df = pd.read_parquet(panel_path(symbol))128 df["session"] = pd.to_datetime(df["session"])129 return df130