How to Use the Release Calendar API to Schedule Indicator Fetches

Al final de esta guía tendrá un programador que utiliza el calendario de lanzamiento de producción FXMacroData para encontrar el siguiente announcement_datetime Para un indicador elegido, entonces llama a los anuncios de coincidencia punto final exactamente cuando la nueva impresión debe estar disponible.

¿Por qué el horario en lugar de la encuesta?

La encuesta cada pocos minutos es simple, pero desperdicia solicitudes e introduce un retraso evitable alrededor del segundo de lanzamiento exacto. El punto final del calendario de producción le da el próximo tiempo de publicación programado directamente, por lo que puede dormir hasta justo antes del evento y hacer una solicitud de seguimiento específica.

Paso 1 - Comprender la respuesta del calendario

El calendario de lanzamiento devuelve un objeto JSON con un data Cada fila incluye el slug de lanzamiento y su marca de tiempo UTC programada.

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

Forma de respuesta:

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

Para las monedas que no sean USD, mantenga la misma ruta y añada api_key Si consulta un calendario de divisas más amplio sin un filtro de indicadores, algunas filas también pueden incluir metadatos de enrutamiento como endpoint_family ¿ Qué ? endpoint_path- ¿ Qué ?

Paso 2 - Encuentra la próxima fecha de lanzamiento

El asistente de abajo pide al calendario de producción un solo indicador y devuelve la primera marca de tiempo futura.

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

Paso 3 - Obtener los datos publicados en el momento adecuado

Cuando llegue la marca de tiempo, llame a los anuncios correspondientes punto final y lea la observación más reciente de los devueltos data - ¿Qué es eso?

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

Paso 4 - Construir un bucle continuo

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 breve pausa posterior a la liberación da al sistema tiempo para adelantar el evento programado antes de la siguiente consulta de calendario.

Paso 5 - Escala a varios relojes

Para seguir varios indicadores, ejecute un programador por par. Utilice las babosas de indicadores canónicos en todo el calendario y las rutas de anuncios permanezcan alineadas.

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

Para obtener más información sobre las búsquedas de seguimiento, véase Tasa de política del AUD¿ Qué ? Desempleo en GBP, y el resto del índice de documentos de anuncio.

Paso 6 - Variante JavaScript / Node.js

El mismo patrón se traduce directamente a Node.js: consulta el calendario con un filtro de indicadores, descansa hasta que la marca de tiempo devuelta, luego busca la ruta de anuncios correspondiente.

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 necesita soportar una cola, cron worker, o temporizador sin servidor, mantenga el mismo patrón de dos pasos: calendario para el tiempo, anuncios para valores.