릴리스 캘린더 API를 사용하여 표시자 검색을 스케줄하는 방법

이 가이드의 끝으로 다음을 찾기 위해 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"
    }
  ]
}

USD가 아닌 통화에 대해서는 같은 경로를 유지하고 더합니다 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"},
]

후속 검색에 대한 현장 세부 사항은 참조하십시오. AUD 정책금리 GBP 실업, 그리고 나머지 발표 문서 인덱스.

단계 6 - 자바스크립트 / Node.js 변종

같은 패턴은 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();
}

If you need to support a queue, cron worker, or serverless timer, keep the same two-step pattern: calendar for timing, announcements for values.