Every production market data system experiences it eventually: a brief network hiccup, a momentary cloud infrastructure blip, or a server maintenance window. The WebSocket connection drops. For 5 minutes, your system is blind. When connectivity returns, the question becomes urgent — not just "how do I reconnect?" but "how do I recover the data I missed without creating gaps, duplicates, or corrupted state?"
This article provides a complete, production-grade solution. We walk through the reconnection architecture, the REST API recovery protocol, timestamp alignment mechanics, and the local merge strategy that ensures your data store remains consistent after every disconnection event.
Why WebSocket Connections Fail in Production
Understanding failure modes is prerequisite to building resilient systems. WebSocket disconnections in financial data pipelines fall into three categories:
Network-level failures include TCP connection resets, NAT timeout expirations, and routing path changes. Cloud providers like AWS and GCP terminate idle WebSocket connections after 60–380 seconds depending on load balancer configuration. A market data subscriber that publishes no messages for 4 minutes may find its connection silently killed.
Infrastructure failures include server restarts, container orchestrator pod evictions, and application crashes. Kubernetes default pod eviction triggers on memory pressure or node resource exhaustion — events that can occur during high-volatility market sessions when your subscriber is consuming maximum memory.
Application-level failures include unhandled exceptions, garbage collection pauses that exceed connection heartbeat intervals, and process restarts triggered by deployment pipelines. Python's GIL-related pauses can occasionally exceed expected latency windows, causing the heartbeat mechanism to fail.
The consequences of ignoring these failure modes are severe. A 5-minute gap during a Federal Reserve announcement or an earnings release represents a permanent hole in your data history — one that cannot be reconstructed from any vendor's real-time stream, only from their historical REST API, and only if you know exactly what to request.
The Reconnection Strategy: Exponential Backoff with Jitter
Before discussing data recovery, we must address the immediate problem: reestablishing the connection without hammering the server or getting trapped in a retry loop.
A naive reconnection strategy — retrying immediately, or retrying on a fixed interval — creates thundering herd problems at scale and violates most real-time API rate limit policies. The correct approach is exponential backoff with jitter.
import random
import time
import asyncio
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class WebSocketReconnectManager:
"""
Manages WebSocket reconnection with exponential backoff and jitter.
Thread-safe for production deployment.
"""
def __init__(
self,
base_delay: float = 1.0,
max_delay: float = 60.0,
max_retries: int = 0, # 0 = unlimited retries
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.retry_count = 0
self.is_connected = False
def compute_delay(self) -> float:
"""
Compute next delay using exponential backoff with full jitter.
Formula: random(0, min(max_delay, base_delay * 2^retry))
"""
exponential_delay = min(
self.max_delay,
self.base_delay * (2 ** self.retry_count)
)
jitter = random.uniform(0, exponential_delay * self.jitter_factor)
return exponential_delay + jitter
async def wait_before_retry(self):
"""Async-safe delay with backoff and jitter."""
delay = self.compute_delay()
logger.info(
f"Reconnecting in {delay:.2f}s "
f"(retry #{self.retry_count})"
)
await asyncio.sleep(delay)
self.retry_count += 1
def reset(self):
"""Call after successful connection to reset retry state."""
self.retry_count = 0
self.is_connected = True
logger.info("Connection established — retry counter reset")
# ⚠️ Production note: This class handles reconnection logic only.
# Actual WebSocket connection establishment, message parsing, and
# state management must be implemented in the calling context.
The jitter is critical. Without it, 1,000 client instances reconnecting simultaneously after an outage will all compute the same delay and flood the server at t+1 second, t+2 seconds, and so on. Randomizing the jitter value spreads reconnection attempts across time, reducing server load and improving your own success rate.
Capturing the Disconnection Window
Effective data recovery requires knowing precisely when the connection was lost and restored. Store these timestamps durably — not in memory, not in local variables — because your process may crash during the disconnection window, leaving you without any recovery context.
import json
import os
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Optional
@dataclass
class ConnectionState:
"""Durable record of connection state for recovery scenarios."""
last_received_timestamp: Optional[int] = None # Unix milliseconds
disconnected_at: Optional[float] = None # Unix seconds
reconnected_at: Optional[float] = None # Unix seconds
last_sequence_id: Optional[int] = None # If API uses sequences
def to_file(self, path: str):
"""Persist state to disk for crash recovery."""
with open(path, "w") as f:
json.dump(asdict(self), f)
@classmethod
def from_file(cls, path: str) -> "ConnectionState":
"""Restore state from disk."""
if not Path(path).exists():
return cls()
with open(path, "r") as f:
data = json.load(f)
return cls(**data)
# ⚠️ Production note: In distributed systems, use a persistent store
# (Redis, PostgreSQL, etcd) instead of a local file. Local files are
# single-instance only and may not survive container restarts on K8s.
When the WebSocket connection drops, record disconnected_at. When the reconnection succeeds, record reconnected_at. The last_received_timestamp field should be updated on every incoming message — this is your recovery anchor.
REST API Recovery: Incremental Fetching by Timestamp
TickDB's REST API supports timestamp-based range queries, making gap recovery straightforward. The recovery workflow follows this sequence:
- Query the last known good timestamp from your local state
- Request data from
last_timestamp + 1mstocurrent_time - Validate response integrity (sequence continuity, no overlaps)
- Merge recovered data into local store
- Resume WebSocket consumption from the verified boundary
Identifying the Correct API Endpoint
Different data types require different recovery endpoints:
| Data type | Recovery endpoint | Key parameter |
|---|---|---|
| OHLCV candles | /v1/market/kline |
start_time, end_time |
| Order book snapshots | /v1/market/depth |
symbol, limit (most recent only) |
| Trade ticks | /v1/market/trades |
symbol, start_time, end_time |
For continuous candle data (klines), the /v1/market/kline endpoint supports time-range queries that align with your last received timestamp. The recovered data will be in the same OHLCV format as your WebSocket stream, simplifying the merge step.
Constructing the Recovery Request
import os
import requests
import time
from typing import Dict, List, Any, Optional
class TickDBRecoveryClient:
"""
Client for recovering missed market data via REST API.
Designed for post-disconnection gap filling.
"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError(
"TICKDB_API_KEY environment variable is required"
)
self.base_url = "https://api.tickdb.ai/v1"
def _headers(self) -> Dict[str, str]:
return {"X-API-Key": self.api_key}
def recover_klines(
self,
symbol: str,
start_time_ms: int,
end_time_ms: int,
interval: str = "1m",
) -> List[Dict[str, Any]]:
"""
Recover OHLCV candles for a time range.
Args:
symbol: Market symbol, e.g., "BTC-USD"
start_time_ms: Start of recovery window (Unix ms, inclusive)
end_time_ms: End of recovery window (Unix ms, inclusive)
interval: Candle interval ("1m", "5m", "1h", "1d")
Returns:
List of OHLCV candle dictionaries
"""
params = {
"symbol": symbol,
"interval": interval,
"start_time": start_time_ms,
"end_time": end_time_ms,
"limit": 1000, # Max per request; paginate if needed
}
response = requests.get(
f"{self.base_url}/market/kline",
headers=self._headers(),
params=params,
timeout=(3.05, 10), # (connect_timeout, read_timeout)
)
# ⚠️ Production note: Implement retry logic here for HTTP 5xx
# errors. HTTP 4xx errors (except 429) indicate bad requests
# that should not be retried.
response.raise_for_status()
data = response.json()
if data.get("code") != 0:
raise RuntimeError(
f"API error {data.get('code')}: {data.get('message')}"
)
return data.get("data", [])
def estimate_recovery_size(
self,
start_time_ms: int,
end_time_ms: int,
interval: str = "1m",
) -> int:
"""
Estimate number of candles in a time range.
Useful for logging and monitoring recovery progress.
"""
interval_ms = {
"1m": 60_000,
"5m": 300_000,
"1h": 3_600_000,
"1d": 86_400_000,
}.get(interval, 60_000)
return max(1, (end_time_ms - start_time_ms) // interval_ms)
# ⚠️ For production HFT workloads, use aiohttp/asyncio for concurrent
# recovery requests across multiple symbols. Sequential REST recovery
# adds latency that may matter at millisecond timescales.
Handling Rate Limits During Recovery
High-volume recovery requests can trigger rate limits. The TickDB API returns HTTP 429 with a Retry-After header when limits are exceeded. Your recovery client must respect this response.
import requests
from requests.exceptions import HTTPError
def fetch_with_rate_limit_handling(client, url, headers, params, max_retries=3):
"""
Fetch with retry logic for rate limit (429) responses.
"""
for attempt in range(max_retries):
response = requests.get(url, headers=headers, params=params)
if response.status_code == 429:
# Rate limit exceeded — respect Retry-After header
retry_after = int(response.headers.get("Retry-After", 5))
print(f"Rate limited. Waiting {retry_after}s before retry...")
time.sleep(retry_after)
continue
response.raise_for_status()
return response
raise RuntimeError(
f"Failed after {max_retries} rate-limit retries"
)
Timestamp Alignment: The Critical Detail
Timestamp alignment is where most recovery implementations fail. The core principle: timestamps must be in the same timezone system before comparison and merging.
TickDB uses Unix milliseconds for all time representations. If your local system stores timestamps in a different format — POSIX seconds, Python datetime objects, or database-native timestamps — you must normalize before comparison.
from datetime import datetime, timezone
from typing import Union
def normalize_to_unix_ms(timestamp: Union[int, float, datetime, str]) -> int:
"""
Convert any timestamp representation to Unix milliseconds.
All TickDB timestamps are in this format.
"""
if isinstance(timestamp, datetime):
# Aware datetime: convert to Unix ms directly
if timestamp.tzinfo is None:
timestamp = timestamp.replace(tzinfo=timezone.utc)
return int(timestamp.timestamp() * 1000)
if isinstance(timestamp, str):
# ISO 8601 string: parse then convert
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
return int(dt.timestamp() * 1000)
if isinstance(timestamp, (int, float)):
# Assume Unix seconds if > 1e10 (i.e., appears to be ms already)
if timestamp > 1e10:
return int(timestamp)
return int(timestamp * 1000)
raise ValueError(f"Unsupported timestamp type: {type(timestamp)}")
def unix_ms_to_datetime(ms: int) -> datetime:
"""Convert Unix milliseconds to UTC datetime for logging."""
return datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
Gap Detection: Before You Recover
Before initiating recovery, check whether a gap actually exists. Requesting data for periods you already have wastes API quota and introduces unnecessary processing overhead.
from typing import List, Tuple, Optional
def detect_data_gaps(
existing_timestamps: List[int],
last_timestamp_ms: int,
current_time_ms: int,
expected_interval_ms: int = 60_000,
allowed_missing_ratio: float = 0.1,
) -> List[Tuple[int, int]]:
"""
Detect gaps between last known data and current time.
Args:
existing_timestamps: Sorted list of Unix ms timestamps you already have
last_timestamp_ms: The most recent timestamp in your local store
current_time_ms: Current server/received time
expected_interval_ms: Expected interval between candles (1m = 60000ms)
allowed_missing_ratio: Allow up to 10% missing before flagging as gap
Returns:
List of (start_ms, end_ms) tuples representing gap windows
"""
gaps = []
gap_start = last_timestamp_ms + expected_interval_ms
expected_count = (current_time_ms - gap_start) // expected_interval_ms
if expected_count <= 0:
return gaps
# Check if we have any data in the gap window
data_in_window = [
ts for ts in existing_timestamps
if gap_start <= ts <= current_time_ms
]
missing_count = expected_count - len(data_in_window)
if missing_count / max(expected_count, 1) > allowed_missing_ratio:
gaps.append((gap_start, current_time_ms))
return gaps
# Example usage:
# last_timestamp = 1712500000000 # 10:26:40 UTC
# current_time = 1712500300000 # 10:31:40 UTC
# gap window: 10:27:00 to 10:31:40 UTC (5 minutes of potential gap)
The Local Merge Strategy
Recovered data must be merged into your local store without creating duplicates or overwriting legitimate data points. The merge strategy depends on your storage backend:
For in-memory stores (e.g., pandas DataFrame, Python dict): deduplicate by timestamp before insertion, then re-sort by timestamp.
For SQLite/PostgreSQL: use INSERT ... ON CONFLICT DO UPDATE (upsert) or INSERT IGNORE with a unique constraint on (symbol, interval, timestamp).
For time-series databases (InfluxDB, TimescaleDB): use the native upsert semantics with retention policies.
import pandas as pd
from typing import Dict, Any
def merge_klines(
existing_df: pd.DataFrame,
recovered_data: List[Dict[str, Any]],
timestamp_col: str = "open_time",
key_cols: List[str] = None,
) -> pd.DataFrame:
"""
Merge recovered candles into existing DataFrame without duplicates.
Args:
existing_df: DataFrame with existing kline data
recovered_data: List of kline dicts from REST recovery
timestamp_col: Name of the timestamp column
key_cols: Columns that define uniqueness (default: timestamp only)
Returns:
Merged DataFrame with duplicates removed
"""
if not recovered_data:
return existing_df
# Normalize to DataFrame
recovered_df = pd.DataFrame(recovered_data)
# Ensure timestamp column is int (Unix ms)
recovered_df[timestamp_col] = recovered_df[timestamp_col].astype(int)
# If existing data exists, concat and deduplicate
if existing_df is not None and len(existing_df) > 0:
combined = pd.concat([existing_df, recovered_df], ignore_index=True)
else:
combined = recovered_df
# Drop duplicates, keeping the most recent value if any conflict
key_cols = key_cols or [timestamp_col]
combined = combined.drop_duplicates(
subset=key_cols,
keep="last", # Recovered data is authoritative
)
# Sort by timestamp to maintain time-series order
combined = combined.sort_values(timestamp_col).reset_index(drop=True)
return combined
# ⚠️ Production note: For high-frequency data (tick-level), pandas may
# introduce unacceptable latency. Consider using numpy structured arrays
# or a custom binary format for tick data.
Complete Recovery Workflow
Putting it all together: the full recovery sequence that should execute after any reconnection event.
import asyncio
import logging
import os
from datetime import datetime, timezone
from typing import List, Dict, Any, Optional
from tickdb_recovery import TickDBRecoveryClient
from websocket_manager import WebSocketReconnectManager, ConnectionState
from merge_utils import merge_klines, normalize_to_unix_ms, detect_data_gaps
logger = logging.getLogger(__name__)
class MarketDataRecoveryManager:
"""
Orchestrates WebSocket reconnection and REST API data recovery.
Maintains durable state across process restarts.
"""
def __init__(
self,
symbol: str,
state_file: str = ".tickdb_connection_state.json",
interval: str = "1m",
):
self.symbol = symbol
self.state_file = state_file
self.interval = interval
self.state = ConnectionState.from_file(state_file)
self.ws_manager = WebSocketReconnectManager()
self.rest_client = TickDBRecoveryClient()
self.local_df = None # Your data store reference
async def handle_disconnection(self):
"""Called when WebSocket connection drops."""
self.state.disconnected_at = time.time()
self.ws_manager.is_connected = False
self.state.to_file(self.state_file)
logger.warning(
f"Connection lost at {datetime.now(timezone.utc).isoformat()}"
)
async def handle_reconnection(self):
"""Called when WebSocket connection is restored."""
self.state.reconnected_at = time.time()
self.ws_manager.reset()
self.state.to_file(self.state_file)
await self._recover_missed_data()
logger.info(
f"Recovery complete. "
f"Resuming from timestamp {self.state.last_received_timestamp}"
)
async def _recover_missed_data(self):
"""
Core recovery logic: fetch missed data via REST, merge locally.
"""
if self.state.last_received_timestamp is None:
logger.warning("No last timestamp — cannot determine recovery window")
return
start_ms = self.state.last_received_timestamp + 1
end_ms = int(time.time() * 1000) # Current time in ms
# Detect if gap actually exists before fetching
gaps = detect_data_gaps(
existing_timestamps=self._get_existing_timestamps(),
last_timestamp_ms=self.state.last_received_timestamp,
current_time_ms=end_ms,
)
if not gaps:
logger.info("No data gap detected — skipping recovery")
return
gap_start, gap_end = gaps[0]
estimated = self.rest_client.estimate_recovery_size(
gap_start, gap_end, self.interval
)
logger.info(
f"Recovering ~{estimated} candles from "
f"{datetime.fromtimestamp(gap_start/1000, tz=timezone.utc).isoformat()} "
f"to {datetime.fromtimestamp(gap_end/1000, tz=timezone.utc).isoformat()}"
)
# Fetch from REST API
recovered = self.rest_client.recover_klines(
symbol=self.symbol,
start_time_ms=gap_start,
end_time_ms=gap_end,
interval=self.interval,
)
if not recovered:
logger.warning("Recovery returned empty data — verify API quota")
return
# Merge into local store
self.local_df = merge_klines(self.local_df, recovered)
# Update last received timestamp to new boundary
new_last = max(candle["open_time"] for candle in recovered)
self.state.last_received_timestamp = new_last
self.state.to_file(self.state_file)
logger.info(f"Successfully merged {len(recovered)} recovered candles")
def _get_existing_timestamps(self) -> List[int]:
"""Extract timestamps from local data store."""
if self.local_df is None or len(self.local_df) == 0:
return []
return self.local_df["open_time"].tolist()
def update_last_timestamp(self, timestamp_ms: int):
"""Called on every WebSocket message to maintain continuity."""
self.state.last_received_timestamp = timestamp_ms
# Persist every N messages to balance I/O and crash recovery accuracy
# For production: batch writes or use a write-ahead log
self.state.to_file(self.state_file)
# ⚠️ Production HFT advisory: This architecture uses synchronous file I/O
# for state persistence. For sub-millisecond requirements, replace with
# async I/O (aiofiles) or a distributed state store (Redis, etcd).
Limitations and Tradeoffs
This recovery strategy works reliably for OHLCV candle data, where the API returns deterministic, immutable records. For tick-level trade data, be aware:
- High-frequency gaps: If your disconnection window exceeds the rate limit quota available for trade recovery, you may not retrieve all missed ticks in a single recovery cycle.
- Order book state: The REST API provides snapshots, not the full order book delta stream. Recovering the exact order book state at a past moment requires replaying the delta stream from your own log — not the vendor's API. If you did not log deltas locally, you can only recover the snapshot, not the full depth.
- Latency during recovery: REST recovery adds synchronous overhead. During the recovery window, your system is running on stale data. Design your trading logic to tolerate this lag or pause strategy execution during recovery.
Summary: The Recovery Checklist
Every production market data subscriber should implement these five components:
- Durable state persistence — store
last_received_timestampto disk (or a distributed store) on every message or batch of messages - Exponential backoff with jitter — prevent thundering herd on reconnection; respect server-side rate limits
- Timestamp normalization — convert all timestamps to Unix milliseconds before comparison
- Gap detection before recovery — avoid unnecessary API calls by checking whether a gap actually exists
- Deduplicating merge — upsert recovered data using timestamp as the uniqueness key; never append blindly
With these components in place, a 5-minute disconnection becomes a recoverable incident rather than a permanent data loss event.
Next Steps
If you want to implement this recovery flow today:
- Sign up at tickdb.ai and generate an API key (free tier available)
- Set the
TICKDB_API_KEYenvironment variable - Clone the reference implementation from the TickDB GitHub repository
- Replace the placeholder data store with your existing pandas DataFrame or time-series database
If you need institutional-grade data reliability (multi-region failover, guaranteed delivery contracts, dedicated support), reach out to enterprise@tickdb.ai for Professional and Enterprise plans that include SLA-backed uptime guarantees.
If you are building AI-assisted trading systems, search for and install the tickdb-market-data SKILL in your AI coding assistant's marketplace. The SKILL includes pre-built recovery templates compatible with this article's architecture.
This article does not constitute investment advice. Market data systems involve engineering complexity; test thoroughly in paper trading environments before production deployment.