كيفية بناء استراتيجية تداول أزواج الأسعار مع FXMacroData

تبدل الفجوة بين العائدات، وليس العائدة نفسها

الفرق بين عائدات السندات الحكومية بين اقتصادين هي من بين القوى الهيكلية الأكثر موثوقية في أسواق العملات الأجنبية. عندما يتحرك عائد الولايات المتحدة لمدة 10 سنوات بشكل كبير فوق ما يعادل البوند الألماني، يميل رأس المال إلى التدفق نحو أصول الدولار الأمريكي و يواجه EUR / USD ضغط بيع مستمر. عندما يضغط هذا الانتشار، يتعافى الزوج. العلاقة ليست ميكانيكية، ولكنها ثابتة بما فيه الكفاية لبناء استراتيجية تداول قائمة على القواعد حوله.

هذا الدليل يسير لك من خلال بناء استراتيجية تداول زوج كاملة للفوائد المنتشرة باستخدام FXMacroData API. بحلول النهاية سيكون لديك نظام بايثون يعمل الذي:

• يحضر سلسلة زمنية من عائدات السندات الحكومية لعملتين عبر نقاط النهاية لثمن سندات FXMacroData
• يحسب فروق العائد ومتوسطها المتداول والانحراف المعياري
• يولد إشارات FX طويلة / قصيرة مدفوعة إحصائياً باستخدام عكس متوسط درجة Z
• تتبع المراكز المفتوحة، وتطبيق ضوابط المخاطر الأساسية، وتظهر التحذيرات الدخول/الخروج

الشروط المسبقة

قبل أن تبدأ، تأكد من أن لديك ما يلي جاهز:

• بايثون 3.9+ جميع المقتطفات تستخدم تعليقات نوعية قياسية
• مفتاح FXMacroData API التسجيل في / اشترك ونسخ مفتاحك من لوحة حسابك
• حزم بايثون. requests- لا pandas- لا numpy

pip install requests pandas numpy

تخزين مفتاح API الخاص بك كمتغير البيئة لا تشفيرها أبدا في ملفات المصدر:

export FXMACRO_API_KEY="YOUR_API_KEY"

الخطوة 1: احصل على بيانات عائد السندات الحكومية

أساس هذه الاستراتيجية هو سلسلة زمنية موثوقة لثمن السندات. توفر FXMacroData عائدات السندة الحكومية لمدة 10 سنوات لجميع كتلة العملات الرئيسية عبر gov_bond_10y كل ملاحظة تحمل date، a 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

يمكنك استخدام أي زوج من العملات المدعومة من قبل الحد النهائي gov_bond_10y تشمل المجموعات الشائعة فروق USD/JPY و AUD/USD و GBP/USD.

الخطوة 2: حساب فروق العائد و Z-Score

يخبرك الفارق الخام (USD ناقص عائد EUR) عن الاتجاه الذي يتحيز به رأس المال من الناحية الهيكلية. يطبق Z-score هذا الفارقة على تاريخها الأخير ، مما يمنحك إشارة غير مقارنة مباشرة عبر الفترات الزمنية وأزواج العملات.

تشير درجة Z أعلى من +1.5 إلى أن الفارق قد اتسع بشكل غير عادي إعداد EUR/USD قصير محتمل (طويل USD). تشير درجة Z أقل من −1.5 إلى ضغط غير طبيعي الإعداد 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))

الخطوة الثالثة: توسيع إلى أزواج متعددة

استراتيجية انتشار واحدة على اليورو/دولار مفيدة، ولكن القوة الحقيقية تظهر عندما تقوم بتشغيل المنطق نفسه عبر عدة أزواج في وقت واحد. يمنحك الجمع بين إشارات انتشار USD/JPY و AUD/USD و GBP/USD محفظة متنوعة مدفوعة بالعموم حيث يتم تحديد حجم كل موقف بشكل مستقل.

يمكنك أيضاً استخدام العائد القصير gov_bond_2y النقطة النهائية حساسة بشكل خاص لتوقعات أسعار الفائدة السياسية، مما يجعلها مؤشرًا رائدًا مقارنةً بالسلسلة العشرية:

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: إضافة ملف التكيف على المنحدرات لمدة عامين مقابل 10 سنوات

يضيف شكل منحنى العائد بعدًا ثانًا إلى الاستراتيجية. عادةً ما يشير منحنى التدفق (الاعائد الطويلة ترتفع أسرع من المقصورة) إلى تحسين توقعات النمو ودعم العملة. غالبًا ما يسبق منحنى الانعكاس أو التسطيح التباطؤات ومحاور البنك المركزي.

اسحب كلاهما سنتان و 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 أو البريد الإلكتروني أو شبكة التداول.

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 الخاص بك، زيارة / اشترك. .