Como criar um bot de alerta de calendário de lançamento (Telegram / Discord)

O que você vai construir

Os lançamentos de dados macro movem os mercados rapidamente. Uma impressão surpresa do IPC, uma decisão inesperada de taxa ou um número de empregos melhor do que o esperado podem mudar o EUR/USD em 50 pips em segundos. Se você não estiver assistindo ao calendário em tempo real, você já está reagindo a um mercado que se moveu. Este guia mostra como criar um bot Python leve que pesquisa o calendário de lançamento do FXMacroData e dispara alertas instantâneos para Telegrama E ... Discordância O momento em que se aproxima um evento de grande impacto ou em que é publicado um resultado.

No final deste artigo você terá um bot que funciona:

• Recolhe os próximos eventos macro do ponto final do calendário de liberação
• Filtros por moeda e nível de impacto para que você receba apenas alertas que sejam importantes para sua lista de observação
• Envia uma mensagem de contagem regressiva pré-lançamento para Telegram e / ou Discord em um tempo de chegada configurável (por exemplo , 5 minutos antes)
• Dispara um alerta de acompanhamento com a leitura real vs. esperada vs. anterior uma vez que a versão é impressa
• Executa-se continuamente como um ciclo programado não é necessário nenhum trabalho cron

Requisitos prévios

Precisa dos seguintes antes de começar:

• Python 3.9+ todos os trechos usam sintaxe de digitação padrão
• Chave da API do FXMacroData Inscreva-se em / subscrever e copiar a chave do painel
• Token do Telegram bot (opcional) criar um bot através de @BotFather no Telegram e anote o seu ID de bate-papo
• URL do webhook do Discord (opcional) criar um webhook em qualquer canal Discord sob Configurações → Integrações → Webhooks
• Pacotes Python- Não . requests- Não . schedule

pip install requests schedule

Armazenar credenciais como variáveis de ambiente nunca token de código rígido em arquivos de origem:

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"

Passo 1: Obtenha o calendário de lançamento

O ponto final do calendário de lançamento retorna todos os eventos macro programados para uma determinada moeda, incluindo o valor esperado, a leitura anterior e uma vez lançado o valor real. announcement_datetime campo é um selo de tempo UTC ISO 8601 até o segundo, que é o que impulsiona a lógica de tempo 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 item dentro . data tem campos incluindo indicator- Não . announcement_datetime- Não . expected- Não . prior, e após a liberação actualUm evento que ainda não foi impresso terá actual: null- Não .

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

Passo 2: Filtrar por lista de vigilância e tempo de execução

A função abaixo filtra eventos para aqueles que se enquadram em uma janela de liderança configurável para que o bot possa enviar um alerta de contagem regressiva antes do lançamento.

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

Passo 3: Enviar Alertas no Telegram

A API do Telegram aceita um simples sendMessage A função abaixo formata uma mensagem curta e legível adequada para uma notificação push móvel e a publica no seu 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))

Passo 4: Enviar alertas de discórdia

Os webhooks do Discord aceitam uma carga útil JSON com um content string e opcional embeds A utilização de um sistema de inserção dá ao alerta uma faixa colorida no lado esquerdo verde para uma surpresa positiva, vermelho para uma falta que facilita a varredura de um canal movimentado num olhar.

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

Passo 5: Conecte o circuito principal

O circuito principal corre a cada minuto.

1. Retorna o calendário para cada moeda na lista de observação
2. Verifica se qualquer evento está dentro da janela de pré-alerta (passo 2) e dispara mensagens de contagem regressiva
3. Verifica se qualquer evento foi impresso desde o resultado anterior do sinal de carraça e fogo
4. Dorme até ao próximo 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()

Passo 6: Execute o Bot

Guarde o guião completo como calendar_bot.py e executá-lo diretamente:

python calendar_bot.py

Para uma implantação de produção, execute-a dentro de um contêiner Docker ou um serviço systemd simples para que ele reinicie em caso de falha.

A execução com 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

Passo 7: Estender o Bot

O loop central é intencionalmente mínimo.

Texto completo

Todos os passos acima combinados em um único arquivo. fetch_calendar, filtros auxiliares, Telegram e Discord remetentes, então run_cycle E ... main e você tem um bot totalmente auto-suficiente em menos de 200 linhas de Python.

O ponto final do calendário de lançamento utilizado neste guia está documentado em /api-data-docs/usd/non_farm_payrolls Para os indicadores USD, as moedas suportadas incluem AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD e USD , cada uma com o seu próprio conjunto de eventos de liberação de alto impacto.

Resumo

Agora tem um bot de alerta pronto para produção que:

• Extrair o calendário de lançamento do FXMacroData usando carimbos de tempo UTC de segundo nível precisos
• Envia um Contagem regressiva de pré-lançamento Para Telegram e / ou Discord um número configurável de minutos antes de cada evento
• Envia um Resultado após a liberação alerta com valores reais, esperados e anteriores codificado por cores para direção surpresa no Discord
• Funciona como um loop Python autônomo com guardas de deduplicação
• Implementação limpa no Docker com configuração variável de ambiente

O próximo passo natural é juntar os horários de lançamento com o ponto final do histórico do indicador Para criar um quadro histórico de pontuação surpresa que as moedas tendem a superar ou falhar as suas expectativas e usar para ponderar o seu posicionamento pré-lançamento. Painel de controlo FX para ideias.