Market data monitoring with Prometheus: detect old prices and silent feeds

A WebSocket connection can stay open while the price on a dashboard stops updating. Monitor the age of the data, the time you last received a frame, and whether the market is expected to be open. Then monitor the monitoring itself: an old market-status response must not silently suppress an alert.

This guide adds Prometheus metrics to an existing SiftingIO subscriber. It includes a small Python adapter, a credential-free demonstration, scrape configuration and three alert rules. You can apply the same pattern to another feed by changing its timestamp and symbol mapping.

What should a market data dashboard measure?#

Export Unix timestamps, then calculate age with time() - metric in PromQL. This is the Prometheus-recommended pattern: if the consumer stops updating a gauge, the calculated age keeps increasing.

SiftingIO tick frames use f: "tick", symbol s and Unix milliseconds in t. Convert t to seconds once at the boundary. Resolve the market from your subscription configuration; do not infer it from a symbol alone. Other tick fields and projections are described in the WebSocket documentation.

A cached value delivered on subscription can already be old. Treat its timestamp honestly. The public tick frame does not contain a snapshot marker, so the adapter counts received frames without claiming to distinguish cached values from live updates.

Run the metrics adapter without an API key#

Use Python 3.10 or later. Save the following as feed_metrics.py, then run:

python -m pip install prometheus-client
python feed_metrics.py --demo

The demonstration binds only to 127.0.0.1:9400. It supplies two synthetic timestamps: EURUSD is three minutes old and BTCUSD is current at startup. Both markets are marked open for the fixture. It never calls SiftingIO or opens a WebSocket. Leave it running while you scrape it; stop it with Ctrl+C.

"""Metrics adapter for an existing consumer; --demo uses synthetic data."""
import argparse
import math
import time
import threading
from prometheus_client import Counter, Gauge, start_http_server

SYMBOLS = {('forex', 'EURUSD'), ('crypto', 'BTCUSD')}
newest = {}
lock = threading.Lock()
LAST_T = Gauge('feed_last_tick_timestamp_seconds', 'Newest valid source t', ['market', 'symbol'])
RECEIVED = Gauge('feed_last_received_timestamp_seconds', 'Local receipt time', ['market', 'symbol'])
SEEN = Gauge('feed_tick_seen', 'At least one valid timestamp received', ['market', 'symbol'])
TICKS = Counter('feed_tick_frames_total', 'Valid timestamp frames, including cached values', ['market', 'symbol'])
BAD = Counter('feed_invalid_timestamp_total', 'Invalid timestamp frames', ['market', 'symbol'])
OPEN = Gauge('feed_market_open', 'Latest market is_open value', ['market'])
STATUS_T = Gauge('feed_market_status_timestamp_seconds', 'Last successful complete status poll', ['market'])
CONNECTED = Gauge('feed_connected', 'Consumer is authenticated and connected')
RECONNECTS = Counter('feed_reconnects_total', 'Attempts after the initial connection')
for market, symbol in SYMBOLS:
    LAST_T.labels(market, symbol).set(0)
    RECEIVED.labels(market, symbol).set(0)
    SEEN.labels(market, symbol).set(0)
    TICKS.labels(market, symbol)
    BAD.labels(market, symbol)
for market in {m for m, _ in SYMBOLS}:
    OPEN.labels(market).set(0)
    STATUS_T.labels(market).set(0)


def record_tick(market, frame, received_at=None):
    symbol = frame.get('s')
    if frame.get('f') != 'tick' or (market, symbol) not in SYMBOLS:
        return False
    value = frame.get('t')
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
        BAD.labels(market, symbol).inc()
        return False
    received = time.time() if received_at is None else received_at
    source = value / 1000.0
    # Keep a late/out-of-order frame from replacing a newer source timestamp.
    with lock:
        key = (market, symbol)
        newest[key] = max(newest.get(key, 0), source)
        LAST_T.labels(market, symbol).set(newest[key])
        RECEIVED.labels(market, symbol).set(received)
        SEEN.labels(market, symbol).set(1)
        TICKS.labels(market, symbol).inc()
    return True


def record_market_status(rows, observed_at=None):
    expected = {m for m, _ in SYMBOLS}
    parsed = {row['market']: row['is_open'] for row in rows
              if row.get('market') in expected and isinstance(row.get('is_open'), bool)}
    if parsed.keys() != expected:
        raise ValueError('Status response is missing a configured market')
    stamp = time.time() if observed_at is None else observed_at
    for market, is_open in parsed.items():
        OPEN.labels(market).set(int(is_open))
        STATUS_T.labels(market).set(stamp)


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--demo', action='store_true')
    args = parser.parse_args()
    if not args.demo:
        parser.error('Import this adapter in your consumer, or pass --demo')
    start_http_server(9400, addr='127.0.0.1')
    now = time.time()
    record_tick('forex', {'f': 'tick', 's': 'EURUSD', 't': int((now - 180) * 1000)})
    record_tick('crypto', {'f': 'tick', 's': 'BTCUSD', 't': int(now * 1000)})
    record_market_status([{'market': 'forex', 'is_open': True}, {'market': 'crypto', 'is_open': True}])
    print('Synthetic metrics: http://127.0.0.1:9400/metrics')
    while True:
        time.sleep(60)

Connect it to your existing subscriber#

Import feed_metrics in the process that owns your WebSocket, and start the metrics server once. The adapter does not create a second connection or send subscriptions. Adjust SYMBOLS to your actual configured watchlist.

import feed_metrics as metrics
from prometheus_client import start_http_server

start_http_server(9400, addr="127.0.0.1")

# In your existing receive handler, after decoding a tick:
metrics.record_tick("forex", frame)

# Set 1 after successful authentication; reset on every exit/error path.
metrics.CONNECTED.set(1)
# In the connection cleanup handler:
metrics.CONNECTED.set(0)
# Increment before each connection attempt after the initial attempt:
metrics.RECONNECTS.inc()

These calls belong in separate lifecycle handlers; the snippet shows their integration points, not a complete connection loop. Keep the subscriber's documented authentication, keepalive, retry and entitlement handling intact.

For market status, request GET https://api.sifting.io/v1/fnd/markets/status using your existing authenticated REST client. Pass the decoded data list to record_market_status. Poll on a timer, for example once every five minutes, rather than on every Prometheus scrape. Five-minute polling uses roughly 8,640 requests in a 30-day month, so include it in your usage budget or share an existing status cache.

# After a successful HTTP response and JSON decoding:
metrics.record_market_status(response_json["data"])

Only call this after a successful response. A network error, invalid JSON or missing configured market must leave the previous status timestamp untouched. The separate status-age alert then identifies the failed gate. The adapter rejects incomplete status responses before applying them.

forex and crypto are the market slugs in this example. If you add equities, decide which sessions your application considers active; an open/closed boolean is not a substitute for a session-specific requirement. The market hours guide explains the status endpoint and sessions.

The adapter rejects missing, non-positive, boolean and non-finite timestamps. It also keeps the newest source timestamp when an older frame arrives out of order. Receipt time still advances, because that older frame really did arrive. A future source timestamp is retained so clock problems stay visible; investigate time() - feed_last_tick_timestamp_seconds < -2 instead of clamping it to zero.

Scrape and alert#

Add this job to prometheus.yml. This address works when Prometheus and the exporter run on the same host. Inside containers, 127.0.0.1 refers to the container itself: use your private service network and bind the exporter appropriately, without exposing its port publicly.

scrape_configs:
  - job_name: market-feed
    scrape_interval: 15s
    static_configs:
      - targets: ["127.0.0.1:9400"]
rule_files:
  - feed-alerts.yaml

Save these rules as feed-alerts.yaml:

groups:
  - name: market-feed
    rules:
      - alert: MarketFeedExporterDown
        expr: up{job="market-feed"} == 0
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Market-feed exporter cannot be scraped"
      - alert: MarketFeedStatusUnknown
        expr: time() - feed_market_status_timestamp_seconds{job="market-feed"} > 600
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "{{ $labels.market }} status has not been refreshed"
      - alert: MarketFeedPriceOldOrMissing
        expr: |
          (
            (time() - feed_last_tick_timestamp_seconds{job="market-feed"} > 120)
            or (feed_tick_seen{job="market-feed"} == 0)
          )
          and on (job, instance, market) (feed_market_open{job="market-feed"} == 1)
          and on (job, instance, market)
            (time() - feed_market_status_timestamp_seconds{job="market-feed"} < 600)
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "{{ $labels.symbol }} has old or missing data during an open market"

The job, instance and market labels keep a healthy replica from masking a failing one. Matching only on market could let another exporter's status suppress or enable the wrong alert. Each exporter emits one status series per configured market.

Two minutes and five minutes are alert persistence windows, not service guarantees. The 120-second age threshold and 600-second status threshold are example operating choices. Tune them to the instrument, polling interval and application. A quiet instrument can legitimately have no price change; an unchanged price alone is not evidence of a broken feed.

The status rule handles both a failed poll and a cold start that never obtains a valid response. The price rule handles a symbol that has never received a tick, because its series is initialized at startup. If scraping fails, the exporter-down rule takes over. If you remove the target from Prometheus configuration, up disappears rather than becoming zero; monitor configuration and target inventory separately.

Read the result in Grafana#

Create a time-series panel using:

time() - feed_last_tick_timestamp_seconds{job="market-feed"}

Set the unit to seconds and the legend to {{market}} {{symbol}}. Place market status and status age beside it. Add time() - feed_last_received_timestamp_seconds to distinguish an old source value from a silent consumer, and increase(feed_reconnects_total[30m]) to spot repeated connection attempts.

In the synthetic demonstration, EURUSD starts around 180 seconds old and BTCUSD around zero. Both ages increase because the fixture sends no more ticks. With the open-status fixture still younger than ten minutes, the old-price alert becomes eligible; after ten minutes, the status alert replaces that assumption. This deliberate failure sequence makes the gate testable without a key or live market.

Before enabling notifications, test four cases: stop the exporter, stop status updates, withhold all ticks for one configured symbol, and mark a market closed while leaving its status fresh. Confirm that each produces the intended signal. Keep symbols limited to your watchlist; never put keys, raw prices or arbitrary server messages into metric labels. Prometheus instrumentation guidance explains why label cardinality matters.

For freshness thresholds and market-specific interpretation, continue with stale market data detection. For connecting the subscriber itself, use the SiftingIO docs.