How to Use the Release Calendar API to Schedule Indicator Fetches

No final deste guia você terá um agendador que usa o calendário de lançamento de produção FXMacroData para encontrar o próximo announcement_datetime Para um indicador escolhido, então chama o anúncios correspondentes ponto final exatamente quando a nova impressão deve estar disponível.

Porque agendar em vez de votar?

A votação a cada poucos minutos é simples, mas desperdiça pedidos e introduz um atraso evitável em torno do segundo exato de lançamento. O ponto final do calendário de produção fornece diretamente o próximo horário de publicação programado, para que você possa dormir até pouco antes do evento e fazer uma solicitação de acompanhamento direcionada.

Passo 1 - Entender a resposta do calendário

O calendário de lançamento retorna um objeto JSON com um data Cada linha inclui o slug de lançamento e seu carimbo de hora UTC programado.

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

Forma da resposta:

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

Para moedas que não sejam USD, mantenha a mesma rota e adicione api_key Se você consultar um cronograma de moeda mais amplo sem um filtro de indicador, algumas linhas também podem incluir metadados de roteamento, como endpoint_family Ou ... endpoint_path- Não .

Passo 2 - Encontre o próximo carimbo de lançamento

O assistente abaixo pede ao calendário de produção para um único indicador e retorna o primeiro carimbo de hora futuro.

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

Passo 3 - Obter os dados liberados no momento certo

Quando o carimbo de hora chegar, chamar o correspondente anúncios ponto final e ler a observação mais recente do retorno data - A linha.

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

Passo 4 - Construir um ciclo contínuo

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)

A breve pausa pós-lançamento dá ao sistema tempo para avançar o evento programado antes da próxima consulta de calendário.

Passo 5 - Escala para vários relógios

Para seguir vários indicadores, execute um agendador por par. Use os sinais canônicos de indicadores em todo o calendário e as rotas de anúncios permaneçam alinhadas.

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

Para mais informações sobre as buscas de acompanhamento, ver Taxa de política do AUD- Não . Desemprego em GBP, e o resto do índice de documentos de anúncios.

Passo 6 - JavaScript / variante Node.js

O mesmo padrão se traduz diretamente para o Node.js: consulta o calendário com um filtro de indicador, descanse até o carimbo de tempo retornado, e depois busque a rota de anúncios correspondentes.

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

Se você precisa suportar uma fila, cron worker, ou temporizador sem servidor, manter o mesmo padrão de duas etapas: calendário para o cronograma, anúncios para valores.