Wie man einen Release Kalender Alert Bot (Telegram / Discord) erstellt

Was du bauen wirst

Makrodatenveröffentlichungen bewegen die Märkte schnell. Ein überraschender Preisindexdruck, eine unerwartete Zinsentscheidung oder eine bessere als erwartete Beschäftigungszahl können EUR/USD in Sekunden um 50 Pips verschieben. Wenn Sie den Kalender nicht in Echtzeit beobachten, reagieren Sie bereits auf einen Markt, der sich bewegt hat. Dieser Leitfaden zeigt Ihnen, wie Sie einen leichten Python-Bot erstellen, der den FXMacroData-Veröffentlichungskalender befragt und sofortige Warnungen an Telegramm Und ... Unterschied Der Zeitpunkt, zu dem sich ein Ereignis mit hoher Wirkung nähert oder ein Ergebnis veröffentlicht wird.

Am Ende dieses Artikels haben Sie einen funktionierenden Bot, der:

• Holt die kommenden Makro-Ereignisse aus dem Endpunkt des Release-Kalenders
• Filter nach Währung und Einflussniveau, damit Sie nur Benachrichtigungen erhalten, die für Ihre Beobachtungsliste wichtig sind
• Sendet eine Nachricht zum Rückzählvorgang an Telegramm und / oder Discord zu einer konfigurierbaren Vorlaufzeit (z. B. 5 Minuten vorher)
• Veröffentlicht eine Nachmeldung mit dem tatsächlichen oder erwarteten oder vorherigen Wert, sobald die Ausgabe gedruckt wurde
• Laufen kontinuierlich als geplante Schleife kein Cron-Job erforderlich

Voraussetzungen

Vor dem Start benötigen Sie Folgendes:

• Python 3.9+ alle Snippets verwenden eine Standard-Syntax
• FXMacroData-API-Schlüssel melden Sie sich an /abonnieren und kopieren Sie Ihren Schlüssel vom Armaturenbrett
• Telegramm-Bot-Token (optional) einen Bot erstellen @BotFather auf Telegramm und notieren Sie Ihre Chat-ID
• URL des Discord-Webhooks (optional) Erstellen Sie einen Webhook in jedem Discord-Kanal unter Einstellungen → Integrationen → Webhooks
• Python-Pakete- Ich weiß . requests- Ich weiß . schedule

pip install requests schedule

Speichern Sie Anmeldeinformationen als Umgebungsvariablen niemals Hardcode-Token in Quelldateien:

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"

Schritt 1: Holen Sie sich den Kalender der Veröffentlichungen

Der Endpunkt des Release-Kalenders gibt jedes geplante Makro-Ereignis für eine bestimmte Währung zurück, einschließlich des erwarteten Wertes, der vorherigen Messung und nach der Veröffentlichung der tatsächlichen Zahl. announcement_datetime Das Feld ist ein UTC ISO 8601 Zeitstempel bis auf die Sekunde, was die Alarmzeitlogik steuert.

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

Jeder Artikel ist hier . data hat Felder einschließlich indicator- Ich weiß . announcement_datetime- Ich weiß . expected- Ich weiß . prior, und nach der Veröffentlichung actualEin Ereignis , das noch nicht gedruckt wurde , wird actual: null- Ich weiß .

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

Schritt 2: Filterung nach Beobachtungsliste und Vorlaufzeit

Die folgende Funktion filtert Ereignisse auf diejenigen, die in ein konfigurierbares Lead-Fenster fallen, so dass der Bot eine Countdown-Warnung senden kann, bevor die Freigabe ausgeht.

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

Schritt 3: Telegramm-Benachrichtigungen senden

Die Bots-API von Telegram akzeptiert eine einfache sendMessage Die folgende Funktion formatiert eine kurze, lesbare Nachricht geeignet für eine mobile Push-Benachrichtigung und postet sie in Ihren Chat.

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

Schritt 4: Versenden von Diskordwarnungen

Discord Webhooks akzeptieren eine JSON-Nutzlast mit einem content String und optional . embeds Die Verwendung eines Einbettungsmoduls gibt der Alarm eine farbige Streifen auf der linken Seite grün für eine positive Überraschung, rot für eine Verfehlung , wodurch es einfach ist, einen beschäftigten Kanal auf einen Blick zu scannen.

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,
    })

Schritt 5: Die Hauptschleife vernetzen

Die Hauptschleife läuft jede Minute.

1. Holt den Kalender für jede Währung auf der Beobachtungsliste zurück
2. Überprüft, ob sich ein Ereignis im Vorwarnfenster befindet (Schritt 2) und schießt Countdown-Nachrichten aus
3. Überprüft, ob seit dem letzten Ticks- und Feuersergebnis eine Veranstaltung gedruckt wurde
4. Schläft bis zum nächsten Zyklus

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

Schritt 6: Führen Sie den Bot aus

Speichern Sie das komplette Skript als calendar_bot.py und es direkt ausführen:

python calendar_bot.py

Für eine Produktionsbereitstellung, laufen Sie es in einem Docker-Container oder einem einfachen systemd-Service, so dass es bei einem Ausfall neu gestartet wird. Der Bot verbraucht einen API-Aufruf pro Währung pro Minute gut innerhalb der Standard-FXMacroData-Plan-Grenzen.

Mit Docker ausgeführt

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

Schritt 7: Erweitern Sie den Bot

Die Kernschleife ist absichtlich minimal. Hier sind einige natürliche Erweiterungen:

Vollständiges Drehbuch

Alle oben genannten Schritte werden in einer einzigen Datei kombiniert. fetch_calendar, Filterhelfer, Telegramm und Discord Absender, dann run_cycle Und ... main und Sie haben einen vollständig eigenständigen Bot in weniger als 200 Zeilen Python.

Der in diesem Leitfaden verwendete Endpunkt des Release-Kalenders ist unter /api-data-docs/usd/Nichtlandwirtschaftliche Gehaltslisten Für USD-Indikatoren sind die unterstützten Währungen AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD und USD , die jeweils mit ihren eigenen Release-Ereignissen mit hoher Wirkung verbunden sind.

Zusammenfassung

Sie haben jetzt einen produktionsfähigen Alarmbot , der:

• Zieht den Release-Kalender von FXMacroData mit präzisen UTC-Zeitstempeln auf der zweiten Ebene ab
• Schickt einen Rückzählung vor der Freigabe Sie können Telegram und / oder Discord eine konfigurierbare Anzahl von Minuten vor jedem Ereignis anzeigen
• Schickt einen Ergebnis nach der Freisetzung Alarm mit tatsächlichen, erwarteten und vorherigen Werten Farbcodiert für Überraschungsrichtung auf Discord
• Ausführt als eigenständige Python-Schleife mit Deduplikationsschutz
• Die Bereitstellung in Docker mit Umgebungsvariablen

Ein natürlicher nächster Schritt ist , die Zeitstempel mit dem Endpunkt der Indikatorgeschichte Sie können sich die Ergebnisse der ersten beiden Wochen der Ausgabe anhand einer Überraschungskarte erstellen, welche Währungen ihre Erwartungen übertreffen oder verfehlen und diese verwenden, um Ihre Positionierung vor der Veröffentlichung zu gewichten. FX-Dashboard Für Ideen.