Cambiar la brecha entre los rendimientos, no los rendimentos en sí

Los diferencial de rendimiento de los bonos del gobierno entre dos economías se encuentran entre las fuerzas estructurales más confiables en los mercados de divisas. Cuando el rendimiento a 10 años de los Estados Unidos se mueve sustancialmente por encima del equivalente al Bund alemán, el capital tiende a fluir hacia los activos USD y el EUR/USD enfrenta una presión de venta sostenida. Cuando ese diferencial se comprime, el par se recupera. La relación no es mecánica, pero es lo suficientemente persistente como para construir una estrategia de negociación basada en reglas a su alrededor.

Esta guía le guía a través de la construcción de una estrategia completa de comercio de pares de rendimiento con el uso de la API FXMacroData.

Obtiene series de tiempo de rendimiento de bonos del gobierno para dos monedas a través de la Puntos finales de rendimiento de los bonos FXMacroData
Calcula el diferencial de rendimiento y su media móvil y desviación estándar
Genera señales de divisas largas/cortas estadísticamente impulsadas utilizando la reversión media de la puntuación Z
Seguimiento de las posiciones abiertas, aplicación de controles de riesgo básicos y alertas de entrada/salida

Tesión central

Los diferenciales de rendimiento se invierten en torno a equilibrios estructuralmente estables. Cuando un diferencial se amplía bruscamente por encima de su promedio reciente, el par de divisas correspondiente está estrictamente estirado y es probable que se retrace.

Los requisitos previos

Antes de comenzar, asegúrese de tener listos los siguientes:

Python 3.9+ todos los fragmentos utilizan anotaciones de tipo estándar
La clave de la API de FXMacroData inscribirse en / suscribirse y copia tu clave desde el panel de la cuenta
Paquetes de Python¿ Qué ? requests¿ Qué ? pandas¿ Qué ? numpy

pip install requests pandas numpy

Almacene su clave de API como una variable de entorno nunca la codifique en archivos fuente:

export FXMACRO_API_KEY="YOUR_API_KEY"

Paso 1: Obtener datos sobre el rendimiento de los bonos del gobierno

La base de esta estrategia es una serie temporal de rendimientos de bonos confiable. gov_bond_10y Cada observación tiene un date¿ Qué ? val (rendimiento en porcentaje), y un announcement_datetime para la marca de tiempo de liberación de segundo nivel.

Aquí extraemos los rendimientos de USD y EUR a 10 años, y luego los ensamblamos en un único DataFrame alineado en fecha:

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

Puede utilizar cualquier par de monedas soportadas por el el punto final de gov_bond_10y Las combinaciones más populares incluyen los diferenciales USD/JPY, AUD/USD y GBP/USD.

Rendimiento a 10 años en dólares estadounidenses frente a euros ilustrativo

Los rendimientos en USD aumentaron más rápido que los equivalentes en EUR hasta 20222024, creando un margen persistentemente amplio que pesó sobre el EUR/USD durante todo el ciclo.

Paso 2: Calcule el diferencial de rendimiento y el puntaje Z

El spread bruto (USD menos rendimiento del EUR) le dice en qué dirección está estructuralmente sesgado el capital.

Un puntaje Z por encima de +1,5 indica que el diferencial se ha ampliado inusualmente una posible configuración corta del EUR/USD (larga del 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))

USDEUR 10Y Diferencia y puntuación Z ilustrativo

Las puntuaciones Z más allá de ±1,5 episodios históricamente marcados en los que la propagación se había movido demasiado lejos demasiado rápido y tendía a revertir la media en las semanas siguientes.

Paso 3: Extienda a varios pares

Una estrategia de spread única en EUR/USD es útil, pero el poder real surge cuando ejecuta la misma lógica en varios pares simultáneamente.

También puedes usar rendimientos de tenor más corto el gov_bond_2y El punto final es especialmente sensible a las expectativas de tipos de interés, lo que lo convierte en un indicador líder en comparación con las series a 10 años:

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

Referencia de la señal

+ 1: diferencias demasiado reducidas esperan una ampliación larga parte de la moneda base (por ejemplo, larga GBP/USD)
- 1: el diferencial es demasiado amplio se espera una compresión la parte baja de la moneda base es corta (por ejemplo, el USD/JPY corto es JPY largo)
0 - - -: dispersión dentro del rango normal sin borde direccional mantenerse plano

Paso 4: Añadir una superposición de pendiente de 2 años vs 10 años

La curva de rendimiento es una curva que se mueve hacia arriba, hacia abajo, hacia arriba y hacia abajo.

Saca las dos . 2 años ¿ Qué ? 10 años las emisiones de los valores de la moneda de origen para calcular la pendiente y utilizarla como filtro de régimen: sólo se toma una señal de dispersión en la dirección alineada con la pendientes de la curva de la divisa de origen.

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

Pendiente de la curva USD y EUR (10Y2Y) ilustrativo

Ambas curvas se invirtieron en 20222023; el diferencial de pendiente relativa todavía proporcionaba una señal negociable incluso cuando las pendientes absolutas eran negativas.

Paso 5: Generar alertas y construir un monitor en vivo

El último paso ensambla la lógica de la señal en un monitor que puede ejecutar en un horario (cerrar diariamente, cada hora, o activado inmediatamente después de un anuncio de rendimiento de bonos a través de la Calendario de lanzamiento de FXMacroDataCuando se dispara una nueva señal, el monitor imprime una alerta estructurada que puede enrutar a Slack, correo electrónico o un webhook comercial.

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

Consejo de programación

Los datos sobre el rendimiento de los bonos se actualizan cuando los gobiernos publican nuevos resultados de emisión y cuando los bancos centrales publican decisiones de política monetaria. Calendario de lanzamiento de FXMacroData para encontrar la próxima subasta de bonos programada o anuncio de política, y activar su actualización de señal inmediatamente después de que el evento se dispara.

Distribución de la señal entre pares ilustrativo

Con umbrales de puntuación Z de ±1,5 en una ventana de rodaje de 60 días, una mayoría significativa de los días cae en la zona plana, concentrando el despliegue de capital en configuraciones de alta convicción.

Resumen y pasos siguientes

Ahora tiene un marco completo de comercio de pares de rendimiento con diferenciales.

Obtención de rendimiento de bonos ¿ Qué pasa ? /announcements/{currency}/gov_bond_10y ¿ Qué ? /announcements/{currency}/gov_bond_2y suministrar la materia prima con marcas de tiempo de anuncio de segundo nivel
Diferencia y puntuación Z la reversión media alrededor de una ventana de rodaje de 60 días genera niveles de entrada y salida objetivos sin ajuste de curva en un único umbral
Filtro de inclinación de la curva el diferencial 10Y2Y actúa como un control de régimen, suprimiendo las señales que van en contra del sesgo estructural de la curva de rendimiento de cada moneda
Alertas en tiempo real el estructurado SpreadAlert la salida es fácil de enrutar a cualquier notificación, registro o tubería de ejecución

Las extensiones naturales incluyen la combinación de diferencias de rendimiento con Diferenciales de tipos de interés ¿ Qué ? Diferenciales del IPC En el caso de los datos de la evaluación de la calidad de los resultados, se puede utilizar el tasa de inflación de equilibrio el cambio de los diferenciales de rendimiento nominales a los diferencia les reales para el posicionamiento ajustado a la inflación.

La documentación completa de todos los puntos finales de rendimiento y tasa disponibles se encuentra en /api-referenciaPara obtener su clave API, visite / suscribirse- ¿ Qué ?

Securely Redirecting...

Cómo construir una estrategia de negociación de pares de rendimiento con FXMacroData