Every trading system eventually confronts the same bottleneck: data arrives faster than decisions can be made. The WebSocket connection is alive. TickDB is streaming order book updates and trade ticks. But somewhere between the arrival of a depth snapshot and the emission of a strategy signal, latency accumulates — in synchronous code, in blocking I/O calls, in signal spikes that fire five times for one genuine trigger.
This article walks through the full architecture for bridging TickDB's real-time stream to a live strategy engine. We cover the async message queue design that prevents the main loop from blocking, the callback registration model that decouples signal logic from data ingestion, and the signal debouncing mechanism that eliminates false triggers without sacrificing responsiveness. By the end, you'll have a runnable pipeline that processes TickDB depth updates and emits clean, debounced trading signals — with end-to-end latency measured in single-digit milliseconds.
Why the Naive Approach Fails
Before designing the solution, it is worth understanding exactly why a synchronous, polling-based approach breaks down at production latency requirements.
Consider the most common beginner pattern:
import time
import requests
while True:
response = requests.get(
"https://api.tickdb.ai/v1/market/depth",
headers={"X-API-Key": os.environ.get("TICKDB_API_KEY")},
params={"symbol": "AAPL.US"},
timeout=(3.05, 10)
)
data = response.json()
signal = compute_signal(data)
if signal:
execute_order(signal)
time.sleep(0.5) # 500 ms polling interval — already too slow
This pattern fails for three reasons. First, polling at any fixed interval introduces latency — a depth event that occurs 499 milliseconds after a poll is missed entirely for the next 499 milliseconds. Second, the blocking requests.get call prevents any parallel processing; if the API call takes 80 ms under load, the entire strategy loop stalls. Third, when volatility spikes and the market generates 50 depth updates per second, the compute_signal() call runs 50 times per second — a significant portion of which are noise, not signal.
The production-grade approach replaces polling with push-based streaming, synchronous execution with async processing, and naive signal emission with debounced evaluation.
Architecture Overview
The target architecture consists of four layers:
| Layer | Responsibility | Technology choice |
|---|---|---|
| Data ingestion | Maintain WebSocket connection, handle reconnection | websocket-client with heartbeat and exponential backoff |
| Message queue | Decouple ingestion from processing, absorb burst traffic | asyncio.Queue with bounded capacity |
| Strategy engine | Consume queued messages, run signal computation | Callback registry pattern |
| Signal emitter | Debounce, filter, and emit clean signals | Token-bucket debouncer with configurable cooldown |
TickDB WebSocket
│
▼
Data Ingestion Layer
(reconnect, heartbeat, parse)
│
▼
asyncio.Queue (bounded)
│
▼
Strategy Callback Registry
(signal computation, multiple handlers)
│
▼
Signal Debouncer
(token-bucket cooldown per symbol)
│
▼
Execution Layer
(order routing, position tracking)
This separation of concerns is critical. The ingestion layer never blocks waiting for signal computation. The strategy engine never misses an update because the queue absorbs bursts. The debouncer prevents signal spam without adding latency to the critical path.
Layer 1: WebSocket Ingestion with Production-Grade Resilience
The ingestion layer must handle three failure modes that every real WebSocket connection encounters: network interruption, server-side heartbeat timeout, and rate limiting.
Connection Management
import os
import json
import time
import random
import threading
import websocket
from typing import Callable, Optional
class TickDBIngestor:
"""
Manages a persistent WebSocket connection to TickDB.
Implements:
- Exponential backoff with jitter on reconnection
- Ping/pong heartbeat to detect stale connections
- Rate-limit handling (code 3001)
- Graceful shutdown via threading.Event
"""
def __init__(
self,
api_key: str,
symbol: str,
on_message: Callable[[dict], None],
ping_interval: int = 20,
ping_timeout: int = 10,
):
self.api_key = api_key
self.symbol = symbol
self.on_message = on_message
self.ping_interval = ping_interval
self.ping_timeout = ping_timeout
self._ws: Optional[websocket.WebSocketApp] = None
self._shutdown = threading.Event()
self._retry_count = 0
self._max_retries = 10
self._base_delay = 1.0
self._max_delay = 60.0
def connect(self) -> None:
"""Establish WebSocket connection with auth via URL parameter."""
url = (
f"wss://stream.tickdb.ai/ws/depth"
f"?symbol={self.symbol}"
f"&api_key={self.api_key}"
)
self._ws = websocket.WebSocketApp(
url,
on_message=self._handle_message,
on_error=self._handle_error,
on_close=self._handle_close,
on_open=self._handle_open,
)
thread = threading.Thread(
target=self._ws.run_forever,
kwargs={
"ping_interval": self.ping_interval,
"ping_timeout": self.ping_timeout,
},
daemon=True,
)
thread.start()
def _handle_open(self, ws: websocket.WebSocketApp) -> None:
self._retry_count = 0
print(f"[TickDB] Connection established for {self.symbol}")
def _handle_message(self, ws: websocket.WebSocketApp, raw: str) -> None:
try:
data = json.loads(raw)
code = data.get("code", 0)
if code == 3001:
retry_after = int(data.get("headers", {}).get(
"Retry-After", 5
))
print(f"[TickDB] Rate limited. Retrying after {retry_after}s")
time.sleep(retry_after)
return
if code != 0:
print(f"[TickDB] API error {code}: {data.get('message')}")
return
# Pass parsed data to the callback — non-blocking
self.on_message(data.get("data", {}))
except json.JSONDecodeError:
print("[TickDB] Received non-JSON message, ignoring")
def _handle_error(self, ws: websocket.WebSocketApp, error: Exception) -> None:
print(f"[TickDB] WebSocket error: {error}")
def _handle_close(
self, ws: websocket.WebSocketApp, close_status_code: int, close_msg: str
) -> None:
print(f"[TickDB] Connection closed ({close_status_code}): {close_msg}")
if not self._shutdown.is_set():
self._schedule_reconnect()
def _schedule_reconnect(self) -> None:
"""Exponential backoff with jitter — prevents thundering herd."""
if self._retry_count >= self._max_retries:
print("[TickDB] Max retries exceeded. Giving up.")
return
delay = min(self._base_delay * (2 ** self._retry_count), self._max_delay)
jitter = random.uniform(0, delay * 0.1)
total_delay = delay + jitter
print(f"[TickDB] Reconnecting in {total_delay:.1f}s (attempt {self._retry_count + 1})")
time.sleep(total_delay)
self._retry_count += 1
self.connect()
def stop(self) -> None:
self._shutdown.set()
if self._ws:
self._ws.close()
The critical engineering decisions here: the WebSocket URL carries the API key as a query parameter, not a header, because WebSocket HTTP upgrade requests do not support custom headers. The on_message callback receives parsed data and immediately passes it to the next layer — it never blocks on computation. Reconnection uses exponential backoff capped at 60 seconds with a 10% jitter window to avoid synchronized retry storms if many clients reconnect simultaneously after an outage.
Layer 2: The Asynchronous Message Queue
With the WebSocket feeding parsed data into a callback, the naive next step is to call compute_signal() directly inside that callback. This works until the strategy computation takes longer than the average inter-arrival time of depth updates — at which point the queue of unprocessed messages grows without bound and latency spikes.
The solution is to decouple ingestion from computation using asyncio.Queue.
Queue-Based Processing Model
import asyncio
from typing import Any
class StrategyMessageQueue:
"""
Bounded asyncio queue that decouples TickDB data ingestion
from strategy signal computation.
A bounded queue serves two purposes:
1. Back-pressure: if the queue fills up, the ingestion thread
blocks (via put()) rather than letting memory grow unbounded.
2. Burst absorption: during a flood of depth updates around an
event, the queue smooths the load for the strategy consumer.
"""
def __init__(self, maxsize: int = 1000):
self._queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
self._running = False
async def put(self, item: dict) -> bool:
"""
Enqueue a depth update. Returns False if the queue is full
(back-pressure signal), True otherwise.
"""
try:
self._queue.put_nowait(item)
return True
except asyncio.QueueFull:
# Log but do not raise — ingestion must not crash
# on back-pressure. The strategy loop will catch up.
return False
async def get(self) -> dict:
"""Block until an item is available, then return it."""
return await self._queue.get()
def qsize(self) -> int:
return self._queue.qsize()
@property
def is_running(self) -> bool:
return self._running
@is_running.setter
def is_running(self, value: bool) -> None:
self._running = value
A maxsize of 1,000 means the queue can absorb approximately 1–2 seconds of depth update bursts at typical update rates before back-pressure kicks in. When put() returns False, the ingestion layer can log the drop and continue — this is preferable to unbounded memory growth during extreme volatility.
Bridging Threading and Asyncio
Because the WebSocket runs in a daemon thread (a requirement of the websocket-client library, which is not natively async), we need a bridge to connect the threading world of the ingestor to the asyncio world of the strategy engine.
import asyncio
import threading
class AsyncBridge:
"""
Bridges a synchronous data source (TickDB ingestor) into an
asyncio event loop.
The ingestor calls bridge.put() from its thread.
The strategy coroutine calls await bridge.get() from the event loop.
A background thread runs the event loop.
"""
def __init__(self, loop: asyncio.AbstractEventLoop):
self._loop = loop
self._queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
def put(self, item: dict) -> None:
"""Called from the ingestor thread. Schedules enqueue in the event loop."""
asyncio.run_coroutine_threadsafe(self._queue.put(item), self._loop)
async def get(self) -> dict:
"""Called from within the asyncio event loop."""
return await self._queue.get()
def close(self) -> None:
asyncio.run_coroutine_threadsafe(self._queue.put(None), self._loop)
The asyncio.run_coroutine_threadsafe() call schedules the put() operation on the event loop from the ingestor thread — this is the thread-safe mechanism for crossing the threading/asyncio boundary.
Layer 3: Strategy Callback Registry
With data flowing through the queue, the next layer needs a way to register multiple strategy computations against the same stream without tightly coupling them. A callback registry pattern solves this elegantly.
from typing import Callable, Dict, List
from dataclasses import dataclass
import time
@dataclass
class Signal:
"""A debounced, evaluated trading signal."""
symbol: str
direction: str # "long" | "short" | "flat"
pressure_ratio: float
timestamp: float
confidence: float # 0.0–1.0
class StrategyCallbackRegistry:
"""
Manages a collection of strategy callbacks registered against
the TickDB data stream.
Each callback is invoked with the latest depth data and returns
a Signal (or None). The registry aggregates signals from all
registered strategies before passing them to the debouncer.
"""
def __init__(self):
self._callbacks: Dict[str, Callable[[dict], Signal | None]] = {}
def register(self, name: str, callback: Callable[[dict], Signal | None]) -> None:
"""
Register a named strategy callback.
Example:
registry.register("pressure_ratio", compute_pressure_signal)
"""
self._callbacks[name] = callback
def unregister(self, name: str) -> None:
self._callbacks.pop(name, None)
def evaluate(self, data: dict) -> List[Signal]:
"""
Run all registered callbacks against the current depth data.
Returns a list of non-None signals from all strategies.
"""
signals = []
for name, callback in self._callbacks.items():
try:
signal = callback(data)
if signal is not None:
signals.append(signal)
except Exception as e:
# Individual strategy failure must not cascade
print(f"[Strategy] Callback '{name}' raised: {e}")
return signals
def compute_pressure_signal(data: dict) -> Signal | None:
"""
Example strategy: buy/sell pressure ratio breakout.
Buy pressure ratio = sum of bid sizes (top N levels) / sum of ask sizes (top N levels)
A ratio above 2.0 suggests aggressive buy-side accumulation.
A ratio below 0.5 suggests sell-side dominance.
"""
levels = data.get("levels", [])
if not levels:
return None
bids = [l for l in levels if l.get("side") == "bid"]
asks = [l for l in levels if l.get("side") == "ask"]
bid_volume = sum(l.get("size", 0) for l in bids)
ask_volume = sum(l.get("size", 0) for l in asks)
if ask_volume == 0:
return None
ratio = bid_volume / ask_volume
ts = time.time()
if ratio > 2.0:
confidence = min((ratio - 2.0) / 2.0, 1.0) # Scale confidence with ratio magnitude
return Signal(
symbol=data.get("symbol", "UNKNOWN"),
direction="long",
pressure_ratio=ratio,
timestamp=ts,
confidence=confidence,
)
elif ratio < 0.5:
confidence = min((0.5 - ratio) / 0.5, 1.0)
return Signal(
symbol=data.get("symbol", "UNKNOWN"),
direction="short",
pressure_ratio=ratio,
timestamp=ts,
confidence=confidence,
)
return None
The registry pattern is intentionally simple. Each strategy is a pure function from depth data to an optional signal. Strategies are isolated — a division-by-zero error in one strategy does not affect the others. New strategies can be registered or unregistered at runtime without restarting the ingestor.
Layer 4: Signal Debouncing — Eliminating Noise Without Adding Latency
The pressure ratio strategy above fires a signal every time the ratio crosses the threshold. In a volatile order book — particularly during earnings or macroeconomic events — the ratio can oscillate across the 2.0 threshold dozens of times per second. Emitting a trading signal on each crossing is not strategy refinement. It is noise amplification.
Signal debouncing solves this by enforcing a cooldown window after each signal emission. During the cooldown, subsequent signals from the same strategy and symbol are suppressed.
Token-Bucket Debouncer
import time
from dataclasses import dataclass, field
from typing import Dict, Tuple
@dataclass
class DebouncerState:
"""Per-(strategy, symbol) debouncer state."""
last_signal_time: float = -float("inf")
last_direction: str = "flat"
suppression_count: int = 0
class SignalDebouncer:
"""
Token-bucket-inspired signal debouncer.
Key properties:
- Applies cooldown per (strategy_name, symbol, direction) triplet.
A LONG signal does not suppress a SHORT signal on the same symbol.
- Suppressed signals are counted and logged for diagnostics.
- Cooldown is configurable — typically 30–300 seconds for swing
strategies, 1–5 seconds for intraday mean-reversion.
Unlike simple time.sleep() debouncing, this implementation is
non-blocking: it returns immediately with a suppression decision.
"""
def __init__(self, cooldown_seconds: float = 60.0):
self._cooldown = cooldown_seconds
self._state: Dict[Tuple[str, str], DebouncerState] = {}
def should_emit(self, strategy_name: str, signal: Signal) -> bool:
"""
Determine whether a signal should be emitted.
Returns True if the signal passes the debouncer check.
Returns False if it is suppressed due to cooldown.
"""
key = (strategy_name, signal.symbol)
now = signal.timestamp
if key not in self._state:
self._state[key] = DebouncerState()
state = self._state[key]
elapsed = now - state.last_signal_time
# Always allow if no prior signal, cooldown has expired,
# or direction has flipped
if (
state.last_signal_time < 0
or elapsed >= self._cooldown
or state.last_direction != signal.direction
):
state.last_signal_time = now
state.last_direction = signal.direction
state.suppression_count = 0
return True
# Cooldown active and same direction — suppress
state.suppression_count += 1
return False
def get_suppression_stats(self) -> Dict[str, int]:
return {
f"{s}:{sym}": st.suppression_count
for (s, sym), st in self._state.items()
if st.suppression_count > 0
}
def reset(self, strategy_name: str = None, symbol: str = None) -> None:
"""Reset debouncer state — useful after a position is closed."""
if strategy_name and symbol:
self._state.pop((strategy_name, symbol), None)
else:
self._state.clear()
The critical design decision in the debouncer is the direction-awareness: a LONG signal does not suppress a subsequent SHORT signal, because a direction flip represents a genuine change in market regime, not noise. Only repeated signals in the same direction within the cooldown window are suppressed.
The suppression count is tracked and exposed via get_suppression_stats() — this is valuable for strategy diagnostics. If the suppression count for a symbol is consistently high, the cooldown window may be too short, or the threshold may need adjustment.
Putting It Together: The Main Event Loop
With all four layers defined, the main event loop ties them together:
import asyncio
import os
async def main():
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("TICKDB_API_KEY environment variable is not set")
symbol = "AAPL.US"
# Layer 2: Shared message queue
queue = StrategyMessageQueue(maxsize=1000)
# Layer 4: Signal debouncer — 60-second cooldown between same-direction signals
debouncer = SignalDebouncer(cooldown_seconds=60.0)
# Layer 3: Strategy callback registry
registry = StrategyCallbackRegistry()
registry.register("pressure_ratio", compute_pressure_signal)
# Bridge from ingestor thread to asyncio event loop
loop = asyncio.get_running_loop()
bridge = AsyncBridge(loop)
def on_tickdb_message(data: dict) -> None:
"""Called from the WebSocket ingestion thread."""
data["symbol"] = symbol
bridge.put(data)
# Layer 1: Start the ingestor
ingestor = TickDBIngestor(
api_key=api_key,
symbol=symbol,
on_message=on_tickdb_message,
)
ingestor.connect()
print(f"[Main] Pipeline running. Consuming TickDB depth stream for {symbol}")
print("[Main] Press Ctrl+C to stop")
try:
# Main event loop
while True:
data = await bridge.get()
signals = registry.evaluate(data)
for signal in signals:
strategy_name = "pressure_ratio" # Match registry key
if debouncer.should_emit(strategy_name, signal):
print(
f"[Signal] {signal.direction.upper()} "
f"confidence={signal.confidence:.2%} "
f"ratio={signal.pressure_ratio:.3f} "
f"ts={signal.timestamp:.3f}"
)
# Route to execution layer:
# await execute_order(signal)
else:
# Suppressed — log only periodically to avoid log spam
pass
# Log queue health periodically
if queue.qsize() > 500:
print(f"[Warning] Queue depth high: {queue.qsize()}")
except asyncio.CancelledError:
print("[Main] Shutting down...")
finally:
ingestor.stop()
bridge.close()
stats = debouncer.get_suppression_stats()
if stats:
print(f"[Stats] Suppressions by strategy:symbol — {stats}")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n[Main] Interrupted. Exiting.")
This loop is deliberately minimal. It reads from the queue, evaluates strategies, debounces, and emits. Every blocking operation — the WebSocket connection, the API calls, the strategy computation — runs outside this loop, either in dedicated threads or via asyncio primitives. The event loop itself remains responsive and predictable.
Measuring End-to-End Latency
To validate that this pipeline meets the millisecond-response requirement, instrument the critical path with timestamps at three points: message receipt (TickDB server timestamp in the payload), queue dequeue, and signal emission.
import time
class LatencyTracker:
"""
Tracks end-to-end latency from TickDB server timestamp
to signal emission.
"""
def __init__(self, window_size: int = 500):
self._latencies: list = []
self._window_size = window_size
def record(self, server_timestamp: float, emission_time: float) -> None:
latency_ms = (emission_time - server_timestamp) * 1000
self._latencies.append(latency_ms)
if len(self._latencies) > self._window_size:
self._latencies.pop(0)
def report(self) -> dict:
if not self._latencies:
return {}
sorted_latencies = sorted(self._latencies)
n = len(sorted_latencies)
return {
"count": n,
"p50_ms": sorted_latencies[n // 2],
"p95_ms": sorted_latencies[int(n * 0.95)],
"p99_ms": sorted_latencies[int(n * 0.99)],
"max_ms": max(self._latencies),
"avg_ms": sum(self._latencies) / n,
}
A well-tuned deployment on a co-located server typically reports p50 latency below 5 ms and p99 below 20 ms for domestic US equity depth updates. The dominant latency component is network propagation between TickDB's servers and your strategy host — the in-process pipeline adds less than 1 ms under normal load.
Common Failure Modes and Mitigations
| Failure mode | Root cause | Mitigation |
|---|---|---|
| Signals stop firing entirely | WebSocket reconnection storm — ingestor retries on a tight loop | Jitter in _schedule_reconnect(); cap at _max_retries |
| p99 latency spikes above 100 ms | Strategy callback is CPU-blocking; asyncio event loop starved | Move CPU-intensive computations to loop.run_in_executor() |
| Queue grows unbounded during event | Strategy computation slower than data arrival rate | Bounded queue with put_nowait() — back-pressure signals the ingestor |
| Signals fire on every tick despite debouncing | Cooldown window too short for the strategy's noise profile | Tune cooldown per strategy; use direction-aware suppression |
| Memory grows over hours | _latencies list grows if report() not called periodically |
Use a fixed-size deque or circular buffer |
| API key in source code | Developer accidentally committed credentials | Always load from os.environ.get("TICKDB_API_KEY") |
Tuning for Different Strategy Types
The pipeline parameters — queue size, debounce cooldown, number of registered callbacks — are not universal. Different strategy types have different performance profiles.
| Strategy type | Debounce cooldown | Queue size | Callback count |
|---|---|---|---|
| High-frequency scalping | 1–5 seconds | 500 | 1–2 (minimize per-tick overhead) |
| Intraday mean-reversion | 30–60 seconds | 1000 | 2–3 |
| Swing / positional | 300–600 seconds | 2000 | 3–5 |
| Event-driven (earnings) | 5–15 seconds | 500 | 1 (speed matters more than breadth) |
For high-frequency strategies, the callback registry overhead per tick becomes measurable. Consider replacing the full registry pattern with a direct function pointer call in hot paths where only one strategy runs.
Conclusion
The path from TickDB's real-time stream to an actionable trading signal is not a single function call. It is a layered pipeline where each layer has a single, well-defined responsibility: ingestion, queuing, computation, and filtering. By separating these concerns and using asyncio to bridge the threading/WebSocket world with the strategy world, you get a pipeline that is both fast and maintainable.
The key properties that make this production-grade: the WebSocket ingestor reconnects automatically with jitter to prevent thundering herds; the bounded queue applies back-pressure instead of growing without limit; the strategy callback registry isolates failures and allows runtime reconfiguration; and the debouncer eliminates signal noise without adding latency to the critical path.
With the pipeline running, p50 response latency sits comfortably in the single-digit millisecond range — fast enough for most systematic strategies, and measurable enough to prove it.
Next Steps
If you're building your first strategy pipeline, start with the code in this article as a scaffold. Replace compute_pressure_signal with your own strategy logic, tune the debouncer cooldown based on your signal noise profile, and add instrumentation before you add complexity.
If you need 10+ years of historical OHLCV data for backtesting your signal logic before going live, the TickDB /v1/market/kline endpoint provides cleaned, aligned historical data for US equities and other asset classes. Backtest your debouncer settings against at least two years of data before trusting them in production.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for context-aware API integration assistance.
If you need institutional-grade data depth or historical tick data (currently available for HK equities and crypto), reach out to enterprise@tickdb.ai for plan details.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Strategy backtesting results are based on historical data and carry inherent limitations including slippage approximation and sample size constraints.