How to Use FXMacroData SSE Streams for Real-Time Macro Data

इस गाइड के अंत तक आपके पास एक कार्यरत एसएसई ग्राहक होगा जो वास्तविक समय में नई एफएक्समैक्रोडाटा रिलीज़ घटनाओं के लिए सुनता है, मुद्राओं और संकेतकों के लिए फ़िल्टर करता है जिनकी आप परवाह करते हैं, और फिर रिलीज प्रकाशित होने के क्षण में REST एपीआई से पूर्ण संकेतक पेलोड खींचता है।

चरण 1 एसएसई धारा क्या प्रदान करती है, समझें

नया एसएसई एंडपॉइंट एक लंबे समय तक चलने वाला HTTP कनेक्शन खोलता है और हर बार जब FXMacroData एक ताजा आर्थिक रिलीज का सेवन करता है तो एक घटना को धक्का देता है। एंडपॉपइंट हैः

https://api.fxmacrodata.com/v1/stream/events

आप दो वैकल्पिक क्वेरी पैरामीटर के साथ फ़ीड को संकुचित कर सकते हैंः

• currencies अल्पविराम से अलग सूची जैसे usd,eur
• indicators अल्पविराम से अलग सूची जैसे inflation,policy_rate

गैर-USD मुद्राओं के लिए, अपनी एपीआई कुंजी को क्वेरी पैरामीटर के रूप में पास करें. उदाहरण के लिए:

curl -N "https://api.fxmacrodata.com/v1/stream/events?currencies=eur,gbp&indicators=inflation,policy_rate&api_key=YOUR_API_KEY"

यदि आप फ़िल्टर के बिना और एक पेशेवर कुंजी के बिना कनेक्ट करते हैं, तो स्ट्रीम स्वचालित रूप से USD घटनाओं तक सीमित है। इससे एसएसई व्यापक कवरेज में तार से पहले एक मुफ्त अवधारणा प्रमाण के लिए भी उपयोगी हो जाता है।

चरण 2 क्लाइंट लिखने से पहले इवेंट प्रारूप को जानें

FXMacroData मानक W3C इवेंटसोर्स फ्रेम उत्सर्जित करता है. एक विशिष्ट संदेश इस तरह दिखता हैः

id: usd_inflation_1772109000
event: announcement
data: {"event_id": "usd_inflation_1772109000", "currency": "usd", "indicator": "inflation", "records_written": 1, "timestamp": 1772109002}

पेलोड क्षेत्र जानबूझकर छोटे और परिचालन में हैं:

• event_id फॉर्म में निर्धारक पहचानकर्ता {currency}_{indicator}_{timestamp}
• currency लघु मुद्रा कोड जैसे usd या eur
• indicator उदाहरण के लिए FXMacroData सूचक स्लग inflation, policy_rate, या non_farm_payrolls
• records_written सेवन के दौरान सहेजे गए नए रिकॉर्ड की संख्या
• timestamp स्ट्रीम पर घटना प्रकाशित होने के लिए यूनिक्स टाइमस्टैम्प

क्योंकि एसएसई संदेश में पूर्ण समय श्रृंखला नहीं होती है, सामान्य पैटर्न हैः घटना के लिए सुनें, फिर एपीआई डेटा डॉक्स नए उपलब्ध रिलीज डेटा को पुनः प्राप्त करने के लिए।

चरण 3 EventSource के साथ ब्राउज़र में स्ट्रीम खोलें

यदि आप एक डैशबोर्ड, ब्राउज़र आधारित अधिसूचना, या आंतरिक निगरानी पृष्ठ बना रहे हैं, तो मूल EventSource स्ट्रीम का उपभोग करने का सबसे सरल तरीका है. auth के लिए क्वेरी स्ट्रिंग का उपयोग करें ताकि कनेक्शन कस्टम हेडर के बिना ब्राउज़र में काम करे.

const apiKey = "YOUR_API_KEY";

const streamUrl = new URL("https://api.fxmacrodata.com/v1/stream/events");
streamUrl.searchParams.set("currencies", "eur,gbp");
streamUrl.searchParams.set("indicators", "inflation,policy_rate");
streamUrl.searchParams.set("api_key", apiKey);

const source = new EventSource(streamUrl);

source.addEventListener("announcement", async (event) => {
  const payload = JSON.parse(event.data);
  console.log("Release received", payload);

  const dataUrl = new URL(
    `https://api.fxmacrodata.com/v1/announcements/${payload.currency}/${payload.indicator}`
  );
  dataUrl.searchParams.set("api_key", apiKey);

  const response = await fetch(dataUrl);
  const records = await response.json();
  const latest = records[records.length - 1];

  console.log("Latest record", latest);
});

source.onerror = (error) => {
  console.error("SSE connection problem", error);
};

मूल निवासी EventSource स्वचालित रूप से पुनः कनेक्ट जब कनेक्शन गिर जाता है. क्योंकि FXMacroData एक भेजता है id: प्रत्येक संदेश के साथ फ़ील्ड, ब्राउज़र स्वचालित रूप से पुनः कनेक्ट के दौरान अंतिम प्राप्त घटना आईडी शामिल करेगा, जो सर्वर को बफर की गई घटनाओं को फिर से खेलने देता है जिन्हें आपने याद किया है।

चरण 4 प्रत्येक अलर्ट के बाद पूर्ण सूचक पेलोड प्राप्त करें

सबसे महत्वपूर्ण डिजाइन बिंदु यह है कि एसएसई आपको बताता है कब लाने के लिए, नहीं सब कुछ एक घटना के आने के बाद, मिलान घोषणा अंत बिंदु कॉल और नवीनतम रिकॉर्ड का निरीक्षण करें.

async function fetchLatestRelease(currency, indicator, apiKey) {
  const url = new URL(`https://api.fxmacrodata.com/v1/announcements/${currency}/${indicator}`);

  if (currency !== "usd") {
    url.searchParams.set("api_key", apiKey);
  }

  const response = await fetch(url);
  if (!response.ok) {
    throw new Error(`Announcement fetch failed: ${response.status}`);
  }

  const records = await response.json();
  return records[records.length - 1] ?? null;
}

source.addEventListener("announcement", async (event) => {
  const payload = JSON.parse(event.data);
  const latest = await fetchLatestRelease(payload.currency, payload.indicator, apiKey);

  if (latest) {
    console.log(`New ${payload.currency.toUpperCase()} ${payload.indicator}`, latest);
  }
});

यह आरईएसटी एंडपॉइंट्स की पूर्ण निष्ठा को बनाए रखते हुए स्ट्रीम को हल्का रखता है। इसका मतलब यह भी है कि एक ही डाउनस्ट्रीम पार्सिंग कोड का पुनः उपयोग किया जा सकता है चाहे आपका ट्रिगर एसएसई, एक शेड्यूलर या मैनुअल एपीआई कॉल से आता हो।

चरण 5 पुनः कनेक्ट पर स्पष्ट रीप्ले के साथ एक पायथन कार्यकर्ता का निर्माण करें

सर्वर-साइड डेमॉन या बॉट के लिए, यह अक्सर पुनः कनेक्ट लॉजिक को सीधे नियंत्रित करने और Last-Event-ID नीचे दिया गया उदाहरण घोषणा घटनाओं के लिए सुनता है, अंतिम देखी आईडी संग्रहीत करता है, और एक डिस्कनेक्ट के बाद याद बफर की घटनाओं को फिर से चलाता है.

import json
import time
import requests

API_KEY = "YOUR_API_KEY"
STREAM_URL = (
    "https://api.fxmacrodata.com/v1/stream/events"
    "?currencies=eur,gbp&indicators=inflation,policy_rate&api_key=" + API_KEY
)


def fetch_latest_release(currency: str, indicator: str) -> dict | None:
    url = f"https://api.fxmacrodata.com/v1/announcements/{currency}/{indicator}"
    params = {"api_key": API_KEY} if currency != "usd" else {}
    response = requests.get(url, params=params, timeout=20)
    response.raise_for_status()
    records = response.json()
    return records[-1] if records else None


def consume_stream() -> None:
    last_event_id = None

    while True:
        headers = {"Accept": "text/event-stream"}
        if last_event_id:
            headers["Last-Event-ID"] = last_event_id

        try:
            with requests.get(STREAM_URL, headers=headers, stream=True, timeout=90) as response:
                response.raise_for_status()

                event = {}
                for raw_line in response.iter_lines(decode_unicode=True):
                    if raw_line is None:
                        continue

                    line = raw_line.strip()

                    if not line:
                        if event.get("event") == "announcement" and event.get("data"):
                            payload = json.loads(event["data"])
                            last_event_id = event.get("id") or payload["event_id"]

                            latest = fetch_latest_release(
                                payload["currency"],
                                payload["indicator"],
                            )
                            print("Announcement event", payload)
                            print("Latest record", latest)

                        event = {}
                        continue

                    if line.startswith(":"):
                        continue

                    field, _, value = line.partition(":")
                    event[field] = value.lstrip()

        except requests.RequestException as exc:
            print(f"Stream disconnected: {exc}. Reconnecting in 3 seconds...")
            time.sleep(3)


if __name__ == "__main__":
    consume_stream()

यह सही पैटर्न है जब आपको अस्थायी आउटेज के माध्यम से निर्धारक पुनःप्ले की आवश्यकता होती है. यदि कोई घटना प्रकाशित होने के बाद कनेक्शन गिर जाता है लेकिन आपके कार्यकर्ता ने इसे संसाधित करने से पहले, अगली अनुरोध में अंतिम प्राप्त आईडी शामिल है और सर्वर बफर की गई घटनाओं को फिर से चलाता है जो इसके बाद आया था.

चरण 6 एक घटना के आने पर आप क्या करना चाहते हैं, यह तय करें

एक बार धारा में तारों में है, असली डिजाइन विकल्प क्या नीचे होता है। आम पैटर्न में शामिल हैंः

चरण 7 कुछ किनारे के मामले पहले से ही संभालें

एसएसई सरल है, लेकिन उत्पादन उपयोग अभी भी कुछ नियमों से लाभान्वित होता हैः

• एक घटना का एक मूल्य के बराबर न मानें। records_written एक से अधिक हो सकता है यदि एक रिलीज में एक से ज्यादा रिकॉर्ड अपडेट किए जाते हैं.
• फिर से कनेक्ट करने की उम्मीद है। ब्राउज़र, प्रॉक्सी और मोबाइल नेटवर्क कभी-कभी लंबे समय तक चलने वाले कनेक्शन को तोड़ देंगे; रिप्ले समर्थन ठीक इसी कारण से मौजूद है।
• जहाँ तक संभव हो, धारा को संकीर्ण रखें। फ़िल्टर द्वारा currencies और indicators शोर को कम करता है और अनावश्यक डाउनस्ट्रीम फ़ॉन्ट्स से बचता है।
• REST को सत्य का स्रोत छोड़ दें। स्ट्रीम आपको बताता है कि एक रिलीज़ हुई; कैनोनिकल रिकॉर्ड अभी भी मिलान घोषणा अंत बिंदु से आता है.

आपने जो बनाया

अब आपके पास FXMacroData वास्तविक समय कार्यप्रवाहों के लिए मुख्य घटना-संचालित पैटर्न हैः सदस्यता लें https://api.fxmacrodata.com/v1/stream/events, फ़िल्टर करें और उन मुद्राओं और संकेतकों पर प्रतिक्रिया करें जो मायने रखते हैं। announcement घटनाओं के रूप में वे आते हैं, और पूर्ण रिलीज डेटा केवल जब स्ट्रीम आपको बताता है कुछ नया उतरा है लाने के लिए।

एक स्वाभाविक अगला कदम एसएसई को रिलीज कैलेंडर अनुसूची गाइड तो आप जानते हैं कि आगे क्या होने वाला है और अभी भी एक तत्काल संकेत प्राप्त करते हैं जब प्रकाशन वास्तव में होता है।

FXMacroData टीम