A system that screams "DISCONNECTED!" every day at noon is not a monitoring system. It's a liability.
At 12:00:00 HKT, a trading system monitoring 47 Hong Kong-listed securities fires an alert. The Slack channel lights up. The on-call engineer grabs their laptop. They open Grafana, trace the WebSocket connection, check the API status page, and verify their credentials — everything looks fine. They watch the dashboard for two minutes, increasingly confused, until the data stream resumes at 13:00:01 HKT and the alert clears itself.
This happens every trading day. It's not a failure. It's lunch.
The Hong Kong Stock Exchange operates a reduced trading session during the lunch break — a quirk rooted in the market's origins when physical trading floors required midday clearing. From 12:00 to 13:00 HKT (04:00–05:00 UTC), the exchange halts continuous trading. No order book updates. No trades. No market data push. For a real-time monitoring system built by someone who assumes markets trade continuously, this one-hour window is a daily false-positive generator.
The engineering challenge is not "how do I receive data during lunch." The exchange doesn't send it. The challenge is: how do I build a monitoring system that understands market structure well enough to distinguish a one-hour scheduled silence from an actual connection failure?
This article provides a production-grade solution — a time-aware monitoring layer that applies market calendar logic to suppress false alerts and escalate genuine incidents.
1. The Problem: Two Kinds of Silence
When a WebSocket data stream goes quiet, the monitoring system faces an inference problem. Silence is ambiguous. It could mean:
| Scenario | Duration | Root cause | Should alert? |
|---|---|---|---|
| Normal lunch break | 60 minutes | Exchange halt (HKEX rule) | No |
| Planned maintenance | Variable | Exchange notification | Conditional |
| Network partition | Indefinite | Network infrastructure failure | Yes — immediately |
| WebSocket timeout | Until reconnect | Client/server heartbeat failure | Yes — after timeout |
| API rate limit | Until cooldown | Client-side throttling | Yes — after cooldown |
| Symbol delisting | Indefinite | Corporate action | Conditional |
The naive approach — alerting after N seconds of silence — cannot distinguish between these scenarios. It generates false positives during every lunch break and potentially misses real incidents that occur during trading hours.
The correct approach is a time-aware monitoring layer that knows the market calendar and adjusts its behavior accordingly.
2. The Architecture: Market-Calendar-Aware Monitoring
The solution consists of three components:
- Market Calendar Service: Maintains a structured schedule of trading sessions, breaks, and special events for each exchange.
- Connection State Machine: Tracks WebSocket state transitions and applies calendar-aware timeout logic.
- Alert Decision Engine: Combines connection state, calendar state, and elapsed time to decide whether to escalate.
2.1 System Architecture Diagram
┌──────────────────────────────────────────────────────────────┐
│ Monitoring Application │
├──────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ │
│ │ WebSocket │───▶│ State │───▶│ Alert │ │
│ │ Manager │ │ Machine │ │ Decision │ │
│ └─────────────┘ └──────────────┘ │ Engine │ │
│ │ ▲ └───────────────┘ │
│ │ │ ▲ │
│ ▼ │ │ │
│ ┌─────────────┐ │ │ │
│ │ Market │────────────┴────────────────────┘ │
│ │ Calendar │ │
│ │ Service │ │
│ └─────────────┘ │
│ │
└──────────────────────────────────────────────────────────────┘
│
▼
┌───────────────┐
│ TickDB API │
│ (Data Source)│
└───────────────┘
2.2 State Machine Definition
The connection lifecycle maps to a state machine with six states:
from enum import Enum, auto
from dataclasses import dataclass
from datetime import datetime, time
from typing import Optional
class ConnectionState(Enum):
"""WebSocket connection states with market-calendar awareness."""
DISCONNECTED = auto() # Initial state or after clean disconnect
CONNECTING = auto() # Attempting handshake
AUTHENTICATED = auto() # Handshake complete, awaiting first data
STREAMING = auto() # Receiving data normally
MARKET_HALT = auto() # Known market break (lunch, holiday)
FAILED = auto() # Unrecoverable error — requires human intervention
Each state has an associated alert policy:
| State | Alert on entry? | Alert on timeout? | Escalation priority |
|---|---|---|---|
| DISCONNECTED | No | N/A | N/A |
| CONNECTING | No | After 30s → FAILED | Medium |
| AUTHENTICATED | No | After 60s → FAILED | Medium |
| STREAMING | No | After 30s → FAILED | High |
| MARKET_HALT | No | No timeout during halt | None |
| FAILED | Yes | N/A | Critical |
3. Production-Grade Code: Market-Calendar-Aware WebSocket Manager
The following implementation provides a complete, production-ready WebSocket manager with market calendar integration. It includes heartbeat, exponential backoff with jitter, rate-limit handling, and time-aware alert suppression.
import os
import json
import time
import asyncio
import logging
import random
import threading
from datetime import datetime, time as dt_time, date, timedelta, timezone
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Callable, Optional, List
import requests
# ⚠️ For production HFT workloads, use aiohttp/asyncio instead of threading
# This synchronous implementation is suitable for monitoring and alerting use cases
logger = logging.getLogger(__name__)
# ============================================================================
# SECTION 1: Market Calendar Service
# ============================================================================
@dataclass
class TradingSession:
"""Defines a single trading session within a trading day."""
name: str
open_time: dt_time # Market local time (no timezone — applied per exchange)
close_time: dt_time
is_continuous: bool = True
@dataclass
class MarketCalendar:
"""
Market calendar with session definitions and holiday handling.
For HKEX:
- Continuous trading: 09:30–12:00 and 13:00–16:00 HKT
- Lunch break: 12:00–13:00 HKT (no data, no trading)
- Timezone: Asia/Hong_Kong (UTC+8)
"""
market_id: str
timezone_name: str
sessions: List[TradingSession]
holidays: List[date] = field(default_factory=list)
# HKEX-specific: lunch break is a scheduled halt, not a session boundary
# We model it as a "halt period" that falls between two trading sessions
halt_periods: List[tuple[dt_time, dt_time]] = field(default_factory=list)
def is_trading(self, check_time: datetime) -> bool:
"""Returns True if the market is in an active trading session."""
local_time = check_time.time()
today = check_time.date()
if today in self.holidays:
return False
for session in self.sessions:
if session.open_time <= local_time < session.close_time:
return True
return False
def is_scheduled_halt(self, check_time: datetime) -> bool:
"""
Returns True if the market is in a scheduled halt (e.g., lunch break).
This is distinct from a market closure — the exchange is aware and
the data feed will resume automatically.
"""
local_time = check_time.time()
today = check_time.date()
if today in self.holidays:
return False # Holidays are not "halt" periods
for halt_start, halt_end in self.halt_periods:
if halt_start <= local_time < halt_end:
return True
return False
def time_to_next_session(self, check_time: datetime) -> Optional[timedelta]:
"""Returns time until next trading session starts, or None if market is open."""
if self.is_trading(check_time):
return None
today = check_time.date()
# Check if we've passed all sessions today
current_time = check_time.time()
next_session = None
for session in sorted(self.sessions, key=lambda s: s.open_time):
if session.open_time > current_time:
next_session = session
break
if next_session:
next_dt = datetime.combine(today, next_session.open_time)
return timedelta(seconds=(next_dt - check_time).total_seconds())
# Market closed for today — return time to first session tomorrow
first_session = min(self.sessions, key=lambda s: s.open_time)
tomorrow = today + timedelta(days=1)
next_dt = datetime.combine(tomorrow, first_session.open_time)
return timedelta(seconds=(next_dt - check_time).total_seconds())
# HKEX Calendar Definition
HKEX_CALENDAR = MarketCalendar(
market_id="HKEX",
timezone_name="Asia/Hong_Kong",
sessions=[
TradingSession(name="Morning", open_time=dt_time(9, 30), close_time=dt_time(12, 0)),
TradingSession(name="Afternoon", open_time=dt_time(13, 0), close_time=dt_time(16, 0)),
],
halt_periods=[
(dt_time(12, 0), dt_time(13, 0)), # Lunch break
],
holidays=[
# Add HKEX public holidays for the year
# Example: date(2026, 1, 1), date(2026, 2, 1), ...
]
)
# ============================================================================
# SECTION 2: Connection State Machine
# ============================================================================
class ConnectionState(Enum):
DISCONNECTED = auto()
CONNECTING = auto()
AUTHENTICATED = auto()
STREAMING = auto()
MARKET_HALT = auto()
FAILED = auto()
@dataclass
class ConnectionMetrics:
"""Tracks connection health metrics for alerting decisions."""
last_message_at: Optional[datetime] = None
last_heartbeat_at: Optional[datetime] = None
consecutive_failures: int = 0
total_reconnects: int = 0
state_entered_at: datetime = field(default_factory=datetime.now)
# ============================================================================
# SECTION 3: Market-Aware WebSocket Manager (Production Implementation)
# ============================================================================
class MarketAwareWebSocketManager:
"""
Production-grade WebSocket manager with market calendar integration.
Key features:
- Heartbeat with ping/pong
- Exponential backoff + jitter on reconnect
- Rate-limit handling (3001 error code + Retry-After header)
- Market calendar awareness: suppresses alerts during scheduled halts
- Time-aware timeout: adjusts thresholds based on market state
"""
# Configuration constants
HEARTBEAT_INTERVAL_SECONDS = 30
BASE_RECONNECT_DELAY_SECONDS = 1.0
MAX_RECONNECT_DELAY_SECONDS = 60.0
JITTER_FACTOR = 0.1
# Timeout thresholds (in seconds) — varies by state
TIMEOUT_THRESHOLDS = {
ConnectionState.CONNECTING: 30,
ConnectionState.AUTHENTICATED: 60,
ConnectionState.STREAMING: 30,
ConnectionState.MARKET_HALT: None, # No timeout during scheduled halt
}
# Rate limit handling
RATE_LIMIT_CODE = 3001
RATE_LIMIT_DEFAULT_WAIT = 5
def __init__(
self,
api_key: str,
symbols: List[str],
market_calendar: MarketCalendar,
on_data: Callable[[dict], None],
on_alert: Callable[[str, str], None], # (severity, message)
):
self.api_key = api_key
self.symbols = symbols
self.market_calendar = market_calendar
self.on_data = on_data
self.on_alert = on_alert
self.state = ConnectionState.DISCONNECTED
self.metrics = ConnectionMetrics()
self._running = False
self._lock = threading.Lock()
self._worker_thread: Optional[threading.Thread] = None
# WebSocket simulation (replace with actual websocket-client library in production)
self._ws = None
self._last_pong_received: Optional[datetime] = None
logger.info(f"MarketAwareWebSocketManager initialized for {len(symbols)} symbols")
# -------------------------------------------------------------------------
# Public API
# -------------------------------------------------------------------------
def start(self):
"""Starts the WebSocket manager in a background thread."""
if self._running:
logger.warning("Manager already running")
return
self._running = True
self._worker_thread = threading.Thread(target=self._run_loop, daemon=True)
self._worker_thread.start()
logger.info("WebSocket manager started")
def stop(self):
"""Gracefully stops the WebSocket manager."""
self._running = False
if self._worker_thread:
self._worker_thread.join(timeout=10)
self._set_state(ConnectionState.DISCONNECTED)
logger.info("WebSocket manager stopped")
def get_status(self) -> dict:
"""Returns current status for health check endpoints."""
return {
"state": self.state.name,
"last_message_seconds_ago": (
(datetime.now() - self.metrics.last_message_at).total_seconds()
if self.metrics.last_message_at else None
),
"reconnects": self.metrics.total_reconnects,
"is_market_halt": self.market_calendar.is_scheduled_halt(datetime.now()),
"is_trading": self.market_calendar.is_trading(datetime.now()),
}
# -------------------------------------------------------------------------
# Internal State Machine
# -------------------------------------------------------------------------
def _set_state(self, new_state: ConnectionState):
"""Thread-safe state transition with logging and alert evaluation."""
old_state = self.state
self.state = new_state
self.metrics.state_entered_at = datetime.now()
logger.info(f"State transition: {old_state.name} → {new_state.name}")
# Evaluate alert conditions on state entry
if new_state == ConnectionState.FAILED:
self.on_alert(
"CRITICAL",
f"WebSocket connection failed after {self.metrics.consecutive_failures} attempts. "
f"Manual intervention required."
)
def _run_loop(self):
"""
Main event loop running in a background thread.
Handles connection lifecycle, heartbeat, reconnection, and market calendar checks.
"""
retry_count = 0
while self._running:
try:
# Check market calendar before any connection attempt
now = datetime.now()
if self.market_calendar.is_scheduled_halt(now):
if self.state != ConnectionState.MARKET_HALT:
self._set_state(ConnectionState.MARKET_HALT)
self.on_alert("INFO", "Entering scheduled market halt (lunch break)")
# Wait until market resumes, then reconnect
time_to_resume = self.market_calendar.time_to_next_session(now)
if time_to_resume:
logger.info(f"Market halt active. Waiting {time_to_resume} until resume.")
time.sleep(min(time_to_resume.total_seconds(), 60)) # Check every minute
continue
# Attempt connection if not already connected
if self.state not in (ConnectionState.STREAMING, ConnectionState.CONNECTING, ConnectionState.AUTHENTICATED):
self._connect(retry_count)
retry_count = 0
# Heartbeat check
self._check_heartbeat()
# Timeout check
self._check_timeout()
# Simulate message processing (replace with actual WebSocket receive in production)
# In a real implementation, this would be: message = self._ws.recv()
# For demonstration, we simulate occasional data receipt during trading hours
self._simulate_message_processing()
time.sleep(1) # Main loop tick rate: 1 second
except Exception as e:
logger.error(f"Error in run loop: {e}", exc_info=True)
retry_count += 1
self._handle_connection_error(e, retry_count)
def _connect(self, retry_count: int):
"""
Establishes WebSocket connection with exponential backoff.
In production, replace the simulation below with:
import websocket
ws_url = f"wss://api.tickdb.ai/v1/stream?api_key={self.api_key}"
self._ws = websocket.create_connection(ws_url, timeout=10)
# Subscribe to symbols
subscribe_msg = {
"cmd": "subscribe",
"params": {
"channels": ["trades", "depth"],
"symbols": self.symbols,
}
}
self._ws.send(json.dumps(subscribe_msg))
"""
self._set_state(ConnectionState.CONNECTING)
# Calculate reconnect delay with exponential backoff + jitter
delay = min(
self.BASE_RECONNECT_DELAY_SECONDS * (2 ** retry_count),
self.MAX_RECONNECT_DELAY_SECONDS
)
jitter = random.uniform(0, delay * self.JITTER_FACTOR)
sleep_time = delay + jitter
logger.info(f"Connecting (attempt {retry_count + 1}) after {sleep_time:.2f}s backoff")
time.sleep(sleep_time)
# Simulate connection establishment
# In production: self._ws = websocket.create_connection(...)
self._set_state(ConnectionState.AUTHENTICATED)
# Simulate authentication
# In production: handle auth response, handle rate limits
self._set_state(ConnectionState.STREAMING)
self.metrics.consecutive_failures = 0
def _check_heartbeat(self):
"""Sends heartbeat and verifies pong response."""
if self.state not in (ConnectionState.STREAMING, ConnectionState.AUTHENTICATED):
return
now = datetime.now()
# Send heartbeat
# In production: self._ws.send(json.dumps({"cmd": "ping"}))
self.metrics.last_heartbeat_at = now
# Simulate pong response check (in production, receive and parse pong)
# If no pong received within 10 seconds, the connection may be dead
if self._last_pong_received:
elapsed = (now - self._last_pong_received).total_seconds()
if elapsed > self.HEARTBEAT_INTERVAL_SECONDS * 2:
logger.warning(f"Heartbeat timeout: {elapsed:.1f}s since last pong")
self._handle_connection_error(
TimeoutError("No pong received within heartbeat interval"),
self.metrics.consecutive_failures + 1
)
def _check_timeout(self):
"""
Checks if the connection has timed out based on current state.
Critical: Timeouts are SUPPRESSED during MARKET_HALT state.
This is the core of the false-positive prevention logic.
"""
if self.state == ConnectionState.MARKET_HALT:
# Do NOT alert during scheduled halt — this is the key differentiator
return
threshold = self.TIMEOUT_THRESHOLDS.get(self.state)
if threshold is None:
return
elapsed = (datetime.now() - self.metrics.state_entered_at).total_seconds()
if elapsed > threshold:
logger.warning(f"Connection timeout: {elapsed:.1f}s > {threshold}s threshold in {self.state.name}")
self._handle_connection_error(
TimeoutError(f"Connection timeout in state {self.state.name}"),
self.metrics.consecutive_failures + 1
)
def _simulate_message_processing(self):
"""
Simulates processing incoming market data messages.
In production, this would be:
while True:
try:
message = self._ws.recv(timeout=1)
data = json.loads(message)
self._handle_message(data)
except websocket.WebSocketTimeoutException:
break # No message available, check market calendar
"""
now = datetime.now()
# Only simulate data receipt during trading hours
if self.market_calendar.is_trading(now):
# Simulate message receipt with random intervals (1-5 seconds)
if random.random() < 0.3: # 30% chance per tick = ~3s average interval
self.metrics.last_message_at = now
# In production: parse actual data and call self.on_data(data)
self.metrics.last_message_at = now
def _handle_message(self, data: dict):
"""Processes incoming market data message."""
self.metrics.last_message_at = datetime.now()
# Handle pong responses
if data.get("type") == "pong":
self._last_pong_received = datetime.now()
return
# Handle rate limit responses
if data.get("code") == self.RATE_LIMIT_CODE:
retry_after = int(data.get("headers", {}).get("Retry-After", self.RATE_LIMIT_DEFAULT_WAIT))
logger.warning(f"Rate limit hit. Waiting {retry_after}s before retry.")
time.sleep(retry_after)
return
# Forward to data handler
self.on_data(data)
def _handle_connection_error(self, error: Exception, retry_count: int):
"""
Handles connection errors with exponential backoff.
Rate-limit errors (3001) are handled differently from network errors.
"""
self.metrics.consecutive_failures = retry_count
self.metrics.total_reconnects += 1
if isinstance(error, TimeoutError):
# Treat timeouts as recoverable — attempt reconnect
logger.warning(f"Timeout error (attempt {retry_count}): {error}")
elif isinstance(error, RuntimeError) and "3001" in str(error):
# Rate limit — wait for Retry-After and retry immediately
logger.warning(f"Rate limit error: {error}")
# The calling code should handle the Retry-After delay
else:
# Unknown error — log and continue with backoff
logger.error(f"Connection error (attempt {retry_count}): {error}")
# Close existing connection
if self._ws:
try:
self._ws.close()
except Exception:
pass
self._ws = None
# Set state based on retry count
if retry_count >= 5:
self._set_state(ConnectionState.FAILED)
else:
self._set_state(ConnectionState.DISCONNECTED)
# ============================================================================
# SECTION 4: Alert Handler Example
# ============================================================================
def create_alert_handler() -> Callable[[str, str], None]:
"""Creates an alert handler that formats and forwards alerts."""
def handle_alert(severity: str, message: str):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
formatted = f"[{severity}] [{timestamp}] {message}"
if severity == "CRITICAL":
# Send to PagerDuty / OpsGenie
logger.critical(formatted)
elif severity == "INFO":
# Log only during informational state changes
logger.info(formatted)
else:
# WARNING and other severities go to Slack
logger.warning(formatted)
return handle_alert
# ============================================================================
# SECTION 5: Usage Example
# ============================================================================
def main():
"""Example usage of the MarketAwareWebSocketManager."""
# Load API key from environment variable
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("TICKDB_API_KEY environment variable not set")
# Define symbols to monitor (Hong Kong stocks)
symbols = ["0700.HK", "9988.HK", "3690.HK", "9618.HK"]
# Data handler
def on_data(data: dict):
logger.info(f"Received data: {data.get('symbol', 'unknown')}")
# Alert handler
alert_handler = create_alert_handler()
# Initialize manager with HKEX calendar
manager = MarketAwareWebSocketManager(
api_key=api_key,
symbols=symbols,
market_calendar=HKEX_CALENDAR,
on_data=on_data,
on_alert=alert_handler,
)
# Start monitoring
manager.start()
# Keep running (in production, this would run as a daemon)
try:
while True:
status = manager.get_status()
logger.info(f"Status: {status}")
time.sleep(60)
except KeyboardInterrupt:
manager.stop()
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
main()
4. Key Algorithm: Time-to-Next-Session Calculation
The critical function that powers the monitoring logic is time_to_next_session. This function is called whenever the system enters MARKET_HALT state to determine how long to suppress alerts:
def calculate_halt_duration(calendar: MarketCalendar, check_time: datetime) -> dict:
"""
Calculates expected halt duration and time until resumption.
Returns:
dict with keys: is_halt, duration_seconds, resume_time, reason
"""
is_halt = calendar.is_scheduled_halt(check_time)
is_trading = calendar.is_trading(check_time)
if is_trading:
return {
"is_halt": False,
"duration_seconds": 0,
"resume_time": None,
"reason": "Market is in active trading session"
}
if is_halt:
# Find the specific halt period
local_time = check_time.time()
for halt_start, halt_end in calendar.halt_periods:
if halt_start <= local_time < halt_end:
# Calculate remaining duration
halt_start_dt = datetime.combine(check_time.date(), halt_start)
halt_end_dt = datetime.combine(check_time.date(), halt_end)
elapsed = (check_time - halt_start_dt).total_seconds()
remaining = (halt_end_dt - check_time).total_seconds()
return {
"is_halt": True,
"duration_seconds": remaining,
"resume_time": halt_end_dt.isoformat(),
"reason": f"Lunch break: {halt_start.strftime('%H:%M')}–{halt_end.strftime('%H:%M')} HKT"
}
# Market closed (after hours or holiday)
time_to_next = calendar.time_to_next_session(check_time)
return {
"is_halt": False,
"duration_seconds": time_to_next.total_seconds() if time_to_next else 0,
"resume_time": (check_time + time_to_next).isoformat() if time_to_next else None,
"reason": "Market closed for the day"
}
# Example output during lunch break:
# check_time = datetime(2026, 4, 15, 12, 30, 0) # 12:30 PM HKT
# Result: {
# "is_halt": True,
# "duration_seconds": 1800,
# "resume_time": "2026-04-15T13:00:00",
# "reason": "Lunch break: 12:00–13:00 HKT"
# }
This calculation enables the monitoring system to:
- Suppress alerts for the exact duration of the lunch break (not forever)
- Set an expected resumption timer — if data doesn't resume at 13:00 HKT, that's a real incident
- Distinguish lunch from after-hours close — after 16:00, the halt is indefinite and requires different alerting behavior
5. Deployment Configuration by Scale
| Use case | Recommended configuration | Key consideration |
|---|---|---|
| Individual quant | Single-threaded manager, 5–20 symbols | Minimal CPU; focus on reliability |
| Trading team | Per-symbol managers in thread pool, 50–200 symbols | Graceful degradation if one symbol fails |
| Institutional | Async implementation (aiohttp), 500+ symbols | HFT-grade latency; avoid GIL contention |
| Market-data vendor | Distributed architecture, multiple API keys | Handle API key rotation and quota management |
For individual users, the synchronous implementation above is sufficient. For teams or institutions processing high-frequency data, replace the threading-based loop with an asyncio-based architecture:
# ⚠️ Asyncio variant for production HFT workloads
async def run_async_manager(manager: MarketAwareWebSocketManager):
"""Async variant for high-throughput deployments."""
import aiohttp
async with aiohttp.ClientSession() as session:
ws_url = f"wss://api.tickdb.ai/v1/stream?api_key={manager.api_key}"
async with session.ws_connect(ws_url) as ws:
await ws.send_json({
"cmd": "subscribe",
"params": {"channels": ["depth"], "symbols": manager.symbols}
})
async for msg in ws:
if msg.type == aiohttp.WSMsgType.PONG:
manager.metrics.last_message_at = datetime.now()
elif msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
manager._handle_message(data)
6. Testing the Calendar Logic
Before deploying, validate the calendar logic against known trading dates:
import unittest
from datetime import datetime, date
class TestHKEXCalendar(unittest.TestCase):
"""Unit tests for HKEX market calendar."""
def test_lunch_break_morning(self):
"""12:00 is within lunch break."""
t = datetime(2026, 4, 15, 12, 0, 0) # 12:00 HKT
self.assertTrue(HKEX_CALENDAR.is_scheduled_halt(t))
self.assertFalse(HKEX_CALENDAR.is_trading(t))
def test_lunch_break_mid(self):
"""12:30 is within lunch break."""
t = datetime(2026, 4, 15, 12, 30, 0)
self.assertTrue(HKEX_CALENDAR.is_scheduled_halt(t))
self.assertFalse(HKEX_CALENDAR.is_trading(t))
def test_market_open_morning(self):
"""10:00 is during morning session."""
t = datetime(2026, 4, 15, 10, 0, 0)
self.assertFalse(HKEX_CALENDAR.is_scheduled_halt(t))
self.assertTrue(HKEX_CALENDAR.is_trading(t))
def test_market_open_afternoon(self):
"""14:00 is during afternoon session."""
t = datetime(2026, 4, 15, 14, 0, 0)
self.assertFalse(HKEX_CALENDAR.is_scheduled_halt(t))
self.assertTrue(HKEX_CALENDAR.is_trading(t))
def test_after_hours(self):
"""17:00 is after market close."""
t = datetime(2026, 4, 15, 17, 0, 0)
self.assertFalse(HKEX_CALENDAR.is_scheduled_halt(t))
self.assertFalse(HKEX_CALENDAR.is_trading(t))
def test_resume_time_during_lunch(self):
"""System should calculate correct resume time."""
t = datetime(2026, 4, 15, 12, 30, 0)
duration = HKEX_CALENDAR.time_to_next_session(t)
self.assertIsNotNone(duration)
self.assertAlmostEqual(duration.total_seconds(), 1800, places=0) # ~30 min
def test_halt_duration_calculation(self):
"""calculate_halt_duration should return correct remaining time."""
t = datetime(2026, 4, 15, 12, 15, 0) # 15 minutes into lunch
result = calculate_halt_duration(HKEX_CALENDAR, t)
self.assertTrue(result["is_halt"])
self.assertAlmostEqual(result["duration_seconds"], 2700, places=0) # ~45 min remaining
if __name__ == "__main__":
unittest.main()
Run the tests before deployment to ensure the calendar logic handles edge cases correctly.
7. Monitoring Dashboard: What to Track
Beyond connection state, track these metrics to build a complete health dashboard:
| Metric | Description | Alert threshold |
|---|---|---|
state |
Current ConnectionState enum value | FAILED |
seconds_since_last_message |
Age of last data receipt | > 30s during trading, no limit during halt |
reconnects_total |
Cumulative reconnect count | > 5 per hour |
consecutive_failures |
Current failure streak | > 3 |
is_market_halt |
Whether market is in scheduled break | N/A — informational only |
is_trading |
Whether market is in active session | N/A — informational only |
time_to_next_session |
Seconds until next session start | N/A — for display only |
Expose these metrics via a /health endpoint for integration with Prometheus, Datadog, or CloudWatch:
@app.route("/health")
def health_check():
status = manager.get_status()
is_healthy = status["state"] != "FAILED" and (
status["is_market_halt"] or status["is_trading"] or status["last_message_seconds_ago"] is None
)
return {
"status": "healthy" if is_healthy else "degraded",
"details": status,
"timestamp": datetime.now().isoformat(),
}
8. Closing
A monitoring system that alerts on every silence is not monitoring — it's noise generation. For Hong Kong stocks, the 12:00–13:00 lunch break is a predictable, scheduled event. The system should know it is coming, adjust its behavior accordingly, and reserve its alerts for genuine incidents.
The market calendar is not a feature. It is the foundational layer that makes the rest of the monitoring system trustworthy.
If you are building real-time data pipelines for HK stocks or any market with non-continuous trading sessions, the principle holds: know the market's schedule before you build its watchdog.
Next Steps
If you are a quant researcher building event-driven strategies on HK stocks, understanding the market's structural breaks is as important as understanding price action. The lunch break creates a predictable information vacuum — use it to reset your state machines, not to panic.
If you need reliable WebSocket market data for HK equities, TickDB provides real-time depth and trade data with native ping/pong heartbeat support, configurable to your monitoring needs.
If you are building a production monitoring system, consider installing the tickdb-market-data SKILL in your AI coding assistant to accelerate development of time-aware connection managers.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.