從 WebSocket tick 流自建 K 線聚合器：分桶時鐘、遲到數據與 REST 對賬

from datetime import datetime, timezone, timedelta

INTERVAL = timedelta(minutes=1)
EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)


def bucket_start(ts: datetime) -> datetime:
    """Floor a UTC timestamp to the interval boundary (left-closed)."""
    return EPOCH + (ts - EPOCH) // INTERVAL * INTERVAL


def parse_at(raw: str) -> datetime:
    """Normalise an RFC3339 tick timestamp to aware UTC."""
    return datetime.fromisoformat(raw.replace("Z", "+00:00"))


class Aggregator:
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.cur = None
        self.bars = []

    def add(self, tick: dict) -> None:
        ts, price = parse_at(tick["at"]), float(tick["last"])
        start = bucket_start(ts)
        if self.cur is None or start > self.cur["openTime"]:
            if self.cur:
                self.bars.append(self.cur)
            self.cur = {"openTime": start, "open": price, "high": price,
                        "low": price, "close": price, "volume": 0.0}
        self.cur["high"] = max(self.cur["high"], price)
        self.cur["low"] = min(self.cur["low"], price)
        self.cur["close"] = price
        self.cur["volume"] += float(tick.get("volume") or 0)

    def flush(self) -> list:
        if self.cur:
            self.bars.append(self.cur)
            self.cur = None
        return self.bars


if __name__ == "__main__":
    ticks = [
        {"at": "2026-09-26T02:00:05Z", "last": 100.0, "volume": 2},
        {"at": "2026-09-26T02:00:41Z", "last": 103.5, "volume": 1},
        {"at": "2026-09-26T02:02:03Z", "last": 101.0, "volume": 4},
    ]
    agg = Aggregator("BTCUSDT")
    for t in ticks:
        agg.add(t)
    for b in agg.flush():
        print(b["openTime"].isoformat(), b["open"], b["high"],
              b["low"], b["close"], b["volume"])

from datetime import datetime, timedelta, timezone

INTERVAL = timedelta(minutes=1)
FIELDS = ("open", "high", "low", "close")


def reconcile(local_bars: list, rest_bars: list, tol: float = 1e-6) -> dict:
    """Diff locally aggregated bars against REST kline bars.

    missing  : REST has the window, we do not (grace period or gap).
    extra    : we have the window, REST does not (timezone or symbol bug).
    mismatch : both have it, a field value differs beyond tolerance.
    """
    index = {b["openTime"]: b for b in rest_bars}
    report = {"missing": [], "extra": [], "mismatch": []}
    for b in local_bars:
        key = b["openTime"]
        r = index.pop(key, None)
        if r is None:
            report["extra"].append(key)
            continue
        for f in FIELDS:
            if abs(float(b[f]) - float(r[f])) > tol:
                report["mismatch"].append((key, f, b[f], r[f]))
    report["missing"] = sorted(index)
    return report


if __name__ == "__main__":
    t0 = datetime(2026, 9, 26, 2, 0, tzinfo=timezone.utc)
    local = [{"openTime": t0, "open": 100.0, "high": 103.5,
              "low": 100.0, "close": 103.5}]
    rest = [{"openTime": t0, "open": 100.0, "high": 103.5,
             "low": 100.0, "close": 103.5},
            {"openTime": t0 + INTERVAL, "open": 103.5, "high": 104.0,
             "low": 103.0, "close": 103.8}]
    print(reconcile(local, rest))