The server sent a disconnect event at 14:23:47 UTC. Your client logged it, notified the monitoring system, and initiated a reconnect sequence. By 14:28:52 UTC, the WebSocket was back open and streaming live data. Five minutes of market activity — including a significant volatility event at 14:25:00 — passed through the air and into nothing. Your system never saw it.
This is not a hypothetical edge case. It is a guaranteed eventual state. Network partitions occur. Cloud functions hit memory limits. Load balancers terminate idle connections. Mobile clients enter tunnels. Every production market data system designed around WebSocket push will encounter the gap problem: how do you recover structured data that arrived while you were disconnected?
The answer is not "reconnect and ask for everything again." That approach is expensive, slow, and often impossible with real-time streams. The correct architecture combines three techniques in sequence: timestamp-based state capture at disconnect, incremental REST backfill, and deterministic local merge with conflict resolution. This article walks through each layer with production-grade Python code that you can adapt to any market data WebSocket feed.
The Gap Problem: Why Polling Everything Is Not the Answer
Market data systems generate high-frequency events. A single US equity ticker during active trading can produce hundreds of depth updates and thousands of trades per minute. Requesting the full order book or complete trade history on every reconnect would overwhelm your rate limits, introduce latency spikes, and generate unnecessary load on the data provider's infrastructure.
The constraint is not just bandwidth. Many real-time APIs expose one-way streaming channels without a corresponding full-state query endpoint. Even when full-state endpoints exist, they return snapshots, not the granular event stream needed for accurate strategy backtesting or audit trails. You need a surgical approach: fetch only the data that was missed, and fetch it in a way that aligns with the state your application believes it is in.
Timestamp alignment is the foundation. Before attempting any recovery, your client must record the timestamp of the last processed message. This value becomes the checkpoint. Every subsequent operation — whether a backfill request or a live subscription resume — is scoped relative to this checkpoint.
Architecture: Three-Layer Recovery System
The recovery architecture consists of three distinct layers, each with a specific responsibility.
Layer 1: Disconnect Detection and State Preservation. When the WebSocket connection drops, the client must immediately capture the last known state: the last message timestamp, the sequence number if available, and any in-memory order book snapshots. This checkpoint is persisted to durable storage — a local file, a Redis key, or a database row — before the reconnect attempt begins.
Layer 2: Incremental Backfill via REST. After reconnecting, the client issues a time-range query to a REST endpoint that returns historical data for the missed window. The query parameters include the checkpoint timestamp as the start point and the current server time as the end point. The response is processed as a batch, not as a stream, and each message is tagged with its origin timestamp for merge logic.
Layer 3: Deterministic Merge with Live Stream. The backfilled batch and the newly arriving live stream now converge on the same local buffer. A merge strategy must handle three scenarios: duplicate messages (same event arrived via backfill and again via live stream), ordering violations (a backfilled message arrives after a newer live message), and gap detection (the backfill response shows the requested range is partially unavailable).
Production-Grade Reconnection with Exponential Backoff
The reconnect logic is where most implementations fall short. A naive reconnect loop — retry immediately, retry every second — will either hammer the server during an outage or give up too quickly during a transient network hiccup. The following implementation uses exponential backoff with jitter, respects rate limit responses, and persists connection state across process restarts.
import os
import time
import json
import logging
import threading
from datetime import datetime, timezone
from typing import Optional, Callable
from websocket import create_connection, WebSocketTimeoutException, WebSocketConnectionClosedException
logger = logging.getLogger(__name__)
class WebSocketReconnectingClient:
"""
WebSocket client with production-grade reconnection logic:
- Exponential backoff with jitter to prevent thundering herd
- Rate limit handling (code 3001 + Retry-After header)
- State persistence across process restarts
- Thread-safe message dispatch
"""
DEFAULT_CONFIG = {
"base_delay": 1.0, # seconds
"max_delay": 60.0, # seconds
"jitter_factor": 0.1, # +/- 10% randomization
"max_retries": None, # None = infinite
"heartbeat_interval": 25, # seconds — keepalive ping
"connection_timeout": 10, # seconds for initial connect
}
def __init__(
self,
url: str,
api_key: Optional[str] = None,
state_file: str = ".ws_state.json",
on_message: Optional[Callable[[dict], None]] = None,
on_connect: Optional[Callable[[], None]] = None,
on_disconnect: Optional[Callable[[str], None]] = None,
**config_overrides
):
self.url = url
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self.state_file = state_file
self.config = {**self.DEFAULT_CONFIG, **config_overrides}
self.on_message = on_message
self.on_connect = on_connect
self.on_disconnect = on_disconnect
self._ws: Optional[Any] = None
self._running = False
self._lock = threading.RLock()
self._retry_count = 0
self._last_message_ts: Optional[str] = None
self._thread: Optional[threading.Thread] = None
self._load_state()
def _load_state(self):
"""Restore last checkpoint from persistent storage."""
try:
with open(self.state_file, "r") as f:
state = json.load(f)
self._last_message_ts = state.get("last_message_ts")
self._retry_count = state.get("retry_count", 0)
logger.info(f"Restored state: last_ts={self._last_message_ts}, retries={self._retry_count}")
except (FileNotFoundError, json.JSONDecodeError):
logger.info("No prior state found — starting fresh")
self._retry_count = 0
def _save_state(self):
"""Persist checkpoint to durable storage."""
state = {
"last_message_ts": self._last_message_ts,
"retry_count": self._retry_count,
"saved_at": datetime.now(timezone.utc).isoformat(),
}
with open(self.state_file, "w") as f:
json.dump(state, f)
def _build_url(self) -> str:
"""Construct WebSocket URL with API key as query parameter."""
separator = "&" if "?" in self.url else "?"
return f"{self.url}{separator}api_key={self.api_key}"
def _calculate_delay(self) -> float:
"""Exponential backoff with uniform jitter."""
delay = min(
self.config["base_delay"] * (2 ** self._retry_count),
self.config["max_delay"]
)
jitter = delay * self.config["jitter_factor"] * (2 * __import__("random").random() - 1)
return max(0, delay + jitter)
def _connect(self) -> bool:
"""Establish WebSocket connection with timeout and heartbeat."""
try:
ws_url = self._build_url()
self._ws = create_connection(
ws_url,
timeout=self.config["connection_timeout"]
)
self._ws.settimeout(self.config["heartbeat_interval"])
logger.info(f"Connected to {self.url}")
self._retry_count = 0
self._save_state()
return True
except Exception as e:
logger.error(f"Connection failed: {type(e).__name__}: {e}")
return False
def _send_ping(self):
"""Heartbeat: keepalive ping to prevent idle timeout disconnects."""
try:
self._ws.send(json.dumps({"cmd": "ping"}))
except Exception as e:
logger.warning(f"Ping failed: {e}")
raise
def _handle_message(self, raw: str):
"""Parse, checkpoint, and dispatch a single message."""
try:
msg = json.loads(raw)
# Check for rate limit response
if msg.get("code") == 3001:
retry_after = int(msg.get("headers", {}).get("Retry-After", 5))
logger.warning(f"Rate limited — waiting {retry_after}s")
time.sleep(retry_after)
return
# Update checkpoint on every valid message
if "ts" in msg or "timestamp" in msg:
self._last_message_ts = msg.get("ts") or msg.get("timestamp")
self._save_state()
# Dispatch to handler
if self.on_message:
self.on_message(msg)
except json.JSONDecodeError as e:
logger.warning(f"Malformed message ignored: {e}")
def _receive_loop(self):
"""Main receive loop with heartbeat and graceful shutdown."""
while self._running:
try:
# Send heartbeat
self._send_ping()
# Receive with timeout
raw = self._ws.recv()
# None = connection closed by server
if raw is None:
reason = "server_closed"
break
self._handle_message(raw)
except WebSocketTimeoutException:
# Timeout is expected — just loop back to send ping
continue
except WebSocketConnectionClosedException:
reason = "connection_closed"
break
except Exception as e:
logger.error(f"Receive error: {type(e).__name__}: {e}")
reason = f"error_{type(e).__name__}"
break
# Ran out of loop — connection is dead
if self._running:
self._running = False
if self.on_disconnect:
self.on_disconnect(reason)
self._schedule_reconnect()
def _schedule_reconnect(self):
"""Calculate delay and schedule next connection attempt."""
if self.config["max_retries"] and self._retry_count >= self.config["max_retries"]:
logger.error(f"Max retries ({self.config['max_retries']}) reached — giving up")
return
delay = self._calculate_delay()
self._retry_count += 1
logger.info(f"Reconnecting in {delay:.1f}s (attempt {self._retry_count})")
def delayed_connect():
time.sleep(delay)
if self._running:
self._connect_and_loop()
threading.Thread(target=delayed_connect, daemon=True).start()
def _connect_and_loop(self):
"""Attempt connection, then start receive loop if successful."""
if not self._connect():
self._schedule_reconnect()
else:
self._running = True
if self.on_connect:
self.on_connect()
self._receive_loop()
def start(self):
"""Start the client in a background thread."""
if self._running:
logger.warning("Client already running")
return
logger.info("Starting WebSocket client")
self._thread = threading.Thread(target=self._connect_and_loop, daemon=True)
self._thread.start()
def stop(self):
"""Graceful shutdown."""
logger.info("Stopping WebSocket client")
self._running = False
if self._ws:
try:
self._ws.close()
except Exception:
pass
The class above handles the connect-and-stay-connected lifecycle. It does not yet handle the data backfill — that is the responsibility of the layer that orchestrates this client. The on_disconnect callback is where you inject the backfill logic before the reconnect succeeds.
Incremental Backfill: Fetching the Missed Window
Once reconnection is confirmed, your client must backfill the gap between the last known checkpoint and the current live stream head. The REST backfill request is a time-range query with three parameters: the symbol or channel, the start timestamp, and an optional end timestamp. The response is a list of historical events sorted by timestamp.
The critical implementation detail is sequencing: you must not process live messages until the backfill is fully ingested and merged. The cleanest way to enforce this is a dual-buffer architecture with a merge coordinator.
import requests
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timezone
from collections import deque
import threading
@dataclass
class MessageEnvelope:
"""
Wrapper for any message (backfilled or live) with canonical ordering metadata.
This ensures deterministic merge regardless of arrival order.
"""
raw: Dict[str, Any]
received_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
origin: str = "unknown" # "backfill" or "live"
class BackfillCoordinator:
"""
Orchestrates dual-phase recovery:
Phase 1: Backfill window from REST API into backfill_buffer
Phase 2: Merge live stream with backfill buffer, deduplicate, dispatch in order
Thread-safe. Handles partial backfill failures gracefully.
"""
def __init__(
self,
rest_base_url: str,
api_key: Optional[str] = None,
buffer_size: int = 10000,
on_output: Optional[callable] = None
):
self.rest_base_url = rest_base_url.rstrip("/")
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self.buffer_size = buffer_size
self.on_output = on_output
self.backfill_buffer: deque[MessageEnvelope] = deque(maxlen=buffer_size)
self.live_buffer: deque[MessageEnvelope] = deque(maxlen=buffer_size)
self.last_merged_ts: Optional[str] = None
self._phase: str = "live" # "backfill" or "live" or "merged"
self._lock = threading.RLock()
self._backfill_complete = threading.Event()
def _rest_get(self, endpoint: str, params: dict, timeout: tuple = (3.05, 10)) -> List[dict]:
"""Issue a REST request with timeout and error handling."""
url = f"{self.rest_base_url}{endpoint}"
headers = {"X-API-Key": self.api_key}
try:
response = requests.get(url, headers=headers, params=params, timeout=timeout)
data = response.json()
if data.get("code") == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
import time as time_module
time_module.sleep(retry_after)
return self._rest_get(endpoint, params, timeout)
if data.get("code") != 0:
raise RuntimeError(f"REST error {data.get('code')}: {data.get('message')}")
return data.get("data", [])
except requests.Timeout:
raise TimeoutError(f"REST request to {endpoint} timed out after {timeout}s")
except requests.RequestException as e:
raise ConnectionError(f"REST request failed: {e}")
def backfill(
self,
channel: str,
start_ts: str,
end_ts: Optional[str] = None,
interval: str = "1m"
) -> int:
"""
Fetch historical data for the missed window via REST.
Returns the count of messages buffered.
⚠️ This is a simplified single-request version. For production use
with large windows, implement pagination or chunked fetching by
subdividing the time range to respect API limit constraints.
"""
params = {
"channel": channel,
"start": start_ts,
"interval": interval,
}
if end_ts:
params["end"] = end_ts
messages = self._rest_get("/v1/market/history", params)
with self._lock:
for raw in messages:
envelope = MessageEnvelope(raw=raw, origin="backfill")
self.backfill_buffer.append(envelope)
# Sort backfill buffer by message timestamp for deterministic output
sorted_buffer = sorted(
self.backfill_buffer,
key=lambda e: e.raw.get("ts") or e.raw.get("timestamp") or ""
)
self.backfill_buffer = deque(sorted_buffer, maxlen=self.buffer_size)
self._backfill_complete.set()
self._phase = "merged"
return len(messages)
def ingest_live(self, raw: Dict[str, Any]):
"""
Accept a message from the live WebSocket stream.
During backfill phase, buffers it. After backfill, dispatches directly.
"""
envelope = MessageEnvelope(raw=raw, origin="live")
with self._lock:
if self._phase == "backfill":
self.backfill_buffer.append(envelope)
return
# Merged phase: interleave live with backfill, output in order
self.live_buffer.append(envelope)
self._drain_merged()
def _drain_merged(self):
"""Drain both buffers in timestamp order, dispatching deduplicated messages."""
# Merge and sort both buffers
all_messages = list(self.backfill_buffer) + list(self.live_buffer)
all_messages.sort(key=lambda e: e.raw.get("ts") or e.raw.get("timestamp") or "")
output_batch = []
seen_timestamps = set()
for envelope in all_messages:
msg_ts = envelope.raw.get("ts") or envelope.raw.get("timestamp")
if msg_ts and msg_ts in seen_timestamps:
# Duplicate — skip
continue
if msg_ts:
seen_timestamps.add(msg_ts)
output_batch.append(envelope.raw)
self.last_merged_ts = msg_ts
# Clear processed entries
self.backfill_buffer.clear()
self.live_buffer.clear()
# Dispatch to output handler
for msg in output_batch:
if self.on_output:
self.on_output(msg)
@property
def last_checkpoint(self) -> Optional[str]:
"""Return the timestamp of the last successfully merged message."""
return self.last_merged_ts
The coordinator class separates concerns cleanly. The WebSocket client calls ingest_live() on every incoming message. The backfill orchestrator calls backfill() once after reconnection. Internally, the coordinator buffers both streams and drains them in a deterministic timestamp order, eliminating duplicates by comparing message IDs or timestamps.
Timestamp Alignment: The Checkpoint Protocol
Timestamp alignment is the thread that connects every layer. The checkpoint is captured at three points: when the WebSocket connection drops, when the backfill completes, and when the merge phase concludes. Each checkpoint serves a different purpose.
When the connection drops, the last processed message timestamp is your backfill start point. This value must be persisted durably before reconnect attempts begin. In the WebSocketReconnectingClient implementation above, _save_state() writes to a JSON file. In a distributed system, you would write to Redis or a database with a TTL. The important property is write-before-retry: you never attempt a reconnect without first saving the checkpoint, because a process crash during the reconnect window would otherwise lose the checkpoint and trigger an expensive full-backfill.
When the backfill completes, you capture the end-of-backfill timestamp. This value becomes the "resume cursor" for the live subscription. When you subscribe to the live stream after backfill, you include the cursor as a parameter: the server knows to suppress any messages with timestamps earlier than the cursor, avoiding a window of duplicate delivery during the handoff from REST to WebSocket.
When the merge phase concludes, you update the checkpoint one final time. The value should match the timestamp of the most recent message dispatched to your application. This is the value that gets restored on the next process restart.
Conflict Resolution: Handling Edge Cases
The merge logic handles the common cases, but production deployments encounter three edge cases that require explicit handling.
Case 1: Partial backfill failure. The REST endpoint returns a 200 response but with fewer records than expected, or the server returns a gap indicator (an empty segment in the time range). Your code must compare the backfill response's time span against the requested window. If the response ends before the requested end timestamp, flag a gap warning and attempt a second backfill request starting from the response's last timestamp plus one interval.
Case 2: Late backfill arrival. A backfilled message arrives via REST after a newer live message has already been dispatched. The MessageEnvelope approach handles this by deferring dispatch until the backfill buffer is fully ingested, but if your implementation processes live messages immediately, you need a holdback window: after reconnection, buffer live messages for N seconds before dispatching, where N is your worst-case backfill latency estimate.
Case 3: Sequence number gaps. Some WebSocket feeds assign monotonically increasing sequence numbers to each message. If the backfill response skips a sequence number that was never delivered live, it is a genuine gap. If the sequence number falls within the backfill window but the server never sent it (legitimate absence, not a gap), you must update your local state machine accordingly. Log the anomaly but do not halt the stream — the data provider may have legitimately cancelled or amended that update.
Deployment Configurations by Scale
The appropriate architecture scales with your use case. For individual quant developers running a single-strategy system, the local-file state persistence and single-threaded coordinator are sufficient. The entire reconnection-and-backfill pipeline runs in one process with no external dependencies beyond the REST API and WebSocket feed.
For team deployments running multiple strategies against the same data source, centralize the stream ingestion into a dedicated gateway process. The gateway maintains the WebSocket connection and REST backfill logic, then publishes normalized messages to an internal message broker (Kafka, Redis Streams, or ZeroMQ). Each strategy process subscribes to the broker and maintains its own checkpoint, independent of the gateway's state. This architecture isolates the reconnection blast radius: if the gateway restarts, individual strategies continue processing from their own checkpoints.
For institutional deployments requiring audit trails, the coordinator should write every received message — both backfilled and live — to an append-only event log before dispatching to application logic. The event log becomes the source of truth for compliance reviews and strategy attribution. If a backfill reveals a discrepancy between the event log and the application state, the event log is authoritative.
Monitoring the Recovery Pipeline
A recovery pipeline without observability is a liability. You need three monitoring signals at minimum.
Reconnection frequency. Track how often the client reconnects per hour. A healthy steady-state should show near-zero reconnections during normal market hours. More than one reconnection per hour is a signal of instability in either your network path or the data provider's infrastructure.
Backfill lag. Measure the elapsed time from reconnection to backfill completion. This value indicates how stale your data was at the moment of reconnection. For latency-sensitive strategies, backfill lag above 30 seconds warrants an alert and an investigation into the REST endpoint's responsiveness.
Gap events. Count how often the backfill response covers a smaller time window than requested. A high gap event rate indicates the data provider is dropping data in certain time ranges — a condition that can silently corrupt your historical record if left unaddressed.
All three metrics should be logged with structured fields (connection ID, timestamp, symbol, gap size in seconds) so they can be aggregated and visualized in your monitoring dashboard.
Closing Thoughts
WebSocket disconnection is not an error to be avoided — it is a condition to be designed for. The five-minute gap is inevitable. What is not inevitable is losing the data that arrived during those five minutes. By capturing checkpoints durably, backfilling incrementally, and merging deterministically, you build a data pipeline that recovers from failure without losing fidelity. The code in this article is a starting framework. Production deployments add pagination across large backfill windows, multi-symbol coordination, and distributed state management. But the core principles — timestamp alignment, incremental fetch, deduplicated merge — remain constant regardless of scale.
The next time your monitoring system fires a reconnection alert at 14:28 UTC, you can respond with confidence: the gap is intentional, the recovery is automated, and the data is intact.
Next Steps
If you're evaluating market data APIs for real-time streaming, ensure the provider offers a corresponding REST backfill endpoint with time-range query support. Without both, your reconnection strategy is incomplete.
If you want to integrate this pattern with TickDB, the platform provides both the WebSocket stream and the REST backfill endpoint under a unified API key. The depth and ticker channels support timestamp-based resumption, and the /v1/market/history endpoint serves incremental backfill requests. Sign up at tickdb.ai to access the free tier — no credit card required.
If you're building a multi-strategy system, consider deploying the backfill coordinator as a shared gateway service that normalizes and republishes market data to your internal message broker. Each strategy process subscribes independently and maintains its own checkpoint, providing fault isolation across your portfolio of strategies.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace to get context-aware code generation for TickDB integration patterns.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. WebSocket connection behavior and API rate limits are subject to change — verify current specifications in the official API documentation before production deployment.