あなた が 築く もの
マクロデータリリースが市場を急速に動かす.サプライズなCPIプリント,予想外のレート決定,または予想以上に良い雇用数字は,秒間でEUR/USDを50ピップに移動させることができます.あなたがリアルタイムでカレンダーを見ていない場合,あなたはすでに動いた市場に反応しています.このガイドは,FXMacroDataのリリースカレンダーの民意調査を行い,即時アラートを発射する軽量パイトンボットの構築方法を示します. テレグラム ほら 矛盾 影響力の高いイベントが近づいているときや結果が公表されるときです.
この記事の終わりまでに,次の作業ボットが作れます.
- 未来マクロイベントを 検索します リリースカレンダーエンドポイント
- 通貨と影響レベルによってフィルタを設定します. 監視リストに重要なアラートのみが表示されます.
- 設定可能なリードタイムで (例えば 5 分前に) Telegram と/または Discord にリリース前のカウントダウンメッセージを送信します
- リリースが印刷されたら,実際の値と予想値と以前の値でフォローアップアラートを発します.
- 予定ループとして連続して実行します. cron 作業は必要ありません.
なぜ二次次タイムスタンプが重要なのか
金融機関が announcement_datetime このフィールドは,各予定リリースに対して2次レベルのUTCタイムスタンプを搭載している.その精度は,ボットが幅広い日々の窓で投票するのではなく,正確に正しいタイミングで目覚めるのを可能にします.競合するプロバイダは通常,日付のみを提供し,一日中盲目的に投票することを強制します.
条件
始める前に次のものが必要です.
- Python 3.9+ すべてのスニペットは標準的なタイプ構文を使用します
- FXMacroData API キーを 登録する / サブスクリプト そしてダッシュボードからあなたの鍵をコピー
- Telegramボットトークン (オプション) 経由でボットを作成
@BotFatherTelegram でチャット ID を記してください - ディスコード Webhook URL (オプション) 上の Discord チャンネルで Webhook を作成します 設定 → 統合 → Webhook
- 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 警告のタイミングを動かすものです 警告を動かし,警告を実行する時,
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: 監視リストとリードタイムでフィルタリング
設定可能なリードウィンドウに該当するイベントにフィルターする機能が表示されます. これにより,ボットがリリース開始前にカウントダウンアラートを送信できます.
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: Telegram アレートを送信する
簡単に メールのボット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 起動するたびに最大で1回警告が発信されるようにします. プロセスを再起動すると,すでにウィンドウに表示されていたイベントの複製が受信される可能性があります. これは意図的です.リリースを見逃すよりも安全です.
ステップ6:ボットを起動
書き込みを保存する calendar_bot.py 直接実行する
python calendar_bot.py生産展開では,Docker コンテナまたは単純な systemd サービス内で実行して,失敗した場合に再起動します.ボットは1分間に1つの通貨1つの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:ボットを拡張する
核ループは意図的に最小限に 拡張されています
多通貨の概要
すべての通貨で次の24時間以内に起こるすべてのイベントを総括し,イベントごとにピングする代わりに,単一の朝のブリーフィングを送信します.
驚異の大きさフィルター
リリース後のメッセージで警告するだけ |actual - expected| / expected 市場を動かす可能性が低いインライン結果をフィルターします.
SQLite の持続状態
記憶を入れ替える _alerted_pre / _alerted_post 復元状態が生き残るため再起動します.
スラックまたはメールのアラート
交換するか追加するか alerts.py メールを送信するモジュールです. メールの送信方法は,SLACKのWebhookに投稿するモジュールを,
脚本が完成した
ブロックを順番にコピーします. 輸入,定数, fetch_calendarフィルターヘルパー Telegram と Discord の送信者 run_cycle ほら main 完全に自立したボットが200行未満の Pythonで作れます
このガイドで使用されるリリースカレンダーのエンドポイントは, /api-data-docs/usd/非農業_給与表 サポートされる通貨には,AUD,BRL,CAD,CHF,CNY,DKK,EUR,GBP,JPY,NZD,PLN,SEK,SGD,USD がそれぞれ独自の大きな影響を与えるリリースイベントがあります.
概要
生産準備のいい警報ボットができました
- FXMacroData からリリースカレンダーを抽出する. 精確な2次元の UTC 時間スタンプを使用する.
- 送る リリース前のカウントダウン Telegram と/または Discord に各イベントの前に設定可能な数分
- 送る 放出後の結果 警告 実際の値,期待値,以前の値 驚異方向のために色でコード
- デュプリケーションガード付きの自立 Python ループとして実行される
- 環境変数設定でDockerできれいに展開する
公開時のスタンプを 配列化します 指標履歴のエンドポイント 予想を上回ったり 落としやがったりする通貨を記録し,その情報を活用して,発売前ポジションを評価します. FX ダッシュボード アイデアを
