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.

Minimal stream probe:

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

Authentication and filters

Use filters to keep the stream focused.

Authentication follows the same plan model as the API. USD announcement events can be streamed without a key. Non-USD currency streams require a valid Professional API key.

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 and available row metadata 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,"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, apiKey) {
  const url = new URL(
    `https://api.fxmacrodata.com/v1/announcements/${currency}/${indicator}`
  );
  url.searchParams.set("limit", "1");
  if (currency !== "usd" && apiKey) {
    url.searchParams.set("api_key", apiKey);
  }

  const response = await fetch(url);
  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.

Browser client

Use EventSource and query-parameter auth.

Native EventSource cannot set custom headers, so browser clients should put the API key in the query string. The browser automatically reconnects and sends the last received event ID when possible.

const apiKey = "YOUR_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");
streamUrl.searchParams.set("api_key", apiKey);

const source = new EventSource(streamUrl);

async function fetchLatestRelease(currency, indicator) {
  const url = new URL(
    `https://api.fxmacrodata.com/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 body = await response.json();
  const records = Array.isArray(body) ? body : (body.data ?? []);
  return records[0] ?? null;
}

source.addEventListener("announcement", async (event) => {
  const payload = JSON.parse(event.data);
  const latest = payload.val === undefined
    ? await fetchLatestRelease(payload.currency, payload.indicator)
    : payload;
  renderRelease(payload, latest);
});

source.onerror = () => {
  setStreamStatus("reconnecting");
};

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",
    "api_key": API_KEY,
}


def fetch_latest_release(currency: str, indicator: str) -> dict | None:
    url = f"https://api.fxmacrodata.com/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()
    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"}
        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.