"""Reference applier for the PulseScore delta stream protocol v2 (stream-api/docs/PROTOCOL.md). Standard library only; Python >= 3.8. from apply import DeltaBoard board = DeltaBoard() async for message in websocket: # any WebSocket library board.apply(message) # JSON text or a parsed dict board.to_list() # events exactly as the HTTP API serves them board.get("123456") # one event, or None Frames handled: connected, snapshot (chunked), snapshot_end, resumed, delta (upsert / remove / patch), hb, error. A patch is applied in the protocol order set -> unset -> mDel -> mAdd -> mSet -> selDel -> sel; every value is absolute, so re-applying a frame is a no-op. ``kind=racing`` streams carry races (runners) instead of events (markets and selections): use RaceBoard for those, or let StreamClient/StreamConnection pick the right board from the stream id. Identity keys must match the server's exactly, so numbers are rendered without exponent and without a trailing ".0". Python keeps big integers exact, so ids above 2**53 are fine here as long as the producer sent them exactly. """ from __future__ import annotations import json import math from decimal import Decimal from typing import Any, Dict, List, Optional PROTOCOL = 2 _IGNORED_TOP = {"_id", "createdAt", "updatedAt"} _IGNORED_MARKET = {"updatedAt", "oddsSig"} def scalar_key(v: Any) -> str: """Render a scalar identity field the way the server does ('' when not scalar).""" if v is None: return "" if isinstance(v, bool): return "true" if v else "false" if isinstance(v, str): return v if isinstance(v, int): return str(v) if isinstance(v, float): if math.isnan(v) or math.isinf(v): return "" if v.is_integer() and abs(v) < 1e15: return str(int(v)) s = repr(v) if "e" in s or "E" in s: s = format(Decimal(s), "f") return s return "" def _compound(obj: Dict[str, Any], fields) -> str: return "|".join(scalar_key(obj.get(f)) for f in fields) def market_key(m: Dict[str, Any]) -> str: """marketId when present and non-empty, else canonicalMarket|period|line|rawName.""" mid = scalar_key(m.get("marketId")) return mid if mid != "" else _compound(m, ("canonicalMarket", "period", "line", "rawName")) def selection_key(s: Dict[str, Any]) -> str: """selectionId when present and non-empty, else canonicalOutcome|rawName|line.""" sid = scalar_key(s.get("selectionId")) return sid if sid != "" else _compound(s, ("canonicalOutcome", "rawName", "line")) class _Market: __slots__ = ("fields", "selections") def __init__(self, m: Dict[str, Any]): self.fields: Dict[str, Any] = {k: v for k, v in m.items() if k != "selections" and k not in _IGNORED_MARKET} self.selections: Optional[Dict[str, Dict[str, Any]]] = None if "selections" in m: self.selections = {} if isinstance(m["selections"], list): for s in m["selections"]: if isinstance(s, dict): self.selections[selection_key(s)] = s def to_dict(self) -> Dict[str, Any]: out = dict(self.fields) if self.selections is not None: out["selections"] = list(self.selections.values()) return out class _Event: __slots__ = ("fields", "markets") def __init__(self, doc: Dict[str, Any]): self.fields: Dict[str, Any] = {k: v for k, v in doc.items() if k != "markets" and k not in _IGNORED_TOP} self.markets: Optional[Dict[str, _Market]] = None if "markets" in doc: self.markets = {} if isinstance(doc["markets"], list): for m in doc["markets"]: if isinstance(m, dict): self.markets[market_key(m)] = _Market(m) def to_dict(self) -> Dict[str, Any]: out = dict(self.fields) if self.markets is not None: out["markets"] = [m.to_dict() for m in self.markets.values()] return out class DeltaBoard: """Client-side state of one /ws/stream connection.""" def __init__(self, strict: bool = False): self.strict = strict self.events: Dict[str, _Event] = {} self.seq = 0 self.ready = False # snapshot_end or resumed seen self.stats: Dict[str, Any] = {"frames": {}, "changes": {}, "bytes": 0, "errors": []} def _fail(self, msg: str) -> None: if self.strict and len(self.stats["errors"]) < 50: self.stats["errors"].append(msg) def apply(self, frame) -> str: """Apply one frame (JSON text/bytes or a dict). Returns the frame type.""" if isinstance(frame, (str, bytes, bytearray)): self.stats["bytes"] += len(frame) frame = json.loads(frame) t = frame.get("t", "") self.stats["frames"][t] = self.stats["frames"].get(t, 0) + 1 seq = frame.get("seq") if isinstance(seq, int) and not isinstance(seq, bool) and seq > 0: if seq < self.seq and t not in ("snapshot", "snapshot_end"): self._fail(f"{t}: seq went backwards {self.seq} -> {seq}") self.seq = seq if t == "snapshot": if not frame.get("part") or frame["part"] <= 1: self.events = {} self.ready = False for doc in frame.get("events") or []: self.events[scalar_key(doc.get("eventId"))] = _Event(doc) elif t == "snapshot_end": self.ready = True count = frame.get("count") if isinstance(count, int) and count != len(self.events): self._fail(f"snapshot_end count {count} but {len(self.events)} events received") elif t == "resumed": self.ready = True elif t == "delta": for c in frame.get("changes") or []: self._change(c) return t def _change(self, c: Dict[str, Any]) -> None: eid = scalar_key(c.get("eventId")) op = c.get("op") self.stats["changes"][op] = self.stats["changes"].get(op, 0) + 1 if op == "upsert": doc = c.get("event") or {} if scalar_key(doc.get("eventId")) != eid: self._fail(f"upsert: change eventId {eid} but event carries {scalar_key(doc.get('eventId'))}") self.events[eid] = _Event(doc) elif op == "remove": if self.events.pop(eid, None) is None: self._fail(f"remove({c.get('reason')}): unknown event {eid}") elif op == "patch": ev = self.events.get(eid) if ev is None: self._fail(f"patch: unknown event {eid}") return self._patch(ev, c, eid) else: self._fail(f"unknown op {op}") def _patch(self, ev: _Event, p: Dict[str, Any], eid: str) -> None: for k, v in (p.get("set") or {}).items(): ev.fields[k] = v for k in p.get("unset") or []: ev.fields.pop(k, None) for k in p.get("mDel") or []: if ev.markets is None or ev.markets.pop(k, None) is None: self._fail(f"patch {eid}: mDel unknown market {k}") if p.get("mAdd"): if ev.markets is None: ev.markets = {} for m in p["mAdd"]: ev.markets[market_key(m)] = _Market(m) for ms in p.get("mSet") or []: mk = ev.markets.get(ms.get("k")) if ev.markets else None if mk is None: self._fail(f"patch {eid}: mSet unknown market {ms.get('k')}") continue for f, v in ms.items(): if f == "k": continue if v is None: mk.fields.pop(f, None) else: mk.fields[f] = v for ref in p.get("selDel") or []: mk = ev.markets.get(ref.get("m")) if ev.markets else None if mk is None or mk.selections is None or mk.selections.pop(ref.get("k"), None) is None: self._fail(f"patch {eid}: selDel unknown selection {ref.get('k')} in market {ref.get('m')}") for s in p.get("sel") or []: mk = ev.markets.get(s.get("m")) if ev.markets else None if mk is None: self._fail(f"patch {eid}: sel unknown market {s.get('m')}") continue if mk.selections is None: mk.selections = {} mk.selections[s.get("k")] = {f: v for f, v in s.items() if f not in ("m", "k")} def get(self, event_id) -> Optional[Dict[str, Any]]: """One event as the HTTP API serves it, or None.""" ev = self.events.get(scalar_key(event_id)) return ev.to_dict() if ev is not None else None def to_list(self) -> List[Dict[str, Any]]: """Every event as the HTTP API serves it (insertion order).""" return [ev.to_dict() for ev in self.events.values()] def __len__(self) -> int: return len(self.events) def canonical(event: Dict[str, Any]) -> str: """Order-independent JSON of an event for comparisons: markets sorted by marketKey, selections by selKey, keys sorted, null market scalars dropped (same rules as the Go diff.Canonical).""" ev = _Event(event) out: Dict[str, Any] = dict(ev.fields) if ev.markets is not None: markets = [] for k in sorted(ev.markets): mk = ev.markets[k] obj = {f: v for f, v in mk.fields.items() if v is not None} if mk.selections is not None: obj["selections"] = [mk.selections[sk] for sk in sorted(mk.selections)] markets.append(obj) out["markets"] = markets return json.dumps(out, sort_keys=True, separators=(",", ":"), ensure_ascii=False) # --------------------------------------------------------------------------- # Racing (kind=racing). Races are not events: no eventId, no markets, no # selections -- a flat race with a list of runners under ``horses`` or # ``greyhounds``. RaceBoard is a deliberate duplicate of DeltaBoard rather # than a shared base class, so nothing here can change how event streams # behave. _RUNNER_FIELDS = ("horses", "greyhounds") def runner_key(r: Dict[str, Any]) -> str: """``id`` when present and non-empty, else number|trap|name. NOT guaranteed unique within a race: bet365 ships races where one id covers several different runners. ``runner_keys`` resolves those collisions. """ rid = scalar_key(r.get("id")) return rid if rid != "" else _identity(r) def _identity(r: Dict[str, Any]) -> str: """A runner's stable attributes: the number/trap it runs under and its name. Nothing here is a price, so a runner keeps its identity while its odds move.""" return _compound(r, ("number", "trap", "name")) def collision_key(r: Dict[str, Any]) -> str: """The key of a runner whose id is shared with another runner in the race.""" rid = scalar_key(r.get("id")) return runner_key(r) if rid == "" else f"{rid}|{_identity(r)}" def runner_keys(runners) -> List[str]: """Assign a key to every runner of one race, in document order. This is THE rule; the server and every applier go through it, so identical arrays give identical keys. 1. ``id`` when only one runner in the race has that id -- the common case, and position never enters it; 2. ``id|number|trap|name`` for every member of a group sharing an id, so a duplicate group keeps its keys when the scraper reorders the array; 3. ``#2``, ``#3``, ... in document order, only for runners no identity field can tell apart (byte-identical duplicates), where a positional key is stable by definition. Dropping a duplicate instead would delete a real runner from the board. """ shared: Dict[str, int] = {} for r in runners: k = runner_key(r) shared[k] = shared.get(k, 0) + 1 taken = set() out: List[str] = [] for r in runners: k = runner_key(r) if shared[k] > 1: k = collision_key(r) if k in taken: n = 2 while f"{k}#{n}" in taken: n += 1 k = f"{k}#{n}" taken.add(k) out.append(k) return out def is_racing_stream(stream: str) -> bool: """True for a ``/racing/`` stream id.""" parts = str(stream).split("/") return len(parts) > 1 and parts[1] == "racing" class _Race: __slots__ = ("fields", "runners_field", "runners") def __init__(self, doc: Dict[str, Any]): self.runners_field: Optional[str] = None self.runners: Dict[str, Dict[str, Any]] = {} for f in _RUNNER_FIELDS: if isinstance(doc.get(f), list): self.runners_field = f rs = [r for r in doc[f] if isinstance(r, dict)] for k, r in zip(runner_keys(rs), rs): self.runners[k] = r break self.fields: Dict[str, Any] = { k: v for k, v in doc.items() if k != self.runners_field and k not in _IGNORED_TOP } def to_dict(self) -> Dict[str, Any]: out = dict(self.fields) if self.runners_field is not None: out[self.runners_field] = list(self.runners.values()) return out class RaceBoard: """Client-side state of one ``kind=racing`` stream. Same surface as DeltaBoard. A patch is applied in the protocol order set -> unset -> rDel -> rSet; every value is absolute, so re-applying a frame is a no-op. board = RaceBoard() async for message in websocket: board.apply(message) board.to_list() # races exactly as the HTTP API serves them board.get("R-4411") # one race by its ``id``, or None """ def __init__(self, strict: bool = False): self.strict = strict self.races: Dict[str, _Race] = {} self.seq = 0 self.ready = False self.stats: Dict[str, Any] = {"frames": {}, "changes": {}, "bytes": 0, "errors": []} def _fail(self, msg: str) -> None: if self.strict and len(self.stats["errors"]) < 50: self.stats["errors"].append(msg) def apply(self, frame) -> str: """Apply one frame (JSON text/bytes or a dict). Returns the frame type.""" if isinstance(frame, (str, bytes, bytearray)): self.stats["bytes"] += len(frame) frame = json.loads(frame) t = frame.get("t", "") self.stats["frames"][t] = self.stats["frames"].get(t, 0) + 1 seq = frame.get("seq") if isinstance(seq, int) and not isinstance(seq, bool) and seq > 0: if seq < self.seq and t not in ("snapshot", "snapshot_end"): self._fail(f"{t}: seq went backwards {self.seq} -> {seq}") self.seq = seq if t == "snapshot": if not frame.get("part") or frame["part"] <= 1: self.races = {} self.ready = False # A snapshot carries the race documents themselves, so the # identity is the race's own ``id``; only the delta envelope # repeats it as ``eventId``. for doc in frame.get("events") or []: self.races[scalar_key(doc.get("id"))] = _Race(doc) elif t == "snapshot_end": self.ready = True count = frame.get("count") if isinstance(count, int) and count != len(self.races): self._fail(f"snapshot_end count {count} but {len(self.races)} races received") elif t == "resumed": self.ready = True elif t == "delta": for c in frame.get("changes") or []: self._change(c) return t def _change(self, c: Dict[str, Any]) -> None: rid = scalar_key(c.get("eventId")) op = c.get("op") self.stats["changes"][op] = self.stats["changes"].get(op, 0) + 1 if op == "upsert": doc = c.get("event") or {} if scalar_key(doc.get("id")) != rid: self._fail(f"upsert: change eventId {rid} but race carries {scalar_key(doc.get('id'))}") self.races[rid] = _Race(doc) elif op == "remove": if self.races.pop(rid, None) is None: self._fail(f"remove({c.get('reason')}): unknown race {rid}") elif op == "patch": rc = self.races.get(rid) if rc is None: self._fail(f"patch: unknown race {rid}") return self._patch(rc, c, rid) else: self._fail(f"unknown op {op}") def _patch(self, rc: _Race, p: Dict[str, Any], rid: str) -> None: for k, v in (p.get("set") or {}).items(): rc.fields[k] = v for k in p.get("unset") or []: rc.fields.pop(k, None) for k in p.get("rDel") or []: if rc.runners.pop(k, None) is None: self._fail(f"patch {rid}: rDel unknown runner {k}") for r in p.get("rSet") or []: k = r.get("k") if not k: self._fail(f"patch {rid}: rSet runner without a key") continue # ``rSet`` replaces the whole runner; an existing key keeps its position. rc.runners[k] = {f: v for f, v in r.items() if f != "k"} if rc.runners_field is None: rc.runners_field = _RUNNER_FIELDS[0] self._fail(f"patch {rid}: rSet on a race with no runner array; assuming {rc.runners_field}") def get(self, race_id) -> Optional[Dict[str, Any]]: """One race as the HTTP API serves it, or None.""" rc = self.races.get(scalar_key(race_id)) return rc.to_dict() if rc is not None else None def to_list(self) -> List[Dict[str, Any]]: """Every race as the HTTP API serves it (insertion order).""" return [rc.to_dict() for rc in self.races.values()] def __len__(self) -> int: return len(self.races) def canonical_race(race: Dict[str, Any]) -> str: """Order-independent JSON of a race for comparisons: runners sorted by runnerKey, keys sorted. The runner array is emitted as ``runners`` plus a ``runnersField`` marker, so the same race stored under a different spelling does not compare equal (same rules as the Go racediff.Canonical).""" rc = _Race(race) out: Dict[str, Any] = dict(rc.fields) if rc.runners_field is not None: out["runners"] = [rc.runners[k] for k in sorted(rc.runners)] out["runnersField"] = rc.runners_field return json.dumps(out, sort_keys=True, separators=(",", ":"), ensure_ascii=False) class StreamClient: """Protocol v2: one socket carries many streams. Routes every frame to the board of its ``stream`` id -- a DeltaBoard, or a RaceBoard when the id says ``kind=racing`` -- created on first sight; frames without a stream (connected, hb, connection-level errors) are counted only. Per-stream error frames (4004, 4031, 1013) never create a board — a 4031 is followed by a fresh snapshot that resets the existing one. client = StreamClient() async for message in websocket: stream, kind = client.apply(message) if kind == "delta": render(stream, client.board(stream).to_list()) """ def __init__(self, strict: bool = False): self.strict = strict self.boards: Dict[str, Any] = {} # DeltaBoard, or RaceBoard for kind=racing self.limits: Optional[Dict[str, int]] = None self.streams: List[Dict[str, Any]] = [] self.stats: Dict[str, Any] = {"frames": {}, "errors": []} def apply(self, frame) -> "tuple[str, str]": if isinstance(frame, (str, bytes, bytearray)): frame = json.loads(frame) t = frame.get("t", "") self.stats["frames"][t] = self.stats["frames"].get(t, 0) + 1 if t == "connected": self.limits = frame.get("limits") self.streams = frame.get("streams") or [] return "", t if t == "error": self.stats["errors"].append({"stream": frame.get("stream", ""), "code": frame.get("code"), "message": frame.get("message"), "validSports": frame.get("validSports"), "retryAfterSec": frame.get("retryAfterSec")}) return frame.get("stream", ""), t stream = frame.get("stream") if not stream: return "", t board = self.boards.get(stream) if board is None: # The stream id is `//`, so the kind # decides which board a stream needs -- racing frames carry # runners, not markets. board = RaceBoard(strict=self.strict) if is_racing_stream(stream) else DeltaBoard(strict=self.strict) self.boards[stream] = board board.apply(frame) return stream, t def board(self, stream: str): """The board of a stream (DeltaBoard, or RaceBoard for kind=racing).""" return self.boards.get(stream) def since(self) -> Dict[str, int]: """Last seq per stream — what to send back as ``since`` on a reconnect.""" return {sid: b.seq for sid, b in self.boards.items()} # --------------------------------------------------------------------------- # Connection keeping (plan P8.1): a socket that survives deploys and drops. #: Close codes after which reconnecting cannot help: bad key, plan without #: streams, plan lost or downgraded, replaced by a newer socket of the same #: account, subscription lapsed, bad declaration, declaration over the caps. FATAL_CLOSE_CODES = frozenset({4001, 4003, 4010, 4011, 4012, 4013, 4032, 4034, 4035}) #: Query parameters of the URL declaration form (dropped once the payload form takes over). _URL_DECLARATION_PARAMS = ("bookmakers", "sports", "kinds", "markets", "from", "to") class StreamConnection: """Owns one socket and keeps it alive across drops: deploys (``1012``), network loss (``1006``), server full (``4029``), the reconnect cooldown (``4429``, honouring ``retryAfterSec``), a slow-consumer close (``4030``). After the first ``connected`` it redeclares the very streams the server listed, in the payload form, with ``since`` per stream — so a reconnect is answered with ``resumed`` + the missed deltas, not a snapshot, and the boards held in the StreamClient stay intact. conn = StreamConnection("wss://api.pulsescore.net/api/stream/ws", key=KEY, streams=[{"bookmaker": "bet365", "kind": "live", "sport": "soccer"}], on_frame=lambda stream, kind, client, raw: ..., on_state=lambda state, info: print(state, info)) asyncio.run(conn.run()) # needs the `websockets` package (>= 12) ``declaration()``, ``next_url()``, ``should_reconnect(code)`` and ``backoff(attempt, retry_after)`` are pure and usable with any library. """ def __init__(self, url: str, key: str = "", streams: Optional[List[Dict[str, Any]]] = None, client: Optional[StreamClient] = None, on_frame=None, on_state=None, min_backoff: float = 0.5, max_backoff: float = 30.0, max_retries: Optional[int] = None, strict: bool = False, random=None): self.url = url self.key = key self.streams = [dict(s) for s in streams] if streams else None self.client = client or StreamClient(strict=strict) self.on_frame = on_frame or (lambda stream, kind, client, raw: None) self.on_state = on_state or (lambda state, info: None) self.min_backoff = min_backoff self.max_backoff = max_backoff self.max_retries = max_retries self._random = random or __import__("random").random self.state = "closed" self.attempt = 0 # consecutive attempts without a `connected` self.declared: Optional[List[Dict[str, Any]]] = None self.last_close: Optional[Dict[str, Any]] = None self._closed = False # -- pure parts ------------------------------------------------------- def declaration(self) -> Optional[Dict[str, Any]]: """Payload declaration of the next attempt (None = URL form): the streams ``connected`` listed (or the constructor's before that) with ``since`` = the seq each board holds.""" src = self.declared if self.declared is not None else self.streams if src is None: return None out = [] for s in src: d = {"bookmaker": s.get("bookmaker"), "kind": s.get("kind"), "sport": s.get("sport")} f = s.get("filters") or s for k in ("league", "from", "to"): if f.get(k): d[k] = f[k] if f.get("markets"): d["markets"] = list(f["markets"]) board = self.client.board(s["stream"]) if s.get("stream") else None seq = board.seq if board is not None else int(s.get("since") or 0) if seq > 0: d["since"] = seq out.append(d) return {"streams": out} def next_url(self) -> str: """URL of the next attempt: ``key`` added, URL-declaration parameters removed once the payload form carries the declaration.""" from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit parts = urlsplit(self.url) q = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True) if k != "key"] if self.declaration() is not None: q = [(k, v) for k, v in q if k not in _URL_DECLARATION_PARAMS] if self.key: q.append(("key", self.key)) return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(q), parts.fragment)) @staticmethod def should_reconnect(code: int) -> bool: return code not in FATAL_CLOSE_CODES def backoff(self, attempt: int, retry_after: Optional[float] = None) -> float: """Seconds before attempt n (1-based): exponential with ±25 % jitter, or the server's ``retryAfterSec``.""" if retry_after and retry_after > 0: return float(retry_after) base = min(self.max_backoff, self.min_backoff * (2 ** max(0, attempt - 1))) return base * (0.75 + 0.5 * self._random()) def retry_after_for(self, code: int) -> Optional[float]: errs = self.client.stats["errors"] if errs and errs[-1].get("code") == code and (errs[-1].get("retryAfterSec") or 0) > 0: return float(errs[-1]["retryAfterSec"]) return None def on_connected(self) -> None: """Bookkeeping when a ``connected`` frame arrived (called by run()).""" self.attempt = 0 self.declared = [dict(s) for s in self.client.streams] self._set_state("open", {"streams": len(self.declared)}) def on_closed(self, code: int, reason: str = "") -> Optional[float]: """Decide what a close means: None = stop (state ``closed``), else the delay in seconds before the next attempt (state ``reconnecting``).""" self.last_close = {"code": code, "reason": reason} if self._closed: self._set_state("closed", dict(self.last_close)) return None if not self.should_reconnect(code): self._set_state("closed", dict(self.last_close, fatal=True)) return None self.attempt += 1 if self.max_retries is not None and self.attempt > self.max_retries: self._set_state("closed", dict(self.last_close, exhausted=True)) return None delay = self.backoff(self.attempt, self.retry_after_for(code)) self._set_state("reconnecting", dict(self.last_close, attempt=self.attempt, delay=delay)) return delay def close(self) -> None: """Close for good: run() returns after the current socket ends.""" self._closed = True def _set_state(self, state: str, info: Dict[str, Any]) -> None: self.state = state self.on_state(state, info) # -- asyncio loop on the `websockets` package --------------------------- async def run(self) -> None: import asyncio import websockets # optional dependency: pip install websockets from websockets.exceptions import ConnectionClosed, InvalidHandshake while not self._closed: self._set_state("connecting", {"attempt": self.attempt}) code, reason = 1006, "" try: async with websockets.connect(self.next_url(), max_size=None) as ws: decl = self.declaration() if decl is not None: await ws.send(json.dumps(decl, separators=(",", ":"))) try: async for message in ws: stream, kind = self.client.apply(message) if kind == "connected": self.on_connected() self.on_frame(stream, kind, self.client, message) if self._closed: await ws.close(1000, "client closed") break except ConnectionClosed as e: code, reason = int(e.code or 1006), e.reason or "" else: code, reason = int(getattr(ws, "close_code", None) or 1000), getattr(ws, "close_reason", "") or "" except (InvalidHandshake, OSError, asyncio.TimeoutError): code, reason = 1006, "handshake failed" delay = self.on_closed(code, reason) if delay is None: return await asyncio.sleep(delay)