SniperGold_ML/engine/aggregate.py

"""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)