SSE release streams

Endpoint contract

Connect directly to the streaming API.

The stream is a long-lived HTTP GET request that returns W3C EventSource frames. Use the direct API host for the stream itself so the connection is not buffered by website routing. The server sends comment frames to confirm the connection; announcement events arrive only when a release is ingested.

Minimal stream probe:

curl -N -H "X-API-Key: YOUR_API_KEY" "https://api.fxmacrodata.com/v1/stream/events?currencies=usd&indicators=inflation&payload=full"

Authentication and filters

Use filters to keep the stream focused.

Live SSE delivery requires an active trial or subscription API key. Public no-key REST access remains available for supported cacheable endpoints, but persistent stream connections require authenticated access.

Response headers

Streaming responses are sent without cache buffering.

Event schema

Each SSE frame is a release notification.

The default compact payload gives enough information to route and deduplicate the release signal, plus an optional latest value for fast display. Use payload=full when your client needs the latest release value, prior value, and any forecast snapshot captured before the official release in the SSE event itself.

id: usd_inflation_1772109000
event: announcement
data: {"event_id":"usd_inflation_1772109000","currency":"usd","indicator":"inflation","records_written":1,"timestamp":1772109002,"latest_announcement":{"date":"2026-02-28","val":3.1,"announcement_datetime":1772109000}}

Full payload mode:

id: usd_unemployment_1782995400
event: announcement
data: {"event_id":"usd_unemployment_1782995400","currency":"usd","indicator":"unemployment","records_written":1,"timestamp":1782995400,"latest_announcement":{"date":"2026-06-30","val":4.2,"announcement_datetime":1782995400},"date":"2026-06-30","val":4.2,"forecast":4.3,"forecast_available":true,"forecast_captured_at":1782907200,"forecast_is_pre_release":true,"previous":4.3,"announcement_datetime":1782995400,"announcement_datetime_local":"2026-07-02T08:30:00-04:00","collected_at_iso":"2026-07-02T12:30:00.123456Z"}

REST follow-up

Use full SSE for execution and REST for confirmation.

Full payload mode can remove the pre-execution REST round trip when the fields your client needs are present in the event. The announcements endpoint remains the source of record for confirmation, storage, metadata, and historical rows.

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

  const response = await fetch(url, {
    headers: { "X-API-Key": process.env.FXMD_API_KEY },
  });
  if (!response.ok) {
    throw new Error(`Announcement fetch failed: ${response.status}`);
  }

  const body = await response.json();
  const rows = Array.isArray(body) ? body : (body.data ?? []);
  return rows[0] ?? null;
}

Reconnects and replay

Expect clean reconnects and dedupe by event ID.

Status and errors

Handle the stream like a long-lived network connection.

Client examples

Connect, parse, execute, and reconcile.

These examples show the full-payload loop: subscribe with filters, parse announcement events, use the event value immediately when present, and keep reconnect state.

Node.js worker

Stream with fetch and an X-API-Key header.

Node.js 18 and later can read the stream with the built-in fetch. Send the key as a header, keep the last processed event ID, and pass it back as Last-Event-ID when you reconnect.

const API_KEY = process.env.FXMD_API_KEY;
const streamUrl = new URL("https://api.fxmacrodata.com/v1/stream/events");
streamUrl.searchParams.set("currencies", "eur,gbp");
streamUrl.searchParams.set("indicators", "inflation,policy_rate");
streamUrl.searchParams.set("payload", "full");

let lastEventId = null;

async function fetchLatestRelease(currency, indicator) {
  const url = `https://api.fxmacrodata.com/v1/announcements/${currency}/${indicator}?limit=1`;
  const response = await fetch(url, { headers: { "X-API-Key": API_KEY } });
  if (!response.ok) {
    throw new Error(`Announcement fetch failed: ${response.status}`);
  }

  const body = await response.json();
  const records = Array.isArray(body) ? body : (body.data ?? []);
  return records[0] ?? null;
}

async function handleFrame(frame) {
  const event = {};
  for (const line of frame.split("\n")) {
    if (!line || line.startsWith(":")) continue;
    const index = line.indexOf(":");
    const field = index === -1 ? line : line.slice(0, index);
    event[field] = index === -1 ? "" : line.slice(index + 1).trimStart();
  }
  if (event.event !== "announcement" || !event.data) return;

  const payload = JSON.parse(event.data);
  lastEventId = event.id ?? payload.event_id;
  const latest = payload.val === undefined
    ? await fetchLatestRelease(payload.currency, payload.indicator)
    : payload;
  process.stdout.write(`${JSON.stringify({ payload, latest })}\n`);
}

async function consumeStream() {
  const headers = { Accept: "text/event-stream", "X-API-Key": API_KEY };
  if (lastEventId) {
    headers["Last-Event-ID"] = lastEventId;
  }

  const response = await fetch(streamUrl, { headers });
  if (!response.ok) {
    throw new Error(`Stream request failed: ${response.status}`);
  }

  const decoder = new TextDecoder();
  let buffer = "";
  for await (const chunk of response.body) {
    buffer += decoder.decode(chunk, { stream: true }).replace(/\r/g, "");
    let boundary;
    while ((boundary = buffer.indexOf("\n\n")) !== -1) {
      const frame = buffer.slice(0, boundary);
      buffer = buffer.slice(boundary + 2);
      await handleFrame(frame);
    }
  }
}

async function run() {
  while (true) {
    try {
      await consumeStream();
    } catch (error) {
      process.stderr.write(`Stream disconnected: ${error.message}\n`);
    }
    await new Promise((resolve) => setTimeout(resolve, 3000));
  }
}

run();

Python worker

Send Last-Event-ID when you control reconnects yourself.

Server-side workers can manage reconnect timing explicitly. Persist last_event_id somewhere durable if missing an event would matter after a process restart.

import json
import time

import requests


API_KEY = "YOUR_API_KEY"
STREAM_URL = "https://api.fxmacrodata.com/v1/stream/events"
PARAMS = {
    "currencies": "eur,gbp",
    "indicators": "inflation,policy_rate",
    "payload": "full",
}


def fetch_latest_release(currency: str, indicator: str) -> dict | None:
    url = f"https://api.fxmacrodata.com/v1/announcements/{currency}/{indicator}"
    response = requests.get(
        url,
        params={"limit": 1},
        headers={"X-API-Key": API_KEY},
        timeout=20,
    )
    response.raise_for_status()
    body = response.json()
    records = body if isinstance(body, list) else body.get("data", [])
    return records[0] if records else None


def consume_stream() -> None:
    last_event_id = None

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

        try:
            with requests.get(
                STREAM_URL,
                params=PARAMS,
                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 = payload
                            if "val" not in payload:
                                latest = fetch_latest_release(
                                    payload["currency"],
                                    payload["indicator"],
                                )
                            print(payload, 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()

Troubleshooting

Common integration checks.

Related docs

Use SSE with the rest of the API surface.