Fix graceful websocket reconnect and release 0.2.2

CHANGELOG.md

@@ -2,6 +2,17 @@
 
 All notable changes to `predxt` are documented here.
 
+## 0.2.2 - 2026-09-11
+
+- fixed graceful websocket stream exhaustion for Polymarket, Kalshi, and
+  Opinion: mark health disconnected, count the reconnect, wait for backoff,
+  and restore the connection and subscription
+- stop Opinion's old heartbeat before reconnecting and restart it on success
+- interrupt retry waits on explicit close and release websocket/heartbeat
+  resources when message iteration is cancelled or closed; repeated close
+  remains safe, and an explicit `connect()` can start a new session
+- preserved public imports and the existing `VenueMessage.raw_data` contract
+
 ## 0.2.1 - 2026-08-26
 
 - fixed Polymarket, Kalshi, and Opinion connection manager shutdown so idle

docs/reconnects.md

@@ -15,6 +15,21 @@ Connection retry behavior uses exponential backoff with jitter. Live systems
 should monitor `last_message_timestamp_ms`, reconnect counts, and process-level
 logs.
 
+## Graceful stream endings
+
+Polymarket, Kalshi, and Opinion recover when the remote websocket iterator
+ends normally as well as when it raises a connection error. Each disconnect
+sets `connected` to false, increments `reconnect_count` once, and waits for
+backoff before reconnecting and restoring the saved subscription. Opinion
+stops its old heartbeat before waiting and starts a new one after connecting.
+A completed socket is never repeatedly read without a retry delay.
+
+`close()` marks the client stopped before closing the socket and interrupts
+pending retry waits. Cancelling the message reader or calling `aclose()` on
+its iterator releases the connection and heartbeat without reconnecting.
+Repeated `close()` calls are safe, including before the first connection.
+After closing, call `connect()` explicitly to start a new session.
+
 ## Connection manager shutdown
 
 Polymarket, Kalshi, and Opinion connection managers own their background

llms-full.txt

@@ -98,3 +98,11 @@ predxt stream polymarket --asset-id 1234567890 --limit 10 --jsonl
 - `docs/cli.md`
 - `docs/security.md`
 - `skills/predxt/SKILL.md`
+
+## Websocket recovery
+
+All three websocket clients recover from graceful stream endings with
+backoff, updated connection health, and restored subscriptions. Opinion stops
+its heartbeat during reconnect. Explicit `close()` interrupts retry waits;
+cancelling or closing the message iterator releases the connection without
+reconnecting. Call `connect()` explicitly to restart a closed client.

llms.txt

@@ -63,3 +63,11 @@ predxt stream polymarket --asset-id 1234567890 --limit 10 --jsonl
 - Prefer `OrderBookState` for simple dashboards; do not claim execution-grade semantics.
 
 More context: `llms-full.txt`, `docs/`, `skills/predxt/SKILL.md`.
+
+## Websocket recovery
+
+All three websocket clients recover from graceful stream endings with
+backoff, updated connection health, and restored subscriptions. Opinion stops
+its heartbeat during reconnect. Explicit `close()` interrupts retry waits;
+cancelling or closing the message iterator releases the connection without
+reconnecting. Call `connect()` explicitly to restart a closed client.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "predxt"
-version = "0.2.1"
+version = "0.2.2"
 description = "Read-only prediction market data clients for Polymarket, Kalshi, and Opinion"
 readme = "README.md"
 requires-python = ">=3.12"

src/predxt/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.2.1"
+__version__ = "0.2.2"
 
 from .base import BaseWsClient, HealthMetrics, VenueMessage, build_venue_message
 from .events import (

src/predxt/kalshi/client.py

@@ -18,7 +18,7 @@
 )
 from predxt.kalshi.auth import build_kalshi_auth_headers
 from predxt.kalshi.parser import parse_message
-from predxt.utils.backoff import ExponentialBackoff
+from predxt.utils.backoff import ExponentialBackoff, _wait_for_backoff
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +35,7 @@ def __init__(
         self._ws_url = ws_url
         self._ws: Any | None = None
         self._connected = False
+        self._closed = asyncio.Event()
         self._connect_time: Optional[float] = None
         self._channels: list[str] = []
         self._params: dict[str, Any] = {}
@@ -44,17 +45,22 @@ def __init__(
         self._max_reconnect_attempts = max_reconnect_attempts
 
     async def connect(self, auth_params: Optional[dict[str, Any]] = None) -> None:
+        self._closed.clear()
         if auth_params is not None:
             self._auth_params = auth_params
         ws_path = urlparse(self._ws_url).path or "/trade-api/ws/v2"
         headers = self._build_auth_headers(self._auth_params, ws_path=ws_path)
         attempts = 0
 
-        while attempts < self._max_reconnect_attempts:
+        while not self._closed.is_set() and attempts < self._max_reconnect_attempts:
             try:
-                self._ws = await websockets.connect(
+                ws = await websockets.connect(
                     self._ws_url, additional_headers=headers
                 )
+                if self._closed.is_set():
+                    await ws.close()
+                    return
+                self._ws = ws
                 self._connected = True
                 self._connect_time = time.time()
                 self._backoff.reset()
@@ -66,8 +72,10 @@ async def connect(self, auth_params: Optional[dict[str, Any]] = None) -> None:
                 attempts += 1
                 self._health.last_error = str(exc)
                 self._health.last_error_timestamp_ms = time.time() * 1000
-                await asyncio.sleep(self._backoff.next_delay())
+                await _wait_for_backoff(self._backoff, self._closed)
 
+        if self._closed.is_set():
+            return
         raise ConnectionError(
             f"Failed to connect to Kalshi after {self._max_reconnect_attempts} attempts"
         )
@@ -104,60 +112,73 @@ async def unsubscribe(self, channels: list[str]) -> None:
         self._health.messages_sent += 1
 
     async def close(self) -> None:
-        if self._ws:
-            await self._ws.close()
+        self._closed.set()
         self._connected = False
         self._health.connected = False
+        ws, self._ws = self._ws, None
+        if ws is not None:
+            await ws.close()
 
     async def messages(self) -> AsyncIterator[VenueMessage]:
         seen_hashes: set[str] = set()
 
-        while True:
-            try:
-                if not self._connected:
-                    await self.connect(self._auth_params)
-                    if self._channels:
-                        await self.subscribe(self._channels, self._params)
-
-                if self._ws is None:
-                    continue
-                async for raw_msg in self._ws:
-                    payload = json.loads(raw_msg)
-                    parsed = parse_message(payload)
-                    if not parsed:
-                        continue
-
-                    dedupe_hash = parsed.get("hash")
-                    if not dedupe_hash:
-                        dedupe_hash = hashlib.sha1(
-                            json.dumps(parsed, sort_keys=True).encode("utf-8")
-                        ).hexdigest()
-
-                    if dedupe_hash in seen_hashes:
-                        continue
-                    seen_hashes.add(dedupe_hash)
-
-                    vm = build_venue_message(
-                        venue="kalshi",
-                        raw_data=parsed,
-                        timestamp_ms=time.time() * 1000,
-                    )
-
-                    self._health.messages_received += 1
-                    self._health.last_message_timestamp_ms = vm.timestamp_ms
-                    yield vm
-
-            except (websockets.exceptions.ConnectionClosed, OSError):
+        try:
+            while not self._closed.is_set():
+                try:
+                    if not self._connected:
+                        await self.connect(self._auth_params)
+                        if self._closed.is_set():
+                            return
+                        if self._channels:
+                            await self.subscribe(self._channels, self._params)
+
+                    if self._closed.is_set():
+                        return
+                    if self._ws is None:
+                        raise OSError("Websocket unavailable")
+                    async for raw_msg in self._ws:
+                        payload = json.loads(raw_msg)
+                        parsed = parse_message(payload)
+                        if not parsed:
+                            continue
+
+                        dedupe_hash = parsed.get("hash")
+                        if not dedupe_hash:
+                            dedupe_hash = hashlib.sha1(
+                                json.dumps(parsed, sort_keys=True).encode("utf-8")
+                            ).hexdigest()
+
+                        if dedupe_hash in seen_hashes:
+                            continue
+                        seen_hashes.add(dedupe_hash)
+
+                        vm = build_venue_message(
+                            venue="kalshi",
+                            raw_data=parsed,
+                            timestamp_ms=time.time() * 1000,
+                        )
+
+                        self._health.messages_received += 1
+                        self._health.last_message_timestamp_ms = vm.timestamp_ms
+                        yield vm
+
+                except (websockets.exceptions.ConnectionClosed, OSError):
+                    pass
+                except Exception:
+                    logger.exception("Unexpected Kalshi websocket error")
+                    raise
+
+                # Both graceful EOF and transport errors require a paced reconnect.
+                if self._closed.is_set():
+                    return
                 self._connected = False
                 self._health.connected = False
                 self._health.reconnect_count += 1
-                await asyncio.sleep(self._backoff.next_delay())
-            except Exception:
-                logger.exception("Unexpected Kalshi websocket error")
-                raise
-
-            if len(seen_hashes) > 10000:
-                seen_hashes.clear()
+                await _wait_for_backoff(self._backoff, self._closed)
+                if len(seen_hashes) > 10000:
+                    seen_hashes.clear()
+        finally:
+            await self.close()
 
     def health(self) -> HealthMetrics:
         self._health.connected = self._connected

src/predxt/opinion/client.py

@@ -16,7 +16,7 @@
     build_venue_message,
 )
 from predxt.opinion.parser import parse_message
-from predxt.utils.backoff import ExponentialBackoff
+from predxt.utils.backoff import ExponentialBackoff, _wait_for_backoff
 
 logger = logging.getLogger(__name__)
 
@@ -38,6 +38,7 @@ def __init__(
         self._max_reconnect_attempts = max_reconnect_attempts
         self._ws: Any | None = None
         self._connected = False
+        self._closed = asyncio.Event()
         self._connect_time: Optional[float] = None
         self._channels: list[str] = []
         self._params: dict[str, Any] = {}
@@ -47,16 +48,21 @@ def __init__(
         self._heartbeat_task: asyncio.Task | None = None
 
     async def connect(self, auth_params: Optional[dict[str, Any]] = None) -> None:
+        self._closed.clear()
         if auth_params is not None:
             self._auth_params = auth_params
         api_key = self._resolve_api_key()
         if not api_key:
             raise ValueError("Opinion websocket requires an API key")
 
         attempts = 0
-        while attempts < self._max_reconnect_attempts:
+        while not self._closed.is_set() and attempts < self._max_reconnect_attempts:
             try:
-                self._ws = await websockets.connect(self._authenticated_url(api_key))
+                ws = await websockets.connect(self._authenticated_url(api_key))
+                if self._closed.is_set():
+                    await ws.close()
+                    return
+                self._ws = ws
                 self._connected = True
                 self._connect_time = time.time()
                 self._backoff.reset()
@@ -69,8 +75,10 @@ async def connect(self, auth_params: Optional[dict[str, Any]] = None) -> None:
                 attempts += 1
                 self._health.last_error = str(exc)
                 self._health.last_error_timestamp_ms = time.time() * 1000
-                await asyncio.sleep(self._backoff.next_delay())
+                await _wait_for_backoff(self._backoff, self._closed)
 
+        if self._closed.is_set():
+            return
         raise ConnectionError(
             f"Failed to connect to Opinion after {self._max_reconnect_attempts} attempts"
         )
@@ -96,41 +104,54 @@ async def unsubscribe(self, channels: list[str]) -> None:
             self._health.messages_sent += 1
 
     async def close(self) -> None:
-        await self._stop_heartbeat()
-        if self._ws:
-            await self._ws.close()
+        self._closed.set()
         self._connected = False
         self._health.connected = False
+        ws, self._ws = self._ws, None
+        await self._stop_heartbeat()
+        if ws is not None:
+            await ws.close()
 
     async def messages(self) -> AsyncIterator[VenueMessage]:
         seen_hashes: set[str] = set()
-        while True:
-            try:
-                if not self._connected:
-                    await self.connect(self._auth_params)
-                    if self._channels:
-                        await self.subscribe(self._channels, self._params)
+        try:
+            while not self._closed.is_set():
+                try:
+                    if not self._connected:
+                        await self.connect(self._auth_params)
+                        if self._closed.is_set():
+                            return
+                        if self._channels:
+                            await self.subscribe(self._channels, self._params)
+
+                    if self._closed.is_set():
+                        return
+                    if self._ws is None:
+                        raise OSError("Websocket unavailable")
+                    async for raw_msg in self._ws:
+                        async for vm in self._handle_raw_message(
+                            raw_msg,
+                            seen_hashes=seen_hashes,
+                        ):
+                            yield vm
+                except (websockets.exceptions.ConnectionClosed, OSError):
+                    pass
+                except Exception:
+                    logger.exception("Unexpected Opinion websocket error")
+                    raise
 
-                if self._ws is None:
-                    continue
-                async for raw_msg in self._ws:
-                    async for vm in self._handle_raw_message(
-                        raw_msg,
-                        seen_hashes=seen_hashes,
-                    ):
-                        yield vm
-            except (websockets.exceptions.ConnectionClosed, OSError):
+                # Both graceful EOF and transport errors require a paced reconnect.
+                if self._closed.is_set():
+                    return
                 self._connected = False
                 self._health.connected = False
                 self._health.reconnect_count += 1
                 await self._stop_heartbeat()
-                await asyncio.sleep(self._backoff.next_delay())
-            except Exception:
-                logger.exception("Unexpected Opinion websocket error")
-                raise
-
-            if len(seen_hashes) > 10000:
-                seen_hashes.clear()
+                await _wait_for_backoff(self._backoff, self._closed)
+                if len(seen_hashes) > 10000:
+                    seen_hashes.clear()
+        finally:
+            await self.close()
 
     def health(self) -> HealthMetrics:
         self._health.connected = self._connected