Trading algorithmique avec des signaux macroéconomiques : Un guide complet de stratégie FX

Pourquoi les signaux macro sont la base du trading FX Algo

Les taux de change ne bougent pas au hasard. Ils reflètent l'attractivité relative de deux économies: quelle banque centrale resserre plus rapidement, où les rendements réels sont plus élevés, quel régime d'inflation détériore le pouvoir d'achat et où le capital circule en conséquence. Chaque tendance majeure de plusieurs semaines dans l'EUR/USD, AUD/JPY ou GBP/USD peut être attribuée à des changements dans ces fondamentaux des changement qui sont annoncés sur un calendrier connu et observables à travers des données publiques.

Les traders discrétionnaires traitent ces signaux manuellement. Les trader algorithmiques les codifient. Ce guide vous guide dans la construction d'une stratégie FX complète basée sur les signaux macro en Python en utilisant FXMacroData comme couche de données. À la fin, vous aurez un système de travail qui:

• Retrouve les taux directeurs, l'inflation, l"emploi et les différentiels de rendement obligataires pour deux devises via l'API FXMacroData
• Calcule un score composite de macro-régime pour identifier le biais directionnel
• Retirer l'historique des taux de change au comptant pour fournir un contexte technique
• Planifier les mises à jour de signaux autour des événements du calendrier de sortie à fort impact
• Émet des signaux longs/courtes/neutraux avec orientation de la taille de la position
• Applique des contrôles de risque de base: stop-loss, limites de taille et pannes de courant dans les fenêtres d'actualités

Pré-requis

Avant de commencer, assurez- vous d'avoir les éléments suivants:

• 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 jamais les identifiants de code dur dans les fichiers source:

export FXMACRO_API_KEY="YOUR_API_KEY"

Les exemples ci-dessous sont des échanges Le taux de change, mais le même schéma s'applique à toute paire disponible dans FXMacroData. EUR Je suis désolé . USD pour les devises que vous voulez modéliser.

Étape 1: Récupérer les indicateurs macro de base

Quatre groupes d'indicateurs déterminent la majorité des mouvements structurels des changes: les taux de la banque centrale, l'inflation des prix à la consommation, la santé du marché du travail et les rendements des obligations publiques. régime macroéconomique pour chaque face d'une paire de devises.

FXMacroData fournit toutes ces informations par le biais d'un point de terminaison REST cohérent: GET /api/v1/announcements/{currency}/{indicator}Chaque observation contient une dateJe suis désolé . val (valeur de l'indicateur) et un announcement_datetime précis à la seconde pour que vous sachiez toujours exactement quand le marché a découvert.

import os
import requests
from datetime import date, timedelta

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY  = os.environ["FXMACRO_API_KEY"]


def fetch_indicator(currency: str, indicator: str, lookback_days: int = 400) -> list[dict]:
    """Fetch the most recent observations for a macro indicator."""
    start = (date.today() - timedelta(days=lookback_days)).isoformat()
    resp = requests.get(
        f"{BASE_URL}/announcements/{currency}/{indicator}",
        params={"api_key": API_KEY, "start": start},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])


# Pull the four core series for both sides of EUR/USD
usd_rate      = fetch_indicator("usd", "policy_rate")
eur_rate      = fetch_indicator("eur", "policy_rate")
usd_inflation = fetch_indicator("usd", "inflation")
eur_inflation = fetch_indicator("eur", "inflation")
usd_nfp       = fetch_indicator("usd", "non_farm_payrolls")
usd_unemp     = fetch_indicator("usd", "unemployment")
usd_yield     = fetch_indicator("usd", "gov_bond_10y")
eur_yield     = fetch_indicator("eur", "gov_bond_10y")

Le taux de réglementation Je suis désolé . Rentabilité des obligations à 10 ans Les taux d'intérêt sont directement liés à la dimension des taux d inflation La série de mesures révélera si une banque centrale devra encore resserrer ses mesures ou si elle a des marges d'assouplissement. Liste des salariés non agricoles Je suis désolé . le chômage La situation de l'emploi est en revanche plus complexe.

Étape 2: calculer un score de macro-régime

Un score de régime regroupe plusieurs indicateurs en un seul signal directionnel. L'approche ici est délibérément simple: pour chaque devise, comparez le dernier taux directeur, le taux d'inflation et le rendement obligataire à leurs propres moyennes sur 12 mois. Une devise dont les fondamentaux sont supérieurs à leur tendance est dans un le régime de renforcement; une en dessous de sa tendance est dans un régime d'affaiblissementL'écart entre les deux scores vous donne le biais directionnel de la paire.

import pandas as pd
import numpy as np


def latest_val(series: list[dict]) -> float | None:
    """Return the most recent value from a sorted indicator series."""
    if not series:
        return None
    return series[-1]["val"]


def rolling_zscore(series: list[dict], window: int = 12) -> float | None:
    """Z-score of the latest value relative to the last `window` observations."""
    vals = [r["val"] for r in series if r.get("val") is not None]
    if len(vals) < 2:
        return None
    arr = np.array(vals[-window:], dtype=float)
    mu, sigma = arr.mean(), arr.std()
    if sigma == 0:
        return 0.0
    return float((arr[-1] - mu) / sigma)


def macro_score(
    policy_rate: list[dict],
    inflation:   list[dict],
    bond_yield:  list[dict],
) -> float:
    """
    Composite macro score for one currency.
    Positive → strengthening macro backdrop.
    Negative → weakening macro backdrop.
    """
    weights = {"policy_rate": 0.40, "inflation": 0.30, "bond_yield": 0.30}
    scores = {
        "policy_rate": rolling_zscore(policy_rate),
        "inflation":   rolling_zscore(inflation),
        "bond_yield":  rolling_zscore(bond_yield),
    }
    total, weight_sum = 0.0, 0.0
    for key, w in weights.items():
        z = scores[key]
        if z is not None:
            total      += w * z
            weight_sum += w
    return total / weight_sum if weight_sum > 0 else 0.0


usd_score = macro_score(usd_rate, usd_inflation, usd_yield)
eur_score = macro_score(eur_rate, eur_inflation, eur_yield)

# Positive → USD macro stronger → bias SHORT EUR/USD
# Negative → EUR macro stronger → bias LONG EUR/USD
regime_spread = usd_score - eur_score
print(f"USD macro score: {usd_score:+.3f}")
print(f"EUR macro score: {eur_score:+.3f}")
print(f"Regime spread (USD − EUR): {regime_spread:+.3f}")

Étape 3: Ajouter le contexte du marché du travail pour le dollar américain

Pour les paires de dollars américains en particulier, le marché du travail annule souvent le signal de taux à court terme. Une impression de paie explosive peut pousser la Fed à mettre en pause les coupes même lorsque l'inflation chute; un bond surprenant du chômage peut accélérer les attentes d'assouplissement.

def employment_score(nfp: list[dict], unemployment: list[dict]) -> float:
    """
    Labour market contribution to the USD score.
    Positive NFP momentum + falling unemployment → bullish.
    """
    nfp_z   = rolling_zscore(nfp)
    unemp_z = rolling_zscore(unemployment)

    if nfp_z is None and unemp_z is None:
        return 0.0
    score = 0.0
    count = 0
    if nfp_z is not None:
        score += 0.60 * nfp_z   # NFP gets more weight
        count += 1
    if unemp_z is not None:
        # Unemployment is inverse: a rising z-score is bearish for USD
        score -= 0.40 * unemp_z
        count += 1
    return score


usd_employment = employment_score(usd_nfp, usd_unemp)

# Rebuild USD score including labour market
usd_score_full = (
    0.35 * (rolling_zscore(usd_rate) or 0.0) +
    0.25 * (rolling_zscore(usd_inflation) or 0.0) +
    0.25 * (rolling_zscore(usd_yield) or 0.0) +
    0.15 * usd_employment
)

regime_spread_full = usd_score_full - eur_score
print(f"USD score (with labour): {usd_score_full:+.3f}")
print(f"Regime spread (full):    {regime_spread_full:+.3f}")

Étape 4: Retirer l'historique des taux de change au comptant

Les scores du régime macro fournissent une conviction directionnelle, mais le timing d'entrée bénéficie toujours d'un filtre basé sur le prix. point final de change fournit des taux de clôture quotidiens que vous pouvez utiliser pour calculer le contexte de prix de base.

def fetch_spot_rates(base: str, quote: str, lookback_days: int = 200) -> pd.Series:
    """Fetch FX spot rate history and return as a date-indexed Series."""
    start = (date.today() - timedelta(days=lookback_days)).isoformat()
    resp = requests.get(
        f"{BASE_URL}/forex/{base}/{quote}",
        params={"api_key": API_KEY, "start": start},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json().get("data", [])
    if not data:
        return pd.Series(dtype=float)
    df = pd.DataFrame(data).set_index("date").sort_index()
    return df["close"].astype(float)


spot = fetch_spot_rates("EUR", "USD")
sma50  = spot.rolling(50).mean()
sma200 = spot.rolling(200).mean()

latest_price  = spot.iloc[-1]
latest_sma50  = sma50.iloc[-1]
latest_sma200 = sma200.iloc[-1]

# Simple trend filter: is price above or below key moving averages?
price_trend = "bullish" if latest_price > latest_sma50 > latest_sma200 else (
              "bearish" if latest_price < latest_sma50 < latest_sma200 else "mixed")

print(f"EUR/USD latest: {latest_price:.5f}  SMA50: {latest_sma50:.5f}  SMA200: {latest_sma200:.5f}")
print(f"Price trend: {price_trend}")

Étape 5: Abonnez-vous au calendrier des sorties

Le moment le plus dangereux pour détenir une position FX ouverte est les 15 minutes autour d'une sortie importante. Les décisions de taux de politique, les impressions de l'IPC et les données de paie comportent toutes le potentiel de 3080 lacunes de pip.

Le point de fin du calendrier de sortie FXMacroData renvoie chaque sortie programmée à venir avec son nom d'indicateur et sa date d'annonce prévue, ce qui facilite la création d'un planificateur de black-out:

from datetime import datetime, timezone

def fetch_upcoming_releases(currency: str, days_ahead: int = 14) -> list[dict]:
    """Return scheduled macro releases for a currency over the next N days."""
    resp = requests.get(
        f"{BASE_URL}/calendar/{currency}",
        params={"api_key": API_KEY},
        timeout=10,
    )
    resp.raise_for_status()
    events = resp.json().get("data", [])
    cutoff = datetime.now(timezone.utc) + timedelta(days=days_ahead)
    upcoming = []
    for evt in events:
        dt_str = evt.get("announcement_datetime") or evt.get("date")
        if not dt_str:
            continue
        try:
            evt_dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
        except ValueError:
            continue
        if datetime.now(timezone.utc) <= evt_dt <= cutoff:
            upcoming.append(evt)
    return upcoming


HIGH_IMPACT = {"policy_rate", "inflation", "non_farm_payrolls", "unemployment", "gdp"}
BLACKOUT_MINUTES = 20  # minutes before/after release to block new entries


def is_in_blackout_window(releases: list[dict], now: datetime | None = None) -> bool:
    """Return True if the current moment falls inside any high-impact release window."""
    if now is None:
        now = datetime.now(timezone.utc)
    window = timedelta(minutes=BLACKOUT_MINUTES)
    for evt in releases:
        if evt.get("indicator") not in HIGH_IMPACT:
            continue
        dt_str = evt.get("announcement_datetime") or evt.get("date")
        if not dt_str:
            continue
        try:
            evt_dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
        except ValueError:
            continue
        if abs(now - evt_dt) <= window:
            return True
    return False


usd_releases = fetch_upcoming_releases("usd")
eur_releases = fetch_upcoming_releases("eur")
all_releases = usd_releases + eur_releases

print(f"Upcoming high-impact releases (next 14 days): {len(all_releases)}")
print(f"Currently in blackout window: {is_in_blackout_window(all_releases)}")

Étape 6: générer un signal en direct

Avec les scores macro, le contexte du taux au comptant et une vérification de l'éteinte de calendrier, vous pouvez les assembler en une seule fonction de signal qui produit une direction, une confiance et une taille de position recommandée à la demande.

from dataclasses import dataclass
from typing import Literal


@dataclass
class Signal:
    direction: Literal["long", "short", "neutral"]
    confidence: float        # 0.0 → 1.0
    regime_spread: float     # positive → USD stronger
    price_trend: str
    in_blackout: bool
    reason: str


REGIME_THRESHOLD = 0.45   # minimum spread magnitude to take a position
TREND_CONFIRMATION = True  # require price trend to agree with regime signal


def generate_signal(
    regime_spread: float,
    price_trend:   str,
    releases:      list[dict],
) -> Signal:
    """
    Combine macro regime spread and price trend into a trade signal.

    regime_spread > 0  → USD stronger → short EUR/USD (quote currency up)
    regime_spread < 0  → EUR stronger → long  EUR/USD (base currency up)
    """
    in_blackout = is_in_blackout_window(releases)

    if in_blackout:
        return Signal("neutral", 0.0, regime_spread, price_trend, True,
                      "Blackout window: high-impact release imminent.")

    magnitude = abs(regime_spread)

    if magnitude < REGIME_THRESHOLD:
        return Signal("neutral", 0.0, regime_spread, price_trend, False,
                      f"Regime spread {regime_spread:+.3f} below threshold {REGIME_THRESHOLD}.")

    # Determine raw macro direction
    macro_dir = "short" if regime_spread > 0 else "long"

    # Price trend confirmation
    if TREND_CONFIRMATION:
        if macro_dir == "short" and price_trend == "bullish":
            return Signal("neutral", 0.20, regime_spread, price_trend, False,
                          "Macro bearish EUR/USD but price trend still bullish — wait for confirmation.")
        if macro_dir == "long" and price_trend == "bearish":
            return Signal("neutral", 0.20, regime_spread, price_trend, False,
                          "Macro bullish EUR/USD but price trend still bearish — wait for confirmation.")

    # Confidence scales with regime magnitude (capped at 0.90)
    confidence = min(0.90, magnitude / 1.5)

    return Signal(macro_dir, confidence, regime_spread, price_trend, False,
                  f"Macro {'USD' if macro_dir == 'short' else 'EUR'} outperformance confirmed by price trend.")


signal = generate_signal(regime_spread_full, price_trend, all_releases)
print(f"Signal: {signal.direction.upper()}  confidence: {signal.confidence:.0%}")
print(f"Reason: {signal.reason}")

Étape 7: Appliquer les contrôles des risques et la taille des positions

La génération de signaux n'est que la moitié du travail. Sans contrôles systématiques des risques, même une source de signal de haute qualité produira des retraits qui dépassent ce que la stratégie peut supporter. Trois contrôles sont essentiels au minimum: une taille de position maximale par rapport au capital du compte, un stop-loss dur en pips et une limite de perte quotidienne qui met en pause la négociation après une série de sessions perdantes.

@dataclass
class RiskConfig:
    account_equity:      float = 10_000.0   # USD
    risk_per_trade_pct:  float = 1.0        # percent of equity risked per trade
    stop_loss_pips:      float = 30.0       # maximum allowed loss in pips
    pip_value_per_lot:   float = 10.0       # USD per pip per standard lot (EUR/USD)
    max_lots:            float = 2.0        # hard cap on position size
    daily_loss_limit_pct: float = 3.0       # pause trading if daily loss exceeds this


def compute_position_size(signal: Signal, config: RiskConfig) -> float:
    """
    Return lot size based on risk per trade and stop-loss.
    Scales with signal confidence — higher confidence allows up to full risk.
    """
    if signal.direction == "neutral":
        return 0.0

    risk_amount = config.account_equity * (config.risk_per_trade_pct / 100)
    # Scale risk by confidence
    adjusted_risk = risk_amount * signal.confidence
    # Max loss per lot at this stop = stop_loss_pips * pip_value_per_lot
    max_loss_per_lot = config.stop_loss_pips * config.pip_value_per_lot
    if max_loss_per_lot == 0:
        return 0.0
    raw_lots = adjusted_risk / max_loss_per_lot
    return round(min(raw_lots, config.max_lots), 2)


risk   = RiskConfig()
lots   = compute_position_size(signal, risk)
dollar_risk = lots * risk.stop_loss_pips * risk.pip_value_per_lot

print(f"Recommended position: {lots} lots  ({signal.direction.upper()} EUR/USD)")
print(f"Max risk at {risk.stop_loss_pips}-pip stop: ${dollar_risk:.2f}")

Étape 8: mettre tout en place La boucle de la stratégie quotidienne

L'étape finale rassemble tout dans une boucle d'exécution qui s'exécute une fois par jour, rafraîchit toutes les données macro, évalue le signal, vérifie le calendrier de sortie et émet une recommandation d'ordre.

import logging
from datetime import date, datetime, timedelta, timezone

logging.basicConfig(level=logging.INFO, format="%(asctime)s  %(message)s")
log = logging.getLogger(__name__)


def run_daily_strategy():
    """Main strategy loop — call once per trading day."""
    log.info("─── Daily macro strategy update ───")

    # 1. Fetch macro data
    log.info("Fetching macro indicators...")
    usd_rate_data  = fetch_indicator("usd", "policy_rate")
    eur_rate_data  = fetch_indicator("eur", "policy_rate")
    usd_inf_data   = fetch_indicator("usd", "inflation")
    eur_inf_data   = fetch_indicator("eur", "inflation")
    usd_nfp_data   = fetch_indicator("usd", "non_farm_payrolls")
    usd_unemp_data = fetch_indicator("usd", "unemployment")
    usd_bond_data  = fetch_indicator("usd", "gov_bond_10y")
    eur_bond_data  = fetch_indicator("eur", "gov_bond_10y")

    # 2. Compute regime scores
    usd_emp    = employment_score(usd_nfp_data, usd_unemp_data)
    usd_s      = (
        0.35 * (rolling_zscore(usd_rate_data) or 0.0) +
        0.25 * (rolling_zscore(usd_inf_data) or 0.0) +
        0.25 * (rolling_zscore(usd_bond_data) or 0.0) +
        0.15 * usd_emp
    )
    eur_s = macro_score(eur_rate_data, eur_inf_data, eur_bond_data)
    spread = usd_s - eur_s
    log.info(f"USD score: {usd_s:+.3f}  EUR score: {eur_s:+.3f}  Spread: {spread:+.3f}")

    # 3. Fetch spot rates and compute trend
    log.info("Fetching spot rates...")
    spot_series = fetch_spot_rates("EUR", "USD")
    sma50_val   = spot_series.rolling(50).mean().iloc[-1] if len(spot_series) >= 50 else None
    sma200_val  = spot_series.rolling(200).mean().iloc[-1] if len(spot_series) >= 200 else None
    last_price  = spot_series.iloc[-1]
    trend = "mixed"
    if sma50_val and sma200_val:
        trend = ("bullish" if last_price > sma50_val > sma200_val else
                 "bearish" if last_price < sma50_val < sma200_val else "mixed")
    log.info(f"EUR/USD {last_price:.5f}  trend: {trend}")

    # 4. Fetch release calendar
    log.info("Fetching release calendars...")
    releases = fetch_upcoming_releases("usd") + fetch_upcoming_releases("eur")
    log.info(f"Upcoming events: {len(releases)}")

    # 5. Generate signal
    sig  = generate_signal(spread, trend, releases)
    lots = compute_position_size(sig, RiskConfig())
    log.info(f"Signal: {sig.direction.upper()}  confidence: {sig.confidence:.0%}  lots: {lots}")
    log.info(f"Reason: {sig.reason}")

    return sig, lots


if __name__ == "__main__":
    run_daily_strategy()

Suivant: élargir la stratégie

Le cadre ci-dessus est intentionnellement léger pour que vous puissiez suivre chaque décision des données brutes à la sortie finale.

• Ajouter plus de devises s'étendre à la livre sterling, au dollar australien, au yen japonais ou au dollar canadien en utilisant les mêmes indicateurs. Taux directeur de la livre sterling Je suis désolé . Inflation en dollars australiens Les séries suivent le même contrat de données.
• Ajouter des données de positionnement COT le positionnement des gros spéculateurs dans le rapport COT de la CFTC est un filtre de sentiment utile. Lorsque le régime macro dit long USD mais que les longs spéculatifs sont déjà extrêmes, le risque/rendement d'une nouvelle entrée est inférieur. FXMacroData fournit des données COT via la même API.
• Test de retour par rapport aux données d'annonce historiques parce que chaque observation FXMacroData porte une announcement_datetime avec une précision de la seconde, vous pouvez reconstruire exactement ce que le marché savait à tout moment et simuler les entrées de stratégie sans biais de la tête de l'observateur.
• Automatiser avec un planificateur enveloppé run_daily_strategy() Les signaux macro typiques ne doivent être mis à jour qu'après les principales versions de données, de sorte que des mises à jour quotidiennes ou même hebdomadaires sont suffisantes pour les positions à moyen terme.
• Connectez-vous à une API de courtier les sorties de signal et de taille de lot sont indépendantes du courtier. direction Je suis désolé . lots pour les ordres de marché dans votre couche d'exécution préférée (OANDA v20, Interactive Brokers TWS ou un simulateur de négociation papier).

Résumé

Les signaux macro ne sont pas une décoration pour un alg ils sont le bord.

1. Retirer les taux directeurs, l'inflation, l"emploi et les rendements obligataires de l'API FXMacroData avec un modèle cohérent
2. Calculer un score composite pour chaque devise en utilisant la normalisation du score Z et le mélange d'indicateurs pondérés
3. Ajouter un filtre de tendance des prix à partir de l'historique des taux de change au comptant pour exiger un accord macro et technique avant d'entrer
4. Utilisez le calendrier de sortie pour éviter le bruit des fenêtres de données à fort impact avec un simple planificateur de black-out
5. Taille des positions proportionnelle à la confiance des signaux et au fonds propres du compte en utilisant un modèle de risque fractionné fixe
6. Assembler tous les composants dans une boucle quotidienne unique et répétable

La couche de données La série d'annonces à horodatage précis et systématiquement structurée de FXMacroData fait le gros du travail de maintenir chaque signal basé sur des faits de marché vérifiables.