FXMacroData で Yield Spread パア取引戦略をどのように構築するか

生産 率 の 差 を 交換 する

外国為替市場における最も信頼性の高い構造的力の一つは,両国の政府債券の利回り差である.米国の10年利回りがドイツ・バンド相当値よりも大幅に上昇すると,資本はUSD資産に流入し,EUR/USDは継続的な販売圧力に直面する.その差が圧縮すると,ペアは回復する.関係は機械的ではないが,その周りにルールベースの取引戦略を構築するのに十分な持続性がある.

このガイドでは,FXMacroData API を使用して完全なリターン・スプレッド・ペア・トレーディング戦略を構築します.

• 通貨の2つの間の政府債券の収益率の時間列を FXMacroData 債券の収益性エンドポイント
• 利回り幅とローリング・ミディアと標準偏差を計算する
• 統計的に動かす長/短FX信号を生成する.
• オープンポジションを追跡し,基本的なリスク管理を実施し,エントリー/アウトリースアラートを表示する

条件

始める前に,次のものを用意してごらん.

• Python 3.9+ すべてのスニペットは標準的なタイプ注釈を使用します
• FXMacroData API キーを 登録する / サブスクリプト 口座のダッシュボードからあなたの鍵をコピー
• Python パッケージありがとうございました requestsほら pandasほら numpy

pip install requests pandas numpy

ソースファイルにハードコードしない:

export FXMACRO_API_KEY="YOUR_API_KEY"

ステップ1: 政府の債券の利回りデータを取得

この戦略の基礎は信頼性の高い債券利回り時間列です.FXMacroDataは,すべての主要通貨ブロックの10年期国債の利回りを, gov_bond_10y 観察が結果として dateほら val 収益率は, announcement_datetime レベル2のリリースタイムスタンプの場合は

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

サポートされている通貨ペアを使用できます. 決済の最終点 人気のある組み合わせには,USD/JPY,AUD/USD,GBP/USDのスプレッドが含まれます.

ステップ2: 収益率の差とZスコアを計算する

粗差 (USDマイナスEURの利回り) は,資本が構造的に偏った方向を教えてくれます.Zスコアは,最近の歴史に対してその分散を正規化し,時間帯や通貨ペアに直接比較可能な無次元信号を提供します.

+1.5以上のZスコアは,スプレッドが異常な幅に拡大したことを示唆する.

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年の傾斜層を重ねる

利回り曲線の形は戦略に第二の次元を追加する.急上昇曲線 (長期端の利回りが短期端よりも速く上昇する) は,通常成長期待の改善をシグナル化し,通貨をサポートする.逆転または平坦化曲線は,しばしば減速と中央銀行のピボットに先行する.

両方の... 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: 警告を発生し,ライブモニタを構築する

監視器に組み込む. 監視はスケジュールで実行できます (日々の閉店,毎時間,または 債券の利回り発表の直後に起動します. FXMacroData のリリース カレンダー新しい信号が発信されると,モニターはSlack,メール,または取引Webフックにルーティングできる構造化されたアラートをプリントします.

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 送信は,通知,ログ,実行パイプラインに簡単にルーティングできます.

生産性差を組み合わせる 政策金利の差 ほら CPI差異 複合マクロスコアに変換したり, 破綻インフレ率 インフレ調整のポジショニングの為の名目利回りから実利回りの利回りを切り替える

利回り率の最終点については, /api参照APIキーを取得するには, / サブスクリプトわかった