Cómo crear un bot de alerta de calendario de lanzamiento (Telegram / Discord)

Lo que construirás

Las publicaciones de datos macro mueven los mercados rápidamente. Una impresión sorpresa del IPC, una decisión inesperada de tasa o una cifra de empleo mejor de lo esperado pueden cambiar el EUR/USD en 50 pips en segundos. Si no está viendo el calendario en tiempo real, ya está reaccionando a un mercado que se ha movido. Esta guía le muestra cómo construir un bot Python ligero que encuesta el calendário de lanzamiento de FXMacroData y dispara alertas instantáneas a El telegrama ¿ Qué ? No hay discordia. el momento en que se acerca un evento de gran impacto o se publica un resultado.

Al final de este artículo tendrás un bot que funciona:

• Obtiene los próximos eventos macro de la punto final del calendario de liberación
• Filtros por moneda y nivel de impacto para que solo reciba alertas que sean importantes para su lista de vigilancia
• Envía un mensaje de cuenta regresiva de pre-lanzamiento a Telegram y / o Discord en un tiempo de entrega configurable (por ejemplo , 5 minutos antes)
• Se dispara una alerta de seguimiento con la lectura real vs. esperada vs. previa una vez que se imprime la versión
• Se ejecuta continuamente como un bucle programado no se requiere trabajo cron

Los requisitos previos

Antes de comenzar, necesitará lo siguiente:

• Python 3.9+ todos los fragmentos utilizan una sintaxis de escritura estándar
• La clave de la API de FXMacroData inscribirse en / suscribirse y copia tu llave del tablero de instrumentos
• El bot de Telegram es un token (opcional) crear un bot a través de @BotFather en Telegram y anote su ID de chat
• URL de la conexión de Discord (opcional) crear un webhook en cualquier canal de Discord bajo Configuración → Integraciones → Webhooks
• Paquetes de Python¿ Qué ? requests¿ Qué ? schedule

pip install requests schedule

Almacenar credenciales como variables de entorno nunca fichas de código duro en archivos fuente:

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"

Paso 1: Busca el calendario de lanzamiento

El punto final del calendario de lanzamiento devuelve todos los eventos macro programados para una moneda determinada, incluido el valor esperado, la lectura previa y una vez lanzado la cifra real. announcement_datetime El campo es una marca de tiempo UTC ISO 8601 hasta el segundo, que es lo que impulsa la lógica de tiempo de alerta.

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

Cada artículo en data tiene campos incluyendo indicator¿ Qué ? announcement_datetime¿ Qué ? expected¿ Qué ? prior, y después de la liberación actualUn evento que aún no se ha impreso tendrá actual: null- ¿ Qué ?

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

Paso 2: Filtrar por lista de vigilancia y tiempo de entrega

Probablemente no quieras alertas para cada indicador menor. La función de abajo filtra los eventos a los que se encuentran dentro de una ventana de liderazgo configurable para que el bot pueda enviar una alerta de cuenta regresiva antes de que se lancen los disparos.

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

Paso 3: Enviar alertas de Telegram

La API de Telegram acepta un simple sendMessage La función de abajo forma un mensaje corto y legible adecuado para una notificación push móvil y lo publica en su 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))

Paso 4: Enviar alertas de discordia

Los webhooks de Discord aceptan una carga útil JSON con un content de cuerda y opcional . embeds El uso de una incrustación da a la alerta una franja de color en el lado izquierdo verde para una sorpresa positiva, rojo para una falta lo que facilita el escaneo de un canal ocupado de un vistazo.

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

Paso 5: Conecte el circuito principal

El circuito principal corre cada minuto.

1. Volver a buscar el calendario para cada moneda en la lista de vigilancia
2. Comprueba si algún evento está dentro de la ventana de prealerta (paso 2) y dispara mensajes de cuenta regresiva
3. Comprueba si se ha impreso algún evento desde el último mensaje de resultado de tick y fires
4. Dorme hasta el siguiente ciclo.

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

Paso 6: ejecuta el bot

Guarde el guión completo como calendar_bot.py y ejecutarlo directamente:

python calendar_bot.py

Para una implementación de producción, ejecutarlo dentro de un contenedor Docker o un servicio simple de systemd para que se reinicie en caso de fallo. El bot consume una llamada de API por moneda por minuto muy dentro de los límites estándar del plan FXMacroData.

Se ejecuta con 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

Paso 7: Extienda el bot

El bucle central es intencionalmente mínimo.

El guión completo

Todos los pasos anteriores se combinan en un solo archivo. fetch_calendar, ayudantes de filtro, remitentes de Telegram y Discord, entonces run_cycle ¿ Qué ? main y tienes un bot completamente autónomo en menos de 200 líneas de Python.

El punto final del calendario de liberación utilizado en esta guía se documenta en /api-data-docs/usd/no_agrícola_payrolls para los indicadores USD. Las monedas admitidas incluyen AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD y USD , cada una con su propio conjunto de eventos de liberación de alto impacto.

Resumen de las actividades

Ahora tienes un bot de alerta listo para la producción que:

• Extrae el calendario de lanzamiento de FXMacroData utilizando marcas de tiempo UTC de segundo nivel precisas
• Envía un Cuenta atrás de la pre-lanzamiento a Telegram y / o Discord un número configurable de minutos antes de cada evento
• Envía un resultado posterior a la liberación alerta con valores reales, esperados y anteriores codificado por color para dirección sorpresa en Discord
• Se ejecuta como un bucle Python autónomo con guardias de deduplicación
• Se implementa limpiamente en Docker con configuración de variables de entorno

Un paso natural es emparejar las marcas de tiempo de lanzamiento con el punto final del historial del indicador Para crear un cuadro de resultados histórico de sorpresa cuáles monedas tienden a superar o fallar sus expectativas y utilizarlo para ponderar su posicionamiento previo al lanzamiento. Cuadro de control de divisas para las ideas.