FXMacroData 및 Webhooks로 매크로 알림을 설정하는 방법

Macro data releases are among the most reliable catalysts in FX markets. A CPI print 0.2% above consensus can send EUR/USD sliding 60 pips before the news headline even loads. If your alert system relies on manual calendar checks or broad timer-based polling, you are already behind. This guide shows you how to build a precise, webhook-driven alert pipeline that detects new FXMacroData releases the moment they appear and delivers notifications to Slack, Discord, or any HTTP endpoint — all in under 80 lines of Python.

pip install requests schedule

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

단계 1 최신 발표 값을 가져오기

- 발표 최종점 모든 지표와 통화에 대한 가장 최근에 공개된 값을 반환합니다. 키 필드는 announcement_datetime — a second-level UTC timestamp that tells you exactly when this value was published. You will use this timestamp as the deduplication key: if it has changed since your last check, a new release has printed.

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

표본 응답은 다음과 같습니다.

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

단계 2 복제된 알림을 피하기 위한 추적 상태

프로세스의 재시작은 이미 보신 출시에 대한 경고를 다시 발사해서는 안 됩니다. announcement_datetime 간단한 JSON 파일에서 관찰된 각 지표에 대한 정보입니다. 시작 시 봇이 이 파일을 로드합니다. 새로운 알림마다 업데이트된 시간표를 다시 작성합니다.

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

단계 3 웹 알림을 보내

슬랙의 온 웹 룩과 디스코드 웹 럭 모두 JSON 페이로드가 있는 HTTP POST를 받아들인다. 아래 함수는 포맷된 메시지를 만들고 전송한다. 슬랙은 을 사용한다. text 이 분야는 디스코드 사용 content. 도움말은 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}")

단계 4 투표 순환에서 여러 지표를 관찰합니다

관심 있는 통화/지표 쌍을 정의하고 주기적으로 체크를 실행하십시오. schedule 라이브러리는 크론 데몬 없이도 간단하게 할 수 있습니다.

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

터미널에서 로봇을 실행하세요.

python macro_alert_bot.py

첫 번째 순환은 꽉 채워집니다 alert_state.json 현재 최신 가치와 함께 미래가 실행되는 것은 뭔가 진정으로 새로운 것이 나타날 때만입니다.

단계 5 출시 전 카운트다운 알림을 추가

해방이 일어났다는 것을 아는 것은 유용합니다. 곧 일어날 것입니다. 가치 있는 것 같아요 발매 달력 최종점 계획된 것을 노출합니다. announcement_datetime 예상된 합의값을 포함한 미래 이벤트에 대한 사전 경고를 발사하기 위해 이것을 사용하십시오.

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

추가해 check_upcoming 투표 일정에 따라 check_all

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)

단계 6 장기 서비스로 배포

생산용으로 사용하려면 봇이 단말기 세션 없이 지속적으로 실행되기를 원합니다. 아래는 두 가지 가장 일반적인 배포 패턴입니다.

옵션 A 시스템 단위 (리눅스 서버 / 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

옵션 B 도커 컨테이너

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

요약

이제 작동하는 매크로 알림 파이프 라인이 있습니다.

• ✅ FXMacroData의 설문조사 발표 최종점 5분마다 새로운 버전의 감시 목록에
• ✅ 재시작 중 중복 경고를 피하기 위해 상태를 지속합니다
• ✅ 포맷된 웹후크 메시지를 Slack, Discord 또는 HTTP 타겟으로 전송합니다.
• ✅ 화재 전 발매 카운트다운 알림 발매 일정은 구성 가능한 전속시간
• ✅ systemd 서비스 또는 Docker 컨테이너로 지속적으로 실행됩니다.