Jupyter Notebook是希望将代码,数据和评论结合到一个可共享的文件中的分析师和开发人员的首选工具.本指南将您介绍一个完整的端到端工作流程:安装FXMacroData Python SDK,设置API键,获取宏观指标系列,执行分析,并在单个笔记本中生成已准备的图表.到最后,您将有一个可工作,可重复的笔记簿,您可以使用FXMacrodata所展示的80多个指标扩展.
你将要建造什么
一个独立的Jupyter笔记本,可以通过FXMacroData REST API进行身份验证,为四个G10货币抽取政策利率和通货膨胀时间序列,计算实际利率差距,并呈现交互式多系列图表准备分享或作为反复报告安排.
预先要求
- 基于Python 3.9或更高版本的
- Jupyter笔记本或JupyterLab 安装与
pip install jupyterlab - 接下来的包装:
requests没有人知道.pandas没有人知道.matplotlib没有人知道.seaborn - 一个FXMacroData API 键 登录 订阅 为了得到一个
你可以在一个命令中安装所有依赖:
pip install requests pandas matplotlib seaborn jupyterlab然后从项目文件启动JupyterLab:
jupyter lab步骤1 确保您的API密钥
笔记本文件中的硬代码凭证永远不要笔记型文件很容易意外地共享,并且泄露的API密钥可能被滥用.最安全的方法是将密钥存储在环境变量中并在运行时读取.
在Linux/macOS上,将这个行添加到你的 ~/.bashrc 没有 ~/.zshrc (然后重新启动终端或运行 source ~/.bashrc没有
export FXMD_API_KEY="your_actual_api_key_here"在Windows (PowerShell) 上:
$env:FXMD_API_KEY = "your_actual_api_key_here"然后在专用设置单元格中阅读笔记本书的顶部的键:
import os
API_KEY = os.environ.get("FXMD_API_KEY")
if not API_KEY:
raise EnvironmentError(
"FXMD_API_KEY environment variable is not set. "
"See https://fxmacrodata.com/subscribe to get a key."
)
print("API key loaded ✓")提示: python-dotenv 对于项目级的秘密
如果您更喜欢 .env 文件每个项目,安装 pip install python-dotenv 并且加上
from dotenv import load_dotenv; load_dotenv() 在 os.environ.get() 电话. 保持 .env 文件在你的 .gitignore现在我们要做什么?
步骤2 写一个可重复使用的获取辅助器
每个FXMacroData指标终点都遵循相同的URL结构:
GET https://fxmacrodata.com/api/v1/announcements/{currency}/{indicator}?api_key=YOUR_API_KEYJSON 响应包含一个 data 每个对象都有一个 date 场上,一个 val 场地,以及 (为了准确的时间) announcement_datetime 下面的辅助器将该响应直接转换为熊猫数据框架:
import requests
import pandas as pd
BASE_URL = "https://fxmacrodata.com/api/v1"
def fetch_indicator(currency: str, indicator: str,
start: str = None, end: str = None) -> pd.DataFrame:
"""Fetch a macro indicator time series from FXMacroData.
Parameters
----------
currency : ISO currency code, e.g. 'usd', 'eur', 'gbp'
indicator : Indicator slug, e.g. 'policy_rate', 'inflation', 'gdp'
start : Optional start date 'YYYY-MM-DD'
end : Optional end date 'YYYY-MM-DD'
Returns
-------
pd.DataFrame with columns: date (datetime64), val (float64),
announcement_datetime (datetime64, nullable),
currency (str), indicator (str)
"""
params = {"api_key": API_KEY}
if start:
params["start"] = start
if end:
params["end"] = end
url = f"{BASE_URL}/announcements/{currency}/{indicator}"
resp = requests.get(url, params=params, timeout=15)
resp.raise_for_status()
rows = resp.json().get("data", [])
if not rows:
return pd.DataFrame(columns=["date", "val", "announcement_datetime",
"currency", "indicator"])
df = pd.DataFrame(rows)
df["date"] = pd.to_datetime(df["date"])
df["val"] = pd.to_numeric(df["val"], errors="coerce")
if "announcement_datetime" in df.columns:
df["announcement_datetime"] = pd.to_datetime(
df["announcement_datetime"], errors="coerce", utc=True
)
df["currency"] = currency.upper()
df["indicator"] = indicator
return df.sort_values("date").reset_index(drop=True)助手举起一个 HTTPError 在4xx/5xx的响应,转换 date 列到一个适当的 datetime64 在下游连接中置.
步骤3 拿出第一套
让我们通过查看自2022年以来美国联邦储备委员会政策利率决定来验证助手:
usd_rate = fetch_indicator("usd", "policy_rate", start="2022-01-01")
usd_rate.head(10)你应该看到一个数据,上面有这样的列:
date val announcement_datetime currency indicator
0 2022-03-16 0.25 2022-03-16T18:00:00+00:00 USD policy_rate
1 2022-05-04 0.75 2022-05-04T18:00:00+00:00 USD policy_rate
2 2022-06-15 1.50 2022-06-15T18:00:00+00:00 USD policy_rate
3 2022-07-27 2.25 2022-07-27T18:00:00+00:00 USD policy_rate
...注意到了 announcement_datetime 列 对于依赖于事件驱动的策略或依赖精确公告时间的后台测试,这为您提供了第二级发布精度. 美国美元/政策_利率现在我们要做什么?
步骤4 获取多种货币
最强大的模式之一是同时为多种货币抽取相同的指标并堆叠结果.下面的循环一次性获取美元,欧元,英和澳元的政策利率:
currencies = ["usd", "eur", "gbp", "aud"]
START = "2022-01-01"
policy_rates = pd.concat(
[fetch_indicator(ccy, "policy_rate", start=START) for ccy in currencies],
ignore_index=True
)
print(f"Fetched {len(policy_rates)} rows across {policy_rates['currency'].nunique()} currencies")
policy_rates.groupby("currency").tail(2)可用的指标
整个指标目录可以在 它们的数据来源:汇率分析的关键系列包括:
policy_rate没有人知道.
inflation没有人知道.
gdp没有人知道.
unemployment没有人知道.
pmi没有
trade_balance每个系列都使用相同的获取模式交换货币代码和指标.
步骤5 结合多个指标
为了计算实际利率,你需要每个货币的政策利率和总通胀.
inflation = pd.concat(
[fetch_indicator(ccy, "inflation", start=START) for ccy in currencies],
ignore_index=True
)
# Stack all observations into one long-form DataFrame
macro = pd.concat([policy_rates, inflation], ignore_index=True)
print(macro.groupby(["currency", "indicator"]).size().to_string())第6步 改造和前
央行决定很少发生,每年发生6-12次.为了与其他月度系列相匹配,建立一个月度日期后脊,并向前填写最后已知值:
import numpy as np
# Floor each observation to month start for alignment
macro["month"] = macro["date"].dt.to_period("M").dt.to_timestamp()
# Pivot to wide: one column per currency+indicator combination
wide = (
macro
.groupby(["month", "currency", "indicator"])["val"]
.last() # latest reading within the month
.unstack(["currency", "indicator"])
)
# Build a complete monthly date spine and forward-fill
date_spine = pd.date_range(START, pd.Timestamp.today(), freq="MS")
wide = wide.reindex(date_spine).ffill()
wide.tail(3)步骤7 计算实际利率差距
实际利率是指指数政策利率减去总通胀.正差表示货币政策限制;负差意味着央行仍然适应消费者价格增长.欧元/美元实际利息差距是欧元美元最强的中期驱动因素之一:
for ccy in [c.upper() for c in currencies]:
try:
wide[(ccy, "real_rate")] = (
wide[(ccy, "policy_rate")] - wide[(ccy, "inflation")]
)
except KeyError:
pass # skip if either series is missing
# EUR minus USD real rate differential
wide[("spread", "eur_usd")] = (
wide[("EUR", "real_rate")] - wide[("USD", "real_rate")]
)
wide[[("USD", "real_rate"), ("EUR", "real_rate"), ("spread", "eur_usd")]].tail(6)步骤8 用Matplotlib可视化
随着数据框架的准备,多面板图只需要几行. 使用步骤图为政策利率它们是离散的阶梯决定,不是连续的系列:
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
fig, axes = plt.subplots(3, 1, figsize=(13, 11), sharex=True)
fig.patch.set_facecolor("#F8FAFC")
COLORS = {"USD": "#2563EB", "EUR": "#16A34A", "GBP": "#7C3AED", "AUD": "#F97316"}
# Panel 1: Policy rates
ax1 = axes[0]
for ccy in [c.upper() for c in currencies]:
try:
series = wide[(ccy, "policy_rate")].dropna()
ax1.step(series.index, series.values, where="post",
label=ccy, color=COLORS[ccy], linewidth=1.8)
except KeyError:
pass
ax1.set_ylabel("Policy rate (%)")
ax1.set_title("Central Bank Policy Rates — G4", fontweight="bold")
ax1.legend(loc="upper left", fontsize=9)
ax1.grid(axis="y", alpha=0.3)
# Panel 2: Real rates
ax2 = axes[1]
for ccy in [c.upper() for c in currencies]:
try:
series = wide[(ccy, "real_rate")].dropna()
ax2.plot(series.index, series.values,
label=ccy, color=COLORS[ccy], linewidth=1.8)
except KeyError:
pass
ax2.axhline(0, color="#94A3B8", linewidth=0.8, linestyle="--")
ax2.set_ylabel("Real rate (%)")
ax2.set_title("Real Interest Rates (Policy Rate − Inflation)", fontweight="bold")
ax2.legend(loc="upper left", fontsize=9)
ax2.grid(axis="y", alpha=0.3)
# Panel 3: EUR–USD spread
ax3 = axes[2]
spread = wide[("spread", "eur_usd")].dropna()
ax3.fill_between(spread.index, spread.values, 0,
where=spread.values >= 0,
color="#16A34A", alpha=0.30, label="EUR favoured")
ax3.fill_between(spread.index, spread.values, 0,
where=spread.values < 0,
color="#2563EB", alpha=0.30, label="USD favoured")
ax3.plot(spread.index, spread.values, color="#374151", linewidth=1.6)
ax3.axhline(0, color="#94A3B8", linewidth=0.8, linestyle="--")
ax3.set_ylabel("Spread (pp)")
ax3.set_title("EUR−USD Real Rate Differential", fontweight="bold")
ax3.legend(loc="upper left", fontsize=9)
ax3.grid(axis="y", alpha=0.3)
ax3.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
ax3.xaxis.set_major_locator(mdates.MonthLocator(interval=4))
plt.setp(ax3.xaxis.get_majorticklabels(), rotation=30, ha="right")
fig.tight_layout(h_pad=1.6)
plt.savefig("macro_analysis.png", dpi=150, bbox_inches="tight")
plt.show()提示:对于政策利率系列使用步骤图
ax.step(..., where="post") 代表中央银行决策的正确性,是单独的阶梯移动而不是平滑的插入线.对于所有其他宏观系列 (通货膨胀,GDP,PMI) 来说,标准 ax.plot() 是合适的.
步骤9 查看发行日历
在安排反复检索之前,很有用知道下一次数据发布的时间. 发布日历终点 返回任何货币的即将发布的日期,以便您可以将笔记本更新时间按每次发布后几分钟运行,而不是按固定时间表进行投票:
def fetch_calendar(currency: str) -> pd.DataFrame:
"""Fetch upcoming release dates for a currency from the FXMacroData calendar."""
url = f"{BASE_URL}/calendar/{currency}"
resp = requests.get(url, params={"api_key": API_KEY}, timeout=15)
resp.raise_for_status()
events = resp.json().get("data", [])
df = pd.DataFrame(events)
if df.empty:
return df
df["release_date"] = pd.to_datetime(df["release_date"], errors="coerce")
return df.sort_values("release_date").reset_index(drop=True)
# Upcoming USD releases
usd_calendar = fetch_calendar("usd")
upcoming = usd_calendar[usd_calendar["release_date"] >= pd.Timestamp.today()]
print(upcoming[["release_date", "indicator"]].head(10).to_string(index=False))步骤10 保存和安排
为了可重复的分析,您可以将宽数据框架导出到CSV,并通过 papermill,它通过参数注入从命令行执行笔记本:
pip install papermill# At the end of your notebook: persist the dataset
wide.to_csv("macro_data.csv")
print(f"Saved {wide.shape[0]} rows × {wide.shape[1]} columns to macro_data.csv")# Execute the notebook from the command line (e.g. from cron or a CI job)
papermill macro_analysis.ipynb macro_analysis_output.ipynb \
-p START "2022-01-01"让您的计划工作只能在实际上预期新指标数据时运行,
一个地方的全笔记本
它们是单独的笔记本.
FXMD_API_KEY 您将在不到两分钟内获得一个完全互动的宏观分析.
总结
您已经建立了一个完整的 Jupyter 笔记本工作流,
- 通过环境变量安全地通过FXMacroData进行身份验证
- 实现可重复使用的
fetch_indicator()帮助者将API响应转换为panda数据 - 拉和存政策利率和通货膨胀系列,用于多个G10货币
- 通过前期填写将稀少的央行数据重新塑造成一个正常的月度日期
- 计算实际利率差距和货币间差距
- 产生一个为报告或共享准备的三面板Matplotlib图表
- 查询发布日历时间反复更新智能
- 输出结果到 CSV 和时间表
papermill
作为下一步,扩展笔记本书,增加FXMacroData系列,如
失业率没有人知道.
贸易平衡没有
美国 为了建立一个更完整的宏观得分表.
fetch_indicator() 只有指标需要更新.
