كيفية دمج FXMacroData مع Dagster

داجستر هي منصة أوركسترات البيانات الأصلية في بايثون بنيت حول مفهوم الأصول المحددة برمجيًا: أدوات البيانات التي تعرف نسبها الخاصة ، يمكن أن تتحقق عند الطلب أو على جدول زمني ، وتعرض بيانات الوصفية الغنية والمكافئ خارج الصندوق. بالنسبة لتدفقات عمل البيانة الكبرى سحب سلسلة المؤشرات ، واكتشاف أحداث الإصدار ، وتخزين لقطات الفوتوغرافية ، وتنبيه على الشذوذات داغستر مناسبة بشكل ممتاز. هذا الدليل يمر عبر بناء خط أنابيب يبتلع مؤشرات FXMacroData في مخزن SQLite المحلي ، ويظهر أحداثة تقويم الإصلاحات القادمة ، ويتم تشغيلها تلقائيًا على جدول يومي.

الخطوة 1 تثبيت Dagster وتبعيات المشروع

قم بإنشاء بيئة افتراضية جديدة وتثبيت Dagster جنبا إلى جنب مع مجموعة صغيرة من المكتبات التي يستخدمها هذا الخط. dagster و dagster-webserver إلى نفس النسخة يتجنب أكثر الفخاخ شيوعا في التثبيت.

python -m venv .venv
source .venv/bin/activate            # Windows: .venv\Scripts\activate

pip install "dagster==1.9.*" "dagster-webserver==1.9.*" \
            requests pandas sqlalchemy

بدء تخطيط مشروع بسيط، وترتيب المخططات من خلال أمر "داغستر" pyproject.toml والذي يسمح لمستخدم الويب المحلي باكتشاف تعريفاتك تلقائياً

dagster project scaffold --name fxmacro_pipeline
cd fxmacro_pipeline

تخزين مفتاح API الخاص بك كمتغير بيئي. لا تضع أرقام الاعتماد في ملفات المصدر.

export FXMACRO_API_KEY="YOUR_API_KEY"

على مضيفات الإنتاج، قم بإدخال المفتاح عبر إدارة مخططك السرية (أسرار إجراءات GitHub، أسرار Kubernetes، متغيرات بيئة Dagster Cloud). يقرأها خط الأنابيب من البيئة في وقت التشغيل لا حاجة إلى تغييرات في الشفرة.

الخطوة 2 تحديد مورد FXMacroData القابل لإعادة الاستخدام

" خنجر " الموارد هي اعتماد مشترك قابل للحقن مماثل لصلة قاعدة بيانات أو جلسة HTTP. تغليف FXMacroData REST API في مورد يعني أن كل أصل يمكن أن يدعوه دون تكرار منطق المصادقة أو معالجة وقت الانقطاع.

إبتكر fxmacro_pipeline/resources.py.

import os
from typing import Any

import requests
from dagster import ConfigurableResource, get_dagster_logger

_BASE_URL = "https://fxmacrodata.com/api/v1"


class FXMacroDataResource(ConfigurableResource):
    """Thin wrapper around the FXMacroData REST API."""

    api_key: str = ""           # overridden at run-time from env
    timeout: int = 15

    def _get(self, path: str, **params: Any) -> dict:
        logger = get_dagster_logger()
        key = self.api_key or os.environ.get("FXMACRO_API_KEY", "")
        resp = requests.get(
            f"{_BASE_URL}{path}",
            params={"api_key": key, **params},
            timeout=self.timeout,
        )
        resp.raise_for_status()
        data = resp.json()
        logger.debug("GET %s → %d records", path, len(data.get("data", [])))
        return data

    def get_announcements(
        self,
        currency: str,
        indicator: str,
        start: str = "2020-01-01",
    ) -> list[dict]:
        """Return time-series records for a macro indicator."""
        result = self._get(
            f"/announcements/{currency.lower()}/{indicator}",
            start=start,
        )
        return result.get("data", [])

    def get_forex(self, base: str, quote: str, start: str = "2023-01-01") -> list[dict]:
        """Return daily FX spot-rate records."""
        result = self._get(f"/forex/{base.lower()}/{quote.lower()}", start=start)
        return result.get("data", [])

    def get_calendar(self, currency: str) -> list[dict]:
        """Return the upcoming release calendar for a currency."""
        result = self._get(f"/calendar/{currency.lower()}")
        return result.get("data", [])

الخطوة 3 تحديد أصول البيانات الكبرى

كل أصل من أصول Dagster يمثل منتج بيانات منطقي واحد. ماديّة أصل يحصل على أحدث البيانات من FXMacroData ويستمرّ في ذلك محليّاً. نحن نحدّد أربعة أصول: سعر الفائدة، ال مؤشر أسعار المستهلكين الأمريكي، ال سعر سعر العملة، وسلسلة فورية لـ EUR/USD. ثم يقرأ أصل خامس تقويم الإصدار حتى يعرف خط الأنابيب متى يجب إعلانات عالية التأثير التالية.

إبتكر fxmacro_pipeline/assets.py.

import os
import sqlite3
from datetime import date, timedelta

import pandas as pd
from dagster import (
    AssetExecutionContext,
    MaterializeResult,
    MetadataValue,
    asset,
)

from .resources import FXMacroDataResource

_DB_PATH = os.environ.get("FXMACRO_DB", "fxmacro_data.db")


def _ensure_table(conn: sqlite3.Connection, table: str) -> None:
    conn.execute(
        f"""
        CREATE TABLE IF NOT EXISTS {table} (
            date TEXT PRIMARY KEY,
            val  REAL,
            announcement_datetime TEXT
        )
        """
    )
    conn.commit()


def _upsert(conn: sqlite3.Connection, table: str, records: list[dict]) -> int:
    rows = [
        (r["date"], r.get("val"), r.get("announcement_datetime"))
        for r in records
        if r.get("date") and r.get("val") is not None
    ]
    conn.executemany(
        f"INSERT OR REPLACE INTO {table} (date, val, announcement_datetime) VALUES (?,?,?)",
        rows,
    )
    conn.commit()
    return len(rows)


@asset(group_name="macro_indicators", compute_kind="api")
def usd_policy_rate(
    context: AssetExecutionContext,
    fxmacrodata: FXMacroDataResource,
) -> MaterializeResult:
    """Latest Fed funds rate series from the FXMacroData announcements endpoint."""
    records = fxmacrodata.get_announcements("usd", "policy_rate", start="2015-01-01")
    with sqlite3.connect(_DB_PATH) as conn:
        _ensure_table(conn, "usd_policy_rate")
        n = _upsert(conn, "usd_policy_rate", records)
    latest = records[0] if records else {}
    context.log.info("Fed rate: %s%% (as of %s)", latest.get("val"), latest.get("date"))
    return MaterializeResult(
        metadata={
            "row_count": MetadataValue.int(n),
            "latest_date": MetadataValue.text(latest.get("date", "n/a")),
            "latest_val": MetadataValue.float(latest.get("val") or 0.0),
        }
    )


@asset(group_name="macro_indicators", compute_kind="api")
def usd_cpi(
    context: AssetExecutionContext,
    fxmacrodata: FXMacroDataResource,
) -> MaterializeResult:
    """US CPI (year-on-year) series. Tracks whether inflation is rising or falling."""
    records = fxmacrodata.get_announcements("usd", "inflation", start="2015-01-01")
    with sqlite3.connect(_DB_PATH) as conn:
        _ensure_table(conn, "usd_cpi")
        n = _upsert(conn, "usd_cpi", records)
    latest = records[0] if records else {}
    context.log.info("US CPI: %s%% (as of %s)", latest.get("val"), latest.get("date"))
    return MaterializeResult(
        metadata={
            "row_count": MetadataValue.int(n),
            "latest_date": MetadataValue.text(latest.get("date", "n/a")),
            "latest_val": MetadataValue.float(latest.get("val") or 0.0),
        }
    )


@asset(group_name="macro_indicators", compute_kind="api")
def eur_policy_rate(
    context: AssetExecutionContext,
    fxmacrodata: FXMacroDataResource,
) -> MaterializeResult:
    """ECB main refinancing rate series."""
    records = fxmacrodata.get_announcements("eur", "policy_rate", start="2015-01-01")
    with sqlite3.connect(_DB_PATH) as conn:
        _ensure_table(conn, "eur_policy_rate")
        n = _upsert(conn, "eur_policy_rate", records)
    latest = records[0] if records else {}
    context.log.info("ECB rate: %s%% (as of %s)", latest.get("val"), latest.get("date"))
    return MaterializeResult(
        metadata={
            "row_count": MetadataValue.int(n),
            "latest_date": MetadataValue.text(latest.get("date", "n/a")),
            "latest_val": MetadataValue.float(latest.get("val") or 0.0),
        }
    )


@asset(group_name="fx_rates", compute_kind="api")
def eurusd_spot(
    context: AssetExecutionContext,
    fxmacrodata: FXMacroDataResource,
) -> MaterializeResult:
    """Daily EUR/USD closing spot rate."""
    records = fxmacrodata.get_forex("eur", "usd", start="2023-01-01")
    with sqlite3.connect(_DB_PATH) as conn:
        _ensure_table(conn, "eurusd_spot")
        n = _upsert(conn, "eurusd_spot", records)
    latest = records[0] if records else {}
    context.log.info("EUR/USD: %s (as of %s)", latest.get("val"), latest.get("date"))
    return MaterializeResult(
        metadata={
            "row_count": MetadataValue.int(n),
            "latest_date": MetadataValue.text(latest.get("date", "n/a")),
            "latest_val": MetadataValue.float(latest.get("val") or 0.0),
        }
    )


@asset(group_name="calendar", compute_kind="api")
def usd_release_calendar(
    context: AssetExecutionContext,
    fxmacrodata: FXMacroDataResource,
) -> MaterializeResult:
    """Upcoming USD macro release dates and times."""
    events = fxmacrodata.get_calendar("usd")
    df = pd.DataFrame(events) if events else pd.DataFrame()
    context.log.info("Found %d upcoming USD events", len(df))
    # Surface a preview in the Dagster UI asset metadata pane
    preview = df.head(5).to_markdown(index=False) if not df.empty else "No events found"
    return MaterializeResult(
        metadata={
            "event_count": MetadataValue.int(len(df)),
            "preview": MetadataValue.md(preview),
        }
    )

الخطوة 4 وضع وظيفة وجدول يومي

" خنجر " العمل يختار الأصول التي سيتم تحقيقها في جولة واحدة. جدول نحن نريد قراءات جديدة كل صباح يوم أسبوعي في الساعة 06:30 بتوقيت الاتحاد الأوروبي قبل فتح فرانكفورت

إبتكر fxmacro_pipeline/jobs.py.

from dagster import (
    AssetSelection,
    ScheduleDefinition,
    define_asset_job,
)

# ── Jobs ──────────────────────────────────────────────────────────────────────

macro_refresh_job = define_asset_job(
    name="macro_refresh",
    description="Refresh all FXMacroData indicator and FX rate assets.",
    selection=AssetSelection.groups("macro_indicators", "fx_rates"),
)

calendar_refresh_job = define_asset_job(
    name="calendar_refresh",
    description="Pull the latest USD release calendar.",
    selection=AssetSelection.groups("calendar"),
)

# ── Schedules ─────────────────────────────────────────────────────────────────

# Weekdays at 06:30 UTC — 30 min before Frankfurt opens
macro_daily_schedule = ScheduleDefinition(
    name="macro_daily_at_0630_utc",
    job=macro_refresh_job,
    cron_schedule="30 6 * * 1-5",
    execution_timezone="UTC",
)

# Refresh the release calendar every Sunday evening so the week's
# upcoming events are loaded before Monday's first run.
calendar_weekly_schedule = ScheduleDefinition(
    name="calendar_weekly_on_sunday",
    job=calendar_refresh_job,
    cron_schedule="0 18 * * 0",
    execution_timezone="UTC",
)

وظيفتان منفصلتان واحدة لسلسلة المؤشر، واحدة لجدول التاريخ الإصدار تسمح لك بمراقبتها وإعادة تشغيلها بشكل مستقل. إذا فشل استرداد جدول التوقيت الإصلاحي (على سبيل المثال خلال عطلة عامة عندما تعيد النقطة النهائية حمولة فارغة) ، فإن وظيفة تحديث المؤشر لا تتأثر.

الخطوة 5 سلك كل شيء إلى كائن التعاريف

(داجستر) Definitions المادة هي نقطة الدخول الوحيدة التي تربط الأصول والموارد والوظائف والجدول الزمني معا. fxmacro_pipeline/__init__.py.

import os

from dagster import Definitions

from .assets import (
    eur_policy_rate,
    eurusd_spot,
    usd_cpi,
    usd_policy_rate,
    usd_release_calendar,
)
from .jobs import (
    calendar_refresh_job,
    calendar_weekly_schedule,
    macro_daily_schedule,
    macro_refresh_job,
)
from .resources import FXMacroDataResource

defs = Definitions(
    assets=[
        usd_policy_rate,
        usd_cpi,
        eur_policy_rate,
        eurusd_spot,
        usd_release_calendar,
    ],
    resources={
        "fxmacrodata": FXMacroDataResource(
            # Falls back to the FXMACRO_API_KEY env var inside the resource
            api_key=os.environ.get("FXMACRO_API_KEY", ""),
        ),
    },
    jobs=[macro_refresh_job, calendar_refresh_job],
    schedules=[macro_daily_schedule, calendar_weekly_schedule],
)

الخطوة 6 إضافة جهاز استشعار للتنبيه عن بصمات تضخم مفاجئة

أجهزة الاستشعار في داجستر استطلاع للظروف الخارجية و تشغيل الزناد أو قنوات التحذير عندما يتغير شيء. usd_cpi التحقق: إذا كانت قراءة مؤشر أسعار المستهلك الأخيرة أعلى من 0.4 نقطة مئوية عن القراءة السابقة، فإنها تطلق تنبيهًا على شبكة الإنترنت حتى تتمكن استراتيجيتك أو نظام المخاطر من التفاعل على الفور.

إضافة fxmacro_pipeline/sensors.py.

import os
import sqlite3

import requests as http
from dagster import (
    RunStatusSensorContext,
    asset_sensor,
    AssetKey,
    EventLogEntry,
    SensorResult,
    SkipReason,
)

_DB_PATH = os.environ.get("FXMACRO_DB", "fxmacro_data.db")
_ALERT_WEBHOOK = os.environ.get("ALERT_WEBHOOK_URL", "")
_SURPRISE_THRESHOLD = 0.4  # percentage points


@asset_sensor(asset_key=AssetKey("usd_cpi"), job_name="macro_refresh")
def cpi_surprise_alert_sensor(
    context: RunStatusSensorContext,
    asset_event: EventLogEntry,
) -> SensorResult | SkipReason:
    """Alert when the latest US CPI print is significantly above the prior reading."""
    try:
        conn = sqlite3.connect(_DB_PATH)
        rows = conn.execute(
            "SELECT date, val FROM usd_cpi ORDER BY date DESC LIMIT 2"
        ).fetchall()
        conn.close()
    except Exception as exc:
        return SkipReason(f"DB read failed: {exc}")

    if len(rows) < 2:
        return SkipReason("Not enough CPI history yet.")

    latest_date, latest_val = rows[0]
    prior_date, prior_val = rows[1]
    surprise = (latest_val or 0) - (prior_val or 0)

    context.log.info(
        "CPI %s: %.2f%% (prior %s: %.2f%%, surprise: %+.2f pp)",
        latest_date, latest_val, prior_date, prior_val, surprise,
    )

    if surprise >= _SURPRISE_THRESHOLD and _ALERT_WEBHOOK:
        payload = {
            "text": (
                f":fire: *USD CPI Surprise* — latest {latest_date}: *{latest_val:.1f}%* "
                f"vs prior {prior_date}: {prior_val:.1f}% "
                f"(surprise: *{surprise:+.2f} pp*)"
            )
        }
        http.post(_ALERT_WEBHOOK, json=payload, timeout=8)
        context.log.warning("CPI surprise alert fired: %+.2f pp", surprise)

    return SensorResult(cursor=str(latest_date))

إدخلوا جهاز الاستشعار __init__.py من خلال استيرادها وإضافتها إلى Definitions sensors قائمة:

from .sensors import cpi_surprise_alert_sensor

defs = Definitions(
    # ... existing fields ...
    sensors=[cpi_surprise_alert_sensor],
)

الخطوة 7 اطلاق واجهة Dagster وتشغيل أول تطبيق

أبدأ خادم التطوير المحلي، سيكتشف (داغستر) Definitions من خلال pyproject.toml نقطة الدخول:

dagster dev

إبحار إلى http://localhost:3000سترى الرسم البياني للأصول يظهر جميع الأصول الخمس المجموعة في مجموعاتها المنطقية (macro_indicators- لا fx_rates- لا calendarانقر على تَمَتَّعوا بكلّ شيء لتشغيل أول مرور استيعاب في غضون ثوانٍ، ستظهر نافذة الأصول أعداد الصفوف، وأحدث التواريخ، وأخيرة القيم التي تم استخراجها مباشرة من واجهة برمجة برمجة بيانات FXMacroData.

sqlite3 fxmacro_data.db
sqlite> SELECT date, val FROM usd_policy_rate ORDER BY date DESC LIMIT 5;
sqlite> SELECT date, val FROM usd_cpi ORDER BY date DESC LIMIT 5;
sqlite> SELECT date, val FROM eurusd_spot ORDER BY date DESC LIMIT 3;

لتفعيل الجدول الزمني، انتقل إلى لمحة عامة → الجداول الزمنية في واجهة Dagster و التبديل macro_daily_at_0630_utc إلى -أركض. سيقوم العملية الديمونية لـ Dagster الآن باستيقظ كل يوم أسبوعي في الساعة 06:30 بتوقيت العالم ويتم تطبيق أصول مؤشر الكلية تلقائيًا.

الخطوة 8 كتابة اختبارات الوحدة مع مورد مزيف

واحدة من المزايا الرئيسية لتصميم Dagster هي أن الأصول هي وظائف Python العادية سهلة الاختبار من خلال الوحدة دون تشغيل كومة التنسيق الكاملة. قم بتبديل الموارد الحية بمزيف للحفاظ على اختبارات سريعة وخارج الاتصال.

# tests/test_assets.py
import sqlite3
import tempfile
import os

from unittest.mock import MagicMock

import pytest
from dagster import build_asset_context, materialize

from fxmacro_pipeline.assets import usd_policy_rate, usd_cpi
from fxmacro_pipeline.resources import FXMacroDataResource


MOCK_RATE_DATA = [
    {"date": "2025-12-18", "val": 4.5, "announcement_datetime": "2025-12-18T19:00:00Z"},
    {"date": "2025-11-07", "val": 4.75, "announcement_datetime": "2025-11-07T19:00:00Z"},
]

MOCK_CPI_DATA = [
    {"date": "2025-12-11", "val": 2.7, "announcement_datetime": "2025-12-11T13:30:00Z"},
    {"date": "2025-11-13", "val": 2.6, "announcement_datetime": "2025-11-13T13:30:00Z"},
]


class MockFXMacroData(FXMacroDataResource):
    def get_announcements(self, currency, indicator, start="2020-01-01"):
        if indicator == "policy_rate":
            return MOCK_RATE_DATA
        if indicator == "inflation":
            return MOCK_CPI_DATA
        return []

    def get_forex(self, base, quote, start="2023-01-01"):
        return [{"date": "2025-12-31", "val": 1.0482, "announcement_datetime": None}]

    def get_calendar(self, currency):
        return []


@pytest.fixture(autouse=True)
def temp_db(tmp_path, monkeypatch):
    monkeypatch.setenv("FXMACRO_DB", str(tmp_path / "test.db"))


def test_usd_policy_rate_materialises():
    result = materialize(
        [usd_policy_rate],
        resources={"fxmacrodata": MockFXMacroData()},
    )
    assert result.success
    mat = result.asset_materializations_for_node("usd_policy_rate")[0]
    assert mat.metadata["row_count"].value == 2
    assert mat.metadata["latest_val"].value == pytest.approx(4.5)


def test_usd_cpi_materialises():
    result = materialize(
        [usd_cpi],
        resources={"fxmacrodata": MockFXMacroData()},
    )
    assert result.success
    mat = result.asset_materializations_for_node("usd_cpi")[0]
    assert mat.metadata["latest_val"].value == pytest.approx(2.7)

pip install pytest
pytest tests/ -v

الخطوة 9 النشر إلى الإنتاج مع Dagster+

للفريق الذي يريد طائرة تحكم مضيفة دون إدارة الديمون الخاص بهم، (داجستر) (تقديم السحابة المدارة) يتطلب خطوة تكوين إضافية واحدة فقط: استبدال dagster dev مع عميل سحابة داجستر، الذي يقرأ نفس الشيء Definitions ويقوم بتنفيذ عمليات في البنية التحتية الخاصة بك.

حدد مفتاح API الخاص بك كمتغير بيئة Dagster + (النشر → المتغيرات البيئيةمع الاسم FXMACRO_API_KEY. سيقوم المصدر باختياره تلقائيًا لا يتطلب أي تغييرات في الشفرة. ينطبق نفس النمط على أي بيئة CI / CD تزرع أسرار كمتغيرات بيئة (GitHub Actions ، GitLab CI ، CircleCI).

لتوزيع على مجموعة Kubernetes المستضافة ذاتياً، قم بتجميع خط الأنابيب كصورة Docker، ودفعه إلى السجل الخاص بك، ووجّه مخطط Dagster Helm إليه. يرتبط مورد FXMacroData بالخارج إلى fxmacrodata.com تأكد من سياسة خروج المجموعة الخاصة بك تسمح HTTPS إلى هذا المضيف.

ما بنيت

من خلال اتباع هذا الدليل لديك الآن خط أنابيب Dagster يعمل الذي:

• يُعرّف المُعدّل الاستخدام، المُحقّق FXMacroDataResource الذي يتعامل مع جميع API التحقق وإدارة الأخطاء في مكان واحد
• يعرض خمسة أصول محددة برمجيات سعر الفيدرالي، ومؤشر أسعار المستهلك الأمريكي، ومعدل البنك المركزي الأوروبي، و EUR/USD الفورية، وتقويم الإصدار للدولار كل منهم استمر في SQLite مع بيانات الوصف الكاملة Dagster
• يجدول تحديثات أيام الأسبوع في الساعة 06:30 بالتوقيت العالمي عبر جدول Dagster المدفوع بواسطة cron حتى يكون لديك دائمًا قراءات محدثة قبل فتح الأسواق الأوروبية
• يطلق تنبيه webhook عندما تظهر مؤشر أسعار المستهلكات مفاجآت إلى الأعلى ، مما يعطي الأنظمة التالية محفزًا مدفوعًا بالأحداث بدلاً من حلقة استطلاع
• يتضمن اختبارات الوحدات السريعة خارج الاتصال باستخدام مورد مزيف ، لذلك لا يقوم خط أنابيب CI أبداً بإجراء مكالمات API مباشرة