Building a Real-Time Forex Dashboard
with WebSocket & Python
in 30 Minutes

What We'll Build

By the end of this guide you'll have a Python script that connects to PipData's WebSocket API and streams:

• Live bid/ask prices with spread calculation for EUR/USD and GBP/USD
• M1 and M5 OHLC candle data as candles complete in real-time
• High-impact economic news alerts from the Forex Factory feed
• A simple spread analyzer that flags unusually wide spreads

This is the exact same foundation we use internally at PipData for our monitoring system. We'll keep it clean and extendable so you can build on top of it.

Prerequisites

Before we start, make sure you have the following ready:

• Python 3.9 or higher
• A PipData API key — grab a free 7-day trial here
• Basic familiarity with asyncio and async/await

Install the required libraries:

pip install websockets asyncio

That's it. PipData's WebSocket API is pure WebSocket — no proprietary SDKs, no vendor lock-in.

Setting Up the Connection

Let's start with the basic connection. PipData uses standard WebSocket with API key authentication via headers:

import asyncio
import websockets
import json
from datetime import datetime

PIPDATA_API_KEY = "your_api_key_here"
WS_URL = "wss://api.pipdata.net/v1/stream"

async def connect_and_subscribe():
    headers = {"X-API-Key": PIPDATA_API_KEY}

    async with websockets.connect(
        WS_URL,
        extra_headers=headers,
        ping_interval=20,      # Keep-alive ping every 20s
        ping_timeout=10        # Disconnect if no pong in 10s
    ) as ws:
        print(f"Connected to PipData at {datetime.utcnow()}")

        # Subscribe to pairs + timeframes
        subscribe_msg = {
            "action": "subscribe",
            "pairs": ["EURUSD", "GBPUSD", "USDJPY"],
            "broker": "icmarkets",          # Your broker
            "timeframes": ["M1", "M5"],
            "include_news": True            # Enable news feed
        }
        await ws.send(json.dumps(subscribe_msg))

        # Confirm subscription
        ack = json.loads(await ws.recv())
        print(f"Subscribed: {ack['subscribed_pairs']} pairs | "
              f"Latency: {ack['server_latency_ms']}ms")

        return ws

asyncio.run(connect_and_subscribe())

The server acknowledges with a confirmation payload showing which pairs were subscribed and the current server latency. This is useful for health checks.

Handling Tick Data

Once subscribed, ticks arrive as JSON messages. Here's how to handle them efficiently:

async def handle_ticks(ws):
    """Process incoming tick messages."""
    async for message in ws:
        try:
            data = json.loads(message)
            msg_type = data.get("type")

            if msg_type == "tick":
                process_tick(data)
            elif msg_type == "candle":
                process_candle(data)
            elif msg_type == "news":
                process_news(data)
            elif msg_type == "heartbeat":
                pass  # Server keep-alive, ignore

        except json.JSONDecodeError:
            print(f"Malformed message: {message[:100]}")
        except Exception as e:
            print(f"Handler error: {e}")

def process_tick(data: dict):
    """Handle a single tick."""
    pair     = data["pair"]        # e.g. "EURUSD"
    bid      = data["bid"]         # e.g. 1.08541
    ask      = data["ask"]         # e.g. 1.08545
    ts       = data["timestamp"]   # Unix ms
    spread   = round((ask - bid) / 0.00001, 1)  # In pips

    dt = datetime.fromtimestamp(ts / 1000)
    print(f"[{dt:%H:%M:%S.%f}] {pair:8} Bid: {bid:.5f}  Ask: {ask:.5f}  Spread: {spread:.1f}pip")

    # Alert on wide spread (> 2 pips = unusual)
    if spread > 2.0:
        print(f"  ⚠ WIDE SPREAD ALERT — {pair} spread is {spread:.1f} pip")

Streaming OHLC Candles

PipData pushes a candle message each time a timeframe completes. You get the complete OHLC data the moment the candle closes:

# In-memory candle store
candles = {"EURUSD": {"M1": [], "M5": []}}

def process_candle(data: dict):
    """Handle a completed OHLC candle."""
    pair   = data["pair"]
    tf     = data["timeframe"]      # "M1", "M5", etc.
    candle = {
        "time":   data["time"],     # Candle open time (Unix sec)
        "open":   data["open"],
        "high":   data["high"],
        "low":    data["low"],
        "close":  data["close"],
        "volume": data["volume"]    # Tick volume
    }

    # Store candle
    if pair in candles and tf in candles[pair]:
        candles[pair][tf].append(candle)
        # Keep last 200 candles in memory
        candles[pair][tf] = candles[pair][tf][-200:]

    dt = datetime.fromtimestamp(candle["time"])
    direction = "▲" if candle["close"] >= candle["open"] else "▼"
    print(f"  {direction} {pair} {tf} | O:{candle['open']:.5f} "
          f"H:{candle['high']:.5f} L:{candle['low']:.5f} "
          f"C:{candle['close']:.5f}")

Adding the News Filter

This is where PipData's Forex Factory integration becomes powerful. You receive news events before they're published (scheduled time), and then the actual result as a second message when it's released:

import asyncio

# Track upcoming events to pause trading
upcoming_high_impact = []

def process_news(data: dict):
    """Handle Forex Factory news events."""
    title    = data["title"]          # "US Non-Farm Payrolls"
    currency = data["currency"]       # "USD"
    impact   = data["impact"]         # "high" | "medium" | "low"
    sched    = data["scheduled_time"] # ISO datetime string
    forecast = data.get("forecast")   # "185K" or None
    actual   = data.get("actual")     # None until published

    if actual is not None:
        # Actual result just published — trade the reaction
        print(f"📊 RESULT: {title}")
        print(f"   Forecast: {forecast}  |  Actual: {actual}")
        surprise = evaluate_surprise(forecast, actual, data["unit"])
        if surprise:
            print(f"   → {surprise['direction']} SURPRISE — consider {surprise['trade']}")

    elif impact == "high":
        # Upcoming high-impact event — pause trading 5 min before
        print(f"📅 UPCOMING: {title} ({currency}) at {sched}")
        upcoming_high_impact.append({
            "title": title,
            "time": sched,
            "currency": currency
        })

def evaluate_surprise(forecast, actual, unit):
    """Evaluate if the actual result is a significant surprise."""
    try:
        # Parse and compare (simplified)
        f = float(str(forecast).replace('K','').replace('%',''))
        a = float(str(actual).replace('K','').replace('%',''))
        pct_diff = abs(a - f) / max(abs(f), 0.001) * 100

        if pct_diff > 5:
            return {
                "direction": "BULLISH" if a > f else "BEARISH",
                "trade": "BUY USD pairs" if a > f else "SELL USD pairs",
                "magnitude": f"{pct_diff:.1f}% vs forecast"
            }
    except (ValueError, TypeError):
        pass
    return None

Complete Dashboard Script

Here's the full production-ready script combining everything above with proper reconnection handling:

import asyncio
import websockets
import json
from datetime import datetime

PIPDATA_API_KEY = "your_api_key_here"
WS_URL          = "wss://api.pipdata.net/v1/stream"
PAIRS           = ["EURUSD", "GBPUSD", "USDJPY"]
BROKER          = "icmarkets"
MAX_RETRIES     = 5

async def run_dashboard():
    """Main dashboard loop with auto-reconnect."""
    retries = 0

    while retries < MAX_RETRIES:
        try:
            headers = {"X-API-Key": PIPDATA_API_KEY}
            async with websockets.connect(
                WS_URL, extra_headers=headers,
                ping_interval=20, ping_timeout=10
            ) as ws:
                retries = 0  # Reset on successful connect

                await ws.send(json.dumps({
                    "action": "subscribe",
                    "pairs": PAIRS,
                    "broker": BROKER,
                    "timeframes": ["M1", "M5"],
                    "include_news": True
                }))

                print(f"✓ Connected | {len(PAIRS)} pairs | {BROKER}")
                print("-" * 50)

                async for message in ws:
                    data = json.loads(message)
                    t = data.get("type")

                    if t == "tick":
                        bid, ask = data["bid"], data["ask"]
                        spread = round((ask-bid)/0.00001, 1)
                        dt = datetime.fromtimestamp(data["timestamp"]/1000)
                        print(f"[{dt:%H:%M:%S}] {data['pair']:8} "
                              f"{bid:.5f}/{ask:.5f}  {spread:.1f}p")

                    elif t == "candle":
                        c = data
                        flag = "▲" if c["close"]>=c["open"] else "▼"
                        print(f"  {flag} {c['pair']} {c['timeframe']} "
                              f"C:{c['close']:.5f}")

                    elif t == "news" and data.get("impact") == "high":
                        print(f"  📰 {data['title']} "
                              f"({data['currency']}) — "
                              f"Forecast: {data.get('forecast','?')}")

        except websockets.ConnectionClosed as e:
            retries += 1
            wait = 2 ** retries
            print(f"⚠ Disconnected (code {e.code}). "
                  f"Retry {retries}/{MAX_RETRIES} in {wait}s...")
            await asyncio.sleep(wait)

        except Exception as e:
            retries += 1
            print(f"✗ Error: {e}. Retry {retries}/{MAX_RETRIES}...")
            await asyncio.sleep(5)

    print("Max retries reached. Exiting.")

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

Next Steps

You now have a solid real-time Forex data foundation. Here's what to build on top of it:

• Add a web UI: Serve the tick data via a Flask WebSocket endpoint and display it in a browser with Chart.js or PipData's charting library.
• SQLite persistence: Write each candle to a local SQLite database for historical analysis and backtesting.
• VPS deployment: Run this script on a €5/month Hetzner VPS in Frankfurt for consistent, low-latency operation near the PipData servers.
• Strategy hooks: Add your entry/exit logic inside the tick and candle handlers. The news filter already gives you the framework for news-aware trading.

The complete code from this guide is available when you subscribe to PipData. Pro subscribers get access to our GitHub repository with extended examples including Flask UI, SQLite storage, and Telegram alert integration.