1"""Frequency-controlled OHLCV features computed per symbol-session from the raw 1-minute export.23Every intraday family is computed identically on bars of length 1m, 5m and 30m, obtained by aggregating the *same* raw41-minute bars within the regular session (09:30 -> closing print). Daily features come from the same aggregation (D-003).5Zero-volume minutes are absent at the source and are never invented; the share of expected buckets present is a feature.6Bar statistics are labelled as such; nothing here is order flow, spread or informed trading.7"""8from __future__ import annotations910import math11from pathlib import Path1213import numpy as np14import polars as pl1516from .calendar import _xnys1718FREQS = {"1m": 1, "5m": 5, "30m": 30}19INTRADAY_FAMILIES = ["rv", "bpv", "rj", "rsneg_share", "sjv", "rskew", "rkurt", "max_share", "ac1", "drift",20 "vol_first30", "vol_last30", "vol_gini", "down_vol_share", "present_share", "rv_mktdown_share",21 "sxy", "syy", "n_bars"]222324def session_bounds_table(start, end) -> pl.DataFrame:25 """session_date, open_utc (09:30 ET bar open), close_utc (closing-print bar open), close_minute (minutes after 09:30)."""26 cal = _xnys()27 s = cal.sessions28 s = s[(s >= str(start)) & (s <= str(end))]29 opens = cal.opens.loc[s] if hasattr(cal, "opens") else None30 rows = []31 for d in s:32 o, c = cal.session_open(d), cal.session_close(d)33 rows.append({"session_date": d.date(), "open_utc": o.tz_convert("UTC").to_pydatetime(),34 "close_utc": c.tz_convert("UTC").to_pydatetime(),35 "close_minute": int((c - o).total_seconds() // 60)})36 df = pl.DataFrame(rows)37 return df.with_columns([pl.col("open_utc").dt.replace_time_zone("UTC"), pl.col("close_utc").dt.replace_time_zone("UTC")])383940def load_regular_session(paths: list[Path], bounds: pl.DataFrame) -> pl.DataFrame:41 """Raw 1m bars restricted to the regular session, with minute-of-session index m (0 = 09:30, close_minute = closing print)."""42 lf = pl.concat([pl.scan_parquet(str(p)) for p in paths]).select(43 ["symbol_as_reported", "timestamp_utc", "session_date", "open", "high", "low", "close", "volume", "quality_flags"])44 df = lf.collect()45 if df.height == 0:46 return df47 # the lake's base+delta layouts (and chunk boundaries) can publish the same bar twice: keep the latest ingestion48 df = pl.concat([pl.scan_parquet(str(p)) for p in paths]).select(49 ["symbol_as_reported", "timestamp_utc", "session_date", "open", "high", "low", "close", "volume", "quality_flags",50 "ingested_at"]).collect().sort("ingested_at").unique(subset=["timestamp_utc"], keep="last").drop("ingested_at")51 df = df.with_columns(pl.col("timestamp_utc").dt.convert_time_zone("UTC"))52 df = df.join(bounds, on="session_date", how="inner")53 df = df.filter((pl.col("timestamp_utc") >= pl.col("open_utc")) & (pl.col("timestamp_utc") <= pl.col("close_utc")))54 df = df.with_columns(((pl.col("timestamp_utc") - pl.col("open_utc")).dt.total_minutes()).cast(pl.Int32).alias("m"))55 df = df.filter(pl.col("quality_flags").list.len() == 0) # drop rows flagged by FinData (ohlc_inconsistent, price_jump, ...)56 return df.sort(["session_date", "m"])575859def aggregate_buckets(reg: pl.DataFrame, freq_min: int) -> pl.DataFrame:60 """Aggregate regular-session 1m bars into buckets of `freq_min` minutes; the closing print joins the last bucket."""61 if freq_min == 1:62 b = reg.with_columns(pl.col("m").alias("b"))63 return b.select(["session_date", "b", "open", "high", "low", "close", "volume", "m", "close_minute"])64 nb = (pl.col("close_minute") / freq_min).ceil().cast(pl.Int32) # buckets per session65 b = reg.with_columns(pl.min_horizontal((pl.col("m") / freq_min).floor().cast(pl.Int32), nb - 1).alias("b"))66 agg = b.group_by(["session_date", "b"], maintain_order=True).agg([67 pl.col("open").first(), pl.col("high").max(), pl.col("low").min(), pl.col("close").last(), pl.col("volume").sum(),68 pl.col("m").first(), pl.col("close_minute").first(),69 ])70 return agg.sort(["session_date", "b"])717273def bucket_returns(bk: pl.DataFrame) -> pl.DataFrame:74 """Within-session log returns: first bucket open->close, then close->close."""75 prev_close = pl.col("close").shift(1).over("session_date")76 r = pl.when(prev_close.is_null()).then((pl.col("close") / pl.col("open")).log()).otherwise((pl.col("close") / prev_close).log())77 return bk.with_columns(r.alias("r"))787980def _gini_expr() -> pl.Expr:81 v = pl.col("volume").sort()82 n = pl.len()83 i = pl.int_range(1, n + 1).cast(pl.Float64)84 return ((2.0 * (i * v).sum()) / (n.cast(pl.Float64) * v.sum() + 1e-12) - (n.cast(pl.Float64) + 1.0) / n.cast(pl.Float64))858687def session_features(bk: pl.DataFrame, freq_min: int, spy: pl.DataFrame | None) -> pl.DataFrame:88 """One row per session with the intraday families for this frequency (prefix f{freq}_)."""89 df = bucket_returns(bk)90 if spy is not None:91 df = df.join(spy.select(["session_date", "b", pl.col("r").alias("r_spy")]), on=["session_date", "b"], how="left")92 else:93 df = df.with_columns(pl.lit(None, dtype=pl.Float64).alias("r_spy"))94 r = pl.col("r")95 r2 = r.pow(2)96 rv = r2.sum()97 n = pl.len().cast(pl.Float64)98 prev = r.shift(1).over("session_date")99 expected = pl.when(pl.lit(freq_min) == 1).then(pl.col("close_minute").first() + 1).otherwise(100 (pl.col("close_minute").first() / freq_min).ceil()).cast(pl.Float64)101 first30 = pl.col("m") < 30102 last30 = pl.col("m") > (pl.col("close_minute") - 30)103 out = df.group_by("session_date", maintain_order=True).agg([104 n.alias("n_bars"),105 rv.alias("rv"),106 ((math.pi / 2.0) * (r.abs() * prev.abs()).sum()).alias("bpv"),107 (r2.filter(r < 0).sum()).alias("rs_neg"),108 (r2.filter(r > 0).sum()).alias("rs_pos"),109 (n.sqrt() * r.pow(3).sum() / (rv.pow(1.5) + 1e-18)).alias("rskew"),110 (n * r.pow(4).sum() / (rv.pow(2) + 1e-18)).alias("rkurt"),111 (r2.max() / (rv + 1e-18)).alias("max_share"),112 pl.corr(r, prev).alias("ac1"),113 (pl.col("close").last() / pl.col("open").first()).log().alias("drift"),114 (pl.col("volume").filter(first30).sum() / (pl.col("volume").sum() + 1e-9)).alias("vol_first30"),115 (pl.col("volume").filter(last30).sum() / (pl.col("volume").sum() + 1e-9)).alias("vol_last30"),116 _gini_expr().alias("vol_gini"),117 (pl.col("volume").filter(r < 0).sum() / (pl.col("volume").sum() + 1e-9)).alias("down_vol_share"),118 (n / expected).alias("present_share"),119 (r2.filter(pl.col("r_spy") < 0).sum() / (rv + 1e-18)).alias("rv_mktdown_share"),120 (r * pl.col("r_spy")).sum().alias("sxy"),121 pl.col("r_spy").pow(2).sum().alias("syy"),122 pl.col("volume").sum().alias("volume"),123 pl.col("open").first().alias("open"), pl.col("close").last().alias("close"),124 pl.col("high").max().alias("high"), pl.col("low").min().alias("low"),125 ])126 out = out.with_columns([127 (pl.max_horizontal(pl.col("rv") - pl.col("bpv"), 0.0) / (pl.col("rv") + 1e-18)).alias("rj"),128 (pl.col("rs_neg") / (pl.col("rv") + 1e-18)).alias("rsneg_share"),129 ((pl.col("rs_pos") - pl.col("rs_neg")) / (pl.col("rv") + 1e-18)).alias("sjv"),130 ])131 pre = f"f{freq_min}m_"132 keep = {c: pre + c for c in INTRADAY_FAMILIES}133 return out.select(["session_date"] + [pl.col(c).alias(keep[c]) for c in INTRADAY_FAMILIES] +134 ([pl.col("volume").alias("d_volume"), pl.col("open").alias("d_open"), pl.col("close").alias("d_close"),135 pl.col("high").alias("d_high"), pl.col("low").alias("d_low")] if freq_min == 1 else []))136137138def daily_block(day: pl.DataFrame, actions: pl.DataFrame | None, symbol: str) -> pl.DataFrame:139 """Daily OHLCV features from the 1m aggregation + split/dividend-adjusted close-to-close returns."""140 d = day.sort("session_date")141 if actions is not None and actions.height:142 a = actions.filter(pl.col("symbol") == symbol).select(["ex_date", "split_ratio", "dividend"])143 d = d.join(a, left_on="session_date", right_on="ex_date", how="left").with_columns([144 pl.col("split_ratio").fill_null(1.0), pl.col("dividend").fill_null(0.0)])145 else:146 d = d.with_columns([pl.lit(1.0).alias("split_ratio"), pl.lit(0.0).alias("dividend")])147 prev_close_adj = (pl.col("d_close").shift(1) / pl.col("split_ratio"))148 d = d.with_columns([149 ((pl.col("d_close") + pl.col("dividend")) / prev_close_adj - 1.0).alias("d_ret_cc"),150 (pl.col("d_open") / prev_close_adj - 1.0).alias("d_ret_co"),151 (pl.col("d_close") / pl.col("d_open") - 1.0).alias("d_ret_oc"),152 ((pl.col("d_high") / pl.col("d_low")).log().pow(2) / (4.0 * math.log(2.0))).alias("d_parkinson"),153 (pl.col("d_volume") * pl.col("d_close")).alias("d_dollar_vol"),154 ])155 return d156157158def compute_symbol_features(paths: list[Path], symbol: str, bounds: pl.DataFrame, spy_buckets: dict[str, pl.DataFrame] | None,159 actions: pl.DataFrame | None) -> pl.DataFrame:160 reg = load_regular_session(paths, bounds)161 if reg.height == 0:162 return pl.DataFrame()163 blocks = []164 for name, fmin in FREQS.items():165 bk = aggregate_buckets(reg, fmin)166 spy = spy_buckets.get(name) if spy_buckets else None167 blocks.append(session_features(bk, fmin, spy))168 out = blocks[0]169 for b in blocks[1:]:170 out = out.join(b, on="session_date", how="left")171 out = daily_block(out, actions, symbol)172 return out.with_columns(pl.lit(symbol).alias("symbol")).sort("session_date")173174175def compute_spy_buckets(paths: list[Path], bounds: pl.DataFrame) -> dict[str, pl.DataFrame]:176 reg = load_regular_session(paths, bounds)177 out = {}178 for name, fmin in FREQS.items():179 out[name] = bucket_returns(aggregate_buckets(reg, fmin)).select(["session_date", "b", "r"]).unique(180 subset=["session_date", "b"], keep="last").sort(["session_date", "b"])181 return out182