The market does not wait for your monitoring system to reconnect.
On September 15, 2022, a tier-1 US equity data vendor suffered a 47-minute outage during the pre-market window. For a systematic strategy running on a 15-second rebalance cycle, 47 minutes meant 188 missed rebalances, a portfolio drift of 3.2%, and a drawdown that took three weeks to recover. The vendor's infrastructure failed. So did the monitoring system watching it. Neither failure was detected until the morning standup.
This is the failure mode that turns a routine infrastructure incident into a career conversation. This article builds the monitoring system that catches it: a 7×24 daemon that survives network partitions, detects latency degradation before it becomes a data gap, and pages the right person before the drift becomes a loss.
We will cover four engineering problems in sequence: connection resilience (exponential backoff with jitter), protocol keepalive (heartbeat discipline), delay detection (sliding window anomaly scoring), and alert delivery (Feishu/Lark webhook integration). Every code block is production-grade — it handles every error condition you will encounter in practice, not just the happy path.
1. The Silent Killers: What Kills Monitoring Systems
Before writing code, we need to name the enemies precisely. A monitoring system running for months without intervention faces a specific failure taxonomy:
| Failure Mode | Symptom | Root Cause | Detection Difficulty |
|---|---|---|---|
| Network partition | Connection drops, no reconnect attempt | Router failure, NAT timeout | Easy — obvious disconnect |
| TCP half-open | Heartbeat sent, but no response | Firewall state timeout, load balancer drop | Hard — connection appears alive |
| Memory leak | Process grows, then OOM killed | Unbounded queue, circular references | Slow — days to manifest |
| Deserialization error | Silent message drop | Schema mismatch, encoding error | Hard — no error, just missing data |
| Clock skew | Stale data timestamps appear fresh | NTP drift on cloud VM | Hard — data looks valid |
| Alert fatigue | PagerDuty storm, ignored | Threshold too tight, no deduplication | Organizational — system still runs |
Most monitoring tutorials address the easy problems. This article addresses all six.
2. System Architecture
The monitoring daemon is a single-process, multi-coroutine application. It does not need Kubernetes. A single $6/month virtual private server can run this reliably for years — the engineering discipline comes from the code, not the infrastructure.
┌─────────────────────────────────────────────────────┐
│ Monitoring Daemon │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────┐ │
│ │ WebSocket │───▶│ Delay │───▶│ Feishu │ │
│ │ Client │ │ Detector │ │ Pusher │ │
│ │ │ │ │ │ │ │
│ │ • reconnect│ │ • sliding │ │ • retry │ │
│ │ • heartbeat│ │ window │ │ • dedup │ │
│ │ • parse │ │ • anomaly │ │ │ │
│ └─────────────┘ │ • score │ └──────────┘ │
│ │ └─────────────┘ ▲ │
│ ▼ │ │ │
│ ┌─────────────┐ ▼ │ │
│ │ State │ ┌─────────────┐ │ │
│ │ Manager │◀───│ Alert │────────┘ │
│ │ │ │ Manager │ │
│ │ • session │ │ │ │
│ │ • symbols │ │ • cooldown │ │
│ │ • last_ts │ │ • routing │ │
│ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────┘
│ │
▼ ▼
TickDB WebSocket Feishu Webhook API
(depth channel) ( Lark Open Platform )
The critical design constraint: every component must be independently restartable without corrupting state. If the delay detector crashes, the WebSocket client continues running. If the Feishu pusher backs up, the delay detector does not know — it just enqueues a message.
3. Connection Resilience: Exponential Backoff with Jitter
The naive reconnection strategy — sleep 5 seconds, retry, repeat — will hammer a TickDB server during a 10-minute outage and get your IP temporarily rate-limited. The correct strategy is exponential backoff with full jitter.
3.1 Why Jitter Matters
Exponential backoff without jitter creates a thundering herd. If 1,000 monitoring daemons all reconnect at the same time (because they all dropped within the same 5-second window), they all retry at second 5, then second 10, then second 20. The server sees a predictable wave of reconnection attempts — which itself becomes a denial-of-service event.
Full jitter randomizes the retry interval within the backoff window. Every client picks a different time, and the server sees a smooth distribution.
3.2 The Algorithm
The formula for full jitter:
base_delay = 1.0 seconds
max_delay = 60.0 seconds
attempt = 0, 1, 2, ...
sleep_time = random.uniform(0, min(max_delay, base_delay * (2 ** attempt)))
This gives retry intervals of roughly: 0.3s, 0.8s, 2.1s, 3.7s, 8.5s, 15.2s, 32s, 60s (capped). After 60 seconds of maxed-out backoff, we continue retrying at 60 seconds indefinitely until the connection restores.
3.3 Production Code: Resilient WebSocket Client
import os
import time
import json
import random
import asyncio
import logging
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional
from websockets.client import connect as ws_connect
from websockets.exceptions import (
ConnectionClosed,
InvalidStatusCode,
)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("tickdb.monitor")
@dataclass
class ReconnectPolicy:
"""Exponential backoff with full jitter for WebSocket reconnection."""
base_delay: float = 1.0
max_delay: float = 60.0
max_attempts: Optional[int] = None # None = infinite
def sleep_interval(self, attempt: int) -> float:
cap = min(self.max_delay, self.base_delay * (2 ** attempt))
return random.uniform(0, cap)
def should_retry(self, attempt: int) -> bool:
if self.max_attempts is None:
return True
return attempt < self.max_attempts
class TickDBMonitor:
"""
Production-grade WebSocket client for 7x24 TickDB data monitoring.
Handles:
- Exponential backoff + full jitter reconnection
- WebSocket ping/pong heartbeat
- Configurable rate-limit response (code 3001)
- Graceful degradation during outages
"""
def __init__(
self,
api_key: Optional[str] = None,
symbols: Optional[list[str]] = None,
channels: Optional[list[str]] = None,
on_message=None, # callback(message: dict)
on_connect=None, # callback()
on_disconnect=None, # callback(reason: str)
on_alert=None, # callback(alert: dict)
):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError(
"TickDB API key required. "
"Set TICKDB_API_KEY environment variable or pass api_key."
)
self.symbols = symbols or ["AAPL.US", "NVDA.US", "TSLA.US"]
self.channels = channels or ["depth", "trades"]
self.ws_url = self._build_ws_url()
self.on_message = on_message
self.on_connect = on_connect
self.on_disconnect = on_disconnect
self.on_alert = on_alert
self._reconnect_policy = ReconnectPolicy()
self._running = False
self._ws = None
self._session_stats = {
"connect_attempts": 0,
"successful_connects": 0,
"messages_received": 0,
"last_message_at": None,
"consecutive_failures": 0,
}
def _build_ws_url(self) -> str:
base = "wss://api.tickdb.ai/ws/market"
params = [
("api_key", self.api_key),
("symbol", ",".join(self.symbols)),
("channel", ",".join(self.channels)),
]
query = "&".join(f"{k}={v}" for k, v in params)
return f"{base}?{query}"
async def _connect_with_reconnect(self) -> bool:
"""Core reconnection loop with exponential backoff + jitter."""
attempt = 0
while self._running and self._reconnect_policy.should_retry(attempt):
self._session_stats["connect_attempts"] += 1
try:
logger.info(
"Connecting to TickDB WebSocket (attempt %d)",
attempt + 1,
)
self._ws = await ws_connect(
self.ws_url,
ping_interval=20, # Send ping every 20s
ping_timeout=10, # Expect pong within 10s
close_timeout=5, # Graceful close within 5s
open_timeout=10, # Connection open timeout
)
self._session_stats["successful_connects"] += 1
self._session_stats["consecutive_failures"] = 0
logger.info("Connected to TickDB WebSocket successfully")
if self.on_connect:
self.on_connect()
return True
except InvalidStatusCode as e:
# Server returned HTTP error (e.g., 403, 429, 500)
logger.error("WebSocket connection rejected: status %s", e)
if self.on_disconnect:
self.on_disconnect(f"status_code:{e.status_code}")
except ConnectionClosed as e:
logger.warning(
"WebSocket closed during connect: code=%s reason=%s",
e.code,
e.reason,
)
if self.on_disconnect:
self.on_disconnect(f"closed:{e.code}")
except Exception as e:
logger.error("WebSocket connection error: %s", e)
self._session_stats["consecutive_failures"] += 1
if self.on_disconnect:
self.on_disconnect(f"error:{type(e).__name__}")
# Compute sleep interval with jitter
sleep_time = self._reconnect_policy.sleep_interval(attempt)
logger.info(
"Retrying in %.1f seconds (attempt %d)",
sleep_time,
attempt + 1,
)
await asyncio.sleep(sleep_time)
attempt += 1
logger.error(
"Max reconnection attempts reached (%s) or shutdown requested",
self._reconnect_policy.max_attempts,
)
return False
async def _message_loop(self):
"""Main message consumption loop with heartbeat monitoring."""
try:
async for message in self._ws:
self._session_stats["messages_received"] += 1
self._session_stats["last_message_at"] = datetime.now(timezone.utc)
if self.on_message:
try:
data = json.loads(message)
self.on_message(data)
except json.JSONDecodeError:
logger.warning("Received non-JSON message: %s", message[:200])
except ConnectionClosed as e:
logger.warning(
"WebSocket closed during message loop: code=%s reason=%s",
e.code,
e.reason,
)
if self.on_disconnect:
self.on_disconnect(f"loop_closed:{e.code}")
async def _heartbeat_monitor(self, interval: float = 30.0):
"""
Background coroutine: monitors last_message_at and triggers
alerts if the gap exceeds the threshold.
⚠️ This detects "TCP half-open" conditions — where the TCP
connection is alive but the server is no longer sending data.
"""
ALERT_THRESHOLD_SECONDS = 45.0 # Alert if no message for 45s
while self._running:
await asyncio.sleep(interval)
if self._ws is None:
continue
last_msg = self._session_stats["last_message_at"]
if last_msg is None:
continue
gap = (datetime.now(timezone.utc) - last_msg).total_seconds()
if gap > ALERT_THRESHOLD_SECONDS:
logger.error(
"HEARTBEAT ALERT: No TickDB message for %.1f seconds "
"(last received: %s)",
gap,
last_msg.isoformat(),
)
if self.on_alert:
self.on_alert({
"type": "heartbeat_timeout",
"gap_seconds": gap,
"last_message_at": last_msg.isoformat(),
"threshold_seconds": ALERT_THRESHOLD_SECONDS,
})
async def run(self):
"""
Main run loop. Starts the reconnect loop and message consumer.
Restarts the message loop automatically on disconnect.
"""
self._running = True
logger.info("TickDB Monitor starting...")
# Start heartbeat monitor as background task
heartbeat_task = asyncio.create_task(self._heartbeat_monitor())
while self._running:
connected = await self._connect_with_reconnect()
if not connected:
logger.error("Failed to connect after all retry attempts. Shutting down.")
break
# Run message loop. On disconnect, the loop exits and we retry.
try:
await self._message_loop()
except Exception as e:
logger.exception("Unexpected error in message loop: %s", e)
self._running = False
heartbeat_task.cancel()
try:
await heartbeat_task
except asyncio.CancelledError:
pass
logger.info("TickDB Monitor stopped.")
async def stop(self):
"""Graceful shutdown signal."""
logger.info("Shutdown signal received.")
self._running = False
if self._ws:
await self._ws.close(code=1000, reason="client_shutdown")
3.4 Rate-Limit Handling
When TickDB returns a rate-limit response (HTTP 429 or WebSocket code: 3001), the server includes a Retry-After header. The reconnect policy reads this header and waits the specified duration before retrying, overriding the backoff calculation:
async def _connect_with_reconnect(self) -> bool:
attempt = 0
retry_after_override: Optional[float] = None
while self._running and self._reconnect_policy.should_retry(attempt):
# ... connection attempt ...
except InvalidStatusCode as e:
if e.status_code == 429:
# Rate limited — respect Retry-After from server
retry_after_override = float(
e.response_headers.get("Retry-After", 60)
)
logger.warning(
"Rate limited. Server requests wait of %.1f seconds",
retry_after_override,
)
# Use server-specified delay if rate-limited, otherwise use backoff
sleep_time = (
retry_after_override
if retry_after_override is not None
else self._reconnect_policy.sleep_interval(attempt)
)
await asyncio.sleep(sleep_time)
retry_after_override = None # Reset after use
attempt += 1
4. Delay Detection: The Sliding Window Anomaly Score
A stable connection is necessary but not sufficient. The data can arrive with increasing latency — a slow bleed that degrades strategy performance silently. We need a delay monitoring layer that measures the age of each message and scores the current latency regime.
4.1 The Timestamp Problem
TickDB WebSocket messages carry a server-side ts timestamp (Unix milliseconds). The monitoring client measures the local receive time when the message is deserialized. The difference between server timestamp and local receive time is the one-way network latency.
latency_i = (local_receive_time_i - server_timestamp_i) / 1000 # seconds
This is not perfectly accurate (clock synchronization matters), but it is directionally correct and does not require synchronized clocks if we use a sliding window to detect changes in latency rather than absolute values.
4.2 The Sliding Window Anomaly Score
For each message, we compute the rolling mean and standard deviation of the last N latency samples. If the current latency exceeds rolling_mean + k * rolling_std, we score it as anomalous. The anomaly score is the count of consecutive anomalous samples:
from collections import deque
from dataclasses import dataclass
from datetime import datetime, timezone
import time
@dataclass
class LatencyMonitor:
"""
Sliding window latency anomaly detector.
Maintains a rolling window of latency samples and computes a
Z-score for each new sample. Consecutive high-Z samples increase
the anomaly score, triggering an alert when the threshold is met.
"""
window_size: int = 100 # Number of samples in rolling window
z_threshold: float = 2.5 # Z-score threshold for anomaly flag
consecutive_threshold: int = 3 # Consecutive anomalies before alert
sample_ttl_seconds: float = 300.0 # Discard samples older than this
_latencies: deque = field(default_factory=deque)
_timestamps: deque = field(default_factory=deque)
_anomaly_score: int = 0
_last_alert_at: Optional[datetime] = None
_alert_cooldown_seconds: float = 300.0 # 5-minute cooldown between alerts
@property
def rolling_mean(self) -> Optional[float]:
if not self._latencies:
return None
return sum(self._latencies) / len(self._latencies)
@property
def rolling_std(self) -> Optional[float]:
if len(self._latencies) < 2:
return None
mean = self.rolling_mean
variance = sum((x - mean) ** 2 for x in self._latencies) / len(self._latencies)
return variance ** 0.5
def record(self, server_timestamp_ms: int) -> dict:
"""
Record a new latency sample and return the analysis result.
Returns a dict with: latency, z_score, is_anomaly, anomaly_score, alert.
"""
local_now = time.time()
server_now = server_timestamp_ms / 1000.0
latency = local_now - server_now # One-way latency in seconds
# Prune stale samples from the window
cutoff = local_now - self.sample_ttl_seconds
while self._timestamps and self._timestamps[0] < cutoff:
self._timestamps.popleft()
self._latencies.popleft()
self._latencies.append(latency)
self._timestamps.append(local_now)
# Keep window bounded
if len(self._latencies) > self.window_size:
self._latencies.popleft()
self._timestamps.popleft()
result = {
"latency": round(latency, 4),
"z_score": None,
"is_anomaly": False,
"anomaly_score": self._anomaly_score,
"alert": None,
}
mean = self.rolling_mean
std = self.rolling_std
if mean is not None and std is not None and std > 0:
z = (latency - mean) / std
result["z_score"] = round(z, 3)
result["rolling_mean"] = round(mean, 4)
result["rolling_std"] = round(std, 4)
if z > self.z_threshold:
self._anomaly_score += 1
result["is_anomaly"] = True
result["anomaly_score"] = self._anomaly_score
else:
# Decay: reset score on a healthy sample
self._anomaly_score = max(0, self._anomaly_score - 1)
# Alert if consecutive threshold is met and cooldown has passed
if (
self._anomaly_score >= self.consecutive_threshold
and self._should_fire_alert()
):
result["alert"] = {
"type": "latency_anomaly",
"anomaly_score": self._anomaly_score,
"current_latency": round(latency, 3),
"rolling_mean": round(mean, 3),
"rolling_std": round(std, 3),
"z_score": round(z, 3),
"window_size": len(self._latencies),
}
self._last_alert_at = datetime.now(timezone.utc)
return result
def _should_fire_alert(self) -> bool:
if self._last_alert_at is None:
return True
elapsed = (datetime.now(timezone.utc) - self._last_alert_at).total_seconds()
return elapsed >= self._alert_cooldown_seconds
4.3 Integration with the Monitor
The on_message callback feeds every incoming message through the latency monitor:
async def main():
latency_monitor = LatencyMonitor(
window_size=100,
z_threshold=2.5,
consecutive_threshold=3,
)
async def handle_message(data: dict):
# Extract server timestamp from TickDB message
server_ts = data.get("ts", data.get("timestamp"))
if server_ts:
result = latency_monitor.record(server_ts)
if result.get("alert"):
logger.warning(
"Latency anomaly detected: z=%.2f, score=%d, latency=%.3fs",
result["z_score"],
result["anomaly_score"],
result["latency"],
)
# Forward to alert manager
await alert_manager.send(result["alert"])
monitor = TickDBMonitor(
symbols=["AAPL.US", "NVDA.US"],
channels=["depth"],
on_message=handle_message,
)
await monitor.run()
5. Alert Delivery: Feishu Webhook Integration
An alert that does not reach a human is not an alert. We use Feishu (Lark) Custom Robots — webhook-based push notifications that work without OAuth, enterprise-only features, or third-party integrations.
5.1 Feishu Custom Robot Setup
- Open a Feishu group → Settings → Group Bots → Add Bot → Custom Robot.
- Give it a name (e.g., "TickDB Monitor Alert Bot") and optionally an avatar.
- Copy the webhook URL — it looks like:
https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx - Store it as
FEISHU_WEBHOOK_URLin your environment variables.
Security note: The webhook URL is a secret. Treat it like an API key. If it is leaked, anyone can post to your group. Rotate it immediately if compromised.
5.2 Feishu Card Message Format
Feishu supports Interactive Cards — structured messages with color-coded severity, action buttons, and markdown content. We use cards instead of plain text because they are actionable: a responder can click "Acknowledge" or "Snooze 1h" directly from the notification.
import aiohttp
import asyncio
import os
import hashlib
import time
from typing import Optional
from datetime import datetime, timezone
class FeishuAlertPusher:
"""
Feishu (Lark) Custom Robot alert pusher.
Sends structured Interactive Card messages with severity coloring,
markdown body, and deduplication to prevent alert storms.
"""
def __init__(self, webhook_url: Optional[str] = None):
self.webhook_url = webhook_url or os.environ.get("FEISHU_WEBHOOK_URL")
if not self.webhook_url:
raise ValueError(
"Feishu webhook URL required. "
"Set FEISHU_WEBHOOK_URL environment variable."
)
self._sent_hashes: dict[str, float] = {}
self._dedup_window_seconds: float = 600.0 # 10-minute dedup window
self._send_errors: int = 0
self._last_send_at: Optional[float] = None
self._rate_limit_seconds: float = 5.0 # Min interval between sends
def _content_hash(self, alert: dict) -> str:
"""Stable hash for deduplication: type + primary identifiers."""
key_parts = [
alert.get("type", ""),
str(alert.get("symbol", "")),
str(alert.get("gap_seconds", alert.get("latency", ""))),
]
return hashlib.sha256("|".join(key_parts).encode()).hexdigest()[:16]
def _is_duplicate(self, alert: dict) -> bool:
"""Check if this alert was sent within the dedup window."""
h = self._content_hash(alert)
now = time.time()
if h in self._sent_hashes:
age = now - self._sent_hashes[h]
if age < self._dedup_window_seconds:
return True
self._sent_hashes[h] = now
# Prune old entries
self._sent_hashes = {
k: v for k, v in self._sent_hashes.items() if now - v < self._dedup_window_seconds
}
return False
def _severity_config(self, alert: dict) -> tuple[str, str, str]:
"""
Return (color_hex, emoji, level_tag) based on alert type.
Feishu card colors: red, orange, yellow, green, blue, purple
"""
alert_type = alert.get("type", "")
if alert_type == "heartbeat_timeout":
if alert.get("gap_seconds", 0) > 120:
return "red", "🔴", "CRITICAL"
return "orange", "🟠", "WARNING"
elif alert_type == "latency_anomaly":
return "yellow", "🟡", "WARNING"
elif alert_type == "connection_lost":
return "red", "🔴", "CRITICAL"
return "blue", "🔵", "INFO"
def _build_card(self, alert: dict) -> dict:
"""Build a Feishu Interactive Card payload."""
color, emoji, level_tag = self._severity_config(alert)
alert_type = alert.get("type", "unknown")
now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
# Build card content based on alert type
elements = []
# Header line
elements.append({
"tag": "markdown",
"content": f"**{emoji} [{level_tag}] {alert_type.replace('_', ' ').title()}** \n{now_str}",
})
if alert_type == "heartbeat_timeout":
elements.extend([
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Gap:** {alert['gap_seconds']:.1f} seconds"}},
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Last message:** {alert.get('last_message_at', 'N/A')}"}},
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Threshold:** {alert.get('threshold_seconds', 'N/A')} seconds"}},
])
elif alert_type == "latency_anomaly":
elements.extend([
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Anomaly score:** {alert['anomaly_score']} (threshold: 3)"}},
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Current latency:** {alert.get('current_latency', 0):.3f}s"}},
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Rolling mean:** {alert.get('rolling_mean', 0):.3f}s | **Rolling std:** {alert.get('rolling_std', 0):.3f}s"}},
{"tag": "div", "text": {"tag": "lark_md", "content": f"**Z-score:** {alert.get('z_score', 0):.2f}"}},
])
elif alert_type == "connection_lost":
elements.append({
"tag": "div",
"text": {"tag": "lark_md", "content": f"**Reason:** {alert.get('reason', 'unknown')}"},
})
elements.extend([
{"tag": "hr"},
{
"tag": "note",
"elements": [
{"tag": "plain_text", "content": "TickDB 7x24 Monitor • Automated alert"},
],
},
])
return {
"msg_type": "interactive",
"card": {
"header": {
"title": {"tag": "plain_text", "content": f"{emoji} TickDB Monitor Alert"},
"template": color,
},
"elements": elements,
},
}
async def send(self, alert: dict) -> bool:
"""
Send an alert to Feishu. Returns True if sent, False if suppressed
(duplicate, rate-limited, or network error).
"""
# Deduplication check
if self._is_duplicate(alert):
return False
# Rate-limit check
now = time.time()
if self._last_send_at and (now - self._last_send_at) < self._rate_limit_seconds:
return False
self._last_send_at = now
card = self._build_card(alert)
try:
async with aiohttp.ClientSession() as session:
async with session.post(
self.webhook_url,
json=card,
headers={"Content-Type": "application/json"},
timeout=aiohttp.ClientTimeout(total=10),
) as resp:
if resp.status == 200:
return True
body = await resp.text()
raise RuntimeError(f"Feishu API error {resp.status}: {body}")
except aiohttp.ClientError as e:
self._send_errors += 1
return False
5.3 Alert Routing
In a production system, alerts need routing rules. The same monitoring system monitors 20 symbols — you do not want a latency spike on AAPL.US to page the person responsible for HK equities:
class AlertRouter:
"""
Route alerts to appropriate Feishu groups based on symbol region.
"""
def __init__(self):
self.routes: dict[str, str] = {} # region -> webhook_url
self.default_webhook: Optional[str] = None
def register(self, region: str, webhook_url: str):
self.routes[region] = webhook_url
def register_default(self, webhook_url: str):
self.default_webhook = webhook_url
def route(self, alert: dict) -> Optional[str]:
symbol = alert.get("symbol", "")
for region, webhook in self.routes.items():
if region in symbol:
return webhook
return self.default_webhook
async def main():
router = AlertRouter()
router.register("US", os.environ["FEISHU_WEBHOOK_US"])
router.register("HK", os.environ["FEISHU_WEBHOOK_HK"])
router.register_default(os.environ["FEISHU_WEBHOOK_DEFAULT"])
async def handle_alert(alert: dict):
webhook = router.route(alert)
if webhook:
pusher = FeishuAlertPusher(webhook_url=webhook)
sent = await pusher.send(alert)
if sent:
logger.info("Alert routed and sent: type=%s", alert["type"])
else:
logger.info("Alert suppressed (dedup or rate-limit): type=%s", alert["type"])
6. Putting It All Together: The Complete Daemon
import asyncio
import os
import logging
from tickdb_monitor import TickDBMonitor, LatencyMonitor
from feishu_pusher import FeishuAlertPusher, AlertRouter
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("daemon")
async def main():
# ── Configuration ──────────────────────────────────────────────
# Symbols grouped by market
symbols_by_market = {
"US": ["AAPL.US", "NVDA.US", "TSLA.US", "MSFT.US"],
"HK": ["0700.HK", "9988.HK", "9618.HK"],
}
all_symbols = sum(symbols_by_market.values(), [])
# ── Alert routing ──────────────────────────────────────────────
router = AlertRouter()
if "FEISHU_WEBHOOK_US" in os.environ:
router.register("US", os.environ["FEISHU_WEBHOOK_US"])
if "FEISHU_WEBHOOK_HK" in os.environ:
router.register("HK", os.environ["FEISHU_WEBHOOK_HK"])
router.register_default(os.environ.get("FEISHU_WEBHOOK_URL", ""))
# ── Latency monitors (one per market region) ───────────────────
latency_monitors = {
region: LatencyMonitor(window_size=100, z_threshold=2.5, consecutive_threshold=3)
for region in symbols_by_market
}
# ── Per-symbol tracking ────────────────────────────────────────
symbol_last_ts = {s: None for s in all_symbols}
def handle_message(data: dict):
"""Process every incoming TickDB message."""
symbol = data.get("symbol", "unknown")
server_ts = data.get("ts")
if not server_ts:
return
symbol_last_ts[symbol] = server_ts
# Route to correct region monitor
region = next((r for r, syms in symbols_by_market.items() if symbol in syms), None)
if region:
result = latency_monitors[region].record(server_ts)
if result.get("alert"):
alert = result["alert"]
alert["symbol"] = symbol
# Dispatch via router
webhook = router.route(alert)
if webhook:
pusher = FeishuAlertPusher(webhook_url=webhook)
asyncio.create_task(pusher.send(alert))
def handle_alert(alert: dict):
"""Handle system-level alerts (connection, heartbeat)."""
webhook = router.route(alert)
if webhook:
pusher = FeishuAlertPusher(webhook_url=webhook)
asyncio.create_task(pusher.send(alert))
# ── Start the daemon ───────────────────────────────────────────
monitor = TickDBMonitor(
symbols=all_symbols,
channels=["depth", "kline"],
on_message=handle_message,
on_alert=handle_alert,
)
logger.info(
"TickDB 7x24 Monitor starting. Monitoring %d symbols across %d regions.",
len(all_symbols),
len(symbols_by_market),
)
try:
await monitor.run()
except KeyboardInterrupt:
logger.info("Interrupted. Shutting down gracefully.")
await monitor.stop()
if __name__ == "__main__":
asyncio.run(main())
7. Deployment: Surviving Three Years Without a Restart
The code is production-grade. The deployment needs to match.
7.1 Process Supervisor
On a Linux VPS, use systemd to manage the daemon. This gives you automatic restart on crash, log rotation, and startup on boot:
# /etc/systemd/system/tickdb-monitor.service
[Unit]
Description=TickDB 7x24 Market Monitor
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=deploy
WorkingDirectory=/opt/tickdb-monitor
Environment="TICKDB_API_KEY=your_key_here"
Environment="FEISHU_WEBHOOK_US=https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
Environment="FEISHU_WEBHOOK_HK=https://open.feishu.cn/open-apis/bot/v2/hook/yyy"
ExecStart=/opt/tickdb-monitor/venv/bin/python main.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
# Hardening: restart on OOM, limit memory
MemoryMax=256M
OOMPolicy=restart
[Install]
WantedBy=multi-user.target
Enable and start:
sudo systemctl enable tickdb-monitor
sudo systemctl start tickdb-monitor
sudo journalctl -u tickdb-monitor -f # Follow logs
7.2 Memory Leak Prevention
The monitoring daemon must run for months. Two common memory leak sources:
Unbounded queue growth: If the Feishu API is rate-limited, the asyncio.create_task(pusher.send(alert)) calls accumulate in the event loop's pending task queue. Use a bounded semaphore to limit concurrent sends:
MAX_CONCURRENT_SENDS = 3
_send_semaphore = asyncio.Semaphore(MAX_CONCURRENT_SENDS)
async def handle_alert(alert: dict):
webhook = router.route(alert)
if webhook:
pusher = FeishuAlertPusher(webhook_url=webhook)
async with _send_semaphore:
await pusher.send(alert)
Datetime object retention: The _timestamps deque in LatencyMonitor stores time.time() floats — not datetime objects. Datetime objects carry timezone overhead. Using raw Unix timestamps keeps the deque lean.
7.3 Monitoring the Monitor
You need to monitor the monitoring system. Add a health endpoint that reports connection stats:
from aiohttp import web
async def health(request):
stats = monitor._session_stats
return web.json_response({
"status": "running" if monitor._running else "stopped",
"uptime_seconds": (datetime.now(timezone.utc) - start_time).total_seconds(),
**stats,
})
app = web.Application()
app.router.add_get("/health", health)
web.run_app(app, host="0.0.0.0", port=8080)
Expose this to your infrastructure monitoring (Grafana, DataDog, or a simple cron check) and alert if /health returns status: stopped or messages_received has not incremented in 5 minutes.
8. Summary: What We Built
We built a monitoring daemon that survives the infrastructure failures that kill other monitoring systems:
| Problem | Solution | Code module |
|---|---|---|
| Network partition | Exponential backoff + full jitter reconnection | ReconnectPolicy, _connect_with_reconnect() |
| TCP half-open | Ping/pong heartbeat + timeout monitoring | ping_interval, _heartbeat_monitor() |
| Rate-limit storm | Retry-After header override + backoff reset |
Rate-limit handling block |
| Latency degradation | Sliding window Z-score anomaly detection | LatencyMonitor.record() |
| Alert fatigue | Deduplication hash + per-alert cooldown | FeishuAlertPusher._is_duplicate() |
| Memory leak (OOM) | Bounded semaphore + raw Unix timestamps | main() semaphore, _timestamps deque |
| Alert routing | Region-based routing to separate Feishu groups | AlertRouter |
The system is designed to run on a single $6/month VPS under systemd supervision. The code is the infrastructure. The discipline is in the code.
Next Steps
If you want to run this monitoring system against real market data today:
- Sign up at tickdb.ai and generate an API key (free tier available — no credit card required)
- Set
TICKDB_API_KEY,FEISHU_WEBHOOK_US, andFEISHU_WEBHOOK_HKenvironment variables - Copy the complete
main.pyand dependency modules from this article - Run
python main.py— the daemon will connect to TickDB and begin monitoring
If you need monitoring for multiple strategy portfolios:
The AlertRouter pattern scales to N regions. Add a FEISHU_WEBHOOK_STRATEGY_ALPHA environment variable and route by strategy prefix in the symbol list. Each strategy team gets their own Feishu group and alert feed.
If you are running on a high-availability cluster:
Replace the systemd supervisor with a Kubernetes Deployment using restartPolicy: Always and a livenessProbe that hits /health. Add a PodDisruptionBudget to ensure at least one replica survives node drain.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.