如何使用FXMacroData建立收益率差距交易策略

换取收益率之间的差距,而不是收益量本身

两国经济体之间的政府债券收益率差异是外汇市场中最可靠的结构力量之一.当美国10年收益量明显超过德国邦德等价时,资本往往流向美元资产,欧元/美元面临持续的销售压力.当这种差距缩小时,这对复苏.这种关系不是机械的,但它足够持久,可以围绕它建立一个基于规则的交易策略.

本指南将通过使用FXMacroData API构建一个完整的收益率幅对交易策略.

• 通过 通过两种货币的政府债券收益率时间序列 汇率指数
• 计算收益率差距及其滚动平均值和标准偏差
• 通过Z分数平均反转生成统计驱动的长/短外汇信号
• 追踪开放的仓位,应用基本风险控制,并显示进入/退出警报

预先要求

在开始之前,请确保您准备好以下东西:

• Python 3.9+ 所有片段使用标准类型注释
• 汇率数据API键 登记在 订阅 并从帐户仪表板复制您的密钥
• Python 包没有人知道. requests没有人知道. pandas没有人知道. numpy

pip install requests pandas numpy

存储您的API密钥作为环境变量 永远不要将其硬编码到源文件中:

export FXMACRO_API_KEY="YOUR_API_KEY"

步骤1:获取政府债券收益率数据

这一战略的基础是可靠的债券收益率时间序列. gov_bond_10y 每个观察都带有一个 date没有 val (成百分比的收益率),以及 announcement_datetime 对于第二级发布时间.

我们将美元和欧元的10年收益率,

import os
import requests
import pandas as pd
from datetime import datetime, timedelta

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


def fetch_yield(currency: str, tenor: str = "gov_bond_10y", start: str = "2022-01-01") -> pd.Series:
    """Fetch a bond yield series from FXMacroData and return as a dated Series."""
    resp = requests.get(
        f"{BASE_URL}/announcements/{currency}/{tenor}",
        params={"api_key": API_KEY, "start": start},
        timeout=15,
    )
    resp.raise_for_status()
    records = resp.json()["data"]
    if not records:
        raise ValueError(f"No data returned for {currency}/{tenor}")
    series = pd.Series(
        {r["date"]: r["val"] for r in records},
        name=f"{currency.upper()}_10Y",
        dtype=float,
    )
    series.index = pd.to_datetime(series.index)
    return series.sort_index()


# Fetch USD and EUR 10-year yields
usd_10y = fetch_yield("usd")
eur_10y = fetch_yield("eur")

# Align on a shared date index (inner join drops days missing in either series)
yields = pd.DataFrame({"USD_10Y": usd_10y, "EUR_10Y": eur_10y}).dropna()

print(yields.tail())
# Output:
#            USD_10Y  EUR_10Y
# 2025-04-08    4.41     2.71
# 2025-04-09    4.39     2.69
# 2025-04-10    4.49     2.70
# 2025-04-11    4.52     2.73
# 2025-04-14    4.47     2.70

您可以使用任何由 总体数据 流行组合包括美元/日元,澳元/美元和英/美元差距.

步骤2:计算收益率差距和Z分数

粗差 (美元减去欧元收益率) 告诉你资本结构偏向的方向. Z-score 将这种差距与其最近的历史相比正常化,为你提供一个无维信号,可以直接比较时间段和货币对.

超过+1.5的Z分数表明差异异异常大可能出现EUR/USD短 (长USD) 设置.低于-1.5的Z点数表明异常压缩可能发生EUR/ USD长设置

def compute_spread_signal(
    yields_df: pd.DataFrame,
    base_col: str,
    quote_col: str,
    lookback: int = 60,
    entry_z: float = 1.5,
    exit_z: float = 0.3,
) -> pd.DataFrame:
    """
    Compute yield spread, rolling Z-score, and directional signals.

    A positive spread means base currency yields are higher → base currency
    is structurally favoured → signal to be long base / short quote FX pair.

    Mean reversion: enter when Z-score is extreme, exit when it normalises.
    """
    df = yields_df.copy()
    df["spread"] = df[base_col] - df[quote_col]

    # Rolling statistics over the lookback window
    roll = df["spread"].rolling(lookback, min_periods=lookback // 2)
    df["spread_mean"] = roll.mean()
    df["spread_std"]  = roll.std()

    # Z-score: how many standard deviations from the rolling mean
    df["zscore"] = (df["spread"] - df["spread_mean"]) / df["spread_std"].replace(0, float("nan"))

    # Signal: +1 = long base/short quote, -1 = short base/long quote, 0 = flat
    df["signal"] = 0
    df.loc[df["zscore"] >  entry_z, "signal"] = -1   # spread too wide → expect compression → short base pair
    df.loc[df["zscore"] < -entry_z, "signal"] = +1   # spread too tight → expect widening → long base pair
    # Exit (override) when Z-score returns toward zero
    df.loc[df["zscore"].abs() < exit_z, "signal"] = 0

    return df.dropna(subset=["zscore"])


analysis = compute_spread_signal(yields, base_col="USD_10Y", quote_col="EUR_10Y")

print(analysis[["spread", "spread_mean", "zscore", "signal"]].tail(10))

步骤3:扩展到多对

单个EUR/USD的传播策略是有用的,但当你同时在几个对中运行相同的逻辑时,真正的力量就会出现. 结合USD/JPY,AUD/USD和GBP/USD传播信号,你就能获得一个多元化的宏观驱动组合,每个位置的规模独立.

您也可以使用短短的子 gov_bond_2y 终点对政策利率预期特别敏感,因此与10年期系列相比,它是领先的指标:

PAIRS = [
    # (base_currency, quote_currency, fx_pair_label)
    ("usd", "eur", "EUR/USD"),
    ("usd", "jpy", "USD/JPY"),
    ("aud", "usd", "AUD/USD"),
    ("gbp", "usd", "GBP/USD"),
]

TENOR = "gov_bond_10y"   # swap for gov_bond_2y for rate-expectation signals

results = {}
for base_ccy, quote_ccy, fx_label in PAIRS:
    try:
        base_series  = fetch_yield(base_ccy, tenor=TENOR)
        quote_series = fetch_yield(quote_ccy, tenor=TENOR)
        df = pd.DataFrame({
            f"{base_ccy.upper()}_10Y":  base_series,
            f"{quote_ccy.upper()}_10Y": quote_series,
        }).dropna()
        signal_df = compute_spread_signal(
            df,
            base_col=f"{base_ccy.upper()}_10Y",
            quote_col=f"{quote_ccy.upper()}_10Y",
        )
        latest = signal_df.iloc[-1]
        results[fx_label] = {
            "spread_pct": round(latest["spread"], 3),
            "zscore":     round(latest["zscore"], 2),
            "signal":     int(latest["signal"]),
        }
        print(f"{fx_label}: spread={latest['spread']:.3f}%, z={latest['zscore']:+.2f}, signal={int(latest['signal']):+d}")
    except Exception as exc:
        print(f"{fx_label}: skipped — {exc}")

# Example output:
# EUR/USD: spread=1.770%, z=+1.21, signal=0
# USD/JPY: spread=4.050%, z=+2.18, signal=-1
# AUD/USD: spread=0.340%, z=-0.55, signal=0
# GBP/USD: spread=0.890%, z=-1.62, signal=+1

步骤4:添加一个2年与10年斜率覆盖

收益率曲线的形状为策略增加了第二个维度. 曲线 (长期收益较短期快速上) 通常表明增长预期改善并支持货币. 倒置或平坦曲线往往会在放缓和央行转移之前出现.

拉两边 两年 现在我 十年 结果是计算斜率,并将其作为一个模式过器:只取一个与本国货币曲线斜率一致的方向的传播信号.

def fetch_curve_slope(currency: str, start: str = "2022-01-01") -> pd.Series:
    """Return the 10Y–2Y slope for a currency (positive = normal/steep, negative = inverted)."""
    y10 = fetch_yield(currency, tenor="gov_bond_10y", start=start)
    y2  = fetch_yield(currency, tenor="gov_bond_2y",  start=start)
    slope = (y10 - y2).dropna()
    slope.name = f"{currency.upper()}_slope"
    return slope


def apply_curve_filter(
    signal_df: pd.DataFrame,
    base_slope: pd.Series,
    quote_slope: pd.Series,
) -> pd.DataFrame:
    """
    Suppress signals that contradict the curve-slope regime.

    Long base / short quote (signal=+1) is only taken when:
      - base curve is steep (positive slope) AND
      - quote curve is flat or inverted (slope < base_slope)

    Short base / long quote (signal=-1) is only taken when:
      - quote curve is steep relative to base
    """
    df = signal_df.copy()
    df = df.join(base_slope.rename("base_slope"), how="left")
    df = df.join(quote_slope.rename("quote_slope"), how="left")
    df[["base_slope", "quote_slope"]] = df[["base_slope", "quote_slope"]].ffill()

    # Slope differential: positive → base is steeper → supportive for base
    df["slope_diff"] = df["base_slope"] - df["quote_slope"]

    # Filter: suppress longs when slope_diff is negative (quote steeper)
    df.loc[(df["signal"] == +1) & (df["slope_diff"] < 0), "signal"] = 0
    # Filter: suppress shorts when slope_diff is positive (base steeper)
    df.loc[(df["signal"] == -1) & (df["slope_diff"] > 0), "signal"] = 0

    return df


# Example for EUR/USD
usd_slope = fetch_curve_slope("usd")
eur_slope = fetch_curve_slope("eur")

analysis_filtered = apply_curve_filter(analysis, base_slope=usd_slope, quote_slope=eur_slope)
print(analysis_filtered[["zscore", "signal", "slope_diff"]].tail(8))

步骤5:生成警报并建立实时监视器

最后一步将信号逻辑组装成一个显示器,你可以按时间表运行 (每天关闭,每小时,或通过 汇率数据发布日历) 当一个新信号发射时,显示器会打印一个结构化的警报,您可以将其路由到Slack,电子邮件或交易网络连接.

from dataclasses import dataclass
from typing import Literal

SignalType = Literal["LONG", "SHORT", "EXIT", "HOLD"]


@dataclass
class SpreadAlert:
    fx_pair:     str
    signal:      SignalType
    spread_pct:  float
    zscore:      float
    slope_diff:  float
    timestamp:   str


def latest_signal(
    base_ccy: str,
    quote_ccy: str,
    fx_pair:   str,
    tenor:     str = "gov_bond_10y",
    lookback:  int = 60,
) -> SpreadAlert:
    base_yields  = fetch_yield(base_ccy,  tenor=tenor)
    quote_yields = fetch_yield(quote_ccy, tenor=tenor)
    df = pd.DataFrame({
        "base":  base_yields,
        "quote": quote_yields,
    }).dropna()

    base_col, quote_col = "base", "quote"
    df = compute_spread_signal(
        df.rename(columns={"base": base_col, "quote": quote_col}),
        base_col=base_col,
        quote_col=quote_col,
        lookback=lookback,
    )

    # Curve filter
    base_slope  = fetch_curve_slope(base_ccy)
    quote_slope = fetch_curve_slope(quote_ccy)
    df = apply_curve_filter(df, base_slope, quote_slope)

    row = df.iloc[-1]
    sig_map = {1: "LONG", -1: "SHORT", 0: "HOLD"}

    return SpreadAlert(
        fx_pair=fx_pair,
        signal=sig_map[int(row["signal"])],
        spread_pct=round(float(row["spread"]), 3),
        zscore=round(float(row["zscore"]), 2),
        slope_diff=round(float(row.get("slope_diff", float("nan"))), 3),
        timestamp=datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
    )


# Run for all pairs
for base_ccy, quote_ccy, fx_label in PAIRS:
    try:
        alert = latest_signal(base_ccy, quote_ccy, fx_label)
        print(
            f"[{alert.timestamp}] {alert.fx_pair:8s} | {alert.signal:5s} | "
            f"spread={alert.spread_pct:+.3f}%  z={alert.zscore:+.2f}  slope_diff={alert.slope_diff:+.3f}"
        )
    except Exception as exc:
        print(f"{fx_label}: error — {exc}")

# Example output:
# [2025-04-14T08:32:11Z] EUR/USD   | HOLD  | spread=+1.770%  z=+1.21  slope_diff=+0.210
# [2025-04-14T08:32:14Z] USD/JPY   | SHORT | spread=+4.050%  z=+2.18  slope_diff=+0.580
# [2025-04-14T08:32:17Z] AUD/USD   | HOLD  | spread=+0.340%  z=-0.55  slope_diff=-0.120
# [2025-04-14T08:32:20Z] GBP/USD   | LONG  | spread=+0.890%  z=-1.62  slope_diff=-0.340

摘要及下一步步骤

现在你有一个完整的收益率差距交易框架.

1. 债券收益率获取 没有 /announcements/{currency}/gov_bond_10y 现在我 /announcements/{currency}/gov_bond_2y 提供第二级公告时间的原材料
2. 差距和Z分数 60天滚动窗口周围的平均反转产生客观的入口和出口水平,而无需在单个值上调整曲线
3. 曲线倾斜波器 10Y2Y差异作为一个模式门,抑制了对每个货币自身收益率曲线结构偏差的信号
4. 现场警报 结构化 SpreadAlert 输出很容易被路由到任何通知,记录或执行管道

自然扩展包括将收益率差距与 政策利率差异 现在我 价格指数差异 通过一个综合的宏观分数, 破产率通胀率 调整通货膨胀的收益率差距从名义上转向实际收益差距.

所有可用的收益率和利率终点的完整文档在 没有任何参考为了获得API密钥,请访问 订阅现在我们要做什么?