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

ما الذي ستبنيه

تحرك إصدارات البيانات الكلية الأسواق بسرعة. يمكن أن تتحرك EUR / USD بنسبة 50 نقطة في ثوانٍ. إذا لم تكن تشاهد التقويم في الوقت الحقيقي ، فأنت بالفعل تتفاعل مع سوق قد تحرك. يوضح لك هذا الدليل كيفية إنشاء روبوت Python خفيف الوزن يقوم بالاستطلاع على تقويم إصدرات FXMacroData ويطلق تنبيهات فورية إلى تلغرام و الخلاف لحظة اقتراب حدث ذو تأثير كبير أو نشر نتيجة.

بحلول نهاية هذه المقالة سيكون لديك روبوت يعمل

• يحضر الأحداث الكبرى القادمة من نقطة نهاية تقويم الإصدار
• مرشحات حسب العملة ومستوى التأثير حتى تحصل فقط على تنبيهات ذات أهمية في قائمة المراقبة الخاصة بك
• يرسل رسالة إعادة العد إلى Telegram و / أو Discord في وقت قياسي قابل للتكوين (مثل 5 دقائق قبل)
• يطلق تنبيه متابعة مع القراءة الفعلية مقابل المتوقعة مقابل القراءات السابقة بمجرد طباعة الإصدار
• يتم تشغيلها بشكل مستمر كحلقة محددة لا حاجة إلى وظيفة كرون

الشروط المسبقة

ستحتاج إلى ما يلي قبل البدء:

• بايثون 3.9+ جميع المقتطفات تستخدم نحو المطبوعات القياسية
• مفتاح FXMacroData API التسجيل في / اشترك ونسخ مفتاحك من لوحة القيادة
• رمز الـ (تليجرام) (اختياري) إنشاء روبوت عبر @BotFather على تلغرام ولاحظوا معرف الدردشة الخاص بكم
• عنوان URL لـ Discord webhook (اختياري) إنشاء شبكة في أي قناة في Discord تحت الإعدادات → التكاملات → شبكات الويب
• حزم بايثون. 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"

الخطوة الأولى: احصل على تقويم الإصدارات

تعيد نقطة نهاية تقويم الإصدار كل حدث ماكرو مخطط له لعملة معينة، بما في ذلك القيمة المتوقعة، القراءة السابقة، و بمجرد الإصلاح الرقم الفعلي. 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

الخطوة الثالثة: إرسال تنبيهات التليجرام

تطبيق برمجيات التشغيل الالكتروني لـ (تليجرام) يقبل بـ 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,
    })

الخطوة الخامسة: سلك الحلقة الرئيسية

الحلقة الرئيسية تعمل كل دقيقة

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

الخطوة السادسة: تشغيل الروبوت

حفظ النص الكامل كما calendar_bot.py و أديرها مباشرة

python calendar_bot.py

بالنسبة لنشر الإنتاج، قم بتشغيله داخل حاوية دوكر أو خدمة سستمد بسيطة حتى يتم إعادة تشغيله عند الفشل. يستهلك الروبوت مكالمة API واحدة لكل عملة في الدقيقة ضمن حدود خطة FXMacroData القياسية.

تشغيل مع Docker

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

الخطوة السابعة: توسيع الروبوت

الحلقة الأساسية هي الحد الأدنى عمداً.

النص الكامل

كل الخطوات أعلاه تجمع في ملف واحد. نسخ الكتل في ترتيب الاستيراد، الثوابت، fetch_calendar، مساعدي الفلتر، مبعثي التليجرام وديسكورد، ثم run_cycle و main و لديك بوت مستقل بالكامل في أقل من 200 سطر من Python

يتم توثيق نقطة نهاية تقويم الإصدار المستخدمة في هذا الدليل في /api-data-docs/usd/غير_فلاح_الرواتب للدولار الأمريكي. تشمل العملات المدعومة الدولار الأسترالي، البريطاني، الكادية، الفرنك السويسري، اليورو، الدينماركي، اليور، الجنيه الإسترالي والين الياباني، الدولار النيوزيلندي، البلنك الروسي، الكرونة السويدية، الديرة الصينية، والدولار الأميركي كل منها مع مجموعة من أحداث الإصدار ذات التأثير العالي.

ملخص

لديك الآن روبوت تنبيه جاهز للإنتاج

• يسحب تقويم الإصدار من FXMacroData باستخدام طوابع زمنية دقيقة من المستوى الثاني UTC
• أرسل العد التنازلي قبل الإفراج إلى Telegram و / أو Discord عددًا قابلًا للتكوين من الدقائق قبل كل حدث
• أرسل نتيجة بعد الإفراج تنبيه مع القيم الفعلية والمتوقعة والقيم السابقة مشفرة بالألوان للتوجيه المفاجئ على Discord
• يعمل كحلقة بايثون ذاتية الحكم مع حراس التكرار
• يتم توزيعها بشكل نظيف في دوكر مع تكوين متغير البيئة

الخطوة التالية الطبيعية هي ربط العلامات الزمنية للخروج مع النقطة النهائية لتاريخ المؤشر لإنشاء بطاقة نتيجة مفاجأة تاريخية والتي تميل العملات إلى التفوق أو تخطي توقعاتها واستخدامها لوزن موقعك قبل الإصدار. استكشاف كتالوج المؤشرات الكامل على لوحة المعلومات من أجل الأفكار