Comment créer une stratégie de trading de paires de rendements avec FXMacroData

Échangez l'écart entre les rendements, pas les rendement eux- mêmes

Les différentiels de rendement des obligations d'État entre deux économies sont parmi les forces structurelles les plus fiables sur les marchés des changes. Lorsque le rendement américain à 10 ans dépasse sensiblement l'équivalent du Bund allemand, les capitaux ont tendance à circuler vers les actifs en USD et l'EUR/USD fait face à une pression de vente soutenue. Lorsque cet écart se compresse, la paire se rétablit. La relation n'est pas mécanique, mais elle est suffisamment persistante pour construire une stratégie de trading basée sur des règles autour d'elle.

Ce guide vous guide dans la construction d'une stratégie complète de trading de paires de rendement à écarts en utilisant l'API FXMacroData.

• Apporte les séries chronologiques de rendement des obligations d'État pour deux devises via le Les résultats de la période de référence sont les suivants:
• Calcule l'écart de rendement et sa moyenne mobile et son écart type
• Génère des signaux longs/courtes de change statistiquement guidés en utilisant la réversion moyenne du score Z
• Suivre les positions ouvertes, appliquer les contrôles de base des risques et faire apparaître les alertes d'entrée/sortie

Pré-requis

Avant de commencer, assurez- vous d'avoir les choses suivantes:

• Python 3.9+ tous les extraits utilisent des annotations de type standard
• Clé de l'API FXMacroData inscrivez-vous à / souscrivez et copiez votre clé du tableau de bord du compte
• Paquets PythonJe suis désolé . requestsJe suis désolé . pandasJe suis désolé . numpy

pip install requests pandas numpy

Stockez votre clé API comme une variable d'environnement ne la codez jamais dans les fichiers source:

export FXMACRO_API_KEY="YOUR_API_KEY"

Étape 1: Récupérer les données sur le rendement des obligations d'État

La base de cette stratégie est une série chronologique fiable de rendements obligataires. gov_bond_10y Chaque observation comporte un dateJe suis désolé . val (rendement en pourcentage), et un announcement_datetime pour l'horodatage de sortie de deuxième niveau.

Ici, nous tirons les rendements USD et EUR à 10 ans, puis les assembler dans un seul DataFrame aligné sur la date:

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

Vous pouvez utiliser n'importe quelle paire de devises supportée par le Le montant de la garantie est calculé en fonction de la valeur de la dette. Les combinaisons les plus populaires sont les spreads USD/JPY, AUD/USD et GBP/USD.

Étape 2: Calculer le spread de rendement et le score Z

Le spread brut (USD moins rendement EUR) vous indique dans quelle direction le capital est structurellement biaisé.

Un score Z supérieur à +1,5 indique que l'écart s'est élargi de manière inhabituelle une configuration potentielle courte EUR/USD (long 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))

Étape 3: étendre à plusieurs paires

Une stratégie de spread unique sur l'EUR/USD est utile, mais la puissance réelle émerge lorsque vous exécutez la même logique sur plusieurs paires simultanément.

Vous pouvez aussi utiliser des rendements de tenor plus court le gov_bond_2y Le point final est particulièrement sensible aux attentes des taux directeurs, ce qui en fait un indicateur de premier plan par rapport aux séries à 10 ans:

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

Étape 4: Ajouter une superposition de pente de 2 ans contre 10 ans

La courbe de rendement est une courbe qui s'allonge en deux dimensions: une courbure de croissance qui augmente plus rapidement que la courbe d'achèvement (qui augmente en long terme) indique généralement une amélioration des attentes de croissance et soutient la monnaie.

Tirez les deux . 2 ans Je suis désolé . 10 ans Les données de référence sont utilisées pour calculer la pente et l'utiliser comme filtre de régime: il suffit de prendre un signal de propagation dans le sens aligné sur la pende de la courbe de la monnaie nationale.

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))

Étape 5: Générer des alertes et construire un moniteur en direct

La dernière étape rassemble la logique du signal dans un moniteur que vous pouvez exécuter sur un calendrier (close quotidienne, heure par heure, ou déclenché immédiatement après une annonce de rendement obligataire via le Calendrier de sortie de FXMacroDataLorsque un nouveau signal se déclenche, le moniteur imprime une alerte structurée que vous pouvez acheminer vers Slack, e-mail ou un webhook de trading.

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

Résumé et prochaines étapes

Vous avez maintenant un cadre complet de négociation de paires de taux de rendement.

1. Résultats des obligations Je suis désolé . /announcements/{currency}/gov_bond_10y Je suis désolé . /announcements/{currency}/gov_bond_2y fournir la matière première avec des horodatages d'annonce de deuxième niveau
2. Différence et score Z la réversion moyenne autour d'une fenêtre mobile de 60 jours génère des niveaux d'entrée et de sortie objectifs sans ajustement de la courbe à un seuil unique
3. Filtre à pente courbe le différentiel 10Y2Y agit comme une barrière de régime, supprimant les signaux qui vont à l'encontre du biais structurel de la courbe de rendement de chaque devise
4. Alertes en direct le structuré SpreadAlert la sortie est facile à acheminer vers n'importe quelle pipeline de notification, d'enregistrement ou d'exécution

Les extensions naturelles incluent la combinaison des écarts de rendement avec différentiels de taux d'intérêt Je suis désolé . Différentiels d'IPC Dans le cas de la taux d'inflation de rupture de rentabilité pour passer des écarts de rendement nominaux à des épargnes de rendements réels pour le positionnement ajusté à l'inflation.

La documentation complète pour tous les critères de rendement et de taux disponibles est disponible à /pi-référencePour obtenir votre clé API, visitez / souscrivezJe suis désolé .