""" Timestamp normalization and sequence gap detection for market data. Exchange timestamps come in various formats (ms since epoch, ISO strings, exchange-specific formats). This module normalizes them to a consistent int64 milliseconds-since-epoch. Gap detection tracks per-channel per-coin sequence numbers and flags missing messages so order books can be re-snapshotted. """ from __future__ import annotations import time from datetime import datetime, timezone # ── Timestamp normalization ─────────────────────────────────── def normalize_timestamp(ts, source: str = "hl") -> int: """Normalize a timestamp to int64 milliseconds since epoch. Args: ts: raw timestamp — can be int (ms), float (seconds), str (ISO 8601) source: 'hl' (Hyperliquid), 'binance', 'bybit', 'okx', 'coinbase', 'deribit' Returns int64 milliseconds since epoch. """ if ts is None: return int(time.time() * 1000) if isinstance(ts, (int, float)): if ts > 1_000_000_000_000: return int(ts) # already ms if ts > 1_000_000_000: return int(ts * 1000) # seconds → ms return int(ts * 1000) # fractional seconds → ms if isinstance(ts, str): return _parse_iso_ms(ts) if isinstance(ts, datetime): if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) return int(ts.timestamp() * 1000) return int(time.time() * 1000) def _parse_iso_ms(s: str) -> int: for fmt in [ "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S", ]: try: dt = datetime.strptime(s.replace("+00:00", "").rstrip("Z"), fmt) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return int(dt.timestamp() * 1000) except ValueError: continue return int(time.time() * 1000) # ── Sequence gap detection ──────────────────────────────────── class SequenceTracker: """Track per-channel per-coin sequence numbers and detect gaps. Usage: tracker = SequenceTracker() gap = tracker.check("l2book", "BTC", seq_num=1042) if gap: print(f"Gap detected: expected {gap['expected']}, got {gap['got']}") """ def __init__(self): self._state: dict[str, int] = {} # key = "channel:coin", value = last_seq self._gap_count: dict[str, int] = {} def check(self, channel: str, coin: str, seq_num: int) -> dict | None: """Check for sequence gap. Returns None if ok, dict if gap.""" key = f"{channel}:{coin}" last = self._state.get(key) if last is None: self._state[key] = seq_num return None expected = last + 1 if seq_num == expected or seq_num > expected: self._state[key] = seq_num if seq_num > expected: gap_size = seq_num - expected self._gap_count[key] = self._gap_count.get(key, 0) + gap_size return {"key": key, "expected": expected, "got": seq_num, "gap_size": gap_size} return None return None def reset(self, channel: str, coin: str): """Reset tracker (call after re-snapshot).""" self._state.pop(f"{channel}:{coin}", None) @property def gap_counts(self) -> dict[str, int]: return dict(self._gap_count) def detect_sequence_gap( current_seq: int, last_seq: int | None, max_gap: int = 10, ) -> int: """Return gap size. 0 = ok, >0 = gap count, -1 = negative gap (dupe/reset).""" if last_seq is None: return 0 diff = current_seq - last_seq if diff == 1: return 0 if diff > 1: return min(diff - 1, max_gap * 100) return -1 # duplicate or reset