How to Build a Release Calendar Alert Bot (Telegram / Discord)

Apa yang Akan Anda Bangun

Rilis data makro bergerak pasar dengan cepat. Cetakan CPI kejutan, keputusan suku bunga yang tidak terduga, atau angka pekerjaan yang lebih baik dari yang diharapkan dapat mengubah EUR / USD sebesar 50 pip dalam hitungan detik. Jika Anda tidak menonton kalender secara real time, Anda sudah bereaksi terhadap pasar yang telah bergerak. Panduan ini menunjukkan kepada Anda cara membangun bot Python ringan yang jajak pendapat kalender rilis FXMacroData dan menembakkan peringatan instan ke Telegram Dan Perbedaan saat peristiwa berdampak tinggi mendekati atau hasil dipublikasikan.

Pada akhir artikel ini Anda akan memiliki robot yang bekerja yang:

• Mengambil peristiwa makro mendatang dari Tanggal akhir rilis
• Filter berdasarkan mata uang dan tingkat dampak sehingga Anda hanya mendapatkan peringatan yang penting untuk daftar pantauan Anda
• Mengirim pesan pre-release countdown ke Telegram dan / atau Discord pada waktu lead yang dapat dikonfigurasi (misalnya 5 menit sebelumnya)
• Menembakkan peringatan tindak lanjut dengan aktual vs. diharapkan vs. pembacaan sebelumnya setelah rilis cetak
• Berjalan terus menerus sebagai loop terjadwal tidak diperlukan pekerjaan cron

Persyaratan

Sebelum memulai, Anda perlu hal berikut:

• Python 3.9+ semua cuplikan menggunakan sintaksis penulisan standar
• Kunci API FXMacroData daftar di /langganan dan menyalin kunci Anda dari dasbor
• Token Telegram bot (opsional) membuat bot melalui @BotFather di Telegram dan catat ID obrolan Anda
• URL webhook Discord (opsional) membuat webhook di saluran Discord manapun di bawah Pengaturan → Integrasi → Webhook
• Paket PythonAku tidak tahu. requestsAku akan pergi. schedule

pip install requests schedule

Simpan kredensial sebagai variabel lingkungan jangan pernah hard-code token di file sumber:

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"

Langkah 1: Dapatkan Kalender Rilis

Titik akhir kalender rilis mengembalikan setiap peristiwa makro yang dijadwalkan untuk mata uang tertentu, termasuk nilai yang diharapkan, pembacaan sebelumnya, dan setelah dirilis angka aktual. announcement_datetime bidang adalah timestamp UTC ISO 8601 ke bawah ke kedua, yang merupakan apa yang mendorong logika waktu peringatan.

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

Setiap item di data memiliki bidang termasuk indicatorAku akan pergi. announcement_datetimeAku akan pergi. expectedAku akan pergi. prior, dan setelah rilis actual. Sebuah acara yang belum dicetak akan memiliki actual: null.

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

Langkah 2: Filter berdasarkan Daftar Pengawasan dan Waktu Pelaksanaan

Anda mungkin tidak ingin peringatan untuk setiap indikator kecil. Fungsi di bawah ini menyaring peristiwa untuk mereka yang jatuh dalam jendela lead yang dapat dikonfigurasi sehingga bot dapat mengirim peringatan hitungan mundur sebelum pembebasan menyala.

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

Langkah 3: Kirim Telegram Alerts

Telegram's Bot API menerima sederhana sendMessage HTTP POST. Fungsi di bawah ini memformat pesan singkat dan dapat dibaca cocok untuk pemberitahuan push seluler dan mempostingnya ke obrolan Anda.

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

Langkah 4: Kirim Peringatan Diskord

Webhook Discord menerima muatan JSON dengan content string dan opsional embeds Menggunakan embed memberikan peringatan strip berwarna di sisi kiri hijau untuk kejutan positif, merah untuk miss yang membuat mudah untuk memindai saluran sibuk dengan sekilas.

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

Langkah 5: Hubungkan Lintas Utama

Lintas utama berjalan setiap menit.

1. Mengambil kembali kalender untuk setiap mata uang dalam daftar pengawasan
2. Memeriksa apakah ada peristiwa dalam jendela pra-peringatan (Langkah 2) dan menembakkan pesan hitungan mundur
3. Memeriksa apakah ada acara telah dicetak sejak sebelumnya tanda centang dan api hasil pesan
4. Tidur sampai siklus berikutnya

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

Langkah 6: Jalankan Bot

Simpan skrip lengkap sebagai calendar_bot.py dan jalankan langsung:

python calendar_bot.py

Untuk penyebaran produksi, jalankan di dalam wadah Docker atau layanan systemd sederhana sehingga akan di-restart pada saat gagal. bot mengkonsumsi satu panggilan API per mata uang per menit dalam batas rencana FXMacroData standar.

Menggunakan 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

Langkah 7: Perluas Bot

Lintas inti sengaja minimal. Berikut beberapa ekstensi alami:

Skripsi lengkap

Semua langkah di atas digabungkan menjadi satu file. fetch_calendar, filter helper, Telegram dan Discord pengirim, kemudian run_cycle Dan main dan Anda memiliki bot sepenuhnya mandiri dalam kurang dari 200 baris Python.

Titik akhir kalender rilis yang digunakan dalam panduan ini didokumentasikan di /api-data-docs/usd/non_farm_payrolls untuk indikator USD. Mata uang yang didukung termasuk AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD, dan USD masing-masing dengan serangkaian peristiwa rilis berdampak tinggi.

Ringkasan

Anda sekarang memiliki robot peringatan siap produksi yang:

• Mengambil kalender rilis dari FXMacroData menggunakan timestamps UTC tingkat kedua yang tepat
• Mengirim Penghitungan mundur pra-pembebasan untuk Telegram dan / atau Discord jumlah menit yang dapat dikonfigurasi sebelum setiap acara
• Mengirim Hasil setelah pelepasan peringatan dengan nilai aktual, diharapkan dan sebelumnya kode warna untuk arah kejutan pada Discord
• Berjalan sebagai loop Python mandiri dengan penjaga deduplikasi
• Dilaksanakan dengan bersih di Docker dengan konfigurasi variabel lingkungan

Langkah selanjutnya adalah mencocokkan waktu rilis dengan Endpoint dari sejarah indikator Untuk membangun sebuah skorcard kejutan historis mata uang mana yang cenderung mengalahkan atau gagal harapan mereka dan menggunakannya untuk menimbang posisi pra-pembukaan Anda. Dashboard FX untuk ide.