How to Use the Release Calendar API to Schedule Indicator Fetches

À la fin de ce guide, vous aurez un planificateur qui utilise le calendrier de sortie de production FXMacroData pour trouver le prochain announcement_datetime Pour un indicateur choisi, il appelle ensuite le point d'arrivée des annonces correspondantes exactement lorsque la nouvelle impression devrait être disponible.

Pourquoi un calendrier au lieu d'un sondage ?

Le sondage toutes les quelques minutes est simple, mais il gaspille des demandes et introduit un retard évitable autour de la seconde de sortie exacte.

Étape 1 - Comprendre la réponse du calendrier

Le calendrier de sortie renvoie un objet JSON avec un data Chaque ligne comprend le slug de sortie et son horodatage UTC.

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

Forme de réponse:

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

Pour les devises autres que l'USD, conservez la même trajectoire et ajoutez api_key Si vous consultez un calendrier de devises plus large sans filtre d'indicateur, certaines lignes peuvent également inclure des métadonnées de routage telles que endpoint_family ou ... endpoint_pathJe suis désolé .

Étape 2 - Trouvez le timestamp de sortie suivant

L'assistant ci-dessous demande au calendrier de production un seul indicateur et renvoie le premier horodatage futur.

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

Étape 3 - Récupérer les données publiées au bon moment

Lorsque l'horodatage arrive, appelez le point d'arrivée des annonces correspondantes et lisez l'observation la plus récente du retour data - Je vous en prie.

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()

Étape 4 - Construire une boucle continue

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)

La courte pause après la sortie donne au système le temps de faire avancer l'événement prévu avant la prochaine requête de calendrier.

Étape 5 - Mise à l'échelle à plusieurs montres

Pour suivre plusieurs indicateurs, exécutez un planificateur par paire. Utilisez des globules d'indicateur canoniques pour que les itinéraires du calendrier et des annonces restent alignés.

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

Pour plus de détails sur les recherches de suivi, voir Taux directeur de l'AUDJe suis désolé . Le chômage en GBP, et le reste de l'index de documents d'annonce.

Étape 6 - Variante JavaScript / Node.js

Le même schéma se traduit directement dans Node.js: interrogez le calendrier avec un filtre d'indicateur, dormez jusqu'à ce que l'horodatage soit retourné, puis récupérez la route des annonces correspondantes.

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();
}

Si vous avez besoin de prendre en charge une file d'attente, un ouvrier cron ou un minuteur sans serveur, conservez le même schéma en deux étapes: calendrier pour le calendrier, annonces pour les valeurs.