TickDB as a Second Feed: Cross-Validating Market Data Quality

“The feed was green, the queue was empty, and the strategy was still trading stale prices.”

That failure rarely arrives as an outage. More often, the primary feed remains connected while its timestamps advance unevenly, its prices diverge from another venue, or entire update intervals disappear without triggering a socket error.

Production market-data systems therefore need an independent observation point. Comparing two feeds does not make either source infallible, but it converts a silent data-quality failure into an observable and actionable condition.

This article presents a production-oriented pattern for using TickDB as a second market-data source. The implementation compares prices, source timestamps, update cadence, and sequence continuity to identify stale data, delayed updates, missing candles, and cross-source price divergence.

1. Failure Modes That Connection Health Cannot Detect

A WebSocket can remain open while the data inside it becomes economically unusable. The most common failure modes are not socket failures at all.

The first failure mode is stale pricing. The connection continues delivering messages, but the newest trade timestamp stops advancing because an upstream gateway has stalled.

The second failure mode is cross-source divergence. Two feeds remain active and exchange the same symbol, yet their prices differ because one source has incorporated a correction or stale routing information.

The third failure mode is an update gap. A sequence number jumps, a candle interval disappears, or the primary source stops producing trades during a period in which the secondary source records normal activity.

The fourth failure mode is timestamp corruption. The server responds normally, but exchange timestamps move backward, arrive in the wrong unit, or become implausibly far from synchronized wall-clock time.

The final failure mode is a broken reference. A validation system can remain operational while its second source has failed, leaving the primary feed incorrectly certified as healthy.

A useful monitor must distinguish these conditions instead of reducing every discrepancy to a generic “data mismatch” alert.

Failure mode	Observable signal	Example alert condition	Initial response
Primary staleness	Primary trade age exceeds SLO	primary_age > 5s	Pause strategy evaluation
Secondary staleness	Reference feed stops advancing	secondary_age > 10s	Mark validation as degraded
Price divergence	Cross-source return differs beyond tolerance	abs(deviation_bps) > 15 for 3 checks	Quarantine affected symbol
Missing update	Expected candle or sequence is absent	2 consecutive intervals missing	Restart or fail over
Timestamp discontinuity	Timestamp moves backward or unit changes	delta_ts <= 0	Reject the update
Time skew	Source age differs persistently from reference	Median skew above 30 ms	Investigate clock synchronization
Secondary failure	Reference endpoint errors repeatedly	3 consecutive API failures	Stop using reference verdicts

These thresholds are examples, not universal defaults. A liquid US equity can justify tighter tolerances than an illiquid crypto pair, and an active-market detector should suppress stale-price alerts during sessions with little trading.

2. Cross-Validation Architecture

The monitoring architecture separates the primary trading path from the validation path. The strategy continues consuming the primary feed, while an independent collector compares the primary observations with a second source.

                         +--------------------------+
                         |     Primary Market Feed |
                         | WebSocket / REST / Venue |
                         +------------+-------------+
                                      |
                                      v
                         +--------------------------+
                         | Primary Normalizer      |
                         | symbol, ts, price, seq  |
                         +------------+-------------+
                                      |
                                      v
+-------------+             +--------------------------+
| TickDB      |------------>| Cross-Source Validator   |
| kline/latest|  reference  | freshness / divergence  |
+-------------+             | gaps / sequence checks  |
                           +------------+-------------+
                                        |
                        +---------------+----------------+
                        |                                |
                        v                                v
              +------------------+             +---------------------+
              | Metrics / Logs   |             | Webhook Alerting    |
              +------------------+             | Slack / Pager / SRE |

2.1 Responsibility Boundaries

The primary feed remains the operational source for execution decisions. The second source validates data quality; it should not silently replace the primary feed unless the system has an explicit failover policy.

The second source has four responsibilities:

1. Confirm that the primary price remains plausible.
2. Detect when the primary update cadence stops advancing.
3. Identify missing intervals or sequence numbers.
4. Test whether the validation source itself remains trustworthy.

A separate source is not useful if the monitor cannot tell which side failed. The validator therefore classifies alerts as primary_suspect, secondary_suspect, both_suspect, or timestamp_alignment_error.

2.2 Why a Kline Reference Works for US Equities

The example uses the TickDB current-candle endpoint:

GET https://api.tickdb.ai/v1/market/kline/latest

The kline endpoint is appropriate for US-equity monitoring because TickDB provides more than 10 years of cleaned and aligned US-equity OHLCV history. The current-candle endpoint should be used for live monitoring, while the standard /v1/market/kline endpoint should be used for completed historical intervals and backtesting.

This design does not claim that TickDB provides US-equity tick-level trades. The trades endpoint does not cover US equities or A-shares. For this scenario, the reference signal comes from OHLCV data rather than a claim of full tick-history equivalence.

For Hong Kong equities or crypto, an implementation can extend the same design with supported trades or depth observations. TickDB currently provides order-book depth for US equities at L1, Hong Kong equities at L1–L10, and crypto at L1–L10. Depth is not available for forex, precious metals, or indices.

3. Selecting Comparable Signals

Cross-source validation begins with a precise definition of equality. Comparing every raw field from two providers usually produces false positives because the sources may define candles, timestamps, and session boundaries differently.

3.1 Normalize Before Comparing

Both sources should pass through a canonical representation:

CanonicalObservation {
    symbol
    source_timestamp
    receipt_timestamp
    open
    high
    low
    close
    volume
    sequence_number
}

The normalizer should perform at least four tasks:

• Convert exchange timestamps to UTC epoch milliseconds.
• Convert prices to a consistent currency and tick size.
• Align candle timestamps to a documented interval boundary.
• Attach the local receipt time separately from the source timestamp.

The distinction between source and receipt timestamps is critical. Source timestamps estimate event time. Receipt timestamps measure the local system’s observation time. Conflating them can hide a transport delay or create a false impression of low latency.

3.2 Establish a Comparable Window

The primary stream and the reference candle must describe the same interval. Comparing a primary 1-minute candle at 10:03:59 with a reference candle beginning at 10:03:00 can produce a divergence even when both sources are correct.

A robust comparison therefore applies three conditions:

comparable =
    same_symbol
    AND same_candle_interval
    AND timestamp_boundary_matches

When the boundaries do not match, the validator should record a timestamp_alignment_error rather than classify the discrepancy as a price failure.

3.3 Synchronize Clocks

Absolute latency measurements require synchronized clocks across hosts. Use chrony, NTP, or PTP where available, and monitor the offset continuously.

If clock synchronization is not possible, the validator can still measure relative source skew:

relative_skew_ms =
    primary_source_timestamp
    - secondary_source_timestamp

Relative skew is useful only when both feeds represent the same event or interval. It is not a substitute for end-to-end wire-latency measurement.

4. The Cross-Validation Algorithm

The core algorithm uses four independent checks. A production implementation should combine them rather than depend on price matching alone.

4.1 Price Deviation

The normalized price deviation is calculated in basis points:

deviation_bps =
    abs(primary_price - secondary_price)
    / secondary_price
    * 10,000

A deviation threshold should have both an instantaneous limit and a persistence rule. For example, the monitor might ignore a single deviation above 10 bps but alert after three consecutive observations exceed 15 bps.

The persistence rule suppresses transient noise around bid-ask crossings, auction prints, and source-specific rounding.

4.2 Freshness Gap

The source-age difference is:

freshness_gap =
    age(primary_source_timestamp)
    - age(secondary_source_timestamp)

A positive gap means the primary observation is older than the comparable reference observation. The result must be interpreted relative to trading activity: a stale quote is expected when the instrument has not traded, while a stale trade during an active interval may indicate a data defect.

4.3 Candle Advancement

The monitor tracks the most recent candle boundary observed from each source:

primary_boundary = latest_primary_candle_boundary
secondary_boundary = latest_secondary_candle_boundary

gap_intervals =
    (secondary_boundary - primary_boundary) / interval_duration

If the secondary source advances by two or more intervals while the primary remains on an older boundary, the primary stream may have missed candles. The monitor should confirm that the symbol was expected to be active before raising a high-severity alert.

4.4 Sequence Continuity

For feeds that expose sequence numbers, continuity provides a direct check for missing updates:

expected_sequence = previous_sequence + 1

if current_sequence != expected_sequence:
    missing_updates = current_sequence - expected_sequence

Not every venue publishes a globally continuous sequence number, and some reset it during the trading session. The monitor must therefore store the venue-specific reset behavior rather than assuming universal semantics.

4.5 Alert Classification

The classifier combines the independent signals:

Primary state	Secondary state	Classification
Stale or lagging	Current	primary_suspect
Current	Stale or erroring	secondary_suspect
Divergent	Divergent	both_suspect
Comparable	Comparable	healthy
Non-comparable timestamps	Any	timestamp_alignment_error

This classification matters operationally. Restarting the primary feed is useful when the primary has stopped, but restarting a healthy primary because the reference API is unavailable can create an unnecessary outage.

5. Production-Grade Python Implementation

The following implementation uses:

• aiohttp for TickDB HTTP requests;
• websockets for the configurable primary feed;
• environment-variable authentication;
• request timeouts;
• TickDB rate-limit handling;
• WebSocket heartbeat messages;
• exponential backoff with jitter;
• candle-level comparison;
• sequence-gap detection;
• webhook alerting.

The primary feed adapter expects canonical JSON messages containing symbol, ts, and price. The normalize_primary_message function is the only section that normally needs modification for a venue-specific feed.

import asyncio
import json
import logging
import os
import random
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional

import aiohttp
import websockets


logging.basicConfig(
    level=os.getenv("LOG_LEVEL", "INFO"),
    format="%(asctime)s %(levelname)s %(message)s",
)
LOG = logging.getLogger("feed-validator")

API_KEY = os.environ.get("TICKDB_API_KEY")
PRIMARY_WS_URL = os.environ.get("PRIMARY_WS_URL")
TICKDB_SYMBOL = os.getenv("TICKDB_SYMBOL", "AAPL.US")
INTERVAL_SECONDS = int(os.getenv("INTERVAL_SECONDS", "60"))
PRIMARY_PING_COMMAND = json.loads(
    os.getenv("PRIMARY_PING_COMMAND", '{"cmd":"ping"}')
)

ALERT_WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL")
PRIMARY_STALE_SECONDS = float(os.getenv("PRIMARY_STALE_SECONDS", "10"))
SECONDARY_STALE_SECONDS = float(os.getenv("SECONDARY_STALE_SECONDS", "10"))
PRICE_DEVIATION_BPS = float(os.getenv("PRICE_DEVIATION_BPS", "15"))
DEVIATION_CONSECUTIVE = int(os.getenv("DEVIATION_CONSECUTIVE", "3"))
MISSING_CANDLE_THRESHOLD = int(os.getenv("MISSING_CANDLE_THRESHOLD", "2"))
MONITOR_INTERVAL_SECONDS = float(os.getenv("MONITOR_INTERVAL_SECONDS", "15"))

TICKDB_LATEST_URL = (
    "https://api.tickdb.ai/v1/market/kline/latest"
)

if not API_KEY:
    raise RuntimeError("TICKDB_API_KEY is required")
if not PRIMARY_WS_URL:
    raise RuntimeError("PRIMARY_WS_URL is required")


@dataclass(frozen=True)
class Candle:
    symbol: str
    timestamp_ms: int
    open: float
    high: float
    low: float
    close: float
    volume: float


@dataclass
class PrimaryState:
    last_price: Optional[float] = None
    last_trade_ts_ms: Optional[int] = None
    last_sequence: Optional[int] = None
    sequence_gaps: int = 0
    candles: dict[int, Candle] = field(default_factory=dict)

    def update(self, observation: dict[str, Any]) -> None:
        symbol = observation["symbol"]
        timestamp_ms = observation["ts"]
        price = float(observation["price"])
        sequence = observation.get("sequence")

        if self.last_sequence is not None and sequence is not None:
            if sequence > self.last_sequence + 1:
                self.sequence_gaps += sequence - self.last_sequence - 1

        if sequence is not None:
            self.last_sequence = sequence

        self.last_price = price
        self.last_trade_ts_ms = timestamp_ms

        boundary = timestamp_ms - (timestamp_ms % (INTERVAL_SECONDS * 1000))
        candle = self.candles.get(boundary)

        if candle is None:
            self.candles[boundary] = Candle(
                symbol=symbol,
                timestamp_ms=boundary,
                open=price,
                high=price,
                low=price,
                close=price,
                volume=float(observation.get("volume", 0.0)),
            )
        else:
            candle.high = max(candle.high, price)
            candle.low = min(candle.low, price)
            candle.close = price
            candle.volume += float(observation.get("volume", 0.0))

        # Keep memory bounded during long-running processes.
        if len(self.candles) > 1_000:
            oldest = min(self.candles)
            del self.candles[oldest]


def epoch_ms_to_datetime(value: int) -> datetime:
    return datetime.fromtimestamp(value / 1000, tz=timezone.utc)


def normalize_timestamp(value: Any) -> int:
    """Normalize epoch seconds, milliseconds, microseconds, or ISO-8601."""
    if isinstance(value, str):
        text = value.replace("Z", "+00:00")
        parsed = datetime.fromisoformat(text)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return int(parsed.timestamp() * 1000)

    timestamp = float(value)

    if timestamp > 1e14:
        return int(timestamp / 1_000)
    if timestamp > 1e11:
        return int(timestamp)

    return int(timestamp * 1_000)


def find_candidate_object(payload: Any) -> Optional[dict[str, Any]]:
    """Locate the first candle-like object in a provider response."""
    if isinstance(payload, dict):
        candle_keys = {"open", "high", "low", "close", "volume"}
        if len(candle_keys.intersection(payload.keys())) >= 4:
            return payload

        for value in payload.values():
            candidate = find_candidate_object(value)
            if candidate is not None:
                return candidate

    if isinstance(payload, list):
        for value in payload:
            candidate = find_candidate_object(value)
            if candidate is not None:
                return candidate

    return None


def parse_tickdb_candle(payload: dict[str, Any]) -> Candle:
    if payload.get("code", 0) != 0:
        raise RuntimeError(payload.get("message", "TickDB returned an error"))

    candidate = find_candidate_object(payload.get("data"))
    if candidate is None:
        raise ValueError("TickDB response did not contain a recognizable candle")

    timestamp_value = (
        candidate.get("timestamp")
        or candidate.get("time")
        or candidate.get("ts")
    )
    if timestamp_value is None:
        raise ValueError("TickDB candle is missing its timestamp")

    symbol = str(
        candidate.get("symbol")
        or candidate.get("s")
        or TICKDB_SYMBOL
    )

    return Candle(
        symbol=symbol,
        timestamp_ms=normalize_timestamp(timestamp_value),
        open=float(candidate["open"]),
        high=float(candidate["high"]),
        low=float(candidate["low"]),
        close=float(candidate["close"]),
        volume=float(candidate.get("volume", 0.0)),
    )


async def fetch_tickdb_latest(
    session: aiohttp.ClientSession,
) -> Candle:
    headers = {"X-API-Key": API_KEY}
    params = {
        "symbol": TICKDB_SYMBOL,
        "interval": f"{INTERVAL_SECONDS}s",
    }

    timeout = aiohttp.ClientTimeout(
        connect=3.05,
        total=10,
    )

    async with session.get(
        TICKDB_LATEST_URL,
        headers=headers,
        params=params,
        timeout=timeout,
    ) as response:
        if response.status == 429:
            retry_after = float(response.headers.get("Retry-After", "5"))
            LOG.warning("TickDB HTTP rate limit; waiting %.1fs", retry_after)
            await asyncio.sleep(retry_after)
            return await fetch_tickdb_latest(session)

        payload = await response.json(content_type=None)
        code = payload.get("code", 0)

        if code == 3001:
            retry_after = float(
                response.headers.get("Retry-After", "5")
            )
            LOG.warning(
                "TickDB rate limit code 3001; waiting %.1fs",
                retry_after,
            )
            await asyncio.sleep(retry_after)
            return await fetch_tickdb_latest(session)

        if code in (1001, 1002):
            raise PermissionError(
                "TickDB rejected the API key; check TICKDB_API_KEY"
            )

        if code == 2002:
            raise LookupError(
                f"TickDB symbol not found: {TICKDB_SYMBOL}"
            )

        if response.status >= 400:
            raise RuntimeError(
                f"TickDB HTTP {response.status}: {payload}"
            )

        if code != 0:
            raise RuntimeError(
                f"TickDB error {code}: {payload.get('message')}"
            )

    return parse_tickdb_candle(payload)


def normalize_primary_message(payload: dict[str, Any]) -> dict[str, Any]:
    """
    Adapter for the configured primary provider.

    Expected input:
        {
            "type": "ticker",
            "symbol": "AAPL.US",
            "ts": 1710000000000,
            "price": 187.42,
            "sequence": 12345,
            "volume": 100
        }
    """
    message_type = payload.get("type", "ticker")
    if message_type not in {"ticker", "trade"}:
        raise ValueError(f"Unsupported primary message type: {message_type}")

    return {
        "symbol": str(payload.get("symbol", TICKDB_SYMBOL)),
        "ts": normalize_timestamp(
            payload.get("ts") or payload.get("timestamp")
        ),
        "price": float(
            payload.get("price")
            if payload.get("price") is not None
            else payload.get("last")
        ),
        "sequence": payload.get("sequence"),
        "volume": float(payload.get("volume", 0.0)),
    }


async def primary_heartbeat(websocket: Any) -> None:
    while True:
        await asyncio.sleep(15)
        await websocket.send(json.dumps(PRIMARY_PING_COMMAND))


async def consume_primary(
    websocket: Any,
    state: PrimaryState,
) -> None:
    async for raw_message in websocket:
        if isinstance(raw_message, bytes):
            raw_message = raw_message.decode("utf-8")

        if raw_message.lower() in {"pong", "ping"}:
            continue

        try:
            payload = json.loads(raw_message)
            observation = normalize_primary_message(payload)
        except (ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
            LOG.warning("Discarding invalid primary message: %s", exc)
            continue

        state.update(observation)


async def primary_stream(state: PrimaryState) -> None:
    retry = 0

    while True:
        heartbeat_task = None

        try:
            LOG.info("Connecting to primary WebSocket")

            async with websockets.connect(
                PRIMARY_WS_URL,
                open_timeout=10,
                close_timeout=5,
                ping_interval=None,
                max_queue=1_000,
            ) as websocket:
                retry = 0
                heartbeat_task = asyncio.create_task(
                    primary_heartbeat(websocket)
                )

                LOG.info("Primary WebSocket connected")
                await consume_primary(websocket, state)

        except asyncio.CancelledError:
            raise
        except Exception as exc:
            LOG.exception("Primary WebSocket failed: %s", exc)

        finally:
            if heartbeat_task is not None:
                heartbeat_task.cancel()
                await asyncio.gather(
                    heartbeat_task,
                    return_exceptions=True,
                )

        delay = min(1.0 * (2 ** retry), 30.0)
        jitter = random.uniform(0.0, delay * 0.1)
        retry += 1

        LOG.warning(
            "Reconnecting primary WebSocket in %.2fs",
            delay + jitter,
        )
        await asyncio.sleep(delay + jitter)


async def send_alert(alert: dict[str, Any]) -> None:
    LOG.warning("ALERT: %s", json.dumps(alert, sort_keys=True))

    if not ALERT_WEBHOOK_URL:
        return

    timeout = aiohttp.ClientTimeout(total=5)
    async with aiohttp.ClientSession() as session:
        async with session.post(
            ALERT_WEBHOOK_URL,
            json=alert,
            timeout=timeout,
        ) as response:
            if response.status >= 400:
                body = await response.text()
                LOG.error(
                    "Alert webhook failed with HTTP %s: %s",
                    response.status,
                    body,
                )


def deviation_bps(left: float, right: float) -> float:
    if right == 0:
        return math_inf(left, right)
    return abs(left - right) / abs(right) * 10_000


def math_inf(left: float, right: float) -> float:
    return float("inf") if left != right else 0.0


async def validate_once(
    session: aiohttp.ClientSession,
    state: PrimaryState,
    consecutive_deviations: int,
) -> int:
    now_ms = int(time.time() * 1000)
    alerts: list[dict[str, Any]] = []

    try:
        reference = await fetch_tickdb_latest(session)
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        LOG.error("Secondary source unavailable: %s", exc)
        await send_alert(
            {
                "classification": "secondary_suspect",
                "symbol": TICKDB_SYMBOL,
                "reason": "secondary_fetch_failed",
                "error": str(exc),
                "timestamp_ms": now_ms,
            }
        )
        return consecutive_deviations

    secondary_age_seconds = max(
        0.0,
        (now_ms - reference.timestamp_ms) / 1000,
    )

    if secondary_age_seconds > SECONDARY_STALE_SECONDS:
        alerts.append(
            {
                "classification": "secondary_suspect",
                "reason": "secondary_stale",
                "secondary_age_seconds": secondary_age_seconds,
            }
        )

    if state.last_trade_ts_ms is None:
        alerts.append(
            {
                "classification": "primary_suspect",
                "reason": "primary_has_no_trade_observation",
            }
        )
        await send_alert(
            {
                "symbol": TICKDB_SYMBOL,
                "alerts": alerts,
                "timestamp_ms": now_ms,
            }
        )
        return consecutive_deviations

    primary_age_seconds = max(
        0.0,
        (now_ms - state.last_trade_ts_ms) / 1000,
    )

    if primary_age_seconds > PRIMARY_STALE_SECONDS:
        alerts.append(
            {
                "classification": "primary_suspect",
                "reason": "primary_stale",
                "primary_age_seconds": primary_age_seconds,
            }
        )

    primary_boundary = (
        state.last_trade_ts_ms
        - (state.last_trade_ts_ms % (INTERVAL_SECONDS * 1000))
    )
    reference_boundary = (
        reference.timestamp_ms
        - (reference.timestamp_ms % (INTERVAL_SECONDS * 1000))
    )

    missing_intervals = max(
        0,
        (reference_boundary - primary_boundary)
        // (INTERVAL_SECONDS * 1000),
    )

    if missing_intervals >= MISSING_CANDLE_THRESHOLD:
        alerts.append(
            {
                "classification": "primary_suspect",
                "reason": "missing_candle_intervals",
                "missing_intervals": missing_intervals,
            }
        )

    if primary_boundary == reference_boundary:
        primary_candle = state.candles.get(primary_boundary)

        if primary_candle is not None:
            price_difference = deviation_bps(
                primary_candle.close,
                reference.close,
            )

            if price_difference > PRICE_DEVIATION_BPS:
                consecutive_deviations += 1
            else:
                consecutive_deviations = 0

            if consecutive_deviations >= DEVIATION_CONSECUTIVE:
                alerts.append(
                    {
                        "classification": "primary_suspect",
                        "reason": "persistent_price_divergence",
                        "deviation_bps": price_difference,
                        "primary_close": primary_candle.close,
                        "reference_close": reference.close,
                    }
                )
    else:
        consecutive_deviations = 0
        alerts.append(
            {
                "classification": "timestamp_alignment_error",
                "reason": "candle_boundaries_do_not_match",
                "primary_boundary_ms": primary_boundary,
                "reference_boundary_ms": reference_boundary,
            }
        )

    if state.sequence_gaps > 0:
        alerts.append(
            {
                "classification": "primary_suspect",
                "reason": "sequence_gap",
                "missing_updates": state.sequence_gaps,
            }
        )
        state.sequence_gaps = 0

    if alerts:
        await send_alert(
            {
                "symbol": TICKDB_SYMBOL,
                "primary_last_trade_ts_ms": state.last_trade_ts_ms,
                "reference_timestamp_ms": reference.timestamp_ms,
                "alerts": alerts,
                "timestamp_ms": now_ms,
            }
        )

    return consecutive_deviations


async def main() -> None:
    state = PrimaryState()
    timeout = aiohttp.ClientTimeout(total=30)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        stream_task = asyncio.create_task(primary_stream(state))
        monitor_task = asyncio.create_task(
            monitor_loop(session, state)
        )

        try:
            await asyncio.gather(stream_task, monitor_task)
        except asyncio.CancelledError:
            pass
        finally:
            stream_task.cancel()
            monitor_task.cancel()
            await asyncio.gather(
                stream_task,
                monitor_task,
                return_exceptions=True,
            )


async def monitor_loop(
    session: aiohttp.ClientSession,
    state: PrimaryState,
) -> None:
    consecutive_deviations = 0

    while True:
        try:
            consecutive_deviations = await validate_once(
                session,
                state,
                consecutive_deviations,
            )
        except asyncio.CancelledError:
            raise
        except Exception:
            LOG.exception("Unexpected monitor failure")

        await asyncio.sleep(MONITOR_INTERVAL_SECONDS)


if __name__ == "__main__":
    asyncio.run(main())

5.1 Configuration

The implementation requires the following environment variables:

Variable	Required	Example	Purpose
TICKDB_API_KEY	Yes	Secret value	REST authentication
PRIMARY_WS_URL	Yes	Provider WebSocket URL	Primary market-data stream
TICKDB_SYMBOL	Yes	AAPL.US	Symbol monitored on both sides
INTERVAL_SECONDS	No	60	Candle comparison interval
PRIMARY_PING_COMMAND	No	{"cmd":"ping"}	Provider-specific heartbeat
ALERT_WEBHOOK_URL	No	Internal webhook URL	Alert delivery
PRIMARY_STALE_SECONDS	No	10	Primary freshness threshold
SECONDARY_STALE_SECONDS	No	10	Secondary freshness threshold
PRICE_DEVIATION_BPS	No	15	Cross-source price tolerance
DEVIATION_CONSECUTIVE	No	3	Persistence required for an alert
MISSING_CANDLE_THRESHOLD	No	2	Missing intervals before escalation
MONITOR_INTERVAL_SECONDS	No	15	Validation polling frequency

For security, TICKDB_API_KEY should be injected through the deployment platform’s secret manager rather than stored in a source file or shell history.

5.2 Engineering Warning

This polling design is appropriate for validation and moderate-frequency monitoring. It is not a substitute for a venue-native, hardware-timestamped feed in a high-frequency execution system. A production HFT path should use lower-overhead clients, kernel-bypass options where available, multicast feeds, and exchange timestamp normalization.

The WebSocket heartbeat and REST timeout also do not prove that the upstream source is fresh. They protect the local connection. The cross-source checks protect the data.

6. Turning Discrepancies into Operational Decisions

A validation alert should specify what the monitoring system knows, what it suspects, and what action is safe.

6.1 Recommended Severity Levels

Severity	Conditions	Typical action
Informational	One isolated deviation or timestamp-boundary mismatch	Record the event; no trading action
Warning	Persistent price divergence, secondary staleness, or sequence gap	Inspect the affected symbol and provider
High	Primary stops advancing while the reference advances	Pause new entries for that symbol
Critical	Both sources diverge materially or the reference fails repeatedly	Degrade to a higher-quality venue or halt strategy

6.2 Avoid Blind Automatic Failover

Automatic failover can be dangerous when the two feeds do not provide equivalent markets, prices, or execution semantics.

Before enabling automatic routing changes, verify that:

• both sources cover the same instrument;
• both use the same currency and adjustment rules;
• both represent the same market session;
• the fallback feed is already warm and subscribed;
• the fallback venue can accept the required order type;
• the reference source’s health state is available to the execution controller.

A validator that cannot distinguish a secondary outage from a primary outage should return unknown, not healthy.

6.3 Active-Market Suppression

Staleness thresholds require a market-activity context. For continuously traded crypto, a five-second-old trade can be abnormal during liquid periods. For a low-liquidity symbol, the same observation may be normal.

Useful activity signals include:

• recent trade count within the symbol;
• reference-source candle volume;
• session state;
• scheduled halts and auctions;
• exchange-reported instrument status;
• rolling volatility and spread observations.

The monitor should also avoid treating the current auction print as a permanent price change. Temporary widening of spreads and aggressive price sweeps can exceed normal deviation thresholds without indicating defective data.

7. Deployment Guide

The appropriate operating model depends on the scale and risk profile of the team.

Segment	Recommended reference cadence	Storage	Alert path	Primary objective
Individual quant developer	15–60 seconds	Local SQLite or JSONL	Console and optional webhook	Fast, understandable validation
Small research team	5–15 seconds	Prometheus metrics plus relational incident log	Slack and email	Shared visibility and audit history
Institutional trading team	Tick-level where supported; 1–5 seconds for REST validation	Time-series store and immutable incident log	PagerDuty or equivalent	Symbol-level controls and failover readiness
Market-data vendor	Continuous independent reference pipeline	Full-resolution event archive	Automated provider health API	Provider-level quality measurement

7.1 Minimum Production Checklist

Before enabling automated actions, confirm that the deployment includes:

• synchronized clocks on both collector hosts;
• explicit source and receipt timestamp fields;
• a documented normalization layer;
• separate health states for the primary and secondary sources;
• alert deduplication and recovery notifications;
• market-session and instrument-status context;
• a manually tested failover procedure;
• a rollback path for monitoring rules;
• immutable storage for every alert decision.

The validator itself should be observable. Track its check duration, last successful reference fetch, current consecutive deviation count, classification state, and alert-delivery status.

8. From Feed Monitoring to Quality-Aware Trading

“The feed was green” was never evidence that the strategy had current prices.

A second market-data source changes the operating question from “Is the socket connected?” to “Do independent observations support the data being consumed?” That distinction turns silent failures into measurable states: stale, divergent, missing, misaligned, or unknown.

TickDB fits this role as an independent reference through authenticated historical and current OHLCV endpoints. The same architecture can incorporate supported trade and order-book data where market coverage permits.

The implementation should remain deliberately conservative. It should suppress noise, distinguish primary and secondary faults, require persistence before escalation, and expose unknown when the two sources are not comparable. Under those conditions, cross-validation becomes a practical disaster-recovery layer rather than another status indicator that stays green after the market has moved on.

Next Steps

If you are a retail investor looking for deeper market insight, subscribe to the TickDB newsletter for weekly supply-chain and microstructure analysis.

If you want to run this validation pattern yourself:

1. Sign up at tickdb.ai for a free API key; no credit card is required.
2. Generate an API key in the dashboard.
3. Set the TICKDB_API_KEY environment variable.
4. Adapt normalize_primary_message to your provider’s WebSocket schema.
5. Run the validator against a liquid instrument and tune the thresholds.

If you need 10+ years of historical OHLCV data for strategy backtesting, contact [email protected] for institutional data and deployment options.

If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your tool’s marketplace.

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.