كيفية إعداد تنبيهات الماكرو مع FXMacroData و Webhooks

إصدارات البيانات الكلية هي من بين أكثر المحفزات موثوقية في أسواق العملات الأجنبية. يمكن أن ترسل طباعة مؤشر أسعار المستهلكين 0.2% فوق الإجماع EUR/USD تتزحلق 60 نقطة قبل حتى تحميل عنوان الأخبار. إذا كان نظام التحذير الخاص بك يعتمد على فحص التقويم اليدوي أو استطلاع استقصائي واسع النطاق يقوم على توقيت، فأنت متخلف بالفعل. يوضح لك هذا الدليل كيفية إنشاء خط أنابيب تحذير دقيق مدفوع بالشبكة يكتشف إصدرات FXMacroData الجديدة في اللحظة التي تظهر فيها ويقدم إشعارات إلى Slack أو Discord أو أي نقطة نهاية HTTP كل ذلك في أقل من 80 سطرًا من 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 طابع زمني من المستوى الثاني UTC يخبرك بالضبط متى تم نشر هذه القيمة. ستستخدم هذا الطابع الزمني كمفتاح الحد من التكرار: إذا تغير منذ آخر فحص لك ، فقد تم طباعة إصدار جديد.

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 إرسال إشعار webhook

كل من Slack webhooks الواردة و Discord webhook تقبل HTTP POST مع JSON حمولة مفيدة. الوظيفة أدناه يبني رسالة تنسيقها ويرسلها. Slack يستخدم text المجال ، يستخدم Discord 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

الخيار ب حاوية Docker

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
• ✅ الحرائق تنشر تحذيرات العد التنازلي قبل الإفراج باستخدام تقويم الإصدار مع وقت التوصيل قابلة للتكوين
• ✅ يتم تشغيله بشكل مستمر كخدمة نظام أو حاوية دوكر