How to Build a Release Calendar Alert Bot (Telegram / Discord)

आप क्या बनाएंगे

मैक्रो डेटा रिलीज़ बाजारों को तेजी से स्थानांतरित करते हैं. एक आश्चर्यजनक सीपीआई प्रिंट, एक अप्रत्याशित दर निर्णय, या एक बेहतर-से-अपेक्षित रोजगार आंकड़ा सेकंड में 50 पिप्स से EUR / USD को स्थानांतरित कर सकता है। यदि आप वास्तविक समय में कैलेंडर नहीं देख रहे हैं, तो आप पहले से ही एक बाजार पर प्रतिक्रिया कर रहे हैं जो स्थानांतरित हो गया है। यह गाइड आपको दिखाता है कि एक हल्के पायथन बॉट कैसे बनाया जाए जो FXMacroData रिलीज कैलेंڈر को पोल करता है और तुरंत अलर्ट निकालता है। टेलीग्राम और विभेद जब कोई उच्च प्रभाव वाली घटना आ रही हो या परिणाम प्रकाशित हो रहा हो।

इस लेख के अंत तक आपके पास एक कामकाजी बॉट होगा जोः

• से आगामी मैक्रो घटनाओं को निकालता है रिलीज कैलेंडर अंत बिंदु
• मुद्रा और प्रभाव स्तर के अनुसार फ़िल्टर करें ताकि आपको केवल आपकी निगरानी सूची के लिए महत्वपूर्ण अलर्ट प्राप्त हों
• कॉन्फ़िगर करने योग्य लीड समय पर टेलीग्राम और / या डिस्कॉर्ड को पूर्व-रिलीज़ उलटी गिनती संदेश भेजता है (उदाहरण के लिए 5 मिनट पहले)
• रिलीज़ प्रिंट होने के बाद वास्तविक बनाम अपेक्षित बनाम पूर्व पढ़ने के साथ एक अनुवर्ती अलर्ट चलाता है
• अनुसूचित लूप के रूप में लगातार चलाता है कोई क्रॉन नौकरी की आवश्यकता नहीं है

पूर्व शर्तें

शुरू करने से पहले आपको निम्नलिखित की आवश्यकता होगी:

• पायथन 3.9+ सभी स्निपेट्स मानक टाइपिंग सिंटैक्स का उपयोग करते हैं
• FXMacroData एपीआई कुंजी पर साइन अप करें /अपना नाम लिखें और डैशबोर्ड से अपनी कुंजी की प्रतिलिपि
• टेलीग्राम बॉट टोकन (वैकल्पिक) के माध्यम से एक बॉट बनाएँ @BotFather टेलीग्राम पर और अपनी चैट आईडी को नोट करें
• डिस्कॉर्ड वेबहूक URL (वैकल्पिक) किसी भी डिस्कॉर्ड चैनल में वेबहुक बनाएँ सेटिंग्स → एकीकरण → वेबहूक्स
• पायथन पैकेज requests, schedule

pip install requests schedule

स्रोत फ़ाइलों में पर्यावरण चर के रूप में क्रेडेंशियल्स स्टोर करें हार्ड-कोड टोकन कभी नहींः

export FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY"
export TELEGRAM_BOT_TOKEN="YOUR_TELEGRAM_BOT_TOKEN"
export TELEGRAM_CHAT_ID="YOUR_TELEGRAM_CHAT_ID"
export DISCORD_WEBHOOK_URL="YOUR_DISCORD_WEBHOOK_URL"

चरण 1: रिलीज कैलेंडर प्राप्त करें

रिलीज़ कैलेंडर एंडपॉइंट किसी दिए गए मुद्रा के लिए प्रत्येक अनुसूचित मैक्रो इवेंट को रिटर्न करता है, जिसमें अपेक्षित मूल्य, पूर्व पढ़ने और एक बार रिलीज होने पर वास्तविक आंकड़ा शामिल है। announcement_datetime यह क्षेत्र UTC ISO 8601 टाइमस्टैम्प है जो सेकंड तक है, जो अलर्ट टाइमिंग लॉजिक को चलाता है।

import os
import requests
from datetime import datetime, timezone

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


def fetch_calendar(currency: str) -> list[dict]:
    """Return upcoming releases for a given currency."""
    resp = requests.get(
        f"{BASE_URL}/calendar/{currency}",
        params={"api_key": API_KEY},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])


# Fetch upcoming events for USD and EUR
usd_events = fetch_calendar("usd")
eur_events = fetch_calendar("eur")

for event in usd_events[:3]:
    print(event["indicator"], event.get("announcement_datetime"), event.get("expected"))

प्रत्येक वस्तु में data सहित क्षेत्रों है indicator, announcement_datetime, expected, prior, और रिलीज के बाद actual. एक घटना जो अभी तक मुद्रित नहीं किया गया है होगा actual: null. .

{
  "indicator": "non_farm_payrolls",
  "announcement_datetime": "2026-05-02T12:30:00Z",
  "expected": 185000,
  "prior": 228000,
  "actual": null
}

चरण 2: निगरानी सूची और लीड टाइम द्वारा फ़िल्टर करें

आप शायद हर मामूली संकेतक के लिए अलर्ट नहीं चाहते हैं. नीचे दिया गया फ़ंक्शन उन घटनाओं को फ़िल्टर करता है जो एक विन्यास योग्य लीड विंडो के भीतर आती हैं ताकि बॉट रिलीज़ फायर से पहले उलटी गिनती अलर्ट भेज सके.

from datetime import timedelta

# Indicators worth alerting on — edit to match your watchlist
HIGH_IMPACT = {
    "usd": ["non_farm_payrolls", "inflation", "policy_rate", "gdp_quarterly", "initial_jobless_claims"],
    "eur": ["inflation", "policy_rate", "gdp_quarterly"],
    "gbp": ["inflation", "policy_rate", "employment"],
    "aud": ["policy_rate", "employment", "inflation"],
    "jpy": ["policy_rate", "inflation"],
}

# How many minutes before the release to send the pre-alert
LEAD_MINUTES = 5


def events_due_soon(
    events: list[dict],
    currency: str,
    now: datetime,
    lead_minutes: int = LEAD_MINUTES,
) -> list[dict]:
    """Return events whose announcement_datetime is within the next lead_minutes."""
    watchlist = HIGH_IMPACT.get(currency.lower(), [])
    results = []
    window_end = now + timedelta(minutes=lead_minutes)

    for event in events:
        if event.get("actual") is not None:
            continue  # already released
        if watchlist and event.get("indicator") not in watchlist:
            continue  # not on watchlist

        ann_str = event.get("announcement_datetime")
        if not ann_str:
            continue

        ann_dt = datetime.fromisoformat(ann_str.replace("Z", "+00:00"))
        if now <= ann_dt <= window_end:
            results.append(event)

    return results


def events_just_released(
    events: list[dict],
    currency: str,
    since: datetime,
) -> list[dict]:
    """Return events that have printed since the last check cycle."""
    watchlist = HIGH_IMPACT.get(currency.lower(), [])
    results = []

    for event in events:
        if event.get("actual") is None:
            continue  # not released yet
        if watchlist and event.get("indicator") not in watchlist:
            continue

        ann_str = event.get("announcement_datetime")
        if not ann_str:
            continue

        ann_dt = datetime.fromisoformat(ann_str.replace("Z", "+00:00"))
        if ann_dt >= since:
            results.append(event)

    return results

चरण 3: टेलीग्राम अलर्ट भेजें

टेलीग्राम का बॉट एपीआई एक सरल sendMessage HTTP POST. नीचे दिया गया फ़ंक्शन एक छोटा, पठनीय संदेश मोबाइल पुश अधिसूचना के लिए उपयुक्त को स्वरूपित करता है और इसे आपकी चैट में पोस्ट करता है.

TELEGRAM_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")


def _fmt_indicator(raw: str) -> str:
    return raw.replace("_", " ").title()


def telegram_send(text: str) -> None:
    """Post a plain-text message to a Telegram chat."""
    if not TELEGRAM_TOKEN or not TELEGRAM_CHAT_ID:
        return

    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
        json={"chat_id": TELEGRAM_CHAT_ID, "text": text, "parse_mode": "Markdown"},
        timeout=8,
    )


def telegram_pre_release(currency: str, event: dict, minutes_left: int) -> None:
    indicator = _fmt_indicator(event["indicator"])
    ann_dt = event["announcement_datetime"]
    expected = event.get("expected")
    prior = event.get("prior")

    lines = [
        f"⏰ *{currency.upper()} — {indicator}* in ~{minutes_left} min",
        f"🕐 Release: `{ann_dt}`",
    ]
    if expected is not None:
        lines.append(f"📌 Expected: `{expected}`")
    if prior is not None:
        lines.append(f"📋 Prior: `{prior}`")

    telegram_send("\n".join(lines))


def telegram_post_release(currency: str, event: dict) -> None:
    indicator = _fmt_indicator(event["indicator"])
    actual = event.get("actual")
    expected = event.get("expected")
    prior = event.get("prior")

    if actual is None:
        return

    surprise = ""
    if expected is not None:
        diff = float(actual) - float(expected)
        surprise = f"  ({'▲' if diff > 0 else '▼'} {abs(diff):.1f} vs exp)"

    lines = [
        f"📣 *{currency.upper()} — {indicator}* RELEASED",
        f"✅ Actual: `{actual}`{surprise}",
    ]
    if expected is not None:
        lines.append(f"📌 Expected: `{expected}`")
    if prior is not None:
        lines.append(f"📋 Prior: `{prior}`")

    telegram_send("\n".join(lines))

चरण 4: असहमति अलर्ट भेजें

डिस्कॉर्ड वेबहूक्स एक के साथ एक JSON पेलोड स्वीकार करते हैं content स्ट्रिंग और वैकल्पिक embeds एम्बेड का उपयोग करने से अलर्ट को बाईं ओर एक रंगीन पट्टी मिलती है सकारात्मक आश्चर्य के लिए हरा, मिस के लिए लाल जो एक नज़र में व्यस्त चैनल को स्कैन करना आसान बनाता है।

DISCORD_WEBHOOK = os.environ.get("DISCORD_WEBHOOK_URL", "")

_COLORS = {
    "neutral": 0x3B82F6,   # blue
    "beat": 0x16A34A,       # green
    "miss": 0xDC2626,       # red
}


def discord_send(embed: dict) -> None:
    """Post an embed to a Discord webhook."""
    if not DISCORD_WEBHOOK:
        return

    requests.post(
        DISCORD_WEBHOOK,
        json={"embeds": [embed]},
        timeout=8,
    )


def discord_pre_release(currency: str, event: dict, minutes_left: int) -> None:
    indicator = _fmt_indicator(event["indicator"])
    ann_dt = event["announcement_datetime"]
    expected = event.get("expected")
    prior = event.get("prior")

    fields = [
        {"name": "Release time (UTC)", "value": f"`{ann_dt}`", "inline": True},
    ]
    if expected is not None:
        fields.append({"name": "Expected", "value": str(expected), "inline": True})
    if prior is not None:
        fields.append({"name": "Prior", "value": str(prior), "inline": True})

    discord_send({
        "title": f"⏰ {currency.upper()} — {indicator} in ~{minutes_left} min",
        "color": _COLORS["neutral"],
        "fields": fields,
    })


def discord_post_release(currency: str, event: dict) -> None:
    indicator = _fmt_indicator(event["indicator"])
    actual = event.get("actual")
    expected = event.get("expected")
    prior = event.get("prior")

    if actual is None:
        return

    color = _COLORS["neutral"]
    surprise_label = ""
    if expected is not None:
        diff = float(actual) - float(expected)
        if abs(diff) > 0:
            color = _COLORS["beat"] if diff > 0 else _COLORS["miss"]
            surprise_label = f" ({'beat' if diff > 0 else 'missed'} by {abs(diff):.1f})"

    fields = [
        {"name": "Actual", "value": f"**{actual}**{surprise_label}", "inline": True},
    ]
    if expected is not None:
        fields.append({"name": "Expected", "value": str(expected), "inline": True})
    if prior is not None:
        fields.append({"name": "Prior", "value": str(prior), "inline": True})

    discord_send({
        "title": f"📣 {currency.upper()} — {indicator} RELEASED",
        "color": color,
        "fields": fields,
    })

चरण 5: मुख्य लूप को तार लगाएँ

मुख्य लूप हर मिनट चलता है.

1. निगरानी सूची में प्रत्येक मुद्रा के लिए कैलेंडर पुनः प्राप्त करता है
2. जांचता है कि क्या कोई घटना पूर्व चेतावनी विंडो (चरण 2) के भीतर है और उलटी गिनती संदेशों को निकालता है
3. जाँच करता है कि क्या पिछले टिक और फायर परिणाम संदेशों के बाद से कोई घटना मुद्रित किया गया है
4. अगले चक्र तक सोता है

import time
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

WATCHLIST_CURRENCIES = list(HIGH_IMPACT.keys())

# Track which pre-release alerts have already been sent this cycle
_alerted_pre: set[str] = set()
# Track which post-release alerts have already been sent
_alerted_post: set[str] = set()


def _event_key(currency: str, event: dict) -> str:
    return f"{currency}:{event['indicator']}:{event.get('announcement_datetime','')}"


def run_cycle() -> None:
    now = datetime.now(timezone.utc)
    logger.info("Running calendar check at %s", now.isoformat())

    for currency in WATCHLIST_CURRENCIES:
        try:
            events = fetch_calendar(currency)
        except Exception as exc:  # noqa: BLE001
            logger.warning("Failed to fetch calendar for %s: %s", currency, exc)
            continue

        # Pre-release alerts
        for event in events_due_soon(events, currency, now, lead_minutes=LEAD_MINUTES):
            key = _event_key(currency, event)
            if key not in _alerted_pre:
                ann_dt = datetime.fromisoformat(
                    event["announcement_datetime"].replace("Z", "+00:00")
                )
                minutes_left = max(1, int((ann_dt - now).total_seconds() / 60))
                logger.info("Pre-release alert: %s", key)
                telegram_pre_release(currency, event, minutes_left)
                discord_pre_release(currency, event, minutes_left)
                _alerted_pre.add(key)

        # Post-release alerts (look back 2 minutes to catch releases on previous tick)
        since = now - timedelta(minutes=2)
        for event in events_just_released(events, currency, since):
            key = _event_key(currency, event)
            if key not in _alerted_post:
                logger.info("Post-release alert: %s", key)
                telegram_post_release(currency, event)
                discord_post_release(currency, event)
                _alerted_post.add(key)

    # Prune the alert sets to avoid unbounded growth (keep last 500 keys)
    if len(_alerted_pre) > 500:
        _alerted_pre.clear()
    if len(_alerted_post) > 500:
        _alerted_post.clear()


def main() -> None:
    logger.info("Release calendar alert bot started.")
    while True:
        run_cycle()
        time.sleep(60)


if __name__ == "__main__":
    main()

चरण 6: बॉट चलाएँ

पूरी स्क्रिप्ट को सहेजें calendar_bot.py और इसे सीधे चलाएँ:

python calendar_bot.py

उत्पादन तैनाती के लिए, इसे एक डॉकर कंटेनर या एक साधारण systemd सेवा के अंदर चलाएं ताकि यह विफलता पर पुनरारंभ हो। बॉट प्रति मिनट प्रति मुद्रा एक एपीआई कॉल का उपभोग करता है मानक FXMacroData योजना सीमाओं के भीतर अच्छी तरह से।

डॉकर के साथ चल रहा है

FROM python:3.11-slim
WORKDIR /app
COPY calendar_bot.py .
RUN pip install --no-cache-dir requests schedule
ENV FXMACRO_API_KEY=""
ENV TELEGRAM_BOT_TOKEN=""
ENV TELEGRAM_CHAT_ID=""
ENV DISCORD_WEBHOOK_URL=""
CMD ["python", "calendar_bot.py"]

docker build -t calendar-bot .
docker run -d \
  -e FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY" \
  -e TELEGRAM_BOT_TOKEN="YOUR_BOT_TOKEN" \
  -e TELEGRAM_CHAT_ID="YOUR_CHAT_ID" \
  -e DISCORD_WEBHOOK_URL="YOUR_WEBHOOK_URL" \
  --name calendar-bot \
  calendar-bot

चरण 7: बॉट का विस्तार करें

कोर लूप जानबूझकर न्यूनतम है. यहाँ कुछ प्राकृतिक विस्तार हैंः

पूर्ण स्क्रिप्ट

उपरोक्त सभी चरणों को एक ही फ़ाइल में जोड़ें। क्रम में ब्लॉक की प्रतिलिपि आयात, स्थिरांक, fetch_calendar, फ़िल्टर सहायक, टेलीग्राम और डिस्कॉर्ड भेजने वाले, फिर run_cycle और main और आप पायथन की 200 पंक्तियों से कम में एक पूरी तरह से आत्मनिर्भर बॉट है।

इस गाइड में प्रयुक्त रिलीज कैलेंडर अंत बिंदु पर प्रलेखित है /api-data-docs/usd/गैर-कृषि_पेरोल USD संकेतकों के लिए समर्थित मुद्राओं में AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD और USD शामिल हैं, जिनमें से प्रत्येक के पास उच्च प्रभाव वाली रिलीज़ घटनाओं का अपना सेट है।

सारांश

अब आपके पास एक उत्पादन के लिए तैयार अलर्ट बॉट है जोः

• FXMacroData से रिलीज़ कैलेंडर को दूसरे स्तर के सटीक UTC टाइमस्टैम्प का उपयोग करके खींचता है
• भेजता है पूर्व-रिलीज़ उलटी गिनती प्रत्येक घटना से पहले विन्यस्त मिनटों की संख्या के लिए Telegram और / या Discord
• भेजता है रिलीज के बाद का परिणाम वास्तविक, अपेक्षित और पूर्व के मूल्यों के साथ चेतावनी डिस्कॉर्ड पर आश्चर्य की दिशा के लिए रंग-कोड
• डिडूप्लिकेशन गार्ड्स के साथ एक स्व-निहित पायथन लूप के रूप में चलाता है
• पर्यावरण चर विन्यास के साथ डॉकर में साफ तैनात करता है

एक स्वाभाविक अगला कदम रिलीज समय के साथ जोड़ी है सूचक इतिहास का अंत बिंदु कौन सी मुद्राएँ अपनी अपेक्षाओं को पार करने या चूकने की प्रवृत्ति रखती हैं और इसका उपयोग अपनी पूर्व-रिलीज़ स्थिति को भारित करने के लिए करें। एफएक्स डैशबोर्ड विचारों के लिए।