A WebSocket connection can look perfectly healthy on the surface—green status light, active TCP session, no error messages—while delivering nothing. The server has already closed its end. The client hasn't noticed. Twelve hours of "live" data have silently become twelve hours of nothing. By the time you check the dashboard, your strategy has been trading blind.
This failure mode—sometimes called silent death—is more dangerous than an obvious connection error. It doesn't trigger exceptions. It doesn't log failures. It simply stops working while appearing to work. For quant traders running event-driven strategies, a missed bar during a volatile session is an unrecoverable gap in your data record.
This article dissects the three-pronged detection strategy used by production-grade WebSocket clients: heartbeat timeout detection, data timestamp validation, and active reconnection with exponential backoff. All code is Python-based and production-ready, including full authentication handling, error recovery, and rate-limit compliance.
Why WebSocket Connections Die Silently
WebSocket is a persistent, bidirectional protocol built on top of TCP. Unlike HTTP request-response cycles, WebSocket connections are designed to stay open indefinitely. This design assumption creates a vulnerability: the protocol has no inherent mechanism to detect whether the remote endpoint is still transmitting data.
Several failure modes produce silent disconnections:
Server-side graceful shutdown: The server sends a WebSocket close frame (0x8XXX), but network latency or intermediate proxies drop it. The client never receives the frame and assumes the connection is still alive.
NAT table expiry: Network Address Translation tables on routers and cloud load balancers expire after periods of inactivity (often 60–300 seconds). The TCP connection remains open at the application layer, but packets stop flowing.
Idle timeout at the application layer: Many financial data providers implement server-side idle timeouts—typically 30–120 seconds of no outbound data. If no market data flows during that window (e.g., pre-market hours), the server closes the connection without warning.
Proxy and load balancer drops: Corporate proxies, cloud load balancers (AWS ALB, GCP load balancers), and CDN layers have their own keepalive policies. They may terminate "idle" connections that haven't seen application-layer traffic within their timeout window.
The common thread: none of these failures generate a client-side exception or error event. The socket remains open. The select/poll/epoll call returns readable. But the receive buffer contains nothing and never will again.
The Detection Trifecta
Production-grade clients detect silent death through three independent mechanisms, each covering failure modes the others miss:
| Mechanism | Detects | Blind spot |
|---|---|---|
| Heartbeat timeout | Server went away without close frame; NAT expiry | Clock skew between client and server; heartbeat gets dropped |
| Data timestamp validation | Data stopped flowing mid-session (partial failure) | Clock skew; only works if data arrives predictably |
| Explicit reconnection trigger | All failure modes | Must be paired with one of the above to avoid thundering herd |
No single mechanism is sufficient. A robust production client layers all three.
Heartbeat Timeout Detection
The heartbeat (ping/pong) is the WebSocket-level keepalive mechanism. The client sends a ping frame; the server responds with a pong. If the pong doesn't arrive within a configurable timeout, the client declares the connection dead and initiates reconnection.
Most WebSocket libraries handle heartbeat at the transport layer, but relying on library defaults is insufficient for trading systems. You need explicit application-level heartbeat management with configurable timeouts and explicit state tracking.
Python Implementation: Heartbeat Manager
import asyncio
import time
import random
import logging
from dataclasses import dataclass, field
from typing import Optional, Callable
import json
logger = logging.getLogger(__name__)
@dataclass
class HeartbeatConfig:
"""Configuration for heartbeat-based connection monitoring."""
ping_interval: float = 15.0 # seconds between ping attempts
pong_timeout: float = 10.0 # seconds to wait for pong response
max_missed_heartbeats: int = 3 # consecutive missed heartbeats before reconnect
jitter_fraction: float = 0.1 # randomize intervals by ±10% to prevent resonance
@dataclass
class ConnectionState:
"""Tracks the state of a WebSocket connection."""
connected_at: Optional[float] = None
last_pong_received: Optional[float] = None
last_data_received: Optional[float] = None
consecutive_missed_heartbeats: int = 0
total_reconnections: int = 0
last_reconnect_at: Optional[float] = None
class WebSocketHeartbeatMonitor:
"""
Manages WebSocket heartbeat detection and reconnection logic.
Detects silent disconnections that survive the transport layer.
Engineering warning: This class handles WebSocket-level reconnection.
For TickDB's REST API fallback during extended outages, implement
a separate health-check loop in your data acquisition layer.
"""
def __init__(
self,
url: str,
api_key: str,
on_data: Callable[[dict], None],
config: HeartbeatConfig = None,
reconnect_max_delay: float = 60.0,
reconnect_base_delay: float = 1.0,
):
self.url = url
self.api_key = api_key
self.on_data = on_data
self.config = config or HeartbeatConfig()
self.reconnect_max_delay = reconnect_max_delay
self.reconnect_base_delay = reconnect_base_delay
self.state = ConnectionState()
self._ws = None
self._running = False
self._shutdown_requested = False
# Heartbeat tracking
self._heartbeat_task: Optional[asyncio.Task] = None
self._last_ping_sent: Optional[float] = None
# Data validation
self._data_validator_task: Optional[asyncio.Task] = None
logger.info(
f"Heartbeat monitor initialized: ping_interval={self.config.ping_interval}s, "
f"pong_timeout={self.config.pong_timeout}s"
)
async def connect(self):
"""Establish WebSocket connection with authentication."""
# WebSocket authentication via URL parameter (TickDB protocol)
auth_url = f"{self.url}?api_key={self.api_key}"
import websockets
self._ws = await websockets.connect(
auth_url,
ping_interval=None, # We manage heartbeat explicitly at the application layer
ping_timeout=None,
close_timeout=5.0,
)
self.state.connected_at = time.time()
self.state.last_pong_received = self.state.connected_at
self.state.last_data_received = self.state.connected_at
self.state.consecutive_missed_heartbeats = 0
logger.info(f"WebSocket connected at {self.state.connected_at:.3f}")
async def _send_ping(self) -> bool:
"""
Send an application-level ping and wait for response.
Returns True if pong received within timeout, False otherwise.
"""
if self._ws is None or self._ws.closed:
return False
self._last_ping_sent = time.time()
try:
# Application-level ping using JSON command (common pattern for financial APIs)
ping_frame = json.dumps({"cmd": "ping", "ts": self._last_ping_sent})
await asyncio.wait_for(
self._ws.send(ping_frame),
timeout=3.0
)
# Wait for pong response
response = await asyncio.wait_for(
self._ws.recv(),
timeout=self.config.pong_timeout
)
data = json.loads(response)
# Validate pong response (structure varies by API; adjust for your provider)
if data.get("cmd") == "pong" or data.get("type") == "pong":
self.state.last_pong_received = time.time()
self.state.consecutive_missed_heartbeats = 0
logger.debug(
f"Pong received. RTT: "
f"{self.state.last_pong_received - self._last_ping_sent:.3f}s"
)
return True
else:
# Got data instead of pong — connection is alive
self.state.last_data_received = time.time()
self._process_data(data)
return True
except asyncio.TimeoutError:
self.state.consecutive_missed_heartbeats += 1
logger.warning(
f"Pong timeout. Missed heartbeats: "
f"{self.state.consecutive_missed_heartbeats}/{self.config.max_missed_heartbeats}"
)
return False
except Exception as e:
logger.error(f"Ping failed with error: {e}")
self.state.consecutive_missed_heartbeats += 1
return False
async def _heartbeat_loop(self):
"""
Main heartbeat monitoring loop.
Runs as an independent task alongside data reception.
"""
while self._running and not self._shutdown_requested:
# Apply jitter to prevent synchronized reconnections across clients
base_interval = self.config.ping_interval
jitter = base_interval * self.config.jitter_fraction
interval = base_interval + random.uniform(-jitter, jitter)
await asyncio.sleep(interval)
if self._shutdown_requested:
break
# Send ping and check response
pong_received = await self._send_ping()
# Check if we've exceeded the missed heartbeat threshold
if self.state.consecutive_missed_heartbeats >= self.config.max_missed_heartbeats:
logger.error(
f"Heartbeat threshold exceeded. "
f"Missed {self.state.consecutive_missed_heartbeats} consecutive pongs. "
f"Initiating reconnection."
)
await self._reconnect()
def _process_data(self, data: dict):
"""Process incoming data and update validation state."""
# Update timestamp tracking for data validation
self.state.last_data_received = time.time()
# Reset consecutive missed heartbeats — any data means connection is alive
# (Some APIs don't support explicit pong; data is the heartbeat)
self.state.consecutive_missed_heartbeats = 0
try:
self.on_data(data)
except Exception as e:
logger.error(f"Data handler raised exception: {e}")
async def _data_receiver_loop(self):
"""
Dedicated task for receiving data.
Runs independently of heartbeat to avoid blocking on slow data.
"""
while self._running and not self._shutdown_requested:
if self._ws is None or self._ws.closed:
await asyncio.sleep(0.5)
continue
try:
data = await asyncio.wait_for(
self._ws.recv(),
timeout=30.0 # If no data in 30s, something is wrong
)
parsed = json.loads(data)
# Skip pong frames — they're handled by heartbeat loop
if parsed.get("cmd") in ("pong", "ping_ack"):
continue
self._process_data(parsed)
except asyncio.TimeoutError:
# No data received within timeout — could be normal market pause
# or silent disconnection. Heartbeat will catch the latter.
logger.debug("No data received for 30 seconds.")
continue
except Exception as e:
logger.error(f"Data receiver error: {e}")
await self._reconnect()
break
async def _reconnect(self):
"""
Execute reconnection with exponential backoff and jitter.
Prevents thundering herd when multiple clients reconnect simultaneously.
"""
self._running = False # Stop other loops during reconnection
# Close existing connection if still open
if self._ws and not self._ws.closed:
try:
await self._ws.close(code=1001, reason="Reconnecting")
except Exception:
pass # Connection may already be dead
retry = 0
max_retries = 10
while retry < max_retries and not self._shutdown_requested:
# Calculate backoff delay with jitter
base_delay = self.reconnect_base_delay * (2 ** retry)
capped_delay = min(base_delay, self.reconnect_max_delay)
jitter = random.uniform(0, capped_delay * 0.3)
delay = capped_delay + jitter
logger.info(
f"Reconnection attempt {retry + 1}/{max_retries} "
f"in {delay:.2f}s"
)
await asyncio.sleep(delay)
try:
await self.connect()
# Successfully reconnected — restart monitoring loops
self._running = True
self.state.total_reconnections += 1
self.state.last_reconnect_at = time.time()
self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
self._data_validator_task = asyncio.create_task(self._data_receiver_loop())
logger.info(
f"Reconnection successful. Total reconnections: "
f"{self.state.total_reconnections}"
)
return
except Exception as e:
logger.error(f"Reconnection attempt {retry + 1} failed: {e}")
retry += 1
if retry >= max_retries:
logger.critical(
f"Max reconnection attempts ({max_retries}) exhausted. "
f"Manual intervention required."
)
# In production: trigger alert, page on-call engineer
async def start(self):
"""Start the WebSocket monitoring system."""
self._running = True
self._shutdown_requested = False
await self.connect()
# Launch monitoring loops as independent tasks
self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
self._data_validator_task = asyncio.create_task(self._data_receiver_loop())
logger.info("WebSocket heartbeat monitor started.")
async def stop(self):
"""Gracefully shutdown the monitoring system."""
self._shutdown_requested = True
self._running = False
if self._heartbeat_task:
self._heartbeat_task.cancel()
if self._data_validator_task:
self._data_validator_task.cancel()
if self._ws and not self._ws.closed:
await self._ws.close(code=1000, reason="Client shutdown")
logger.info("WebSocket heartbeat monitor stopped.")
Data Timestamp Validation: Detecting Partial Failures
Heartbeat timeout catches the case where the connection has died entirely. But there's a subtler failure mode: partial disconnection, where the connection remains open but data flow has stopped for a specific channel or data stream.
This happens when:
- A specific market data feed drops (e.g., options chain data stops, but equity quotes continue)
- The server-side subscription expires and isn't renewed
- A load balancer routes traffic to a stale node
In these cases, the WebSocket ping/pong succeeds—there's a live TCP connection—but no market data arrives. Your strategy believes it has current quotes when it doesn't.
Data timestamp validation detects this by tracking the temporal freshness of incoming data against expected arrival patterns.
Timestamp Validation Implementation
import asyncio
from datetime import datetime, timezone
from typing import Dict, Optional
from collections import defaultdict
@dataclass
class DataStreamConfig:
"""Configuration for a specific data stream's expected cadence."""
stream_name: str
expected_interval: float # seconds between expected updates
max_staleness: float # seconds before declaring data stale
check_interval: float = 5.0 # how often to check staleness
@dataclass
class StreamState:
"""Tracks staleness state for a data stream."""
last_update: Optional[float] = None
is_stale: bool = False
stale_since: Optional[float] = None
consecutive_stale_checks: int = 0
class DataTimestampValidator:
"""
Validates data freshness across multiple streams.
Detects partial disconnections where the connection is alive
but data has stopped flowing for specific channels.
Engineering note: This validator assumes server timestamps
in the data payload. If your data provider doesn't include
server-side timestamps, use local receive timestamps instead,
but expect false positives during network latency spikes.
"""
def __init__(
self,
on_stale: Callable[[str, float], None] = None,
on_recovered: Callable[[str], None] = None,
):
"""
Args:
on_stale: Callback(stream_name, seconds_since_last_update) when data goes stale
on_recovered: Callback(stream_name) when stale data resumes
"""
self._streams: Dict[str, DataStreamConfig] = {}
self._stream_states: Dict[str, StreamState] = {}
self._check_task: Optional[asyncio.Task] = None
self._running = False
self.on_stale = on_stale or (lambda s, t: None)
self.on_recovered = on_recovered or (lambda s: None)
def register_stream(self, config: DataStreamConfig):
"""Register a data stream for staleness monitoring."""
self._streams[config.stream_name] = config
self._stream_states[config.stream_name] = StreamState()
logger.info(
f"Registered stream '{config.stream_name}': "
f"expected interval={config.expected_interval}s, "
f"max staleness={config.max_staleness}s"
)
def record_update(self, stream_name: str, server_timestamp: Optional[float] = None):
"""
Record a data update for a stream.
Args:
stream_name: Identifier for the data stream
server_timestamp: Unix timestamp from the server payload.
If None, uses current local time.
"""
if stream_name not in self._stream_states:
logger.warning(f"Update recorded for unregistered stream: {stream_name}")
return
# Use server timestamp if available; otherwise local time
update_time = server_timestamp if server_timestamp else time.time()
state = self._stream_states[stream_name]
# If stream was stale, notify on recovery
if state.is_stale:
logger.info(f"Stream '{stream_name}' recovered after "
f"{update_time - state.stale_since:.1f}s of staleness")
self.on_recovered(stream_name)
state.last_update = update_time
state.is_stale = False
state.stale_since = None
state.consecutive_stale_checks = 0
async def _staleness_check_loop(self):
"""Periodic check for stale data streams."""
while self._running:
await asyncio.sleep(1.0) # Check every second for responsiveness
current_time = time.time()
for stream_name, config in self._streams.items():
state = self._stream_states[stream_name]
if state.last_update is None:
# No data ever received — likely subscription issue
continue
age = current_time - state.last_update
if age > config.max_staleness and not state.is_stale:
# Data just went stale
state.is_stale = True
state.stale_since = current_time
state.consecutive_stale_checks = 1
logger.warning(
f"Stream '{stream_name}' is stale: "
f"{age:.1f}s since last update (max: {config.max_staleness}s)"
)
self.on_stale(stream_name, age)
elif state.is_stale:
state.consecutive_stale_checks += 1
# Re-notify every 30 seconds of continuous staleness
if state.consecutive_stale_checks % 30 == 0:
logger.error(
f"Stream '{stream_name}' still stale: "
f"{current_time - state.stale_since:.1f}s total"
)
self.on_stale(stream_name, current_time - state.stale_since)
async def start(self):
"""Start the staleness monitoring loop."""
self._running = True
self._check_task = asyncio.create_task(self._staleness_check_loop())
logger.info("Data timestamp validator started.")
async def stop(self):
"""Stop the staleness monitoring loop."""
self._running = False
if self._check_task:
self._check_task.cancel()
logger.info("Data timestamp validator stopped.")
def get_status(self) -> Dict:
"""Return current status of all monitored streams."""
current_time = time.time()
return {
name: {
"last_update": state.last_update,
"age_seconds": current_time - state.last_update if state.last_update else None,
"is_stale": state.is_stale,
"stale_duration": current_time - state.stale_since if state.stale_since else None,
}
for name, state in self._stream_states.items()
}
Integration: Combining Heartbeat and Timestamp Validation
The two systems serve different purposes but share state. Here's how to integrate them into a unified monitoring solution:
import os
async def main():
"""Integrated WebSocket monitoring with heartbeat and timestamp validation."""
# Configuration for different data streams
stream_configs = [
DataStreamConfig(
stream_name="quote",
expected_interval=0.1, # Quotes expected every 100ms
max_staleness=2.0, # 2 seconds without quote = stale
),
DataStreamConfig(
stream_name="depth",
expected_interval=0.5, # Depth updates every 500ms
max_staleness=5.0, # 5 seconds without depth = stale
),
DataStreamConfig(
stream_name="trade",
expected_interval=1.0, # Trades vary; 1s is aggressive threshold
max_staleness=10.0, # 10 seconds without trade (for liquid names)
),
]
# Initialize timestamp validator
timestamp_validator = DataTimestampValidator(
on_stale=lambda stream, age: logger.error(f"ALERT: {stream} stale for {age:.1f}s"),
on_recovered=lambda stream: logger.info(f"RECOVERED: {stream}"),
)
for config in stream_configs:
timestamp_validator.register_stream(config)
# Initialize heartbeat monitor
monitor = WebSocketHeartbeatMonitor(
url="wss://api.tickdb.ai/ws/v1/market", # WebSocket endpoint
api_key=os.environ.get("TICKDB_API_KEY"),
on_data=lambda data: _route_data(data, timestamp_validator),
config=HeartbeatConfig(
ping_interval=15.0,
pong_timeout=10.0,
max_missed_heartbeats=3,
),
reconnect_max_delay=60.0,
reconnect_base_delay=1.0,
)
# Start both systems
await timestamp_validator.start()
await monitor.start()
# Keep running
try:
while True:
await asyncio.sleep(30)
status = timestamp_validator.get_status()
logger.info(f"Stream status: {status}")
finally:
await monitor.stop()
await timestamp_validator.stop()
def _route_data(data: dict, validator: DataTimestampValidator):
"""Route incoming data to appropriate handler and validator."""
# Extract stream type from message structure
# (Adjust field names based on your data provider's protocol)
stream_type = data.get("channel") or data.get("type", "unknown")
server_ts = data.get("ts") or data.get("timestamp")
# Record timestamp for validation
validator.record_update(stream_type, server_ts)
# Process data for your strategy
if stream_type == "quote":
_handle_quote(data)
elif stream_type == "depth":
_handle_depth(data)
elif stream_type == "trade":
_handle_trade(data)
def _handle_quote(data: dict):
"""Process quote data for your strategy."""
# Implementation depends on your strategy
pass
def _handle_depth(data: dict):
"""Process order book depth data."""
pass
def _handle_trade(data: dict):
"""Process trade data."""
pass
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
)
asyncio.run(main())
Testing the Detection System
A monitoring system is only as good as its test coverage. Use chaos testing to verify your detection logic handles each failure mode:
import pytest
import asyncio
from unittest.mock import AsyncMock, patch
class TestSilentDisconnectionDetection:
"""Test suite for silent disconnection detection."""
@pytest.mark.asyncio
async def test_heartbeat_timeout_triggers_reconnect(self):
"""Verify reconnection triggers after missed heartbeats."""
monitor = WebSocketHeartbeatMonitor(
url="wss://test.example.com/ws",
api_key="test_key",
on_data=lambda d: None,
config=HeartbeatConfig(
ping_interval=0.5,
pong_timeout=0.3,
max_missed_heartbeats=2,
),
)
reconnect_called = False
original_reconnect = monitor._reconnect
async def mock_reconnect():
nonlocal reconnect_called
reconnect_called = True
monitor._reconnect = mock_reconnect
with patch.object(monitor, 'connect', new_callable=AsyncMock):
await monitor.start()
# Wait for heartbeat loop to run
await asyncio.sleep(2.0)
# Verify reconnect was called
assert reconnect_called, "Reconnection should have triggered after missed heartbeats"
await monitor.stop()
@pytest.mark.asyncio
async def test_data_arrival_resets_heartbeat_counter(self):
"""Verify data reception resets the missed heartbeat counter."""
monitor = WebSocketHeartbeatMonitor(
url="wss://test.example.com/ws",
api_key="test_key",
on_data=lambda d: None,
config=HeartbeatConfig(max_missed_heartbeats=3),
)
await monitor.connect()
# Simulate heartbeat misses
monitor.state.consecutive_missed_heartbeats = 2
# Simulate data receipt
monitor._process_data({"type": "quote", "price": 150.0})
# Counter should reset
assert monitor.state.consecutive_missed_heartbeats == 0
await monitor.stop()
@pytest.mark.asyncio
async def test_timestamp_validator_detects_stale_stream(self):
"""Verify timestamp validator flags stale data."""
validator = DataTimestampValidator()
stale_detected = []
validator.on_stale = lambda s, t: stale_detected.append((s, t))
validator.register_stream(DataStreamConfig(
stream_name="test",
expected_interval=1.0,
max_staleness=0.5,
check_interval=0.1,
))
# Record an update
validator.record_update("test", server_timestamp=time.time())
await validator.start()
# Wait for staleness to be detected
await asyncio.sleep(1.0)
await validator.stop()
assert len(stale_detected) > 0, "Staleness should have been detected"
assert stale_detected[0][0] == "test"
@pytest.mark.asyncio
async def test_reconnection_uses_exponential_backoff(self):
"""Verify reconnection delays follow exponential backoff pattern."""
monitor = WebSocketHeartbeatMonitor(
url="wss://test.example.com/ws",
api_key="test_key",
on_data=lambda d: None,
reconnect_base_delay=1.0,
reconnect_max_delay=60.0,
)
delays = []
original_sleep = asyncio.sleep
async def mock_sleep(duration):
delays.append(duration)
with patch('asyncio.sleep', mock_sleep):
with patch.object(monitor, 'connect', side_effect=Exception("Connection failed")):
await monitor._reconnect()
# Verify delays increase exponentially (with some jitter tolerance)
assert len(delays) >= 2, "Should have multiple retry attempts"
for i in range(1, len(delays)):
# Each delay should be roughly 2x the previous (before jitter)
ratio = delays[i] / delays[i-1]
assert 1.5 < ratio < 3.0, (
f"Delay {i} ({delays[i]:.2f}s) should be ~2x delay {i-1} ({delays[i-1]:.2f}s), "
f"got ratio {ratio:.2f}"
)
Deployment Considerations by Market
Different markets have different data cadences, which affects your staleness thresholds:
| Market | Quote frequency | Recommended max_staleness | Notes |
|---|---|---|---|
| US equities (liquid) | 10–100ms | 2.0s | High-frequency updates; 2s gap is significant |
| US equities (illiquid) | 1–10s | 30.0s | Sparse updates; be lenient |
| Crypto | 10–500ms | 2.0s | Often higher frequency than equities |
| HK equities | 100ms–1s | 5.0s | Moderate frequency |
| Forex (majors) | 50–500ms | 2.0s | High frequency; 2s gap is material |
| Forex (exotics) | 1–30s | 60.0s | Sparse; adjust thresholds accordingly |
Adjust your DataStreamConfig instances based on the asset class and individual symbol liquidity. A stale threshold that's too aggressive generates false positives during normal market microstructure. One that's too lenient delays detection of real disconnections.
Monitoring Dashboard: Exposing Connection Health
A WebSocket monitoring system is only useful if you can observe its state. Expose the following metrics for your operational dashboard:
def get_connection_health_report(monitor: WebSocketHeartbeatMonitor, validator: DataTimestampValidator) -> dict:
"""
Generate a health report for operational monitoring.
Integrate this with Prometheus, Grafana, or your alerting system.
"""
timestamp_validator_state = validator.get_status()
heartbeat_state = monitor.state
# Calculate overall health score
stale_count = sum(1 for s in timestamp_validator_state.values() if s["is_stale"])
total_streams = len(timestamp_validator_state)
health_score = 1.0 - (stale_count / max(total_streams, 1))
return {
"overall_health_score": health_score,
"connection": {
"connected": monitor._ws is not None and not monitor._ws.closed,
"connected_at": heartbeat_state.connected_at,
"uptime_seconds": (
time.time() - heartbeat_state.connected_at
if heartbeat_state.connected_at else None
),
"total_reconnections": heartbeat_state.total_reconnections,
"last_reconnect_at": heartbeat_state.last_reconnect_at,
},
"heartbeat": {
"last_pong_at": heartbeat_state.last_pong_received,
"seconds_since_last_pong": (
time.time() - heartbeat_state.last_pong_received
if heartbeat_state.last_pong_received else None
),
"consecutive_missed_heartbeats": heartbeat_state.consecutive_missed_heartbeats,
},
"data_streams": timestamp_validator_state,
"alerts": {
"connection_unhealthy": health_score < 0.5,
"reconnection_loop_detected": (
heartbeat_state.total_reconnections > 10 and
(time.time() - heartbeat_state.last_reconnect_at) < 300
),
"stale_streams": [name for name, s in timestamp_validator_state.items() if s["is_stale"]],
}
}
Key Takeaways
A WebSocket connection that looks alive can be silently dead. The three detection mechanisms each cover different failure modes:
- Heartbeat timeout catches connection-level failures where the server has disappeared without a close frame
- Data timestamp validation catches partial failures where a specific data stream has stopped while the overall connection persists
- Exponential backoff with jitter prevents thundering herd when reconnecting and guards against repeated connection attempts during extended outages
No single mechanism is sufficient. Production systems require all three, with operational monitoring to detect reconnection loops, stale streams, and overall health degradation.
For TickDB users, the WebSocket endpoint at wss://api.tickdb.ai/ws/v1/market supports application-level ping/pong commands. Use the code in this article as a starting point, but calibrate your staleness thresholds to the asset class and liquidity profile of your target symbols.
Next Steps
If you're debugging an existing WebSocket integration, instrument your client with the health report function above. A health score below 0.7 within 30 minutes of startup indicates a problem worth investigating.
If you need a production-ready WebSocket client with full reconnection logic:
- Sign up at tickdb.ai (free, no credit card required)
- Generate an API key in the dashboard
- Set the
TICKDB_API_KEYenvironment variable, then adapt the code from this article to your specific data channels
If you're running multiple strategies across different markets, consider deploying the timestamp validator per-strategy rather than per-process. Stale detection thresholds differ by strategy aggressiveness and market microstructure.
If you need institutional-grade reliability with SLA-backed uptime guarantees, reach out to enterprise@tickdb.ai for dedicated infrastructure and priority reconnection handling.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. WebSocket connection behavior may vary based on network infrastructure, geographic location, and data provider implementation.