What You Will Build
Macro data releases move markets fast. A surprise CPI print, an unexpected rate decision, or a better-than-expected employment figure can shift EUR/USD by 50 pips in seconds. If you are not watching the calendar in real time, you are already reacting to a market that has moved. This guide shows you how to build a lightweight Python bot that polls the FXMacroData release calendar and fires instant alerts to Telegram and Discord the moment a high-impact event is approaching or a result is published.
By the end of this article you will have a working bot that:
- Fetches upcoming macro events from the release calendar endpoint
- Filters by currency and impact level so you only get alerts that matter to your watchlist
- Sends a pre-release countdown message to Telegram and / or Discord at a configurable lead time (e.g. 5 minutes before)
- Fires a follow-up alert with the actual vs. expected vs. prior reading once the release prints
- Runs continuously as a scheduled loop — no cron job required
Why second-level timestamps matter
FXMacroData's 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.
Prerequisites
You will need the following before starting:
- Python 3.9+ — all snippets use standard typing syntax
- FXMacroData API key — sign up at /subscribe and copy your key from the dashboard
- Telegram bot token (optional) — create a bot via
@BotFatheron Telegram and note your chat ID - Discord webhook URL (optional) — create a webhook in any Discord channel under Settings → Integrations → Webhooks
- Python packages:
requests,schedule
pip install requests schedule
Store credentials as environment variables — never hard-code tokens in source files:
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"
Step 1: Fetch the Release Calendar
The release calendar endpoint returns every scheduled macro event for a given currency, including the expected value,
the prior reading, and — once released — the actual figure. The announcement_datetime field is a UTC
ISO 8601 timestamp down to the second, which is what drives the alert timing logic.
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"))
Each item in data has fields including indicator, announcement_datetime,
expected, prior, and — after the release — actual. An event that has not
yet printed will have actual: null.
Example calendar item (JSON)
{
"indicator": "non_farm_payrolls",
"announcement_datetime": "2026-05-02T12:30:00Z",
"expected": 185000,
"prior": 228000,
"actual": null
}
Step 2: Filter by Watchlist and Lead Time
You probably do not want alerts for every minor indicator. The function below filters events to those that fall within a configurable lead window so the bot can send a countdown alert before the release fires.
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
Step 3: Send Telegram Alerts
Telegram's Bot API accepts a simple sendMessage HTTP POST. The function below formats a short, readable
message — suitable for a mobile push notification — and posts it to your chat.
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))
How to find your Telegram chat ID
Send any message to your bot, then visit
https://api.telegram.org/bot<TOKEN>/getUpdates in a browser.
The chat.id field in the response is the value to export as TELEGRAM_CHAT_ID.
Step 4: Send Discord Alerts
Discord webhooks accept a JSON payload with a content string and optional
embeds array. Using an embed gives the alert a coloured strip on the left side —
green for a positive surprise, red for a miss — which makes it easy to scan a busy channel at a glance.
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,
})
Step 5: Wire Up the Main Loop
The main loop runs every minute. On each tick it:
- Re-fetches the calendar for every currency on the watchlist
- Checks whether any event is within the pre-alert window (Step 2) and fires countdown messages
- Checks whether any event has printed since the previous tick and fires result messages
- Sleeps until the next cycle
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()
Deduplication note
The _alerted_pre and _alerted_post sets ensure each alert fires at most once per
bot restart. If you restart the process you may receive a duplicate for events that were already in the window —
this is intentional; it is safer than missing a release.
Step 6: Run the Bot
Save the complete script as calendar_bot.py and run it directly:
python calendar_bot.py
For a production deployment, run it inside a Docker container or a simple systemd service so it restarts on failure. The bot consumes one API call per currency per minute — well within the standard FXMacroData plan limits.
Running with 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
Step 7: Extend the Bot
The core loop is intentionally minimal. Here are a few natural extensions:
Multi-currency summary digest
Aggregate all events due in the next 24 hours across all currencies and send a single morning briefing instead of per-event pings.
Surprise magnitude filter
Only alert on post-release messages when |actual - expected| / expected exceeds a threshold —
filter out in-line results that are unlikely to move the market.
Persistent state with SQLite
Replace the in-memory _alerted_pre / _alerted_post sets with a small SQLite
table so the deduplication state survives restarts.
Slack or email alerts
Swap or add an alerts.py module to post to a Slack Incoming Webhook or send an email via
SMTP using the same formatted event dict.
Complete Script
All steps above combine into a single file. Copy the blocks in order — imports, constants, fetch_calendar,
filter helpers, Telegram and Discord senders, then run_cycle and main — and you have a
fully self-contained bot in under 200 lines of Python.
The release calendar endpoint used in this guide is documented at /api-data-docs/usd/non_farm_payrolls for USD indicators. Supported currencies include AUD, BRL, CAD, CHF, CNY, DKK, EUR, GBP, JPY, NZD, PLN, SEK, SGD, and USD — each with its own set of high-impact release events.
Summary
You now have a production-ready alert bot that:
- Pulls the release calendar from FXMacroData using precise second-level UTC timestamps
- Sends a pre-release countdown to Telegram and / or Discord a configurable number of minutes before each event
- Sends a post-release result alert with actual, expected, and prior values — colour-coded for surprise direction on Discord
- Runs as a self-contained Python loop with deduplication guards
- Deploys cleanly in Docker with environment-variable configuration
A natural next step is to pair the release timestamps with the indicator history endpoint to build a historical surprise scorecard — which currencies tend to beat or miss their expectations — and use that to weight your pre-release positioning. Explore the full indicator catalogue on the FX Dashboard for ideas.