Làm thế nào để tích hợp FXMacroData với Dagster

Dagster là một nền tảng dàn nhạc dữ liệu gốc Python được xây dựng xung quanh khái niệm Tài sản được xác định bằng phần mềm: các đồ vật dữ liệu biết dòng dõi của riêng họ, có thể được thực hiện theo yêu cầu hoặc theo lịch trình, và phơi bày siêu dữ liệu phong phú và khả năng quan sát ra khỏi hộp. Đối với các luồng công việc dữ liệu macro kéo chuỗi chỉ số, phát hiện các sự kiện phát hành, lưu trữ ảnh chụp tức thời và cảnh báo về sự bất thường Dagster là một sự phù hợp tuyệt vời. Hướng dẫn này đi qua việc xây dựng một đường ống dẫn nhập các chỉ số FXMacroData vào một kho SQLite địa phương, bề mặt các sự việc lịch phát hành sắp tới và chạy tự động theo lịch hàng ngày.

Bước 1 Cài đặt Dagster và các phụ thuộc dự án

Tạo một môi trường ảo mới và cài đặt Dagster bên cạnh bộ thư viện nhỏ mà đường ống này sử dụng. dagster và dagster-webserver để cùng một phiên bản tránh những trở ngại cài đặt phổ biến nhất.

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

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

Đặt đầu một bố cục dự án tối thiểu. lệnh giàn giáo Dagster tạo ra cấu trúc thư mục và một pyproject.toml cho phép giao diện người dùng địa phương phát hiện các định nghĩa của bạn tự động.

dagster project scaffold --name fxmacro_pipeline
cd fxmacro_pipeline

Không bao giờ mã hóa mật khẩu trong tệp nguồn.

export FXMACRO_API_KEY="YOUR_API_KEY"

Trên máy chủ sản xuất, chèn khóa thông qua quản lý bí mật của trình lên lịch của bạn (bí mật hành động GitHub, bí mật Kubernetes, biến môi trường Dagster Cloud).

Bước 2 Định nghĩa một tài nguyên FXMacroData có thể sử dụng lại

Một Dagster. tài nguyên là một sự phụ thuộc được chia sẻ, có thể tiêm tương tự như kết nối cơ sở dữ liệu hoặc phiên HTTP. Bọc API REST FXMacroData vào một tài nguyên có nghĩa là mọi tài sản có thể gọi nó mà không cần trùng lặp logic xác thực hoặc xử lý thời gian hết.

Tạo 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", [])

Bước 3 Định nghĩa các tài sản dữ liệu vĩ mô

Mỗi tài sản Dagster đại diện cho một sản phẩm dữ liệu logic. Việc hiện thực hóa tài sản lấy dữ liệu mới nhất từ FXMacroData và giữ lại nó tại địa phương. Chúng tôi xác định bốn tài sản: Lãi suất chính sách của Fed, CPI của Hoa Kỳ, Lãi suất chính của ECB, và chuỗi EUR/USD. Một tài sản thứ năm sau đó đọc lịch phát hành để đường ống biết khi nào các thông báo có tác động cao tiếp theo sẽ đến.

Tạo 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),
        }
    )

Bước 4 Xây dựng một công việc và một lịch trình hàng ngày

Một Dagster. công việc chọn tài sản nào sẽ được thực hiện trong một lần chạy. lịch trình Chúng tôi muốn đọc macro mới mỗi buổi sáng vào ngày làm việc lúc 06:30 UTC trước khi mở Frankfurt để bất kỳ chỉ số nào được phát hành trong đêm được ghi lại trước khi giao dịch châu Âu bắt đầu.

Tạo 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",
)

Hai công việc riêng biệt một cho chuỗi chỉ số, một cho lịch phát hành cho phép bạn quan sát và chạy lại chúng một cách độc lập. Nếu việc lấy lịch phát triển không thành công (ví dụ trong một ngày lễ công cộng khi điểm cuối trả về tải trọng trống), công việc làm mới chỉ số không bị ảnh hưởng.

Bước 5 Đưa mọi thứ vào đối tượng Definitions

Dagster's. Definitions object là điểm nhập đơn liên kết tài sản, tài nguyên, công việc và lịch trình với nhau. 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],
)

Bước 6 Thêm một cảm biến để cảnh báo về dấu ấn lạm phát đáng ngạc nhiên

Các cảm biến trong Dagster thăm dò các điều kiện bên ngoài và kích hoạt chạy hoặc kênh cảnh báo khi có gì đó thay đổi. usd_cpi thực hiện: nếu chỉ số CPI mới nhất cao hơn 0,4 điểm phần trăm so với chỉ số trước đó, nó sẽ kích hoạt cảnh báo webhook để chiến lược hoặc hệ thống rủi ro của bạn có thể phản ứng ngay lập tức.

Thêm 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))

Nhập cảm biến vào. __init__.py bằng cách nhập khẩu nó và thêm nó vào Definitions sensors danh sách:

from .sensors import cpi_surprise_alert_sensor

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

Bước 7 Khởi động Dagster UI và chạy sự hiện thực hóa đầu tiên của bạn

Bắt đầu các máy chủ phát triển địa phương. Definitions qua pyproject.toml Điểm nhập cảnh:

dagster dev

Chuẩn bị cho http://localhost:3000- Anh sẽ thấy... Biểu đồ tài sản cho thấy tất cả năm tài sản được nhóm thành các nhóm hợp lý của họ (macro_indicators fx_rates calendarNhấp Làm vật chất tất cả để chạy lần đầu tiên nhập trong vòng vài giây, khung tài sản sẽ hiển thị số hàng, ngày gần nhất và giá trị mới nhất được lấy trực tiếp từ FXMacroData API.

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;

Để kích hoạt lịch trình, hãy điều hướng đến Tổng quan → Lịch trình trong Dagster UI và chuyển đổi macro_daily_at_0630_utc Chạy. Quá trình Daemon của Dagster sẽ thức dậy mỗi ngày trong tuần lúc 06:30 UTC và tự động thực hiện các tài sản chỉ số macro.

Bước 8 Viết các bài kiểm tra đơn vị với một tài nguyên giả

Một trong những lợi ích thiết kế chính của Dagster là các tài sản là các hàm Python thông thường dễ dàng thử nghiệm đơn vị mà không cần chạy toàn bộ ngăn xếp dàn nhạc.

# 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

Bước 9 triển khai sản xuất với Dagster+

Đối với các nhóm muốn một máy bay điều khiển được lưu trữ mà không cần quản lý daemon của riêng họ, Dagster+ (các dịch vụ đám mây được quản lý) chỉ cần một bước cấu hình bổ sung: thay thế dagster dev với Dagster Cloud, cũng giống như vậy. Definitions đối tượng và thực hiện chạy trong cơ sở hạ tầng của bạn.

Đặt khóa API của bạn như một biến môi trường Dagster + (Việc triển khai → Các biến môi trường), với tên FXMACRO_API_KEY. Tài nguyên sẽ tự động chọn nó không cần thay đổi mã. Mô hình tương tự áp dụng cho bất kỳ môi trường CI / CD nào tiêm bí mật dưới dạng biến môi trường (GitHub Actions, GitLab CI, CircleCI).

Để triển khai trên một cụm Kubernetes tự lưu trữ, gói đường ống như một hình ảnh Docker, đẩy nó đến đăng ký của bạn, và chỉ biểu đồ Dagster Helm vào nó. fxmacrodata.com đảm bảo chính sách thoát của cụm của bạn cho phép HTTPS đến máy chủ đó.

Những gì anh đã xây dựng

Bằng cách làm theo hướng dẫn này, bạn có một đường ống Dagster hoạt động:

• Định nghĩa một tái sử dụng, tiêm FXMacroDataResource xử lý tất cả các API xác thực và xử lý lỗi ở một nơi
• Thơi ra năm tài sản được xác định bởi phần mềm tỷ lệ Fed, CPI của Mỹ, tỷ lệ ECB, EUR/USD spot và lịch phát hành USD mỗi tài sản vẫn tồn tại với SQLite với đầy đủ siêu dữ liệu Dagster
• Lịch trình ngày trong tuần làm mới lúc 06:30 UTC thông qua lịch trình Dagster dựa trên cron để bạn luôn có các bài đọc cập nhật trước khi thị trường châu Âu mở cửa
• Bắn một cảnh báo webhook khi một ấn bản CPI bất ngờ lên, cung cấp cho hệ thống hạ lưu một sự kiện điều khiển kích hoạt thay vì một vòng lặp thăm dò
• Bao gồm các thử nghiệm đơn vị ngoại tuyến nhanh bằng cách sử dụng một tài nguyên giả, vì vậy đường ống dẫn CI không bao giờ thực hiện cuộc gọi API trực tiếp