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 pa7from findata_schema import BARS_SCHEMA, INTERVAL_SECONDS, Interval8910def aggregate_bars(table: pa.Table, target: Interval, source: Interval = Interval.m1) -> pa.Table:11 if INTERVAL_SECONDS[target] % INTERVAL_SECONDS[source] != 0:12 raise ValueError(f"{target} is not a multiple of {source}")13 df = pl.from_arrow(table)14 if df.height == 0:15 return table.cast(BARS_SCHEMA)16 every = f"{INTERVAL_SECONDS[target]}s"17 # Truncate in local exchange time, then convert back: bucket boundaries follow the wall clock.18 local = pl.col("timestamp_utc").dt.convert_time_zone("America/New_York")19 df = df.with_columns(local.dt.truncate(every).dt.convert_time_zone("UTC").alias("bucket"))20 df = df.sort(["symbol_as_reported", "timestamp_utc"])21 agg = (df.group_by(["instrument_id", "symbol_as_reported", "venue", "asset_class", "session_date", "currency",22 "adjustment_type", "source_id", "bucket"], maintain_order=True)23 .agg(pl.col("open").first(), pl.col("high").max(), pl.col("low").min(), pl.col("close").last(),24 pl.col("volume").sum(), pl.len().cast(pl.Int32).alias("trade_count"),25 ((pl.col("close") * pl.col("volume")).sum() / pl.col("volume").sum()).alias("vwap"),26 pl.col("ingested_at").max(), pl.col("quality_flags").list.explode(keep_nulls=False, empty_as_null=False).drop_nulls().unique().alias("quality_flags"),27 pl.col("revision_id").max()))28 agg = agg.rename({"bucket": "timestamp_utc"}).with_columns(pl.lit(target.value).alias("interval"))29 agg = agg.with_columns(pl.when(pl.col("volume") > 0).then(pl.col("vwap")).otherwise(None).alias("vwap"))30 out = agg.select(BARS_SCHEMA.names).to_arrow()31 return out.cast(BARS_SCHEMA)32