1"""Deterministic bar aggregation (1m → 5m/30m/1h). Buckets are aligned on the ET wall clock so that the209:30 bucket covers 09:30:00–09:34:59 regardless of DST. Never fills empty buckets."""3from __future__ import annotations45import polars as pl6import pyarrow as pa78from findata_schema import BARS_SCHEMA, INTERVAL_SECONDS, Interval91011def aggregate_bars(table: pa.Table, target: Interval, source: Interval = Interval.m1) -> pa.Table:12 if INTERVAL_SECONDS[target] % INTERVAL_SECONDS[source] != 0:13 raise ValueError(f"{target} is not a multiple of {source}")14 df = pl.from_arrow(table)15 if df.height == 0:16 return table.cast(BARS_SCHEMA)17 every = f"{INTERVAL_SECONDS[target]}s"18 # Truncate in local exchange time, then convert back: bucket boundaries follow the wall clock.19 local = pl.col("timestamp_utc").dt.convert_time_zone("America/New_York")20 df = df.with_columns(local.dt.truncate(every).dt.convert_time_zone("UTC").alias("bucket"))21 df = df.sort(["symbol_as_reported", "timestamp_utc"])22 agg = (df.group_by(["instrument_id", "symbol_as_reported", "venue", "asset_class", "session_date", "currency",23 "adjustment_type", "source_id", "bucket"], maintain_order=True)24 .agg(pl.col("open").first(), pl.col("high").max(), pl.col("low").min(), pl.col("close").last(),25 pl.col("volume").sum(), pl.len().cast(pl.Int32).alias("trade_count"),26 ((pl.col("close") * pl.col("volume")).sum() / pl.col("volume").sum()).alias("vwap"),27 pl.col("ingested_at").max(), pl.col("quality_flags").list.explode(keep_nulls=False, empty_as_null=False).drop_nulls().unique().alias("quality_flags"),28 pl.col("revision_id").max()))29 agg = agg.rename({"bucket": "timestamp_utc"}).with_columns(pl.lit(target.value).alias("interval"))30 agg = agg.with_columns(pl.when(pl.col("volume") > 0).then(pl.col("vwap")).otherwise(None).alias("vwap"))31 out = agg.select(BARS_SCHEMA.names).to_arrow()32 return out.cast(BARS_SCHEMA)33