How to Use the Release Calendar API to Schedule Indicator Fetches

By the end of this guide you will have a scheduler that uses the production FXMacroData release calendar to find the next announcement_datetime for a chosen indicator, then calls the matching announcements endpoint exactly when the new print should be available.

Why schedule instead of poll?

Polling every few minutes is simple, but it wastes requests and introduces avoidable lag around the exact release second. The production calendar endpoint gives you the next scheduled publication time directly, so you can sleep until just before the event and make one targeted follow-up request.

Step 1 - Understand the calendar response

The release calendar returns a JSON object with a data array. Each row includes the release slug and its scheduled UTC timestamp.

curl "https://fxmacrodata.com/api/v1/calendar/usd?indicator=inflation"

Response shape:

{
  "currency": "USD",
  "indicator": "inflation",
  "data": [
    {
      "announcement_datetime": 1773077400,
      "release": "inflation"
    }
  ]
}

For non-USD currencies, keep the same route and add api_key as a query parameter. If you query a broader currency schedule without an indicator filter, some rows may also include routing metadata such as endpoint_family or endpoint_path.

Step 2 - Find the next release timestamp

The helper below asks the production calendar for a single indicator and returns the first future timestamp.

import time
import requests

BASE = "https://fxmacrodata.com/api/v1"

def next_release(currency: str, indicator: str, api_key: str | None = None) -> float | None:
    url = f"{BASE}/calendar/{currency}"
    params = {"indicator": indicator}
    if api_key:
        params["api_key"] = api_key

    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()

    now = time.time()
    for row in response.json().get("data", []):
        ts = row.get("announcement_datetime")
        if row.get("release") == indicator and ts and float(ts) > now:
            return float(ts)
    return None

Step 3 - Fetch the released data at the right time

When the timestamp arrives, call the matching announcements endpoint and read the most recent observation from the returned data array.

def wait_and_fetch(currency: str, indicator: str, api_key: str | None = None) -> dict | None:
    release_ts = next_release(currency, indicator, api_key)
    if release_ts is None:
        print(f"No upcoming release found for {currency}/{indicator}.")
        return None

    wake_at = release_ts - 2
    time.sleep(max(0.0, wake_at - time.time()))

    url = f"{BASE}/announcements/{currency}/{indicator}"
    params = {"api_key": api_key} if api_key else {}
    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()
    return response.json()

Step 4 - Build a continuous loop

import time
import requests

BASE = "https://fxmacrodata.com/api/v1"
API_KEY = None
CURRENCY = "usd"
INDICATOR = "inflation"

def on_release(payload: dict) -> None:
    rows = payload.get("data", [])
    latest = rows[-1] if rows else {}
    print(
        "New release:",
        latest.get("date"),
        "value=", latest.get("val"),
        "announced_at=", latest.get("announcement_datetime"),
    )

while True:
    release_ts = next_release(CURRENCY, INDICATOR, API_KEY)
    if release_ts is None:
        print("No release found in the current calendar window. Retrying in 24 hours.")
        time.sleep(86_400)
        continue

    sleep_seconds = max(0.0, release_ts - 2 - time.time())
    print(f"Next {CURRENCY.upper()} {INDICATOR} release in {sleep_seconds / 3600:.2f}h")
    time.sleep(sleep_seconds)

    payload = wait_and_fetch(CURRENCY, INDICATOR, API_KEY)
    if payload:
        on_release(payload)

    time.sleep(5)

The short post-release pause gives the system time to roll the scheduled event forward before the next calendar query.

Step 5 - Scale to multiple watches

To follow several indicators, run one scheduler per pair. Use canonical indicator slugs throughout so the calendar and announcements routes stay aligned.

WATCHES = [
    {"currency": "usd", "indicator": "inflation"},
    {"currency": "usd", "indicator": "gdp"},
    {"currency": "aud", "indicator": "policy_rate"},
    {"currency": "gbp", "indicator": "unemployment"},
]

For field details on the follow-up fetches, see AUD policy rate, GBP unemployment, and the rest of the announcement docs index.

Step 6 - JavaScript / Node.js variant

The same pattern translates directly to Node.js: query the calendar with an indicator filter, sleep until the returned timestamp, then fetch the matching announcements route.

const BASE = "https://fxmacrodata.com/api/v1";

async function nextRelease(currency, indicator, apiKey) {
  const url = new URL(`${BASE}/calendar/${currency}`);
  url.searchParams.set("indicator", indicator);
  if (apiKey) url.searchParams.set("api_key", apiKey);

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

  const payload = await response.json();
  const now = Date.now() / 1000;
  return payload.data.find((row) => row.release === indicator && row.announcement_datetime > now) ?? null;
}

async function fetchAnnouncement(currency, indicator, apiKey) {
  const url = new URL(`${BASE}/announcements/${currency}/${indicator}`);
  if (apiKey) url.searchParams.set("api_key", apiKey);

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

If you need to support a queue, cron worker, or serverless timer, keep the same two-step pattern: calendar for timing, announcements for values.