How to Use the Release Calendar API to Schedule Indicator Fetches

Am Ende dieses Leitfadens haben Sie einen Zeitplaner, der den Produktionskalender FXMacroData verwendet, um die nächste announcement_datetime Die Daten werden dann an den entsprechenden Endpunkt der Ankündigungen angefordert, wenn der neue Druck verfügbar sein soll.

Warum Zeitplan statt Umfrage?

Die Umfrage alle paar Minuten ist einfach, aber sie verschwendet Anfragen und führt zu vermeidbaren Verzögerungen um die genaue Veröffentlichungs-Sekunde. Der Endpunkt des Produktionskalenders gibt Ihnen direkt die nächste geplante Veröffentlichtungszeit, so dass Sie bis kurz vor der Veranstaltung schlafen und eine gezielte Nachfrage stellen können.

Schritt 1 - Verständnis für die Antwort des Kalenders

Der Release-Kalender gibt ein JSON-Objekt mit einem zurück . data Jede Zeile enthält den Release-Slug und seinen geplanten UTC-Zeitstempel.

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

Antwortform:

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

Für andere Währungen als USD, halten Sie die gleiche Route und addieren api_key Wenn Sie einen breiteren Währungsplan ohne Indikatorfilter abfragen, können einige Zeilen auch Routing-Metadaten enthalten, wie z. B. endpoint_family Oder ... endpoint_path- Ich weiß .

Schritt 2 - Finden Sie den nächsten Zeitstempel

Der Hilfsprogramm unten fragt den Produktionskalender nach einem einzelnen Indikator und gibt den ersten zukünftigen Zeitstempel zurück.

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

Schritt 3 - Holen Sie die veröffentlichten Daten zur richtigen Zeit ab

Wenn der Zeitstempel ankommt, rufen Sie die passenden Ankündigungen Endpunkt und lesen Sie die jüngste Beobachtung von den zurückgegebenen data - Das ist nicht wahr.

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

Schritt 4 - Ein kontinuierliches Kreislaufwerk

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)

Die kurze Pause nach der Veröffentlichung gibt dem System Zeit, das geplante Ereignis vor der nächsten Kalenderanfrage vorzubringen.

Schritt 5 - Skalieren auf mehrere Uhren

Um mehreren Indikatoren zu folgen, führen Sie einen Zeitplaner pro Paar aus. Verwenden Sie kanonische Indikator-Schläger, damit der Kalender und die Ankündigungsrouten ausgerichtet bleiben.

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

Für weitere Informationen zu den Nachholzügen siehe Kurs der AUD- Ich weiß . GBP Arbeitslosigkeit, und der Rest des Ankündigungsdokuments Index.

Schritt 6 - JavaScript / Node.js Variante

Das gleiche Muster übersetzt sich direkt in Node.js: Abfrage des Kalenders mit einem Indikatorfilter, Schlaf bis zum zurückgegebenen Zeitstempel, dann holen Sie die passenden Ankündigungen 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();
}

Wenn Sie eine Warteschlange, einen Cron-Worker oder einen serverlosen Timer unterstützen müssen, halten Sie das gleiche zwei-Stufen-Muster: Kalender für die Zeitplanung, Ankündigungen für Werte.