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

你将要建造什么

宏观数据发布会迅速推动市场的发展.一个意外的CPI打印,一个意想不到的利率决定,或者一个比预期更好的就业数据,可以在几秒钟内使欧元/美元变动50个点.如果你没有实时观看日历,你已经对一个已经移动的市场做出了反应.本指南告诉你如何构建一个轻量级的Python机器人,它会调查FXMacroData发布日历并发出即时警报. 电报 现在我 没有关系 时间是大影响力事件即将发生或结果公布的时刻.

在本文结束时,您将有一个工作机器人:

• 从 获取即将到来的宏观事件 发布日历终点
• 根据货币和影响水平进行过,以便您只收到与您的观察名单相关的警报
• 在可配置的预期时间 (例如5分钟前) 发送预发倒计时消息到 Telegram 和/或 Discord
• 发射一个跟踪警报,当发布打印后实际与预期与先前读取
• 作为一个计划循环持续运行 不需要cron工作

预先要求

在开始之前,您需要以下:

• Python 3.9+ 所有片段使用标准的打字语法
• 汇率数据API键 登记在 订阅 复制你的钥匙从仪表板
• 电讯机器人代币 (可选) 通过创建机器人 @BotFather 在 Telegram 上,并记住您的聊天ID
• 离线网链URL (可选) 在任何 Discord 频道中创建网链 设置 → 集成 → 网页
• Python 包没有人知道. 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现在我们要做什么?

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

步骤2:按观察名单和预期时间进行过

您可能不希望每个小指标都会被提醒. 下面的函数将事件过到可配置的引擎窗口内的事件,以便机器人在发布发射之前发送倒计时警报.

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 下面的功能格式化一个简短,可读的消息,适合移动推送通知,并将其发布到您的聊天.

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

步骤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:连接主环

主循环每分钟运行.

1. 检查列表中的每个货币的日历
2. 检查是否有任何事件在预警窗口 (步骤2) 中,并发出倒计时消息
3. 检查是否有任何事件已经打印自前一个标和火灾结果消息
4. 睡到下一个周期.

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

步骤6:运行机器人

保存完整的脚本为 calendar_bot.py 并且直接运行:

python calendar_bot.py

对于生产部署,运行它在Docker容器或简单的systemd服务中,以便在故障时重新启动.机器人每分钟每货币消耗一个API调用在标准FXMacroData计划的限制范围内.

使用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

步骤7:扩展机器人

核心循环是故意最小的.

完整脚本

所有以上步骤将合并成一个文件. 复制按顺序的块进口,常数, fetch_calendar过助手,电讯和Discord发送者,然后 run_cycle 现在我 main 并且你有一个完全自给自足的机器人,

在本指南中使用的发布日历终点记录在 农业工资人数 对于美元指标,支持货币包括澳元,西元人民币,加元,瑞郎,中国元,丹麦皇冠,欧元,英,日元,新西兰元,波兰,瑞典,瑞银和美元. 每个货币都有自己的高影响性发布事件.

总结

现在你有一个生产准备的警报机器人:

• 通过精确的第二级UTC时间从FXMacroData中提取发布日历
• 发送一个 发布前倒计时 在每一个事件之前,可以配置数分钟的 Telegram 和/或 Discord
• 发送一个 放出后的结果 警告与实际,预期和以前的值 颜色编码在 Discord 上的意外方向
• 运行为一个自主 Python 循环,具有除重保护
• 在 Docker 中清洁部署环境变量配置

接下来的自然步骤是将发行时间与 标志器历史终点 建立一个历史惊喜分数表 哪些货币往往超过或错过了预期,并使用它来权衡您的发布前定位. 汇率仪表板 为了创意.