Forex WebSocket API in Python: Is EURUSD Still Updating?

For Python developers building a forex quote monitor, this guide uses TickDB to subscribe to EURUSD and save each quote with its data timestamp and local receipt time. A bounded script writes raw messages, a quote CSV and a run summary. Quote counts, receipt gaps and changes in data timestamps help distinguish an open connection from continuing quote updates. The recorded 30-second observation received 38 EURUSD messages. The same code lets you inspect a new sample in your own environment.

Give a frozen price something you can trace

Your EURUSD monitor displays a price, the terminal is quiet, and the connection appears open. Then the price stops changing. Has the quote stayed the same, or has your application stopped reading updates? A single number on the screen leaves little to investigate later.

Keep the price, the timestamp carried by the message, and the time your script read it. Then inspect whether target quotes continue arriving and whether their data timestamps advance. TickDB supplies the continuing forex quote stream; the Python application filters EURUSD, adds the receipt time and writes the files. Your displayed price can now be traced to a specific received message.

Where TickDB fits

TickDB provides a unified market data interface for developers. Here we use its WebSocket ticker subscription to receive continuing EURUSD quote messages. A REST or MCP query can return an on-demand snapshot; this task needs a subscription that keeps receiving subsequent updates.

The official ticker documentation lists last_price, a Unix-millisecond timestamp, and bid/ask fields when available. The data timestamp comes from the message. The local receipt time, file storage and diagnostic summaries are added by this script.

The recorded observation

On October 28, 2026, the connection and observation ran between 02:10:30 and 02:11:02 UTC. A 30-second receive window started after connection. It captured 38 EURUSD quote messages, with 37 changes between adjacent data timestamps and no observed backward timestamps. The first three rows were:

Data time UTC	Receipt time UTC	last_price
2026-10-28T02:10:33.305+00:00	2026-10-28T02:10:33.405+00:00	1.11242
2026-10-28T02:10:33.706+00:00	2026-10-28T02:10:33.816+00:00	1.11245
2026-10-28T02:10:33.803+00:00	2026-10-28T02:10:33.905+00:00	1.11240

This count describes one short observation. It is not a fixed update rate. Prices and message counts will change on another run.

What to save, and why

Field	Origin	Use
last_price	The quote message	Preserve the price string without an early float conversion
data_time_ms, data_time_utc	Message timestamp	Check whether data timestamps change or move backward
received_time_ms, received_time_utc	Local wall clock after reading a message	Trace when this application read the message
receive_gap_seconds	Local monotonic clock	Inspect gaps between target quote receipts without directly depending on wall-clock adjustments

Subtracting the two UTC timestamps mixes source timestamp semantics, transport, buffering, scheduling and clock differences. Do not label that difference network latency. This script records both times without producing a latency metric.

Run a bounded subscription

Use Python 3.11 or later:

python3 -m pip install websockets certifi

Save the code below as eurusd_monitor.py. It reads TICKDB_API_KEY if set, otherwise prompts for your key without echoing it. Keep keys out of source code and screenshots. Access to the requested symbol depends on your account permissions.

python3 eurusd_monitor.py --seconds 30 --out eurusd_run_01

Choose an output directory that does not yet exist. The script preserves earlier observations by refusing to overwrite one.

"""Python 3.11+; pip install websockets certifi. Bounded observation, not trading."""
import argparse
import asyncio
import csv
import getpass
import json
import os
import ssl
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlencode
import certifi
import websockets


def utc(ms):
    return datetime.fromtimestamp(ms / 1000, timezone.utc).isoformat(timespec='milliseconds')


async def observe(seconds, out):
    key = os.environ.get('TICKDB_API_KEY') or getpass.getpass('TickDB API key (hidden): ')
    if not key.strip():
        raise ValueError('API key required')
    out.mkdir(parents=True, exist_ok=False)
    fields = ['symbol', 'last_price', 'bid_price', 'ask_price', 'data_time_ms',
              'data_time_utc', 'received_time_ms', 'received_time_utc', 'receive_gap_seconds']
    summary = {'started_utc': utc(time.time_ns() // 1000000), 'requested_seconds': seconds,
               'symbol': 'EURUSD', 'status': 'starting', 'quote_count': 0,
               'data_time_changes': 0, 'backward_data_times': 0, 'last_receive_age_seconds': None,
               'max_receive_gap_seconds': None, 'handshake_http_status': None}
    previous_mono = previous_data = last_receive_mono = None
    gaps = []
    try:
        uri = 'wss://api.tickdb.ai/v1/realtime?' + urlencode({'api_key': key})
        context = ssl.create_default_context(cafile=certifi.where())
        async with websockets.connect(uri, ssl=context, open_timeout=10, close_timeout=3,
                                      ping_interval=20, ping_timeout=20) as ws:
            summary['status'] = 'connected'
            summary['handshake_http_status'] = 101
            await ws.send(json.dumps({'cmd': 'subscribe', 'data': {
                'channel': 'ticker', 'symbols': ['EURUSD'], 'type': 'forex'}}))
            deadline = time.monotonic() + seconds
            with (out / 'messages.jsonl').open('w') as raw, (out / 'quotes.csv').open('w', newline='') as f:
                writer = csv.DictWriter(f, fieldnames=fields)
                writer.writeheader()
                while time.monotonic() < deadline:
                    try:
                        message = await asyncio.wait_for(ws.recv(), timeout=min(1, max(.001, deadline-time.monotonic())))
                    except asyncio.TimeoutError:
                        continue
                    received_ms = time.time_ns() // 1000000
                    received_mono = time.monotonic()
                    try:
                        decoded = json.loads(message)
                    except (ValueError, UnicodeDecodeError):
                        summary['invalid_json_count'] = summary.get('invalid_json_count', 0) + 1
                        continue
                    raw.write(json.dumps({'received_time_ms': received_ms, 'message': decoded}).replace(key, '[REDACTED]')+'\n')
                    raw.flush()
                    for item in decoded if isinstance(decoded, list) else [decoded]:
                        if not isinstance(item, dict):
                            continue
                        if item.get('cmd') == 'error':
                            summary['api_error_code'] = item.get('code')
                            raise RuntimeError('API subscription error')
                        data = item.get('data')
                        if item.get('cmd') != 'ticker' or not isinstance(data, dict) or data.get('symbol') != 'EURUSD':
                            continue
                        stamp = data.get('timestamp')
                        if isinstance(stamp, bool) or not isinstance(stamp, int) or not isinstance(data.get('last_price'), str):
                            summary['invalid_quote_count'] = summary.get('invalid_quote_count', 0) + 1
                            continue
                        gap = None if previous_mono is None else round(received_mono-previous_mono, 6)
                        if gap is not None:
                            gaps.append(gap)
                        if previous_data is not None:
                            summary['data_time_changes'] += int(stamp != previous_data)
                            summary['backward_data_times'] += int(stamp < previous_data)
                        row = dict(symbol='EURUSD', last_price=data['last_price'],
                                   bid_price=data.get('bid_price'), ask_price=data.get('ask_price'),
                                   data_time_ms=stamp, data_time_utc=utc(stamp),
                                   received_time_ms=received_ms, received_time_utc=utc(received_ms),
                                   receive_gap_seconds=gap)
                        writer.writerow(row)
                        f.flush()
                        summary['quote_count'] += 1
                        previous_mono = last_receive_mono = received_mono
                        previous_data = stamp
            summary['status'] = 'observed_quotes' if summary['quote_count'] else 'connected_without_target_quotes'
    except Exception as exc:
        summary['status'] = 'failed'
        summary['error_type'] = type(exc).__name__
        response = getattr(exc, 'response', None)
        summary['handshake_http_status'] = getattr(response, 'status_code', None)
        # Never print exception text: it may contain the authenticated URL.
    finally:
        summary['ended_utc'] = utc(time.time_ns() // 1000000)
        summary['last_receive_age_seconds'] = None if last_receive_mono is None else round(time.monotonic()-last_receive_mono, 6)
        summary['max_receive_gap_seconds'] = max(gaps) if gaps else None
        (out / 'summary.json').write_text(json.dumps(summary, indent=2)+'\n')
        print(json.dumps(summary, indent=2))
    return summary['status'] == 'observed_quotes'


if __name__ == '__main__':
    p = argparse.ArgumentParser()
    p.add_argument('--seconds', type=int, default=30)
    p.add_argument('--out', type=Path, default=Path('eurusd_observation'))
    a = p.parse_args()
    if not 1 <= a.seconds <= 300:
        p.error('--seconds must be 1..300')
    raise SystemExit(0 if asyncio.run(observe(a.seconds, a.out)) else 1)

The script closes after the requested receive window. It uses WebSocket protocol Ping/Pong checks. There is no need to copy an older example that sends an application ping every second; the current connection guidance distinguishes protocol heartbeats from optional application checks.

The run writes messages.jsonl for received messages, quotes.csv for valid EURUSD quote rows, and summary.json for status, counts and times. A handshake failure may produce only the summary. Read its status instead of treating a missing CSV as a successful run with zero quotes. Exceptions are reported without printing an authenticated connection URL.

Decide what to investigate next

observed_quotes means target quotes were read during the observation. Inspect the count, data_time_changes and the CSV to see the pattern over the window. A changing timestamp with an unchanged price can still be an update; a monitor should not require every message to carry a different price.

connected_without_target_quotes means the connection opened but no valid EURUSD quotes were read within the window. Check the requested symbol, permissions and raw control messages, then consider whether quotes were active during that period. Connection heartbeats do not establish that target quotes are updating.

failed identifies an error during connection, subscription or receipt. Preserve that status even if some quotes were saved before the interruption. Address authentication or permission failures before retrying.

A short stream of received messages does not establish that no messages were missed. An operational alert needs a silence threshold appropriate to the application, timestamp checks, reconnection and resubscription handling. This example is an investigation starting point; it does not implement recovery or trading.

Start with your own credentials from TickDB, record one EURUSD run, and inspect whether the final data timestamps advance and the final receipt is close to the observation end. Connect that record to your monitor so its displayed price has a traceable source.