كيفية إنشاء لوحة لوحة أداة ماكرو في بايثون باستخدام باندا & Plotly

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

الشروط المسبقة

ستحتاج إلى ما يلي قبل البدء:

• بايثون 3.9+
• مفتاح FXMacroData API التسجيل في / اشترك ونسخ مفتاحك من لوحة القيادة
• حزم بايثون. requests- لا pandas- لا plotly

قم بتثبيت التبعيات في أمر واحد:

pip install requests pandas plotly

تخزين مفتاح API الخاص بك كمتغير بيئة أبداً إثباتات التعليمات الشخصية الصلبة في النصوص:

export FXMD_API_KEY="YOUR_API_KEY"

الخطوة 1 فهم شكل نقطة نهاية المؤشر

يتبع كل مؤشر FXMacroData نفس نمط REST ، مما يجعل من التعميم إلى وظيفة إحضار واحدة تافهًا. طلب سعر السياسة للدولار الأمريكي يبدو هكذا:

GET https://fxmacrodata.com/api/v1/announcements/usd/policy_rate?api_key=YOUR_API_KEY&start=2020-01-01

استجابة JSON هي كائن مسطح مع data صف:

{
  "data": [
    { "date": "2025-03-19", "val": 4.25, "announcement_datetime": "2025-03-19T18:00:00Z" },
    { "date": "2025-01-29", "val": 4.25, "announcement_datetime": "2025-01-29T19:00:00Z" },
    { "date": "2024-12-18", "val": 4.25, "announcement_datetime": "2024-12-18T19:00:00Z" }
  ]
}

كل سجل يحمل date (سنة سنة-ميلاد-ميا) ، رقم val، و عند توفر مستوى UTC الثاني announcement_datetimeالشكل المتسق عبر جميع المؤشرات هو ما يسمح لوظيفة واحدة لخدمة كل لوحة في لوحة القيادة.

الخطوة 2 اكتب وظيفة استرجاع قابلة لإعادة الاستخدام

قم بإنشاء وحدة تدعى macro_fetch.pyتطلب الدالة أدناه أي مؤشر لأي عملة، وتحول الرد إلى سلسلة باندا، وتعيدها مؤشر بتاريخ جاهز للخروج مباشرة إلى إطار بيانات.

import os
import requests
import pandas as pd

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY = os.environ["FXMD_API_KEY"]


def fetch_indicator(currency: str, indicator: str, start: str = "2020-01-01") -> pd.Series:
    """
    Fetch a single indicator series from FXMacroData and return it as a
    pandas Series with a DatetimeIndex, named '{currency.upper()}_{indicator}'.
    """
    url = f"{BASE_URL}/announcements/{currency}/{indicator}"
    resp = requests.get(url, params={"api_key": API_KEY, "start": start}, timeout=15)
    resp.raise_for_status()

    records = resp.json().get("data", [])
    if not records:
        return pd.Series(name=f"{currency.upper()}_{indicator}", dtype=float)

    series = (
        pd.DataFrame(records)
        .assign(date=lambda df: pd.to_datetime(df["date"]))
        .set_index("date")["val"]
        .sort_index()
        .rename(f"{currency.upper()}_{indicator}")
    )
    return series

الخطوة 3 احضر جميع المؤشرات لوحة المعلومات

مع مساعد البحث في مكانه، سحب أربعة مؤشرات لأربع عملات هو حلقة موجزة. إضافة لفافة محاولة جديدة صغيرة للتعامل مع اختناقات الشبكة العابرة:

import time

CURRENCIES = ["usd", "eur", "gbp", "aud"]
INDICATORS = {
    "policy_rate": "Policy Rate (%)",
    "inflation": "CPI Inflation (% YoY)",
    "unemployment": "Unemployment Rate (%)",
    "pmi": "Manufacturing PMI",
}


def fetch_all(start: str = "2020-01-01", retries: int = 3) -> dict[str, pd.DataFrame]:
    """
    Return a dict mapping each indicator slug to a wide DataFrame where
    each column is one currency (e.g. USD_policy_rate, EUR_policy_rate).
    """
    frames: dict[str, list[pd.Series]] = {ind: [] for ind in INDICATORS}

    for currency in CURRENCIES:
        for indicator in INDICATORS:
            for attempt in range(retries):
                try:
                    s = fetch_indicator(currency, indicator, start=start)
                    frames[indicator].append(s)
                    break
                except requests.HTTPError as exc:
                    if attempt == retries - 1:
                        print(f"Warning: could not fetch {currency}/{indicator}: {exc}")
                    else:
                        time.sleep(1.5 ** attempt)

    return {ind: pd.concat(series, axis=1) for ind, series in frames.items() if series}

يمكنك فحص أي إطار بيانات بشكل تفاعلي للتحقق من مظهر البيانات قبل تقديمها:

data = fetch_all(start="2021-01-01")
print(data["policy_rate"].tail())
# Output (example):
#             USD_policy_rate  EUR_policy_rate  GBP_policy_rate  AUD_policy_rate
# date
# 2025-01-29             4.25              NaN              NaN              NaN
# 2025-02-06              NaN              NaN             4.50              NaN
# 2025-02-18              NaN              NaN              NaN             4.10
# 2025-03-06              NaN             2.50              NaN              NaN
# 2025-03-19             4.25              NaN              NaN              NaN

الخطوة 4 تشكيل البيانات لرسوم البيانة

- التآمر أفضل مع ملئ من قبل سلسلة من المخططات الخطية، لذلك يتم نقل القيمة المعروفة الأخيرة إلى الأمام حتى الإصدار التالي. أضف خطوة ملء إلى الأمامه قبل رسم الرسوم البيانية:

def prepare_for_chart(df: pd.DataFrame) -> pd.DataFrame:
    """
    Forward-fill each column so lines in the chart step at each release date
    rather than showing gaps between announcements.
    Resample to a monthly frequency for a cleaner visual.
    """
    return (
        df
        .ffill()
        .resample("ME")      # month-end
        .last()
        .dropna(how="all")
    )

بالنسبة لمؤشر مؤشر الأسهم، وهو مؤشر شهري، فإن التعبئة المسبقة أقل أهمية ولكن الوظيفة تتعامل معها براقة من خلال مجرد تمرير البيانات الشهرية بالفعل دون تغيير.

الخطوة 5 بناء لوحة المعلومات

-بلوتلـي) ؟) make_subplots يسمح لك هذا البرنامج بتنظيم مخططات متعددة في كائن واحد، يمكنك عرضها في المتصفح، أو تصديرها كـ HTML، أو تضمينها في دفتر ملاحظات Jupyter.

import plotly.graph_objects as go
from plotly.subplots import make_subplots

# Palette aligned with FXMacroData brand colours
COLORS = {
    "usd": "#3B82F6",   # finance blue
    "eur": "#D97706",   # gold
    "gbp": "#16A34A",   # green
    "aud": "#7C3AED",   # purple
}

SUBPLOT_TITLES = list(INDICATORS.values())


def build_dashboard(data: dict[str, pd.DataFrame]) -> go.Figure:
    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=SUBPLOT_TITLES,
        shared_xaxes=False,
        vertical_spacing=0.12,
        horizontal_spacing=0.08,
    )

    positions = [(1, 1), (1, 2), (2, 1), (2, 2)]

    for (indicator, label), (row, col) in zip(INDICATORS.items(), positions):
        df = prepare_for_chart(data[indicator])
        for col_name in df.columns:
            currency = col_name.split("_")[0].lower()
            fig.add_trace(
                go.Scatter(
                    x=df.index,
                    y=df[col_name],
                    mode="lines",
                    name=currency.upper(),
                    line=dict(color=COLORS[currency], width=2),
                    legendgroup=currency,
                    showlegend=(indicator == "policy_rate"),  # one legend entry per currency
                    hovertemplate=f"%{{x|%b %Y}}: %{{y:.2f}}{عملة.أعلى() }",
                ),
                row=row, col=col,
            )

    fig.update_layout(
        title=dict(
            text="G4 Central Bank Macro Dashboard",
            font=dict(size=22, color="#1e3a5f"),
            x=0.5,
        ),
        paper_bgcolor="#f8fafc",
        plot_bgcolor="#f1f5f9",
        font=dict(family="Inter, system-ui, sans-serif", size=12, color="#334155"),
        legend=dict(
            orientation="h",
            yanchor="bottom",
            y=1.04,
            xanchor="center",
            x=0.5,
        ),
        height=700,
        margin=dict(t=100, b=50, l=60, r=40),
    )

    fig.update_xaxes(showgrid=True, gridcolor="#e2e8f0", zeroline=False)
    fig.update_yaxes(showgrid=True, gridcolor="#e2e8f0", zeroline=False)

    return fig

الخطوة 6 تشغيل لوحة التحكم

سلك كل شيء معا في dashboard.py نص نقطة الدخول fig.show() يفتح لوحة القيادة في متصفحك الافتراضي. fig.write_html() يحفظ ملف HTML مستقل يمكنك مشاركته أو تضمينه في أي مكان.

if __name__ == "__main__":
    print("Fetching macro data …")
    data = fetch_all(start="2021-01-01")

    print("Building dashboard …")
    fig = build_dashboard(data)

    # Option A: open in browser
    fig.show()

    # Option B: save as portable HTML file
    fig.write_html("macro_dashboard.html", include_plotlyjs="cdn")
    print("Saved macro_dashboard.html")

أطلقها من المحطة

python dashboard.py

ستفتح Plotly نافذة متصفح تظهر لوحة تحكم من صفين وعمدتين مع أربع لوحات حية واحدة لكل مؤشر مدونة باللون حسب العملة. إذا قمت بملاحظة أي سطر، فستظهر التاريخ والقيمة الدقيقة لهذا الإصدار.

الخطوة 7 إضافة ملاحظات توقيت الإصدار (اختياري)

واحدة من أكثر تحسينات لوحة القيادة فائدة هي التداخل علامات الإفراج خطوط عمودية أو نقاط تظهر بالضبط متى تم إعلان كل إعلام. announcement_datetime طوابع زمنية، حتى تتمكن من إضافتها دون أي تخمين:

def fetch_release_datetimes(currency: str, indicator: str, start: str) -> pd.Series:
    """Return a Series of UTC announcement datetimes for a given indicator."""
    url = f"{BASE_URL}/announcements/{currency}/{indicator}"
    resp = requests.get(url, params={"api_key": API_KEY, "start": start}, timeout=15)
    resp.raise_for_status()
    records = resp.json().get("data", [])
    if not records:
        return pd.Series(dtype="datetime64[ns, UTC]")
    df = pd.DataFrame(records)
    if "announcement_datetime" not in df.columns:
        return pd.Series(dtype="datetime64[ns, UTC]")
    return pd.to_datetime(df["announcement_datetime"], utc=True)


def add_release_markers(fig: go.Figure, currency: str, indicator: str,
                         start: str, row: int, col: int) -> None:
    """Overlay vertical dashed lines on a subplot at each release datetime."""
    datetimes = fetch_release_datetimes(currency, indicator, start)
    for dt in datetimes:
        fig.add_vline(
            x=dt.timestamp() * 1000,  # Plotly uses ms since epoch for datetime axes
            line_width=1,
            line_dash="dot",
            line_color=COLORS[currency],
            opacity=0.35,
            row=row, col=col,
        )

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

الخطوة 8 تصدير اللوحات كصور فردية

إذا كنت ترغب في تضمين مخططات فردية في تقرير أو شريط شرائح، يمكن أن تصدر كل مخططة فرعية كPNG عبر الكاليدو.

pip install kaleido

fig.write_image("macro_dashboard.png", width=1400, height=700, scale=2)

بالنسبة للصادرات لكل لوحة، قم ببناء كل مؤشر كشيء مستقل go.Figure باستخدام نفس go.Scatter تتبع و اتصل write_image وظائف الاستيلاء على الشكل والشكل التي تم تطويرها في الخطوات السابقة تعمل دون تغيير لشخصيات لوحة واحدة.

مرجع كامل للنص

أدناه النص الكامل المكتمل الذي يجمع بين جميع الخطوات المذكورة أعلاه dashboard.py، وضعت FXMD_API_KEYو أديرها

"""
macro_dashboard.py — G4 Central Bank Macro Dashboard
Requires: requests, pandas, plotly
Usage:    FXMD_API_KEY=your_key python macro_dashboard.py
"""
import os
import time
import requests
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY = os.environ["FXMD_API_KEY"]

CURRENCIES = ["usd", "eur", "gbp", "aud"]
INDICATORS = {
    "policy_rate": "Policy Rate (%)",
    "inflation": "CPI Inflation (% YoY)",
    "unemployment": "Unemployment Rate (%)",
    "pmi": "Manufacturing PMI",
}
COLORS = {"usd": "#3B82F6", "eur": "#D97706", "gbp": "#16A34A", "aud": "#7C3AED"}


def fetch_indicator(currency: str, indicator: str, start: str = "2020-01-01") -> pd.Series:
    url = f"{BASE_URL}/announcements/{currency}/{indicator}"
    resp = requests.get(url, params={"api_key": API_KEY, "start": start}, timeout=15)
    resp.raise_for_status()
    records = resp.json().get("data", [])
    if not records:
        return pd.Series(name=f"{currency.upper()}_{indicator}", dtype=float)
    return (
        pd.DataFrame(records)
        .assign(date=lambda df: pd.to_datetime(df["date"]))
        .set_index("date")["val"]
        .sort_index()
        .rename(f"{currency.upper()}_{indicator}")
    )


def fetch_all(start: str = "2020-01-01", retries: int = 3) -> dict[str, pd.DataFrame]:
    frames: dict[str, list[pd.Series]] = {ind: [] for ind in INDICATORS}
    for currency in CURRENCIES:
        for indicator in INDICATORS:
            for attempt in range(retries):
                try:
                    frames[indicator].append(fetch_indicator(currency, indicator, start))
                    break
                except requests.HTTPError as exc:
                    if attempt == retries - 1:
                        print(f"Warning: {currency}/{indicator}: {exc}")
                    else:
                        time.sleep(1.5 ** attempt)
    return {ind: pd.concat(series, axis=1) for ind, series in frames.items() if series}


def prepare_for_chart(df: pd.DataFrame) -> pd.DataFrame:
    return df.ffill().resample("ME").last().dropna(how="all")


def build_dashboard(data: dict[str, pd.DataFrame]) -> go.Figure:
    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=list(INDICATORS.values()),
        vertical_spacing=0.12,
        horizontal_spacing=0.08,
    )
    for (indicator, _), (row, col) in zip(INDICATORS.items(), [(1,1),(1,2),(2,1),(2,2)]):
        df = prepare_for_chart(data[indicator])
        for col_name in df.columns:
            currency = col_name.split("_")[0].lower()
            fig.add_trace(
                go.Scatter(
                    x=df.index, y=df[col_name], mode="lines",
                    name=currency.upper(),
                    line=dict(color=COLORS[currency], width=2),
                    legendgroup=currency,
                    showlegend=(indicator == "policy_rate"),
                    hovertemplate=f"%{{x|%b %Y}}: %{{y:.2f}}{عملة.أعلى() }",
                ),
                row=row, col=col,
            )
    fig.update_layout(
        title=dict(text="G4 Central Bank Macro Dashboard", font=dict(size=22, color="#1e3a5f"), x=0.5),
        paper_bgcolor="#f8fafc", plot_bgcolor="#f1f5f9",
        font=dict(family="Inter, system-ui, sans-serif", size=12),
        legend=dict(orientation="h", yanchor="bottom", y=1.04, xanchor="center", x=0.5),
        height=700, margin=dict(t=100, b=50, l=60, r=40),
    )
    fig.update_xaxes(showgrid=True, gridcolor="#e2e8f0", zeroline=False)
    fig.update_yaxes(showgrid=True, gridcolor="#e2e8f0", zeroline=False)
    return fig


if __name__ == "__main__":
    print("Fetching macro data …")
    data = fetch_all(start="2021-01-01")
    print("Building dashboard …")
    fig = build_dashboard(data)
    fig.show()
    fig.write_html("macro_dashboard.html", include_plotlyjs="cdn")
    print("Saved macro_dashboard.html")

ملخص

لديك الآن لوحة أداة عمل ماكرو:

• يحضر أسعار الفائدة والضرائب والبطالة و PMI للدولار الأمريكي واليورو والجنيه الإسترليني والدولار الأسترالي من نقطة نهاية إعلانات FXMacroData
• يعيد تشكيل البيانات إلى باندا متواءمة ومملوءة للأمام
• يعطي لوحة تشغيل تفاعلية من أربع لوحات مع خطوط العملات المرموقة بالألوان
• تصدير ملف HTML مستقل يمكنك مشاركته أو تضمينه

من هنا يمكنك توسيع لوحة القيادة في عدة اتجاهات: إضافة المزيد من العملات، التداخل البطالة معدل التداول المشترك على محورين، أو ربط البيانات في وظيفة محددة تحديث صادرات HTML كل صباح. أداة بيانات api-data-docs.