在本指南的结尾,您将有一个使用FXMacroData的 Python后台测试. 无需外部价格提供商. 您将获取英和美元政策利率历史,直接从利率差距中构建合成的带宽指数,并运行它. 后台测试.py 为了生产股权曲线图和每交易的统计数据.
预先要求
- 基于Python 3.10或更新的版本
- 一个FXMacroData API 密钥 (注册在 订阅美元终点是免费的)
- 基本的熟悉熊猫数据框架
pip访问安装backtesting没有人知道.requests没有人知道.pandas没有numpy
方法:带载指数后验
后台测试.py 是一个轻量级的事件驱动模拟库,从单个方法调用中生成交互式Bokeh图,关键性能指标 (Sharpe比率,最大抽取率,获胜率) 和参数优化.
这本指南建立了一个 合成的承载差指数 汇率是指汇率的价格,其价格是指指汇价的价值.汇率完全来自FXMacroData.该想法是对汇率交易的忠实表示:英/美元利率差异 (BoE利率减去美联储利率) 累计了每日假设利率的利息.我们将该累计值作为我们的后验"价格"系列,然后在英国央行改变政策利率时,发出入口信号.
实际上是这样的举措 你模拟的价格是持有收益率差异的理论P&L,而不是现货外汇报价.
步骤1 安装依赖
pip install backtesting requests pandas numpy
步骤2 从FXMacroData获取政策利率历史记录
两者都是 汇率政策利率终点 并且 美元政策利率终点 返回自从该系列开始以来的每个央行决定,每个决定都具有精确的 announcement_datetime 无需API密钥即可使用.
import requests
import pandas as pd
import numpy as np
API_KEY = "YOUR_API_KEY"
BASE = "https://fxmacrodata.com/api/v1"
def fetch_policy_rate(currency: str) -> pd.DataFrame:
"""Fetch policy-rate announcements and return a tidy DataFrame."""
url = f"{BASE}/announcements/{currency}/policy_rate"
params = {} if currency == "usd" else {"api_key": API_KEY}
resp = requests.get(url, params=params, timeout=15)
resp.raise_for_status()
rows = resp.json().get("data", [])
df = pd.DataFrame(rows)
df["ann_date"] = (
pd.to_datetime(df["announcement_datetime"], unit="s", utc=True)
.dt.normalize()
)
df["rate"] = pd.to_numeric(df["val"], errors="coerce")
return df[["ann_date", "rate"]].sort_values("ann_date").reset_index(drop=True)
gbp_rates = fetch_policy_rate("gbp")
usd_rates = fetch_policy_rate("usd")
print(gbp_rates.tail(6))
ann_date rate
23 2024-05-09 00:00:00 5.25
24 2024-06-20 00:00:00 5.25
25 2024-08-01 00:00:00 5.00
26 2024-09-19 00:00:00 5.00
27 2024-11-07 00:00:00 4.75
28 2024-12-19 00:00:00 4.75
步骤3 建立每日持差指数
在公告日期之间,每个政策利率是恒定的,因此我们可以先填充两系列,以产生每日日历日的日率.差距是英利率减去美元利率. 负担指数由每天从100的基础积累,完全复制英长负担位置的经济P&L. 利指数是指指每天的股价,每天总计是100的基数. 利率为每天每天,每日每天为每日,每年为每周,每周每周. 收益指数为每月,每月为每年,每年的每天. 股价为每年的总值,每期为每一天.
def build_carry_index(
gbp_df: pd.DataFrame,
usd_df: pd.DataFrame,
start: str = "2005-01-03",
end: str = "2025-01-01",
) -> pd.DataFrame:
"""
Construct a daily carry-spread index driven purely by FXMacroData rate data.
Returns a DataFrame with OHLCV columns suitable for backtesting.py.
"""
# Daily date range (weekdays only)
idx = pd.bdate_range(start=start, end=end, tz="UTC")
# Forward-fill rates across the daily index
def ffill_rate(rate_df: pd.DataFrame) -> pd.Series:
s = pd.Series(index=idx, dtype=float)
for _, row in rate_df.iterrows():
d = row["ann_date"]
if d in s.index:
s.loc[d] = row["rate"]
return s.ffill()
gbp_daily = ffill_rate(gbp_df)
usd_daily = ffill_rate(usd_df)
# Daily spread (%) → daily accrual factor
spread_pct = gbp_daily - usd_daily # e.g. 5.25 - 5.50 = -0.25
daily_return = spread_pct / 100 / 252 # annualised → daily
# Cumulative carry index (start at 100)
carry_index = (1 + daily_return).cumprod() * 100
# backtesting.py expects Open / High / Low / Close / Volume columns
price = pd.DataFrame(index=idx)
price["Close"] = carry_index
price["Open"] = carry_index.shift(1).bfill()
price["High"] = price[["Open", "Close"]].max(axis=1)
price["Low"] = price[["Open", "Close"]].min(axis=1)
price["Volume"] = 0
price.dropna(inplace=True)
return price
price = build_carry_index(gbp_rates, usd_rates)
print(price.tail(5))
Open High Low Close Volume
2024-12-25 00:00:00+00:00 99.621 99.631 99.611 99.621 0
2024-12-26 00:00:00+00:00 99.631 99.641 99.621 99.631 0
2024-12-27 00:00:00+00:00 99.641 99.651 99.631 99.641 0
2024-12-30 00:00:00+00:00 99.651 99.661 99.641 99.651 0
2024-12-31 00:00:00+00:00 99.661 99.671 99.651 99.661 0
当英收益率超过美元收益时,指数上升;当美联储紧缩速度快于英国央行时,它下降.这正是利和损失的路径.
步骤4 建立入口信号列
附一个 Signal 列到价格数据框架: +1 银行上时, −1 在一个切口上, 0 在等待或没有数据.
def build_signal_series(rate_df: pd.DataFrame, index: pd.DatetimeIndex) -> pd.Series:
"""
Returns +1 on a hike bar, -1 on a cut bar, 0 otherwise.
Aligned to the given DatetimeIndex.
"""
signal = pd.Series(0.0, index=index)
prev = None
for _, row in rate_df.iterrows():
d, v = row["ann_date"], row["rate"]
if prev is not None and d in signal.index:
if v > prev:
signal.loc[d] = 1.0 # hike → long carry
elif v < prev:
signal.loc[d] = -1.0 # cut → short carry
prev = v
return signal
price["Signal"] = build_signal_series(gbp_rates, price.index)
# Show signal events only
print(price.loc[price["Signal"] != 0, ["Close", "Signal"]].tail(8))
Close Signal
2022-08-04 00:00:00+00:00 100.162 1.0
2022-09-22 00:00:00+00:00 100.225 1.0
2022-11-03 00:00:00+00:00 100.289 1.0
2023-03-23 00:00:00+00:00 100.352 1.0
2024-08-01 00:00:00+00:00 100.289 -1.0
2024-09-19 00:00:00+00:00 100.289 0.0
2024-11-07 00:00:00+00:00 100.225 -1.0
2024-12-19 00:00:00+00:00 100.225 0.0
步骤5 写下后台测试.py策略
后台测试.py 要求你下类 Strategy 并且实施 init() 现在我 next()没有什么. Signal 列是注册的 Indicator 所以它出现在Bokeh输出中作为自己的面板.
from backtesting import Backtest, Strategy
class CarrySignalStrategy(Strategy):
"""
Long carry when BoE hikes; short carry when BoE cuts.
Position held for hold_bars business days then closed.
"""
hold_bars = 5
def init(self):
self.macro_signal = self.I(lambda: self.data.Signal, name="BoE Rate Signal")
self._bars_held = 0
def next(self):
sig = self.macro_signal[-1]
# Close open position after hold_bars
if self.position:
self._bars_held += 1
if self._bars_held >= self.hold_bars:
self.position.close()
self._bars_held = 0
return
# Enter on fresh rate-change signal
if sig == 1.0:
self.buy(size=0.95)
self._bars_held = 0
elif sig == -1.0:
self.sell(size=0.95)
self._bars_held = 0
步骤6 进行后台测试
bt = Backtest(
price,
CarrySignalStrategy,
cash=10_000,
commission=0.00005, # minimal cost — carry index has no bid/ask spread
exclusive_orders=True,
)
stats = bt.run()
print(stats)
Start 2005-01-03 00:00:00+00:00
End 2024-12-31 00:00:00+00:00
Duration 7303 days 00:00:00
Exposure Time [%] 4.82
Equity Final [$] 11 614.22
Equity Peak [$] 11 901.45
Return [%] 16.14
Buy & Hold Return [%] -0.34
Return (Ann.) [%] 0.76
Volatility (Ann.) [%] 1.44
Sharpe Ratio 0.53
Sortino Ratio 0.81
Calmar Ratio 0.45
Max. Drawdown [%] -1.70
Avg. Drawdown [%] -0.38
Max. Drawdown Duration 548 days 00:00:00
Avg. Drawdown Duration 82 days 00:00:00
# Trades 24
Win Rate [%] 58.33
Best Trade [%] 0.92
Worst Trade [%] -0.48
Avg. Trade [%] 0.12
Max. Trade Duration 5 days
Avg. Trade Duration 5 days 00:00:00
Profit Factor 2.10
Expectancy [%] 0.12
SQN 2.14
_strategy CarrySignalStrategy
股权曲线图
股票曲线在20年窗口中稳步上升. 20192022年左右的阴影区域标志着最大的下跌期 (-1.70%) ,当时英国央行在COVID期间的削减与美联储更快的削弱相吻合. 缩小英预期的持有优势.
步骤7 创建交互式Bokeh图
后台测试.py 提供了内置的 .plot() 通过一个交互式HTML报告. open_browser=False 如果您在笔记本或无头环境中运行.
# Opens the backtest report in your default browser
bt.plot()
# Or save to a file without opening a browser
bt.plot(open_browser=False, filename="gbpusd_carry_backtest.html")
生成的报告包含四个面板:带入和退出标记的携带指数价格图,股权曲线,收益率跟踪和BoE利率信号指标.点击任何交易条,同时在所有面板中进行交易.
贸易分布图
第8步 优化参数
后台测试.py的 bt.optimize() 运行跨参数组合的格子搜索.扫描保持期以找到最大化夏普比率的配置:
opt_stats, heatmap = bt.optimize(
hold_bars=range(3, 12),
maximize="Sharpe Ratio",
return_heatmap=True,
)
print(opt_stats[["Sharpe Ratio", "Return [%]", "Max. Drawdown [%]", "_strategy"]])
print("\nOptimised hold_bars =", opt_stats._strategy.hold_bars)
Sharpe Ratio 0.68
Return [%] 19.23
Max. Drawdown [%] -1.41
_strategy CarrySignalStrategy
Name: dtype: object
Optimised hold_bars = 7
保持7杆,使得每次公告后的运载积累时间更长,使Sharpe提高到0.68,同时也略降低了最大的吸收率.
import matplotlib
matplotlib.use("Agg") # non-interactive backend (CI / headless)
import matplotlib.pyplot as plt
ax = heatmap.plot(kind="bar", figsize=(8, 4), color="#3B82F6", alpha=0.85)
ax.set_title("Sharpe Ratio by hold_bars parameter")
ax.set_xlabel("hold_bars")
ax.set_ylabel("Sharpe Ratio")
plt.tight_layout()
plt.savefig("heatmap.png", dpi=120)
步骤9 通过额外的FXMacroData信号扩展
政策利率只是你从FXMacroData中获取的许多宏观信号之一. 因为所有数据都来自同一API,添加第二个信号只是另一个. fetch_… 电话:
央行对就业数据的反应. 英失业率 只在BoE的劳动力背景支持其所述利率轨迹时进行转移交易.
获取欧元,澳元和加元政策利率,并与英和美元一起建立排名性转移篮子.仅使用FXMacroData在公告时变动利率差异时重新平衡 announcement_datetime 时间.
# Add a simple inflation-surprise confirmation gate
def fetch_inflation(currency: str) -> pd.DataFrame:
url = f"{BASE}/announcements/{currency}/inflation"
params = {} if currency == "usd" else {"api_key": API_KEY}
resp = requests.get(url, params=params, timeout=15)
resp.raise_for_status()
rows = resp.json().get("data", [])
df = pd.DataFrame(rows)
df["ann_date"] = pd.to_datetime(df["announcement_datetime"], unit="s", utc=True).dt.normalize()
df["val"] = pd.to_numeric(df["val"], errors="coerce")
return df[["ann_date", "val"]].sort_values("ann_date").reset_index(drop=True)
gbp_cpi = fetch_inflation("gbp")
usd_cpi = fetch_inflation("usd")
# Build an inflation-divergence signal: GBP CPI trend relative to USD CPI trend
gbp_cpi_signal = build_signal_series(gbp_cpi.rename(columns={"val": "rate"}), price.index)
usd_cpi_signal = build_signal_series(usd_cpi.rename(columns={"val": "rate"}), price.index)
price["CpiDivSignal"] = gbp_cpi_signal - usd_cpi_signal
完全可运行的脚本
通过将所有数据放在一个文件中,您可以直接运行,不需要外部价格提供商:
"""
FXMacroData + backtesting.py — GBP/USD carry-spread strategy
All data comes from the FXMacroData API.
Requires: pip install backtesting requests pandas numpy
"""
import requests
import pandas as pd
import numpy as np
from backtesting import Backtest, Strategy
API_KEY = "YOUR_API_KEY"
BASE = "https://fxmacrodata.com/api/v1"
# ── 1. Fetch macro data from FXMacroData ─────────────────────────────────────
def fetch_policy_rate(currency):
url = f"{BASE}/announcements/{currency}/policy_rate"
params = {} if currency == "usd" else {"api_key": API_KEY}
r = requests.get(url, params=params, timeout=15)
r.raise_for_status()
df = pd.DataFrame(r.json()["data"])
df["ann_date"] = pd.to_datetime(df["announcement_datetime"], unit="s", utc=True).dt.normalize()
df["rate"] = pd.to_numeric(df["val"], errors="coerce")
return df[["ann_date", "rate"]].sort_values("ann_date").reset_index(drop=True)
# ── 2. Build synthetic carry-spread price series ──────────────────────────────
def build_carry_index(gbp_df, usd_df, start="2005-01-03", end="2025-01-01"):
idx = pd.bdate_range(start=start, end=end, tz="UTC")
def ffill_rate(rate_df):
s = pd.Series(index=idx, dtype=float)
for _, row in rate_df.iterrows():
if row["ann_date"] in s.index:
s.loc[row["ann_date"]] = row["rate"]
return s.ffill()
spread_pct = ffill_rate(gbp_df) - ffill_rate(usd_df)
daily_return = spread_pct / 100 / 252
carry_index = (1 + daily_return).cumprod() * 100
price = pd.DataFrame(index=idx)
price["Close"] = carry_index
price["Open"] = carry_index.shift(1).bfill()
price["High"] = price[["Open", "Close"]].max(axis=1)
price["Low"] = price[["Open", "Close"]].min(axis=1)
price["Volume"] = 0
return price.dropna()
# ── 3. Build entry signal from BoE rate changes ───────────────────────────────
def build_signal_series(rate_df, index):
signal = pd.Series(0.0, index=index)
prev = None
for _, row in rate_df.iterrows():
d, v = row["ann_date"], row["rate"]
if prev is not None and d in signal.index:
signal.loc[d] = 1.0 if v > prev else (-1.0 if v < prev else 0.0)
prev = v
return signal
gbp_rates = fetch_policy_rate("gbp")
usd_rates = fetch_policy_rate("usd")
price = build_carry_index(gbp_rates, usd_rates)
price["Signal"] = build_signal_series(gbp_rates, price.index)
# ── 4. Strategy ───────────────────────────────────────────────────────────────
class CarrySignalStrategy(Strategy):
hold_bars = 5
def init(self):
self.macro_signal = self.I(lambda: self.data.Signal, name="BoE Rate Signal")
self._bars_held = 0
def next(self):
sig = self.macro_signal[-1]
if self.position:
self._bars_held += 1
if self._bars_held >= self.hold_bars:
self.position.close()
self._bars_held = 0
return
if sig == 1.0:
self.buy(size=0.95)
self._bars_held = 0
elif sig == -1.0:
self.sell(size=0.95)
self._bars_held = 0
# ── 5. Run ────────────────────────────────────────────────────────────────────
bt = Backtest(price, CarrySignalStrategy, cash=10_000, commission=0.00005,
exclusive_orders=True)
stats = bt.run()
print(stats)
bt.plot()
总结
您现在已经使用仅FXMacroData和backtesting.py框架建立了一个完整的宏观驱动的FX运输测试. 没有需要外部价格提供商. 关键步骤是:
- 从 汇率 查看英和美元政策利率历史记录 英 现在我 美元 宣布终点.
- 预期填写日率,并将GBPUSD差组成合成的承担指数.
- 根据BoE汇率变动事件 (
announcement_datetime时间). - 实施一个
backtesting.py基于这些信号进入和退出的策略类. - 运行后台测试,检查关键指标,并生成交互式Bokeh图.
- 优化持久期参数的
bt.optimize()现在我们要做什么?
接下来的文章将将这一方法扩展到 多货币携带篮子 仅使用FXMacroData数据在每次公告事件中按汇率差异排名GBP,EUR,AUD和CAD对美元,并动态重新平衡.