FXMacroData और Webhooks के साथ मैक्रो अलर्ट कैसे सेट करें

मैक्रो डेटा रिलीज़ विदेशी मुद्रा बाजारों में सबसे विश्वसनीय उत्प्रेरक में से हैं। आम सहमति से 0.2% ऊपर एक सीपीआई प्रिंट EUR/USD स्लाइडिंग 60 पिप्स भेज सकता है इससे पहले कि समाचार शीर्षक भी लोड हो। यदि आपकी चेतावनी प्रणाली मैनुअल कैलेंडर जांच या व्यापक टाइमर-आधारित मतदान पर निर्भर करती है, तो आप पहले से ही पीछे हैं। यह गाइड आपको दिखाता है कि कैसे एक सटीक, वेबहूक-संचालित चेतावनी पाइपलाइन का निर्माण करें जो नए FXMacroData रिलीज का पता लगाता है जब वे दिखाई देते हैं और स्लैक, डिस्कॉर्ड या किसी भी HTTP एंडपॉइंट को सूचनाएं वितरित करते हैं सभी पायथन की 80 लाइनों से कम में।

pip install requests schedule

export FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY"
export WEBHOOK_URL="https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"

चरण 1 नवीनतम घोषणा मान प्राप्त करें

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

import os
import requests

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY  = os.environ["FXMACRO_API_KEY"]


def fetch_latest(currency: str, indicator: str) -> dict | None:
    """Return the most recent announcement record, or None on failure."""
    try:
        resp = requests.get(
            f"{BASE_URL}/announcements/{currency}/{indicator}",
            params={"api_key": API_KEY},
            timeout=10,
        )
        resp.raise_for_status()
        data = resp.json().get("data", [])
        return data[0] if data else None
    except requests.RequestException as exc:
        print(f"[WARN] fetch failed for {currency}/{indicator}: {exc}")
        return None

एक नमूना प्रतिक्रिया इस तरह दिखती हैः

{
  "date": "2026-04-02",
  "val": 4.35,
  "prior": 4.10,
  "announcement_datetime": "2026-04-02T02:30:00Z",
  "currency": "aud",
  "indicator": "policy_rate"
}

चरण 2 दोहराए गए अलर्ट से बचने के लिए ट्रैक स्टेटस

आपकी प्रक्रिया को पुनः आरंभ करने से पहले से देखी गई रिलीज़ के लिए अलर्ट फिर से नहीं चलना चाहिए. announcement_datetime एक साधारण JSON फ़ाइल में प्रत्येक देखा संकेतक के लिए. स्टार्टअप पर बॉट इस फ़ाइल लोड करता है; प्रत्येक नए अलर्ट के बाद यह अद्यतन समयस्टैम्प वापस लिखता है.

import json
from pathlib import Path

STATE_FILE = Path("alert_state.json")


def load_state() -> dict:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}


def save_state(state: dict) -> None:
    STATE_FILE.write_text(json.dumps(state, indent=2))


def is_new_release(state: dict, key: str, record: dict) -> bool:
    """Return True if the announcement_datetime is newer than what we last saw."""
    last_seen = state.get(key)
    current   = record.get("announcement_datetime")
    return current is not None and current != last_seen

चरण 3 वेबहूक सूचना भेजें

स्लैक इनकमिंग वेबहूक्स और डिस्कॉर्ड वेबहूक दोनों एक जेएसओएन पेलोड के साथ एक HTTP पोस्ट स्वीकार करते हैं. नीचे दिया गया फ़ंक्शन एक स्वरूपित संदेश बनाता है और इसे भेजता है. स्लैक्स एक का उपयोग करता है text क्षेत्र; डिस्कॉर्ड उपयोग करता है content. सहायक स्वचालित रूप से यूआरएल उपसर्ग के आधार पर अनुकूलित होता है.

def send_webhook(webhook_url: str, record: dict, currency: str, indicator: str) -> None:
    """POST a formatted macro-alert message to a Slack or Discord webhook."""
    value    = record.get("val")
    prior    = record.get("prior")
    dt       = record.get("announcement_datetime", "")
    ccy      = currency.upper()
    ind      = indicator.replace("_", " ").title()

    lines = [
        f"📣  *{ccy} {ind}* just printed",
        f"  Value : *{value}*   |   Prior: {prior}",
        f"  Released: {dt}",
    ]
    message = "\n".join(lines)

    # Discord uses "content", Slack uses "text"
    if "discord.com" in webhook_url:
        payload = {"content": message.replace("*", "**")}
    else:
        payload = {"text": message}

    try:
        resp = requests.post(webhook_url, json=payload, timeout=10)
        resp.raise_for_status()
        print(f"[OK] alert sent for {ccy} {ind}")
    except requests.RequestException as exc:
        print(f"[ERROR] webhook delivery failed: {exc}")

चरण 4 एक मतदान चक्र में कई संकेतकों का निरीक्षण करें

आप जिन मुद्रा/संकेतक जोड़े की परवाह करते हैं, उन्हें परिभाषित करें और नियमित अंतराल पर चेक करें। schedule पुस्तकालय यह एक क्रॉन डेमॉन के बिना सीधा बनाता है.

import schedule
import time

WEBHOOK_URL = os.environ["WEBHOOK_URL"]

# Pairs to monitor — add or remove as needed
WATCHLIST = [
    ("usd", "policy_rate"),
    ("usd", "inflation"),
    ("usd", "non_farm_payrolls"),
    ("eur", "policy_rate"),
    ("eur", "inflation"),
    ("aud", "policy_rate"),
    ("gbp", "policy_rate"),
]


def check_all(state: dict) -> None:
    for currency, indicator in WATCHLIST:
        key    = f"{currency}/{indicator}"
        record = fetch_latest(currency, indicator)
        if record and is_new_release(state, key, record):
            send_webhook(WEBHOOK_URL, record, currency, indicator)
            state[key] = record["announcement_datetime"]
            save_state(state)


def main():
    state = load_state()
    # Run immediately on start, then every 5 minutes
    check_all(state)
    schedule.every(5).minutes.do(check_all, state=state)
    while True:
        schedule.run_pending()
        time.sleep(30)


if __name__ == "__main__":
    main()

अपने टर्मिनल से बॉट चलाएँ:

python macro_alert_bot.py

पहली दौड़ में लोग आते हैं alert_state.json वर्तमान नवीनतम मूल्यों के साथ तो भविष्य केवल आग चलाता है जब कुछ वास्तव में नया दिखाई देता है।

चरण 5 रिलीज से पहले उलटी गिनती अलर्ट जोड़ें

यह जानना कि कोई रिहाई हुई है उपयोगी है; यह जानना उपयोगी है होने वाला मूल्यवान है। रिलीज कैलेंडर अंत बिंदु निर्धारित announcement_datetime भविष्य की घटनाओं के लिए, अपेक्षित आम सहमति मानों सहित. प्रिंट से पहले एक विन्यस्त मिनट की संख्या में पूर्व चेतावनी को फायर करने के लिए इसका उपयोग करें.

from datetime import datetime, timezone, timedelta

LEAD_MINUTES = 10   # fire a pre-alert this many minutes before the release


def fetch_calendar(currency: str) -> list[dict]:
    try:
        resp = requests.get(
            f"{BASE_URL}/calendar/{currency}",
            params={"api_key": API_KEY},
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json().get("data", [])
    except requests.RequestException as exc:
        print(f"[WARN] calendar fetch failed for {currency}: {exc}")
        return []


def check_upcoming(state: dict) -> None:
    now = datetime.now(tz=timezone.utc)
    for currency, _ in WATCHLIST:
        for event in fetch_calendar(currency):
            scheduled = event.get("announcement_datetime")
            if not scheduled:
                continue
            try:
                event_dt = datetime.fromisoformat(scheduled.replace("Z", "+00:00"))
            except ValueError:
                continue
            # Fire pre-alert if the event is within the lead window and not yet fired
            pre_key = f"pre:{currency}/{event.get('indicator')}:{scheduled}"
            delta   = event_dt - now
            if timedelta(0) < delta <= timedelta(minutes=LEAD_MINUTES):
                if pre_key not in state:
                    send_pre_alert(WEBHOOK_URL, event, currency, delta)
                    state[pre_key] = True
                    save_state(state)


def send_pre_alert(webhook_url: str, event: dict, currency: str, delta: timedelta) -> None:
    ccy      = currency.upper()
    ind      = event.get("indicator", "").replace("_", " ").title()
    expected = event.get("expected")
    prior    = event.get("prior")
    mins     = int(delta.total_seconds() / 60)

    lines = [
        f"⏰  *{ccy} {ind}* due in ~{mins} min",
        f"  Expected: {expected}   |   Prior: {prior}",
    ]
    message = "\n".join(lines)
    if "discord.com" in webhook_url:
        payload = {"content": message.replace("*", "**")}
    else:
        payload = {"text": message}
    try:
        resp = requests.post(webhook_url, json=payload, timeout=10)
        resp.raise_for_status()
        print(f"[OK] pre-alert sent for {ccy} {ind}")
    except requests.RequestException as exc:
        print(f"[ERROR] pre-alert delivery failed: {exc}")

जोड़ें check_upcoming मतदान कार्यक्रम के साथ-साथ check_all

def main():
    state = load_state()
    check_all(state)
    check_upcoming(state)
    schedule.every(5).minutes.do(check_all,      state=state)
    schedule.every(5).minutes.do(check_upcoming, state=state)
    while True:
        schedule.run_pending()
        time.sleep(30)

चरण 6 दीर्घकालिक सेवा के रूप में तैनात करें

उत्पादन उपयोग के लिए आप चाहते हैं कि बॉट टर्मिनल सत्र के बिना लगातार चलाए। नीचे दो सबसे आम तैनाती पैटर्न हैं।

विकल्प A सिस्टम इकाई (लिनक्स सर्वर / VPS)

# /etc/systemd/system/macro-alert-bot.service
[Unit]
Description=FXMacroData Alert Bot
After=network.target

[Service]
Type=simple
User=youruser
WorkingDirectory=/opt/macro-alert-bot
ExecStart=/opt/macro-alert-bot/.venv/bin/python macro_alert_bot.py
Restart=always
RestartSec=30
Environment="FXMACRO_API_KEY=YOUR_FXMACRODATA_KEY"
Environment="WEBHOOK_URL=https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now macro-alert-bot

विकल्प B डॉकर कंटेनर

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY macro_alert_bot.py .
CMD ["python", "macro_alert_bot.py"]

docker build -t macro-alert-bot .
docker run -d --name macro-alert-bot --restart unless-stopped \
  -e FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY" \
  -e WEBHOOK_URL="https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK" \
  macro-alert-bot

requests>=2.31
schedule>=1.2

सारांश

अब आपके पास एक काम कर रहे मैक्रो अलर्ट पाइपलाइन है।

• ✅ FXMacroData के सर्वेक्षण घोषणाएं अंत बिंदु हर 5 मिनट में आपकी निगरानी सूची में नई रिलीज़ के लिए
• ✅ पुनः आरंभ के दौरान दोहराए जाने वाले अलर्ट से बचने के लिए राज्य पर कायम है
• ✅ स्लैक, डिस्कॉर्ड या किसी भी HTTP लक्ष्य को स्वरूपित वेबहूक संदेश भेजता है
• ✅ आग से पहले रिलीज उलटी गिनती अलर्ट का उपयोग कर रिलीज कैलेंडर विन्यास योग्य नेतृत्व समय के साथ
• ✅ सिस्टम सेवा या डॉकर कंटेनर के रूप में लगातार चलाता है