How to Use FXMacroData SSE Streams for Real-Time Macro Release Alerts

By the end of this guide you will have a working SSE subscriber that listens for new FXMacroData release events in real time, filters for the currencies and indicators you care about, and then pulls the full indicator payload from the REST API the moment a release is published.

Step 1 — Understand what the SSE stream delivers

The new SSE endpoint opens a long-lived HTTP connection and pushes an event every time FXMacroData ingests a fresh economic release. The endpoint is:

https://fxmacrodata.com/api/v1/stream/events

You can narrow the feed with two optional query parameters:

• currencies — a comma-separated list such as usd,eur
• indicators — a comma-separated list such as inflation,policy_rate

For non-USD currencies, pass your API key as a query parameter. For example:

curl -N "https://fxmacrodata.com/api/v1/stream/events?currencies=eur,gbp&indicators=inflation,policy_rate&api_key=YOUR_API_KEY"

If you connect without filters and without a Professional key, the stream is automatically restricted to USD events. That makes SSE useful even for a free proof-of-concept before you wire in broader coverage.

Step 2 — Know the event format before you write a client

FXMacroData emits standard W3C EventSource frames. A typical message looks like this:

id: usd_inflation_1772109000
event: announcement
data: {"event_id": "usd_inflation_1772109000", "currency": "usd", "indicator": "inflation", "records_written": 1, "timestamp": 1772109002}

The payload fields are intentionally small and operational:

• event_id — deterministic identifier in the form {currency}_{indicator}_{timestamp}
• currency — lowercase currency code such as usd or eur
• indicator — the FXMacroData indicator slug, for example inflation, policy_rate, or non_farm_payrolls
• records_written — number of new records saved during ingestion
• timestamp — Unix timestamp for when the event was published onto the stream

Because the SSE message does not carry the full time series, the normal pattern is: listen for the event, then call the relevant endpoint under API data docs to retrieve the newly available release data.

Step 3 — Open the stream in a browser with EventSource

If you are building a dashboard, browser-based notifier, or internal monitoring page, native EventSource is the simplest way to consume the stream. Use the query string for auth so the connection works in the browser without custom headers.

const apiKey = "YOUR_API_KEY";

const streamUrl = new URL("https://fxmacrodata.com/api/v1/stream/events");
streamUrl.searchParams.set("currencies", "eur,gbp");
streamUrl.searchParams.set("indicators", "inflation,policy_rate");
streamUrl.searchParams.set("api_key", apiKey);

const source = new EventSource(streamUrl);

source.addEventListener("announcement", async (event) => {
  const payload = JSON.parse(event.data);
  console.log("Release received", payload);

  const dataUrl = new URL(
    `https://fxmacrodata.com/api/v1/announcements/${payload.currency}/${payload.indicator}`
  );
  dataUrl.searchParams.set("api_key", apiKey);

  const response = await fetch(dataUrl);
  const records = await response.json();
  const latest = records[records.length - 1];

  console.log("Latest record", latest);
});

source.onerror = (error) => {
  console.error("SSE connection problem", error);
};

Native EventSource automatically reconnects when the connection drops. Because FXMacroData sends an id: field with each message, the browser will automatically include the last received event ID during reconnects, which lets the server replay buffered events you missed.

Step 4 — Fetch the full indicator payload after each alert

The most important design point is that SSE tells you when to fetch, not everything to display. After an event arrives, call the matching announcement endpoint and inspect the freshest record.

async function fetchLatestRelease(currency, indicator, apiKey) {
  const url = new URL(`https://fxmacrodata.com/api/v1/announcements/${currency}/${indicator}`);

  if (currency !== "usd") {
    url.searchParams.set("api_key", apiKey);
  }

  const response = await fetch(url);
  if (!response.ok) {
    throw new Error(`Announcement fetch failed: ${response.status}`);
  }

  const records = await response.json();
  return records[records.length - 1] ?? null;
}

source.addEventListener("announcement", async (event) => {
  const payload = JSON.parse(event.data);
  const latest = await fetchLatestRelease(payload.currency, payload.indicator, apiKey);

  if (latest) {
    console.log(`New ${payload.currency.toUpperCase()} ${payload.indicator}`, latest);
  }
});

This keeps the stream lightweight while preserving the full fidelity of the REST endpoints. It also means the same downstream parsing code can be reused whether your trigger comes from SSE, a scheduler, or a manual API call.

Step 5 — Build a Python worker with explicit replay on reconnect

For a server-side daemon or bot, it is often useful to control reconnect logic directly and send the Last-Event-ID header yourself. The example below listens for announcement events, stores the last seen ID, and replays missed buffered events after a disconnect.

import json
import time
import requests

API_KEY = "YOUR_API_KEY"
STREAM_URL = (
    "https://fxmacrodata.com/api/v1/stream/events"
    "?currencies=eur,gbp&indicators=inflation,policy_rate&api_key=" + API_KEY
)


def fetch_latest_release(currency: str, indicator: str) -> dict | None:
    url = f"https://fxmacrodata.com/api/v1/announcements/{currency}/{indicator}"
    params = {"api_key": API_KEY} if currency != "usd" else {}
    response = requests.get(url, params=params, timeout=20)
    response.raise_for_status()
    records = response.json()
    return records[-1] if records else None


def consume_stream() -> None:
    last_event_id = None

    while True:
        headers = {"Accept": "text/event-stream"}
        if last_event_id:
            headers["Last-Event-ID"] = last_event_id

        try:
            with requests.get(STREAM_URL, headers=headers, stream=True, timeout=90) as response:
                response.raise_for_status()

                event = {}
                for raw_line in response.iter_lines(decode_unicode=True):
                    if raw_line is None:
                        continue

                    line = raw_line.strip()

                    if not line:
                        if event.get("event") == "announcement" and event.get("data"):
                            payload = json.loads(event["data"])
                            last_event_id = event.get("id") or payload["event_id"]

                            latest = fetch_latest_release(
                                payload["currency"],
                                payload["indicator"],
                            )
                            print("Announcement event", payload)
                            print("Latest record", latest)

                        event = {}
                        continue

                    if line.startswith(":"):
                        continue

                    field, _, value = line.partition(":")
                    event[field] = value.lstrip()

        except requests.RequestException as exc:
            print(f"Stream disconnected: {exc}. Reconnecting in 3 seconds...")
            time.sleep(3)


if __name__ == "__main__":
    consume_stream()

This is the right pattern when you need deterministic replay across temporary outages. If the connection drops after an event was published but before your worker processed it, the next request includes the last received ID and the server replays buffered events that came after it.

Step 6 — Decide what you want to do when an event lands

Once the stream is wired in, the real design choice is what happens downstream. Common patterns include:

Step 7 — Handle a few edge cases up front

SSE is simple, but production use still benefits from a few rules:

• Do not assume one event equals one value. records_written may be greater than one if a release updates more than a single record.
• Expect reconnects. Browsers, proxies, and mobile networks will occasionally break long-lived connections; replay support exists for exactly this reason.
• Keep the stream narrow where possible. Filtering by currencies and indicators reduces noise and avoids unnecessary downstream fetches.
• Leave REST as the source of truth. The stream tells you a release happened; the canonical record still comes from the matching announcement endpoint.

What you built

You now have the core event-driven pattern for FXMacroData realtime workflows: subscribe to /api/v1/stream/events, filter to the currencies and indicators that matter, react to announcement events as they arrive, and fetch the full release data only when the stream tells you something new has landed.

A natural next step is to combine SSE with the release calendar scheduling guide so you know what is due next and still receive an immediate signal when publication actually happens.

— The FXMacroData Team