당신 이 건축 할 것

매크로 데이터 발표는 시장을 빠르게 움직입니다. 깜짝 CPI 인쇄, 예상치 못한 비율 결정, 또는 예상보다 더 나은 고용 수치는 초에 50 피프로 EUR / USD를 이동 할 수 있습니다. 당신이 실시간으로 달력을 보지 않으면, 당신은 이미 이동 한 시장에 반응합니다. 이 가이드는 당신이 어떻게 만들지 보여줍니다 가벼운 파이썬 봇 FXMacroData 발표 달력을 조사하고 즉각적인 알림을 발사하는 텔레그램 그리고 불합리 큰 영향력을 가진 사건이 다가오고 있거나 결과가 발표되는 순간입니다.

이 문서의 끝에는 다음과 같은 기능을 하는 로봇이 있습니다.

에서 다가오는 매크로 이벤트를 가져옵니다 발매 달력 최종점
화폐와 영향 수준에 따라 필터링하여 감시 목록에 중요한 알림을만 받습니다.
설정 가능한 리드 타임 (예: 5분 전) 에 Telegram 및 / 또는 Discord에 사전 출시 카운트다운 메시지를 전송합니다.
발매 인쇄가 끝나면 실제 대 예상 대 이전 판독으로 후속 알림을 발사합니다.
일정 루프로 연속적으로 실행됩니다. 크론 작업이 필요하지 않습니다.

두 번째 수준의 시간표가 중요한 이유

FXMacroData의 announcement_datetime field carries a second-level UTC timestamp for every scheduled release. That precision is what lets a bot wake up at exactly the right moment rather than polling on a broad daily window. Competing providers typically supply only a date, forcing you to poll blindly throughout the day.

필수 조건

시작 하기 전 에 다음 과 같은 것 들 이 필요 합니다.

파이썬 3.9+ 모든 스니펙트들은 표준 타이핑 문법을 사용합니다
FXMacroData API 키 등록하세요 / 가입 그리고 대시보드에서 키를 복사
텔레그램 봇 토큰 (선택) 을 통해 봇을 만들 @BotFather Telegram에서 채팅 ID를 입력하고
디스코드 웹 룩 URL (선택) 아래의 모든 Discord 채널에서 웹후크를 만들 수 있습니다 설정 → 통합 → 웹 룩
파이썬 패키지 requests schedule

pip install requests schedule

환경 변수로 인증서를 저장합니다.

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"

단계 1: 발매 일정을 찾아

릴리스 캘린더 엔드포인트에서는 예상 값, 이전 판독 값, 릴리즈 한 번 실제 수치를 포함하여 주어진 화폐에 대한 모든 예정된 매크로 이벤트를 반환합니다. announcement_datetime 이 필드는 UTC ISO 8601 시간표로 초까지 정렬되어 있습니다.

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

모든 물건 data 를 포함하는 필드가 있습니다. indicator announcement_datetime expected prior, 그리고 출시 후 actual아직 출판되지 않은 이벤트는 actual: null-

예제 달력 항목 (JSON)

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

단계 2: 감시 목록 및 진행 시간 기준으로 필터링

당신은 아마도 모든 작은 지표에 대한 경고를 원하지 않을 것입니다. 아래의 기능은 설정 가능한 리드 창 안에 있는 이벤트에 필터로 bot가 출시 발사 전에 카운트다운 경고를 보낼 수 있습니다.

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

단계 3: 텔레그램 알림을 보내십시오

텔레그램의 봇 API는 간단한 sendMessage HTTP POST. 아래의 함수는 짧은 읽기 쉬운 메시지를 포맷합니다. 모바일 푸시 알림에 적합합니다. 그리고 그것을 채팅에 게시합니다.

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

Telegram 채팅 ID를 찾는 방법

어떤 메시지를 보트로 보내면 방문합니다 https://api.telegram.org/bot<TOKEN>/getUpdates 브라우저에서 chat.id 응답의 필드는 로 수출하는 값입니다. TELEGRAM_CHAT_ID-

단계 4: 불화 경고 를 보내

디스코드 웹후크는 JSON 페이로드를 content 문자열과 선택 embeds 배열. 임베드 사용은 경고를 왼쪽에있는 색상 스트립으로 만듭니다. 긍정적 인 놀라움에 대해 녹색, 미스에 대해 빨간색으로 바쁜 채널을 한 눈에 쉽게 스캔합니다.

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

단계 5: 주회로를 연결

주 순환은 매 분마다 실행됩니다.

감시 목록에 있는 모든 통화에 대한 달력을 다시 가져옵니다
어떤 이벤트도 사전 경고 창 안에 있는지 확인 (단계 2) 그리고 카운트다운 메시지를 발사
이전 틱 및 화염 결과 메시지가 인쇄 된 이벤트 있는지 확인
다음 주기로 잠자리

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

중복 해제 메모

- _alerted_pre 그리고 _alerted_post 세트는 bot 재시작에 한 번 이상 각 경고 화염을 보장합니다. 프로세스를 다시 시작하면 이미 창에 있었던 이벤트의 중복을 받을 수 있습니다. 이것은 의도적입니다.

단계 6: 봇을 실행합니다

전체 스크립트를 저장합니다 calendar_bot.py 그리고 바로 실행:

python calendar_bot.py

생산 배포를 위해, 도커 컨테이너 또는 간단한 systemd 서비스 내부에서 실행 하 여 실패에 재시작 합니다. 보트는 표준 FXMacroData 계획 한계 내에서 분당 한 화폐 당 API 호출을 소비 합니다.

도커를 실행합니다

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

단계 7: 봇을 확장

핵심 루프는 의도적으로 최소화되어 있습니다.

다화폐 요약

모든 통화에서 다음 24시간 내에 발생할 모든 이벤트를 집계하고 이벤트별 핑 대신 단일 아침 브리핑을 보내십시오.

놀라움의 크기 필터

릴리스 후 메시지 경고만 |actual - expected| / expected 한 임계치를 초과하면 시장이 움직일 가능성이 없는 직선 결과를 필터링합니다.

SQLite와 함께 지속적인 상태

내 메모리를 교체해 _alerted_pre / _alerted_post 작은 SQLite 테이블을 가지고 세트 그래서 deduplication 상태는 생존 재시작.

슬랙 또는 이메일 알림

교환하거나 추가 alerts.py 모듈을 Slack Incoming Webhook에 게시하거나 SMTP를 통해 동일한 형식 이벤트 디크트를 사용하여 이메일을 보내십시오.

전체 스크립트

위의 모든 단계가 하나의 파일로 결합됩니다. fetch_calendar, 필터 보조자, 텔레그램 및 디스코드 발신자, 그 다음 run_cycle 그리고 main 그리고 여러분은 파이썬의 200줄 이하의 자율적인 봇을 갖게 됩니다.

이 가이드에서 사용 된 출시 달력 최종점은 /api-data-docs/usd/non_farm_payrolls 이 아닌 농업인 USD 지표에 대해 지원되는 통화로는 AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD 및 USD 이 포함되며 각 하나마다 자체적으로 큰 영향을 미치는 릴리스 이벤트가 있습니다.

요약

이제 생산 준비가 된 경고 봇이 있습니다.

두 번째 레벨의 정확한 UTC 시간표를 사용하여 FXMacroData에서 릴리스 캘린더를 가져옵니다.
보내 발매 전 역수계 Telegram 및 / 또는 Discord에 각 이벤트 전에 구성 가능한 몇 분
보내 방출 후의 결과 실제, 예상 및 이전 값과 함께 경고 색상 코딩으로 디스코드에서 깜짝 방향
디듀플리케이션 가드와 함께 자립 파이썬 루프로 실행
환경 변수 구성으로 도커에서 깔끔하게 배포

다음 단계로 자연스럽게 릴리스 타임 스탬프를 지표 역사의 최종점 어떤 화폐가 기대치를 뛰어넘거나 놓치는 경향이 있는지에 대한 역사적인 놀라움 스코어카드를 만들고 출시 전 포지셔닝을 가중화하는 데 사용하십시오. FX 대시보드 아이디어에 대한

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