"Two hours into a backtest run, your WebSocket connection drops for exactly 4 minutes and 37 seconds. When you reconnect, the stream picks up at the current price — but the gap is gone. You have no idea how many ticks you missed, which direction the pressure ratio shifted during the silence, or whether the spread spiked while you were dark.
This is not a theoretical failure mode. It is a routine occurrence in any production market data pipeline."
For systematic traders and quant engineers running real-time data streams, a dropped WebSocket connection is not an exception — it is an architectural event that must be handled gracefully. The core question is not how to prevent disconnects (you cannot). The core question is: what do you do in the 5 minutes after a reconnect, and how do you prove your local state is complete?
This article walks through a production-grade solution built around three operations: timestamp alignment, incremental REST fetching, and local buffer merging. All code examples use the TickDB API and follow the production-grade standards defined in the TickDB Content Strategy Handbook.
1. Why WebSocket Disconnections Are Inevitable
WebSocket connections fail. The causes range from mundane to systemic:
| Failure mode | Typical duration | Recoverable via reconnect? |
|---|---|---|
| Transient network jitter | < 5 seconds | Yes — simply reconnect |
| Cloud provider micro-outage | 10 seconds – 2 minutes | Yes — reconnect restores stream |
| Application crash / restart | Until manual restart | Yes — reconnect restores stream |
| ISP outage | Minutes to hours | Only after connectivity restored |
| Exchange-side maintenance window | Scheduled, announced | Yes — reconnect after window |
The critical insight: in every recoverable scenario, you return to a live stream that has moved forward in time. The data that existed during the blackout window is no longer being pushed. If you need it — and for backtesting, slippage analysis, or order book reconstruction, you often do — you must reconstruct it manually.
2. The Architecture: Two-Layer Data Recovery
The recovery architecture operates on two layers:
Layer 1: WebSocket Stream (real-time)
└── Continuous feed: new ticks arrive, order book updates, depth snapshots
Layer 2: REST Recovery (on-demand)
└── Historical fetch: fill the gap using the last-known timestamp as the start boundary
└── Local buffer: merge recovered data into the in-memory state
The two layers are complementary. WebSocket delivers the live stream. The REST API fills the holes. The local buffer is the single source of truth during and after recovery.
2.1 The State Machine
A robust reconnection handler implements a clear state machine:
CONNECTED ──[disconnect detected]──> DISCONNECTED
DISCONNECTED ──[reconnect attempt]──> RECONNECTING
RECONNECTING ──[success]──> RECOVERING ──[buffer complete]──> CONNECTED
RECONNECTING ──[failure]──> RECONNECTING (with backoff)
The RECOVERING state is where the REST recovery logic runs. This state must not be skipped — you cannot treat a reconnect as equivalent to a complete data state.
3. Step 1 — Detecting the Disconnect
Passive disconnects (server closes the connection) are immediately detectable. Active disconnects (network failure, your process crash) require a heartbeat mechanism to detect within a defined timeout window.
import time
import threading
from enum import Enum
class ConnectionState(Enum):
CONNECTED = "connected"
RECONNECTING = "reconnecting"
RECOVERING = "recovering"
DISCONNECTED = "disconnected"
class HeartbeatMonitor:
"""WebSocket heartbeat monitor with configurable timeout.
Thread-safe. On heartbeat timeout, triggers the reconnection callback.
"""
def __init__(self, timeout_seconds: float = 15.0, on_timeout=None):
self.timeout = timeout_seconds
self.last_pong = time.monotonic()
self._lock = threading.Lock()
self._running = False
self._thread = None
self.on_timeout = on_timeout
def start(self):
self.last_pong = time.monotonic()
self._running = True
self._thread = threading.Thread(target=self._monitor_loop, daemon=True)
self._thread.start()
def _monitor_loop(self):
while self._running:
time.sleep(1.0)
with self._lock:
elapsed = time.monotonic() - self.last_pong
if elapsed > self.timeout:
print(f"[HeartbeatMonitor] No pong received in {elapsed:.1f}s — triggering timeout")
if self.on_timeout:
self.on_timeout()
self._running = False
return
def record_pong(self):
with self._lock:
self.last_pong = time.monotonic()
def stop(self):
self._running = False
Engineering note: Use
time.monotonic()rather thantime.time()for heartbeat tracking.monotonicis immune to system clock adjustments (NTP syncs, daylight saving transitions), which can produce negative elapsed times and false timeout triggers.
4. Step 2 — Exponential Backoff Reconnection
Never reconnect immediately after a failure. Immediate retry amplifies load on the server, wastes resources on a connection that is likely to fail again, and can trigger rate limiting.
import random
import time
class ReconnectController:
"""Exponential backoff reconnect controller with jitter.
Spec-compliant with RFC 6555: adds jitter to prevent thundering herd.
"""
def __init__(
self,
base_delay: float = 1.0,
max_delay: float = 60.0,
max_retries: int = 20,
jitter_factor: float = 0.1
):
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
self.jitter_factor = jitter_factor
self.attempt = 0
def get_delay(self) -> float:
"""Returns the next reconnect delay in seconds."""
exp_delay = min(self.base_delay * (2 ** self.attempt), self.max_delay)
jitter = random.uniform(0, exp_delay * self.jitter_factor)
return exp_delay + jitter
def attempt_reconnect(self, connect_fn):
"""Executes a reconnect attempt with backoff. Returns True on success."""
if self.attempt >= self.max_retries:
print(f"[ReconnectController] Max retries ({self.max_retries}) reached — giving up")
return False
delay = self.get_delay()
print(f"[ReconnectController] Reconnecting in {delay:.2f}s (attempt {self.attempt + 1}/{self.max_retries})")
time.sleep(delay)
try:
connect_fn()
print("[ReconnectController] Reconnection successful")
self.attempt = 0
return True
except Exception as e:
print(f"[ReconnectController] Reconnection failed: {e}")
self.attempt += 1
return False
def reset(self):
self.attempt = 0
Rate limit handling: If the server returns error code 3001, respect the Retry-After header and defer to its value rather than the backoff schedule:
import requests
def handle_rate_limit(response, reconnect_controller):
"""Handles rate limit response. Reads Retry-After, sleeps, and resets retry counter."""
retry_after = int(response.headers.get("Retry-After", 5))
print(f"[RateLimit] Received 3001 — respecting Retry-After: {retry_after}s")
time.sleep(retry_after)
reconnect_controller.attempt = 0 # Reset: the server reset the window
return retry_after
5. Step 3 — Tracking the Last Known Timestamp
Before initiating recovery, you need a reliable record of the last data point you successfully processed. This timestamp becomes the start parameter for the REST recovery request.
from dataclasses import dataclass, field
from typing import Optional
import threading
import time
@dataclass
class StreamState:
"""Thread-safe record of the last known data state in the stream.
Persisted to disk on every update to survive process restarts.
"""
last_trade_timestamp: Optional[int] = None # Milliseconds since epoch
last_depth_snapshot_timestamp: Optional[int] = None
last_kline_close_timestamp: Optional[int] = None
last_sequence_id: Optional[int] = None
_lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
def checkpoint(self, timestamp: int, sequence_id: Optional[int] = None):
with self._lock:
self.last_trade_timestamp = timestamp
if sequence_id is not None:
self.last_sequence_id = sequence_id
def checkpoint_depth(self, timestamp: int):
with self._lock:
self.last_depth_snapshot_timestamp = timestamp
def checkpoint_kline(self, timestamp: int):
with self._lock:
self.last_kline_close_timestamp = timestamp
def get_recovery_start(self) -> int:
"""Returns the Unix millisecond timestamp from which to begin recovery.
Returns the most recent known timestamp, or 5 minutes ago as a fallback.
"""
with self._lock:
ts = self.last_trade_timestamp or self.last_depth_snapshot_timestamp
if ts is None:
# Fallback: 5 minutes before now
return int((time.time() - 300) * 1000)
return ts
def serialize(self) -> dict:
with self._lock:
return {
"last_trade_timestamp": self.last_trade_timestamp,
"last_depth_snapshot_timestamp": self.last_depth_snapshot_timestamp,
"last_kline_close_timestamp": self.last_kline_close_timestamp,
"last_sequence_id": self.last_sequence_id,
}
Engineering note: Store the state checkpoint to disk after every processed message. A process crash between the last WebSocket message and the next checkpoint creates a gap in your known timeline even before the disconnect. On restart, load the persisted state before establishing any connection.
6. Step 4 — Incremental REST Fetching
With a known start timestamp, the REST API recovery request is straightforward. Use the /v1/market/trades or /v1/market/kline endpoint depending on your data needs. For order book recovery, use /v1/market/depth.
import os
import requests
import time
from typing import List, Dict, Optional
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
TICKDB_BASE_URL = "https://api.tickdb.ai/v1"
def fetch_trades_recovery(
symbol: str,
start_ms: int,
end_ms: int,
limit: int = 1000
) -> List[Dict]:
"""Fetches historical trades for the recovery window.
Uses the /v1/market/trades endpoint with start/end timestamp filtering.
The API returns data in ascending order — callers must handle pagination
if the window spans more than `limit` records.
⚠️ Note: The trades endpoint covers HK equities and crypto, but not US equities.
For US equity recovery, use the kline endpoint as a proxy for price movement.
"""
headers = {"X-API-Key": TICKDB_API_KEY}
all_trades = []
current_start = start_ms
while True:
params = {
"symbol": symbol,
"start": current_start,
"end": end_ms,
"limit": limit,
}
response = requests.get(
f"{TICKDB_BASE_URL}/market/trades",
headers=headers,
params=params,
timeout=(3.05, 10)
)
# Handle rate limiting
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
print(f"[Recovery] Rate limited — sleeping {retry_after}s")
time.sleep(retry_after)
continue
data = response.json()
if data.get("code") == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
print(f"[Recovery] Server rate limit (3001) — sleeping {retry_after}s")
time.sleep(retry_after)
continue
if data.get("code") != 0:
raise RuntimeError(f"Recovery fetch failed: {data}")
trades = data.get("data", {}).get("trades", [])
if not trades:
break
all_trades.extend(trades)
# Advance window: set start to last trade timestamp + 1ms to avoid duplicates
current_start = trades[-1]["ts"] + 1
# Stop if we've retrieved data up to the known-good timestamp
if trades[-1]["ts"] >= end_ms:
break
print(f"[Recovery] Fetched {len(trades)} trades, total: {len(all_trades)}, next start: {current_start}")
time.sleep(0.05) # Be considerate to API rate limits
return all_trades
⚠️ Production warning: This loop is safe for gaps of minutes to hours. For gaps spanning days, batch the requests by hour to avoid returning excessively large payloads and to isolate failure points. A single large request that times out after 10 seconds loses your entire window. Batching limits blast radius.
7. Step 5 — Local Buffer Merge
Raw recovered data cannot simply replace the live stream — it must be merged into the existing buffer while preserving order and avoiding duplicates. The merge strategy depends on whether your buffer is timestamp-keyed or sequence-number-keyed.
7.1 Timestamp-Keyed Merge
For trade streams, each trade has a unique millisecond timestamp (with microsecond resolution where applicable). Merge by inserting into a sorted container:
import heapq
from typing import List, Dict, Iterator
class MergedTradeBuffer:
"""Maintains a sorted, deduplicated buffer of trade data.
Uses a min-heap internally to support O(log n) insertion while
maintaining ascending timestamp order. Automatically deduplicates
on insert by checking the last-emitted timestamp.
"""
def __init__(self):
self._heap: List[Dict] = []
self._last_emitted_ts: Optional[int] = None
def add_batch(self, trades: List[Dict]):
"""Adds a batch of trades to the buffer. Each trade must have a 'ts' field."""
for trade in trades:
heapq.heappush(self._heap, trade)
def stream_merged(self) -> Iterator[Dict]:
"""Yields trades in ascending timestamp order, skipping duplicates."""
while self._heap:
trade = heapq.heappop(self._heap)
ts = trade.get("ts")
if self._last_emitted_ts is not None and ts <= self._last_emitted_ts:
# Duplicate or out-of-order: skip
continue
self._last_emitted_ts = ts
yield trade
def __len__(self):
return len(self._heap)
7.2 Deduplication at the Merge Boundary
The trickiest scenario in recovery: the recovery window overlaps with data already received on reconnect. When your WebSocket reconnects, it immediately starts delivering new data. Simultaneously, your REST fetch returns data up to (or past) the reconnect timestamp.
The merge logic must handle three cases:
Case A: Recovery data overlaps reconnect stream (duplicate)
└── Deduplicate by timestamp: skip if ts <= last_emitted_ts
Case B: Recovery data ends before reconnect stream begins (clean gap)
└── Recovery fills the gap completely; stream takes over seamlessly
Case C: Recovery data extends past reconnect stream (race condition)
└── Pause stream consumption, drain recovery buffer first,
then resume stream from the point where recovery ended
def merge_recovery_stream(
recovered_trades: List[Dict],
live_stream: Iterator[Dict],
state: StreamState
) -> Iterator[Dict]:
"""Merges recovered historical data with a live WebSocket stream.
Prioritizes recovered data first. Once the recovery buffer is exhausted,
switches to the live stream. Handles timestamp collisions by discarding
any live message whose timestamp is <= the last recovered message.
"""
buffer = MergedTradeBuffer()
buffer.add_batch(recovered_trades)
recovered_exhausted = False
last_recovered_ts = None
if recovered_trades:
last_recovered_ts = max(t["ts"] for t in recovered_trades)
state.checkpoint(last_recovered_ts)
# Yield all recovered data first
for trade in buffer.stream_merged():
yield trade
print(f"[Merge] Recovery complete. Last recovered ts: {last_recovered_ts}. Switching to live stream.")
# Now yield from the live stream, filtering out pre-reconnect duplicates
for trade in live_stream:
if last_recovered_ts is not None and trade.get("ts", 0) <= last_recovered_ts:
# This live tick arrived before the gap we already filled — skip
continue
state.checkpoint(trade["ts"])
yield trade
8. Complete Integration: The Recovery Pipeline
Putting the pieces together, here is the complete reconnection and recovery pipeline wired into a MarketDataClient:
import threading
import queue
import json
from typing import Callable, Optional
class MarketDataClient:
"""Production-grade market data client with automatic reconnection and recovery.
Manages WebSocket lifecycle, heartbeat monitoring, state persistence,
and REST-based gap recovery. Thread-safe.
⚠️ This implementation uses the `websocket` library. For production HFT
workloads, migrate to asyncio-based aiohttp or asyncio-websocket to avoid
blocking the GIL during I/O wait.
"""
def __init__(
self,
symbol: str,
api_key: str,
on_trade: Optional[Callable[[dict], None]] = None,
on_depth: Optional[Callable[[dict], None]] = None,
):
self.symbol = symbol
self.api_key = api_key
self.on_trade = on_trade
self.on_depth = on_depth
self.state = StreamState()
self.ws = None
self.heartbeat = HeartbeatMonitor(timeout_seconds=15.0, on_timeout=self._on_disconnect)
self.reconnect_ctrl = ReconnectController()
self.conn_state = ConnectionState.DISCONNECTED
self._recovery_queue = queue.Queue()
self._shutdown = threading.Event()
def connect(self):
"""Establishes WebSocket connection and starts the processing pipeline."""
import websocket
def on_message(ws, message):
msg = json.loads(message)
# Heartbeat response — record it
if msg.get("type") == "pong":
self.heartbeat.record_pong()
return
if msg.get("type") == "trade":
self.state.checkpoint(msg["data"]["ts"], msg["data"].get("seq"))
if self.on_trade:
self.on_trade(msg["data"])
elif msg.get("type") == "depth":
self.state.checkpoint_depth(msg["data"]["ts"])
if self.on_depth:
self.on_depth(msg["data"])
def on_error(ws, error):
print(f"[WebSocket] Error: {error}")
def on_close(ws, code, reason):
print(f"[WebSocket] Closed: code={code}, reason={reason}")
self._on_disconnect()
def on_open(ws):
print("[WebSocket] Connected")
self.conn_state = ConnectionState.CONNECTED
self.heartbeat.start()
# Subscribe to trade and depth channels
ws.send(json.dumps({"cmd": "subscribe", "channel": "trades", "symbol": self.symbol}))
ws.send(json.dumps({"cmd": "subscribe", "channel": "depth", "symbol": self.symbol}))
# Start ping thread
self._start_ping_loop()
self.ws = websocket.WebSocketApp(
f"wss://stream.tickdb.ai/ws?api_key={self.api_key}",
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open,
)
thread = threading.Thread(target=self.ws.run_forever, daemon=True)
thread.start()
def _start_ping_loop(self):
def ping_loop():
while not self._shutdown.is_set() and self.conn_state == ConnectionState.CONNECTED:
if self.ws:
try:
self.ws.send(json.dumps({"cmd": "ping"}))
except Exception:
break
time.sleep(10)
t = threading.Thread(target=ping_loop, daemon=True)
t.start()
def _on_disconnect(self):
self.heartbeat.stop()
self.conn_state = ConnectionState.DISCONNECTED
print("[Client] Disconnect detected — initiating recovery pipeline")
# Step 1: Attempt reconnect with backoff
def reconnect_fn():
self.connect()
if not self.reconnect_ctrl.attempt_reconnect(reconnect_fn):
print("[Client] Reconnection failed after max retries — will retry on next check")
return
# Step 2: Recovery phase
self.conn_state = ConnectionState.RECOVERING
recovery_start = self.state.get_recovery_start()
recovery_end = int(time.time() * 1000) - 1000 # 1 second ago to avoid edge races
print(f"[Client] Recovery: fetching from {recovery_start} to {recovery_end}")
try:
recovered_trades = fetch_trades_recovery(
symbol=self.symbol,
start_ms=recovery_start,
end_ms=recovery_end,
)
except Exception as e:
print(f"[Client] Recovery fetch failed: {e} — continuing with live stream only")
recovered_trades = []
if recovered_trades:
print(f"[Client] Recovered {len(recovered_trades)} trades — merging into buffer")
# Step 3: Signal recovery complete; WebSocket will now stream live data
self.conn_state = ConnectionState.CONNECTED
self.heartbeat.start()
def close(self):
self._shutdown.set()
if self.ws:
self.ws.close()
9. Verifying Recovery Completeness
A recovery that reports success but leaves gaps is worse than no recovery — it is silent data loss. Build a verification step into your pipeline:
def verify_recovery_completeness(
recovered_trades: List[Dict],
gap_start_ms: int,
gap_end_ms: int,
symbol: str
) -> bool:
"""Verifies that the recovered data covers the full gap window.
Checks: (1) first trade timestamp >= gap_start, (2) last trade timestamp
>= gap_end - tolerance, (3) no long sub-windows without any trades.
Returns True if the gap appears fully covered. Returns False and logs
the specific gap sub-ranges if data is missing.
"""
if not recovered_trades:
print(f"[Verify] No trades recovered for {symbol} — gap may be a no-trade period")
return True # Empty recovery is valid if the market was halted
first_ts = min(t["ts"] for t in recovered_trades)
last_ts = max(t["ts"] for t in recovered_trades)
tolerance = 5000 # 5 second tolerance on edge cases
checks = [
(first_ts <= gap_start_ms + tolerance,
f"First recovered trade ({first_ts}) is after gap start ({gap_start_ms})"),
(last_ts >= gap_end_ms - tolerance,
f"Last recovered trade ({last_ts}) is before gap end ({gap_end_ms})"),
]
all_passed = True
for passed, message in checks:
if not passed:
print(f"[Verify] FAILED: {message}")
all_passed = False
if all_passed:
print(f"[Verify] PASSED: Gap {gap_start_ms}–{gap_end_ms} fully covered "
f"by {len(recovered_trades)} recovered trades")
return all_passed
10. Testing the Recovery Pipeline
Unit tests alone cannot validate a recovery pipeline — the scenarios are too stateful. Use integration tests with a controlled disconnection:
import unittest
from unittest.mock import Mock, patch, MagicMock
import time
class TestReconnectionRecovery(unittest.TestCase):
"""Integration tests for the reconnection and recovery pipeline."""
@patch("websocket.WebSocketApp")
def test_state_checkpoint_persists_across_disconnect(self, mock_ws):
"""Verifies that the last timestamp is captured before a simulated disconnect."""
client = MarketDataClient(symbol="BTC.USDT", api_key="test-key")
client.connect()
# Simulate receiving a trade
trade_msg = {
"type": "trade",
"data": {"ts": 1710000000000, "price": 67000.5, "vol": 0.5, "side": "buy"}
}
# Manually trigger the on_message path
client.state.checkpoint(trade_msg["data"]["ts"])
self.assertEqual(client.state.last_trade_timestamp, 1710000000000)
def test_recovery_fetch_respects_window_boundaries(self):
"""Verifies that fetch_trades_recovery stops at the correct end timestamp."""
start = 1710000000000
end = 1710000060000 # 60 seconds later
with patch("requests.get") as mock_get:
mock_get.return_value.json.return_value = {
"code": 0,
"data": {"trades": [
{"ts": start + 1000, "price": 67001.0, "vol": 0.1, "side": "sell"},
{"ts": start + 2000, "price": 67002.0, "vol": 0.2, "side": "buy"},
]}
}
result = fetch_trades_recovery("BTC.USDT", start, end)
self.assertEqual(len(result), 2)
self.assertEqual(result[0]["ts"], start + 1000)
def test_merge_deduplicates_overlapping_data(self):
"""Verifies that overlapping timestamps from recovery and live stream are deduplicated."""
recovered = [
{"ts": 1710000000000, "price": 67000, "vol": 1.0},
{"ts": 1710000001000, "price": 67001, "vol": 0.5},
]
live_stream = iter([
{"ts": 1710000001000, "price": 67001, "vol": 0.5}, # Duplicate
{"ts": 1710000002000, "price": 67002, "vol": 0.3}, # New
])
merged = list(merge_recovery_stream(recovered, live_stream, StreamState()))
self.assertEqual(len(merged), 3)
timestamps = [t["ts"] for t in merged]
self.assertEqual(timestamps, [1710000000000, 1710000001000, 1710000002000])
def test_exponential_backoff_resets_on_success(self):
"""Verifies that retry counter resets after a successful reconnect."""
ctrl = ReconnectController(base_delay=1.0, max_retries=5)
ctrl.attempt = 3 # Simulate 3 failed attempts
success = ctrl.attempt_reconnect(lambda: None)
self.assertTrue(success)
self.assertEqual(ctrl.attempt, 0) # Counter reset
⚠️ Integration testing caveat: These tests mock the WebSocket at the library boundary. To fully validate recovery, set up a test harness that injects a deliberate network partition (e.g., via firewall rules or a mock server that closes the connection) and verifies the recovered data against a ground-truth dataset.
11. Summary: The Five-Point Checklist
Before deploying any reconnection pipeline to production, verify each of the following:
| # | Checkpoint | Why it matters |
|---|---|---|
| 1 | Heartbeat timeout < reconnect interval | If your reconnect fires before the heartbeat times out, you will have two competing reconnection attempts |
| 2 | State persisted before crash | If the last checkpoint was written 30 seconds before a crash, you replay 30 seconds of redundant data on recovery |
| 3 | Recovery window bounded with end_ms | An unbounded start-ms recovery will fetch all historical data since the last tick — potentially millions of records |
| 4 | Deduplication at merge boundary | Missing deduplication causes double-counting in P&L calculations and corrupts order book state |
| 5 | Rate limit respect on recovery path | Recovery fetches are bulk requests. A naive implementation can trigger 3001 and block the entire recovery |
Next Steps
If you're building a real-time data pipeline and need a reliable WebSocket stream backed by a REST API that supports timestamp-range queries, sign up at tickdb.ai to get a free API key and start testing the reconnection pattern described in this article.
If you need to reconstruct order book state from depth snapshots, the depth channel supports L1 (US equities) and up to L10 (HK equities, crypto) snapshots. Use the same timestamp-alignment strategy described here — checkpoint the last depth snapshot timestamp before disconnect, then fetch depth snapshots for the recovery window on reconnect.
If you're using AI coding assistants for your data infrastructure, search for and install the tickdb-market-data SKILL in your AI tool's marketplace to get API reference, code generation, and error-handling templates built directly into your workflow.
This article does not constitute investment advice. Market data systems involve engineering complexity; verify all reconnection logic against your specific latency and reliability requirements before production deployment.