How to Set Up Macro Alerts with FXMacroData and Webhooks

Rilis data makro adalah salah satu katalis yang paling dapat diandalkan di pasar FX. CPI cetak 0,2% di atas konsensus dapat mengirim EUR/USD meluncur 60 pip sebelum berita berita bahkan memuat. Jika sistem peringatan Anda bergantung pada pemeriksaan kalender manual atau jajak pendapat berbasis timer luas, Anda sudah tertinggal. Panduan ini menunjukkan kepada Anda cara membangun pipa peringatan webhook yang tepat yang mendeteksi rilis FXMacroData baru pada saat mereka muncul dan memberikan pemberitahuan ke Slack, Discord, atau titik akhir HTTP semuanya dalam kurang dari 80 baris Python.

pip install requests schedule

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

Langkah 1 Dapatkan nilai pengumuman terbaru

- Apa? pengumuman titik akhir mengembalikan nilai yang dirilis terbaru untuk indikator dan mata uang apapun. announcement_datetime timestamp tingkat dua UTC yang memberi tahu Anda persis kapan nilai ini dipublikasikan. Anda akan menggunakan timestam ini sebagai kunci deduplikasi: jika telah berubah sejak pemeriksaan terakhir Anda, rilis baru telah dicetak.

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

Sebuah sampel respon terlihat seperti:

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

Langkah 2 Track status untuk menghindari peringatan duplikat

Memulai kembali proses Anda seharusnya tidak memicu peringatan untuk rilis yang sudah Anda lihat. announcement_datetime Pada saat startup bot memuat file ini; setelah setiap peringatan baru, ia menulis kembali time stamp yang diperbarui.

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

Langkah 3 Kirim pemberitahuan webhook

Kedua Slack webhook masuk dan webhooks Discord menerima HTTP POST dengan muatan JSON. Fungsi di bawah ini membangun pesan yang diformat dan mengirimnya. Slack menggunakan text bidang; Discord menggunakan content. Pembantu menyesuaikan secara otomatis berdasarkan URL prefiks.

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

Langkah 4 Perhatikan beberapa indikator dalam lingkaran jajak pendapat

Tentukan pasangan mata uang/indikator yang Anda minati dan lakukan pemeriksaan pada interval yang teratur. schedule library membuat ini mudah tanpa cron daemon.

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

Jalankan bot dari terminal Anda:

python macro_alert_bot.py

Perjalanan pertama penuh. alert_state.json Dengan nilai-nilai terbaru saat ini sehingga masa depan berjalan api hanya ketika sesuatu yang benar-benar baru muncul.

Langkah 5 Tambahkan peringatan hitungan mundur pra-pembebasan

Mengetahui bahwa pembebasan terjadi berguna; Mengetahuinya adalah akan terjadi adalah berharga. Tanggal akhir rilis Mengungkap jadwal announcement_datetime Untuk peristiwa di masa depan, termasuk nilai konsensus yang diharapkan. Gunakan ini untuk menembakkan peringatan pra-jumlah menit yang dapat dikonfigurasi sebelum cetak.

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

Tambahkan check_upcoming untuk jadwal pemungutan suara di samping check_allAku tidak tahu.

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)

Langkah 6 Mengerahkan sebagai layanan jangka panjang

Untuk penggunaan produksi Anda akan ingin bot berjalan terus-menerus tanpa sesi terminal.

Opsi A unit systemd (Linux server / 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

Pilihan B Docker container

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

Ringkasan

Anda sekarang memiliki pipa peringatan makro yang bekerja.

• ✅ Survei FXMacroData pengumuman titik akhir setiap 5 menit untuk rilis baru di daftar pengawasan Anda
• ✅ Persistence state untuk menghindari peringatan duplikat di seluruh restart
• ✅ Mengirim pesan webhook berformat ke Slack, Discord, atau target HTTP lainnya
• ✅ Api pra-release countdown peringatan menggunakan Kalender rilis dengan waktu lead yang dapat dikonfigurasi
• ✅ Berjalan terus menerus sebagai layanan systemd atau Docker container