Comment créer un bot d'alerte de calendrier de sortie (Telegram / Discord)

Ce que vous allez construire

Les communiqués de données macro déplacent les marchés rapidement. Une impression surprise de l'IPC, une décision de taux inattendue ou un chiffre d'emploi meilleur que prévu peuvent changer l'EUR/USD de 50 pips en quelques secondes. Si vous ne regardez pas le calendrier en temps réel, vous réagissez déjà à un marché qui a changé. Ce guide vous montre comment construire un bot Python léger qui étudie le calendriel de diffusion de FXMacroData et envoie des alertes instantanées à Télégramme Je suis désolé . La discorde Le moment où un événement à fort impact approche ou où un résultat est publié.

À la fin de cet article, vous aurez un robot qui fonctionne:

• Apporte les événements macro à venir à partir de la point final du calendrier de sortie
• Filtres par devise et niveau d'impact pour que vous ne receviez que les alertes qui comptent pour votre liste de surveillance
• Envoie un message de compte à rebours de pré-version à Telegram et / ou Discord à une heure de livraison configurable (par exemple 5 minutes avant)
• Déclenche une alerte de suivi avec la lecture réelle ou prévue une fois la sortie imprimée
• Exécute en continu comme une boucle programmée pas de travail cron requis

Pré-requis

Vous aurez besoin de ce qui suit avant de commencer:

• Python 3.9+ tous les extraits utilisent une syntaxe de saisie standard
• Clé de l'API FXMacroData inscrivez-vous à / souscrivez et copiez votre clé du tableau de bord
• Le jeton du bot Telegram (facultatif) créer un bot via @BotFather sur Telegram et notez votre ID de chat
• URL du lien web de Discord (facultatif) créer un webhook dans n'importe quel canal Discord sous Paramètres → Intégrations → Webhooks
• Paquets PythonJe suis désolé . requestsJe suis désolé . schedule

pip install requests schedule

Stockez les identifiants comme variables d'environnement ne jamais mettre des jetons de code dur dans les fichiers source:

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"

Étape 1: Trouvez le calendrier des sorties

Le point d'extrémité du calendrier de sortie renvoie chaque événement macro prévu pour une devise donnée, y compris la valeur attendue, la lecture précédente et une fois publié le chiffre réel. announcement_datetime Le champ est un horodatage UTC ISO 8601 jusqu'à la seconde, ce qui est ce qui conduit la logique de chronométrage d'alerte.

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

Chaque article est à l' intérieur . data Il y a des champs comme indicatorJe suis désolé . announcement_datetimeJe suis désolé . expectedJe suis désolé . prior, et après la sortie actualUn événement qui n'a pas encore été publié aura actual: nullJe suis désolé .

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

Étape 2: Filtrer par liste de surveillance et temps de traitement

Vous ne voulez probablement pas d'alertes pour chaque indicateur mineur. La fonction ci-dessous filtre les événements à ceux qui tombent dans une fenêtre de plomb configurable afin que le bot puisse envoyer une alerte de compte à rebours avant le lancement des tirs.

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

Étape 3: Envoyer des alertes par télégramme

L'API du bot de Telegram accepte une simple sendMessage La fonction ci-dessous formatera un message court et lisible adapté à une notification push mobile et le publiera dans votre 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))

Étape 4: Envoyer des alertes de désaccord

Les webhooks Discord acceptent une charge utile JSON avec un content - Une chaîne et une option . embeds L'utilisation d'un embed donne à l'alerte une bande colorée sur le côté gauche verte pour une surprise positive, rouge pour une manquement ce qui facilite la numérisation d'une chaîne occupée en un coup d'œil.

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

Étape 5: Branchez la boucle principale

La boucle principale tourne toutes les minutes.

1. Retrouve le calendrier pour chaque devise de la liste de surveillance
2. Vérifie si un événement se trouve dans la fenêtre de pré-alerte (étape 2) et déclenche des messages de compte à rebours
3. Vérifie si un événement a été imprimé depuis le précédent tick et les messages de résultat de feu
4. Il dort jusqu'au prochain cycle.

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

Étape 6: Exécuter le bot

Enregistrez le script complet comme calendar_bot.py et l' exécuter directement:

python calendar_bot.py

Pour un déploiement de production, exécutez-le à l'intérieur d'un conteneur Docker ou d'une simple solution systemd afin qu'il redémarre en cas de panne.

Exécuter avec 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

Étape 7: étendre le bot

La boucle de base est intentionnellement minimale.

Scénario complet

Toutes les étapes ci-dessus sont combinées en un seul fichier. fetch_calendar, les aides de filtrage, les expéditeurs de Telegram et Discord, puis run_cycle Je suis désolé . main et vous avez un bot entièrement autonome en moins de 200 lignes de Python.

Le point final du calendrier de sortie utilisé dans ce guide est documenté à /api-data-docs/usd/non_farm_payrolls Les données sur les salaires des travailleurs agricoles pour les indicateurs USD. Les monnaies prises en charge comprennent AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD et USD , chacune avec son propre ensemble d'événements de libération à fort impact.

Résumé

Vous avez maintenant un robot d'alerte prêt à être produit qui:

• Extrait le calendrier de sortie de FXMacroData en utilisant des horodatages UTC de deuxième niveau précis
• Il envoie un compte à rebours de pré-libération à Telegram et / ou Discord un nombre configurable de minutes avant chaque événement
• Il envoie un Résultat après la libération Alerte avec les valeurs réelles, attendues et antérieures codées par couleur pour une direction surprise sur Discord
• Exécute une boucle Python autonome avec des gardiens de déduplication
• Déploiement propre dans Docker avec configuration variable environnement

Une prochaine étape naturelle est de coupler les horodatages de sortie avec le point final de l'historique des indicateurs Pour créer un tableau de bord historique de surprise sur les devises qui ont tendance à dépasser ou à déjouer leurs attentes et l'utiliser pour peser votre positionnement avant la sortie. Tableau de bord FX Pour des idées.