"""Sequential deterministic bar aggregation (Layer 5, spec 11). UTC clock boundaries; mid_u2 = bid_u + ask_u OHLC (PRICE_SCALE_BAR = 2*PRICE_SCALE); spread min/max/sum/avg (floor division); carry-aware; EOF partial bars are flushed with is_final=False; no empty-bar inflation; bar_idx is a global ordinal per timeframe; period_id strictly increases per TF which implies (TF, period_id) uniqueness. This module is the producer aggregator; engine/verify/vaggregate.py is the independent counterpart and must agree on every row. """ from . import versions def _new_carry(): return None def start_carry(tick, period_id, period_ms): ts, bid, ask, spread, vol, src_line, _rec = tick mid = bid + ask return { "period_id": period_id, "start_ms": period_id * period_ms, "open_u2": mid, "high_u2": mid, "low_u2": mid, "close_u2": mid, "tick_count": 1, "vol_sum": vol, "spread_min_u": spread, "spread_max_u": spread, "spread_sum_u": spread, "last_ts_ms": ts, "last_src_line": src_line, } def update_carry(carry, tick): ts, bid, ask, spread, vol, src_line, _rec = tick mid = bid + ask carry["high_u2"] = max(carry["high_u2"], mid) carry["low_u2"] = min(carry["low_u2"], mid) carry["close_u2"] = mid carry["tick_count"] += 1 carry["vol_sum"] += vol carry["spread_min_u"] = min(carry["spread_min_u"], spread) carry["spread_max_u"] = max(carry["spread_max_u"], spread) carry["spread_sum_u"] += spread carry["last_ts_ms"] = ts carry["last_src_line"] = src_line return carry def finalize_bar(carry, bar_idx, period_ms, is_final): row = ( bar_idx, carry["period_id"], carry["start_ms"], carry["start_ms"] + period_ms, carry["open_u2"], carry["high_u2"], carry["low_u2"], carry["close_u2"], carry["tick_count"], carry["vol_sum"], carry["spread_min_u"], carry["spread_max_u"], carry["spread_sum_u"], carry["spread_sum_u"] // carry["tick_count"], # floor; tick_count >= 1 is_final, carry.get("first_src_line") if "first_src_line" in carry else None, carry["last_src_line"], ) return row class Aggregator: """Carry-aware sequential aggregator over ordered canonical ticks.""" def __init__(self, cfg, carries=None, bar_counts=None, last_final_period=None): self.cfg = cfg tfs = list(cfg["timeframes"]) self.timeframes = tfs self.periods = {tf: versions.TIMEFRAMES[tf] for tf in tfs} self.carries = dict(carries or {}) self.bar_counts = dict(bar_counts or {tf: 0 for tf in tfs}) self.last_final_period = dict(last_final_period or {}) self._buckets = {tf: [] for tf in tfs} self._resume_checked = False # -- streaming ---------------------------------------------------------- def consume(self, tick): """Feed one canonical tick (ts,bid,ask,spread,vol,src_line,rec_ord). On the first tick after construction with restored carries, spec 11.4 requires first input ts_ms >= every carried last_ts_ms; a violation is a material carry mismatch (FAILED, F11). """ ts = tick[0] if not self._resume_checked: for tf in self.timeframes: c = self.carries.get(tf) if c is not None and ts < c.get("last_ts_ms", -1): raise ValueError( "carry continuity violation (%s): first ts %d < " "carried last_ts %d" % (tf, ts, c.get("last_ts_ms"))) self._resume_checked = True for tf in self.timeframes: period_ms = self.periods[tf] period_id = ts // period_ms carry = self.carries.get(tf) if carry is None: self.carries[tf] = start_carry(tick, period_id, period_ms) self.carries[tf]["first_src_line"] = tick[5] continue if period_id == carry["period_id"]: update_carry(carry, tick) continue # period advanced -> finalize the old bar if period_id < carry["period_id"]: raise ValueError("aggregation order violation: period regressed") last = self.last_final_period.get(tf) if last is not None and carry["period_id"] <= last: raise ValueError("(TF, period_id) duplicate finalization") self.last_final_period[tf] = carry["period_id"] row = finalize_bar(carry, self.bar_counts[tf], period_ms, True) self.buckets_tf(tf).append(row) self.bar_counts[tf] += 1 self.carries[tf] = start_carry(tick, period_id, period_ms) self.carries[tf]["first_src_line"] = tick[5] def buckets_tf(self, tf): return self._buckets[tf] def finish_workload(self, tf): """Return and clear the finalized-bar bucket for one timeframe.""" rows = self._buckets[tf] self._buckets[tf] = [] return rows def finish_eof(self, tf): """Flush the EOF partial bar (is_final=False) into the bucket, then return and clear the bucket. Empty-bar inflation never happens: a carry exists only when the timeframe saw at least one tick.""" carry = self.carries.get(tf) if carry is not None: last = self.last_final_period.get(tf) if last is not None and carry["period_id"] <= last: raise ValueError("(TF, period_id) duplicate finalization at EOF") self.last_final_period[tf] = carry["period_id"] row = finalize_bar(carry, self.bar_counts[tf], self.periods[tf], False) self._buckets[tf].append(row) self.bar_counts[tf] += 1 self.carries[tf] = None rows = self._buckets[tf] self._buckets[tf] = [] return rows def has_partial(self, tf): return self.carries.get(tf) is not None def carr_state(self): """Serializable carry state for checkpointing.""" return {tf: (dict(self.carries.get(tf)) if self.carries.get(tf) else None) for tf in self.timeframes} def validate_resume_continuity(self): """Cross-check carries vs bar counts (spec 11.5 / 13.5). bar_idx counts must be >= number of finalized bars; each carry's period must be strictly greater than the last finalized period. Raises ValueError (material F11) on mismatch.""" for tf in self.timeframes: carry = self.carries.get(tf) if carry is None: continue last = self.last_final_period.get(tf) if last is not None and carry["period_id"] <= last: raise ValueError("carry/%s continuity mismatch" % tf)