Comment configurer les alertes macro avec FXMacroData et Webhooks

Les communiqués de données macro sont parmi les catalyseurs les plus fiables sur les marchés des changes. Une impression de l'IPC de 0,2% au-dessus du consensus peut faire glisser l'EUR/USD de 60 pips avant même que le titre de la nouvelle ne se charge. Si votre système d'alerte repose sur des vérifications manuelles du calendrier ou un vaste sondage basé sur un chronomètre, vous êtes déjà en retard. Ce guide vous montre comment créer un pipeline d'avertissement précis, basé en webhook, qui détecte les nouvelles communiqués FXMacroData au moment où ils apparaissent et envoie des notifications à Slack, Discord ou tout autre point de terminaison HTTP en moins de 80 lignes de Python.

pip install requests schedule

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

Étape 1 Récupérer la dernière valeur d'annonce

Le annonces point final renvoie la valeur la plus récente pour tout indicateur et devise. announcement_datetime un horodatage UTC de deuxième niveau qui vous indique exactement quand cette valeur a été publiée. Vous utiliserez cet horodatagé comme clé de déduplication: s'il a changé depuis votre dernière vérification, une nouvelle version a été imprimée.

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

Une réponse d'échantillon ressemble à:

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

Étape 2 État de la piste pour éviter les signalements en double

Un redémarrage de votre processus ne devrait pas réactiver les alertes pour les sorties que vous avez déjà vues. announcement_datetime Pour chaque indicateur observé dans un simple fichier JSON. Au démarrage, le bot charge ce fichiers; après chaque nouvelle alerte, il écrit l'horodatage mis à jour.

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

Étape 3 Envoyer une notification de webhook

Les deux webhooks entrants de Slack et les webhocks Discord acceptent un HTTP POST avec une charge utile JSON. La fonction ci-dessous construit un message formaté et l'envoie. Slack utilise un text Le champ de discorde utilise contentL'aide s'adapte automatiquement en fonction du préfixe URL.

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

Étape 4 Observez plusieurs indicateurs dans une boucle de sondage

Définissez les paires de devises/indicateurs qui vous intéressent et effectuez la vérification à intervalles réguliers. schedule La bibliothèque rend cela simple sans un démon cron.

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

Exécutez le robot depuis votre terminal:

python macro_alert_bot.py

La première course est pleine . alert_state.json Avec les dernières valeurs actuelles, le futur ne fonctionne que lorsque quelque chose de vraiment nouveau apparaît.

Étape 5 Ajouter des alertes de compte à rebours avant la sortie

Savoir qu'une libération s'est produite est utile; le savoir l'est C' est sur le point de se produire. Il est précieux. point final du calendrier de sortie Il est prévu que le announcement_datetime Pour les événements futurs, y compris les valeurs de consensus attendues, utilisez ceci pour déclencher une pré-alerte un nombre configurable de minutes avant l'impression.

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

Ajoutez . check_upcoming à l' agenda des élections à côté check_allJe suis désolé .

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)

Étape 6 Déploiement en tant que service à long terme

Pour une utilisation en production, vous voulez que le bot fonctionne en continu sans session de terminal.

Option A unité système (serveur Linux / 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

Option B conteneur Docker

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

Résumé

Vous avez maintenant un pipeline d'alerte macro fonctionnel.

• ✅ Les sondages effectués par FXMacroData annonces point final toutes les 5 minutes pour les nouvelles versions de votre liste de surveillance
• ✅ État persistant pour éviter les alertes en double lors des redémarrages
• ✅ Envoie des messages webhook formatés à Slack, Discord ou à toute autre cible HTTP
• ✅ Les incendies pré-libération des alertes de compte à rebours en utilisant le calendrier de sortie avec un délai de livraison configurable
• ✅ Exécute en continu sous forme de service systemd ou de conteneur Docker