How to Use the Release Calendar API to Schedule Indicator Fetches

在本指南的结尾,您将有一个使用FXMacroData发布日历来找到下一个发布时间的计划器. announcement_datetime 对于所选的指标,然后调用匹配的公告终点,

为什么要安排时间,而不是投票?

每几分钟进行一次投票是很简单的,但它浪费了请求,并引入了可以避免的延迟,在准确的发布秒.生产日历终点直接给你下一个计划的发布时间,所以你可以睡到事件发生前一刻,并进行一个有针对性的后续请求.

步骤1 - 了解日历的响应

发布日历返回一个JSON对象, data 每一行包含发布子及其计划的UTC时间.

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

响应形式:

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

对于非美元货币,保持相同的路线,并加上 api_key 如果您查询一个更广泛的货币表,没有指标过器,一些行可能还包含路由元数据,如 endpoint_family 没有 endpoint_path现在我们要做什么?

步骤2 - 找到下一个发布时间

下面的辅助程序要求生产日历单个指标,并返回第一个未来时间.

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

步骤3 - 在适当的时间获取发布的数据

查看时间,然后查看回来的最新观察结果. data 没有任何问题.

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

第4步 - 建立一个连续循环

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)

发布后的短暂暂暂停给系统时间,在下一个日历查询之前推进预定的事件.

步骤5 - 扩展到多个钟表

为了跟踪多个指标,每对运行一个调度器. 使用全程的定制指标,以便日历和公告路线保持一致.

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

关于后续采集的详细信息请参见 澳元政策利率没有人知道. 英失业率其他公告文件索引.

步骤6 - JavaScript / Node.js变体

查询日历, 休息到返回时间, 然后取出匹配的公告路线.

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

如果您需要支持队列,cron工作者或无服务器定时器,请保持相同的两步模式:时间表,价值的公告.