Wie man eine Yield Spread Pair Trading Strategie mit FXMacroData erstellt

Die Lücke zwischen den Erträgen, nicht die Erträge selbst

Die Differenzen zwischen den Renditen von Staatsanleihen zwischen zwei Volkswirtschaften gehören zu den zuverlässigsten strukturellen Kräften auf den Devisenmärkten. Wenn sich die 10-jährige Rendite der USA wesentlich über dem deutschen Bund bewegt, fließt Kapital in Richtung USD-Assets und EUR/USD steht unter anhaltendem Verkaufsdruck. Wenn dieser Spread komprimiert, erholt sich das Paar. Die Beziehung ist nicht mechanisch, aber sie ist hartnäckig genug, um eine regelbasierte Handelsstrategie um sie herum aufzubauen.

Dieser Leitfaden führt Sie durch den Aufbau einer vollständigen Rendite-Spread-Paar-Handelsstrategie mit der FXMacroData API.

• Holt die Zeitreihen der Rendite von Staatsanleihen für zwei Währungen über die Endpunkte der Rendite von FXMacroData-Anleihen
• Berechnet den Rendite-Spread und seinen gleitenden Durchschnitt und die Standardabweichung
• Erzeugt statistisch gesteuerte lange/kurze FX-Signale unter Verwendung der Z-Score-Mittelumkehrung
• Verfolgt offene Positionen, wendet grundlegende Risikokontrollen an und stellt Einstiegs-/Ausstiegswarnungen auf

Voraussetzungen

Bevor Sie anfangen, stellen Sie sicher, dass Sie Folgendes bereit haben:

• Python 3.9+ alle Snippets verwenden Standard-Typ-Annotationen
• FXMacroData-API-Schlüssel melden Sie sich an /abonnieren und kopieren Sie Ihren Schlüssel aus dem Konto-Dashboard
• Python-Pakete- Ich weiß . requests- Ich weiß . pandas- Ich weiß . numpy

pip install requests pandas numpy

Speichern Sie Ihren API-Schlüssel als Umgebungsvariable niemals in Quelldateien hardcodieren:

export FXMACRO_API_KEY="YOUR_API_KEY"

Schritt 1: Erhebung von Renditen von Staatsanleihen

Das Fundament dieser Strategie sind zuverlässige Zeitreihen der Anleihenrendite. FXMacroData liefert 10-jährige Renditen von Staatsanleihen für alle wichtigen Währungsblöcke über die gov_bond_10y Jede Beobachtung trägt eine date- Ich weiß . val (Rendite in Prozent) und announcement_datetime für den Zeitstempel der Freigabe der zweiten Ebene.

Hier ziehen wir USD und EUR 10-Jahresrenditen, dann versammeln sie in einem einzigen DataFrame ausgerichtet auf Datum:

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

Sie können jedes Währungspaar verwenden , das von der endpunkt von gov_bond_10y Zu den beliebtesten Kombinationen gehören USD/JPY, AUD/USD und GBP/USD-Spreads.

Schritt 2: Berechnen Sie den Yield Spread und den Z-Score

Der Rohspread (USD minus EUR Rendite) zeigt an, in welche Richtung das Kapital strukturell verzerrt ist. Der Z-Score normalisiert diesen Spread gegenüber seiner jüngsten Geschichte und gibt Ihnen ein dimensionloses Signal, das direkt über Zeiträume und Währungspaare hinweg vergleichbar ist.

A Z-score above +1.5 indicates the spread has widened unusually far — a potential short EUR/USD (long USD) setup. A Z-score below −1.5 indicates abnormal compression — a potential long EUR/USD setup.

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

Schritt 3: Erweitern Sie auf mehrere Paare

Eine einzige Spread-Strategie für EUR/USD ist nützlich, aber die wirkliche Macht entsteht, wenn Sie die gleiche Logik gleichzeitig über mehrere Paare hinweg ausführen.

Sie können auch kürzere Tenor-Erträge verwenden die gov_bond_2y Der Endpunkt ist besonders empfindlich gegenüber den Erwartungen an die Leitzinsen, was ihn im Vergleich zu den 10-Jahresreihen zu einem führenden Indikator macht:

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

Schritt 4: Hinzufügen einer 2-Jahres- vs. 10-Jahr-Spannung

Die Form der Rendite-Kurve verleiht der Strategie eine zweite Dimension. Eine steilene Kurve (langen Renditen schneller steigen als kurzen) signalisiert typischerweise eine Verbesserung der Wachstumserwartungen und unterstützt die Währung. Eine umgekehrte oder flachmachende Kurve geht oft Verlangsamungen und Zentralbank-Pivoten voraus.

Zieh beide . 2 Jahre Und ... 10 Jahre Die Ausgabe der Ausgabe ist in der Regel in der folgenden Form zu erstellen:

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

Schritt 5: Erstellen von Warnungen und Erstellen eines Live-Monitors

Der letzte Schritt ist die Zusammenstellung der Signallogik in einen Monitor, den Sie nach einem Zeitplan ausführen können (täglich geschlossen, stündlich oder sofort nach einer Ankündigung über eine Anleihe durch die FXMacroData-VeröffentlichungskalenderWenn ein neues Signal ausgelöst wird, druckt der Monitor eine strukturierte Warnung aus, die Sie an Slack, E-Mail oder einen Handels-Webhook weiterleiten können.

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

Zusammenfassung und weitere Schritte

Sie haben nun ein komplettes Rahmenwerk für den Handel mit Zinsspannen.

1. Anleiheertrag Ich bin nicht hier . /announcements/{currency}/gov_bond_10y Und ... /announcements/{currency}/gov_bond_2y Liefern Sie den Rohstoff mit Zeitstempeln der zweiten Ebene
2. Spread und Z-Score Durchschnittliche Rückkehr um ein 60-tägiges Rollfenster erzeugt objektive Ein- und Ausstiegsniveaus ohne Kurvenanpassung an einen einzigen Schwellenwert
3. Filter für die Steigung der Kurve Die Differenz von 10Y2Y wirkt wie ein Regime-Gate und unterdrückt Signale, die gegen die strukturelle Verzerrung der Renditekurve jeder Währung verstoßen
4. Live-Warnungen die strukturierte SpreadAlert Ausgabe ist leicht in jede Benachrichtigung, Protokollierung oder Ausführung Pipeline zu leiten

Natürliche Erweiterungen umfassen die Kombination von Ertragsspannen mit Zinsdifferenzen Und ... Preisindexdifferenzen Die Ergebnisse werden in einem zusammengesetzten Makroscore oder mit Hilfe der Brückennutzeninflation die Nominalrenditenspreads für die inflationsausgeglichenen Positionierungen auf die realen zu verlagern.

Die vollständige Dokumentation für alle verfügbaren Rendite- und Ratenendpunkte finden Sie unter /api-VerweisUm Ihren API-Schlüssel zu erhalten, besuchen Sie /abonnieren- Ich weiß .