宏观仪表板将最重要的央行和经济指标放在一个地方,因此您可以在几秒钟内扫描市场状况,而不是在数据终端之间跳转. 本指南将您通过Python中建立一个完全交互式的多面板仪表盘. 熊猫并且将其与 为了你每个面板都会从一个API键和几个函数调用中更新.
你将要建造什么
一个自主 Python 脚本,可以为四种主要外汇货币 (美元,欧元,英,澳元) 抽取政策利率,通胀,失业率和 PMI,为每个指标组装一个干净的熊猫数据框架,并用不到 150 行代码染一个四面板 Plotly 仪表板.
预先要求
在开始之前,您需要以下:
- Python 3.9+
- 汇率数据API键 登记在 订阅 复制你的钥匙从仪表板
- Python 包没有人知道.
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-01JSON响应是一个带有 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为什么每一个指标都需要一个系列?
不同货币的指标在不同的日期发布,因此它们从来没有共享完美对齐的指数. pd.concat(..., axis=1) 让熊猫自动处理日期对齐, NaN 没有报告过货币.
步骤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")
)对于PMI来说,即一个月度指标,前期填充不太重要,但功能通过不变的已经月度数据来处理它.
步骤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点将打开浏览器窗口,显示一个两行,两列的仪表板,有四个活跃面板,每个指标都有一个,按货币编码. 通过点任何一行,显示该发布的确切日期和值.
在几秒钟内添加更多指标
仪表板是可扩展的. INDICATORS 举个例子,
"core_inflation": "Core CPI (% YoY)" 没有 "gdp_quarterly": "GDP Growth (% QoQ)" 查询,形状和图表步骤都会自动采集. 应用程序文件现在我们要做什么?
步骤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 作为单个图像导出面板
如果您想将单个图表包含在报告或幻灯片库中, 鱼没有人知道.
pip install kaleidofig.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 汇率数据的终点
- 将数据重新塑造成对齐,向前填充的熊猫
- 呈现一个四面板互动的图案仪表板,配有彩色编码的货币线
- 导出一个自主 HTML 文件,可以共享或嵌入
您可以在多个方向扩展仪表板:添加更多的货币,覆盖 失业率 总的指标目录包括贸易平衡,信贷增长和COT定位,可在 鱼数据文件现在我们要做什么?
