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"
    }
  ]
}

गैर-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"},
]

अनुवर्ती लाती पर क्षेत्र विवरण के लिए, देखें एयूडी नीतिगत दर, GBP बेरोजगारी, और शेष घोषणा दस्तावेज सूचकांक।

चरण 6 - जावास्क्रिप्ट / नोड.जेएस संस्करण

एक ही पैटर्न सीधे नोड.जेएस में अनुवाद करता हैः एक संकेतक फिल्टर के साथ कैलेंडर क्वेरी, वापस समय टिकट तक नींद, फिर मिलान घोषणाओं मार्ग लाने।

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

यदि आपको एक कतार, क्रोन कार्यकर्ता, या सर्वरलेस टाइमर का समर्थन करने की आवश्यकता है, तो एक ही दो-चरण पैटर्न रखेंः समय के लिए कैलेंडर, मूल्यों के लिए घोषणाएं।