Algorithmic Trading mit Makrosignalen: Ein vollständiger FX-Strategie-Leitfaden

Warum Makro-Signale die Grundlage des FX Algo-Handels sind

Die Wechselkurse bewegen sich nicht zufällig. Sie spiegeln die relative Attraktivität zweier Volkswirtschaften wider: welche Zentralbank schneller zusammengeht, wo die realen Renditen höher sind, welches Inflationsregime die Kaufkraft verschlechtert und wo infolgedessen Kapital fließt. Jeder wichtige mehrwöchige Trend in EUR/USD, AUD/JPY oder GBP/USD kann auf Verschiebungen in diesen Fundamentaldaten zurückgeführt werden Verschiebung, die in einem bekannten Zeitplan angekündigt und durch öffentliche Daten beobachtbar sind.

Die Handlung wird von den algorithmischen Handlern manuell verarbeitet. Diese Handbuch führt Sie durch den Aufbau einer vollständigen Makro-Signal-basierten FX-Strategie in Python mit FXMacroData als Datenschicht. Am Ende haben Sie ein funktionierendes System, das:

• Holt die Leitzinsen, die Inflation, die Beschäftigung und die Differenzen der Anleihenzinsentwicklung für zwei Währungen über die FXMacroData API ab
• Berechnet einen zusammengesetzten Makroregierungswert zur Identifizierung der Richtungsverzerrung
• Zieht den Kursverlauf der Devisenmarktkurse ab, um den technischen Kontext zu ermitteln
• Planung von Signal-Aktualisierungen rund um hochwirksame Release-Kalenderereignisse
• Ausstrahlt lange/kurze/neutrale Signale mit Positionstärkenanleitung
• Anwendung grundlegender Risikokontrollen: Stop-Loss, Größenlimits und News-Window-Blackouts

Voraussetzungen

Bevor du anfängst, solltest du dir die folgenden Dinge vornehmen:

• 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 Hard-Code-Anmeldeinformationen in Quelldateien:

export FXMACRO_API_KEY="YOUR_API_KEY"

Die folgenden Beispiele sind Handelsfragen. EUR/USD, aber das gleiche Muster gilt für jedes Paare, das in FXMacroData verfügbar ist. EUR Und ... USD für die Währungen, die Sie modellieren möchten.

Schritt 1: Holen Sie sich die Kernmakroindikatoren

Die meisten Strukturwandelindikatoren sind vier Gruppen: Zentralbankpolitik, Verbraucherpreisinflation, Arbeitsmarktgesundheit und Staatsanleihenrenditen. Makro-Regelung für jede Seite eines Währungspaares.

FXMacroData stellt all dies über einen konsistenten REST-Endpunkt bereit: GET /api/v1/announcements/{currency}/{indicator}Jede Beobachtung enthält eine date- Ich weiß . val (Wert des Indikators) und announcement_datetime Sie wissen immer genau, wann der Markt es herausgefunden hat.

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

Die ... Zinssatz Und ... 10-Jahres-Anleihenrendite Die Zinssätze werden in der Regel direkt erfasst. Inflation Die Zahl der Banken, die in der Serie "Einheitliche Währung" (EVS) aufgenommen werden, zeigt, ob eine Zentralbank weiter anpassen muss oder Raum für Lockerungen hat. Nichtlandwirtschaftliche Lohnlisten Und ... Arbeitslosigkeit Die Entwicklung der Arbeitslosigkeit ist in den letzten Jahren deutlich gestiegen.

Schritt 2: Berechnen Sie einen Makroverlauf

Eine Regime-Score bündelt mehrere Indikatoren in ein einziges Richtungssignal. Der Ansatz ist hier bewusst einfach: Vergleiche für jede Währung den jüngsten Leitzins, die Inflationsrate und die Anleiherendite mit ihren eigenen 12-Monats-Durchschnitten. Eine Währung, deren Fundamentals über ihrem Trend liegen, ist in einer Stärkung der RegelungEine unter dem Trend liegt in einem SchwächungsregimeDie Differenz zwischen den beiden Punkten gibt Ihnen die Richtungsneigung des Paares.

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

Schritt 3: Hinzufügen des Arbeitsmarktkontexts für USD

Für USD-Paare speziell überschreitet der Arbeitsmarkt oft das Zinssignal kurzfristig. Ein Ausbruch der Lohnlisten kann die Fed dazu bringen, Kürzungen zu pausieren, selbst wenn die Inflation sinkt; ein überraschender Anstieg der Arbeitslosigkeit kann die Erleichterungserwartungen beschleunigen. Die Einbeziehung einer Beschäftigungskomponente schärft die USD-Score um hochwirksame Datenfenster.

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

Schritt 4: Erfassen Sie die Kursgeschichte der FX-Spotkurse

Die Makro-Regime-Scores liefern eine Richtungsüberzeugung, aber der Eintrittszeitpunkt profitiert immer noch von einem preisbasierten Filter. Endpunkt für den Devisenhandel bietet tägliche Schlusskurse, mit denen Sie den Basispreiskontext berechnen können.

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

Schritt 5: Abonnieren Sie den Kalender der Veröffentlichungen

Die gefährlichste Zeit, um eine offene Devisenposition zu halten, ist die 15 Minuten um eine große geplante Veröffentlichung herum. Richtungsentscheidungen, CPI-Drucke und Lohnlistendaten tragen alle das Potenzial für 3080 Pip-Lücken. Ein gut disziplinierter Algo vermeidet das Eintreten neuer Positionen in diesen Fenstern und kann optional bestehende schließen oder absichern.

Der Endpunkt des Release-Kalenders von FXMacroData gibt jede bevorstehende geplante Veröffentlichung mit ihrem Indikatornamen und dem erwarteten Ankündigungsdatum zurück, wodurch es einfach ist, einen Blackout-Scheduler zu erstellen:

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

Schritt 6: Erstellen Sie ein Signal

Mit Makroscoring, Kontakt zu Spot-Rate und einem Kalender-Blackout-Check können Sie sie in eine einzige Signalfunktion zusammenfügen, die nach Bedarf eine Richtung, Vertrauen und empfohlene Positionsgröße erzeugt.

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

Schritt 7: Risikokontrollen und Größenpositionen anwenden

Die Signalgenerierung ist nur die Hälfte der Arbeit. Ohne systematische Risikokontrollen wird selbst eine hochwertige Signalquelle Drawdowns erzeugen, die über das hinausgehen, was die Strategie ertragen kann. Mindestens drei Kontrollen sind essentiell: eine maximale Positionsgröße im Verhältnis zum Kontokapital, ein harter Stop-Loss in Pips und ein tägliches Verlustlimit, das den Handel nach einer Reihe von Verlustsitzungen unterbricht.

@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}")

Schritt 8: Alles zusammenstellen Die tägliche Strategie-Schleife

Der letzte Schritt ist die Zusammenstellung von allen Daten in eine Ausführungsschleife, die einmal täglich ausgeführt wird, alle Makrodaten aktualisiert, das Signal auswertet, den Release-Kalender überprüft und eine Bestellempfehlung ausgibt.

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

Nächste Schritte: Erweiterung der Strategie

Das obige Framework ist absichtlich schlank, so dass Sie jede Entscheidung von Rohdaten bis zur Endleistung verfolgen können.

• Mehr Währungen hinzufügen die Angabe der Wechselkurse für GBP, AUD, JPY oder CAD mit den gleichen Indikatorenendpunkten erweitern. Kurs der GBP Und ... AUD-Inflation Die Daten sind in der Regel in einer Reihe zusammengefasst.
• Hinzufügen von COT-Positionsdaten Die Position von Großspekulanten aus dem CFTC-COT-Bericht ist ein nützlicher Stimmungsfilter. Wenn das Makro-Regime lange USD sagt, aber spekulative Lange bereits extrem sind, ist das Risiko/Rendite eines neuen Eintrags geringer. FXMacroData liefert COT-Daten über die gleiche API.
• Backtest gegenüber historischen Ankündigungsdaten weil jede FXMacroData-Beobachtung eine announcement_datetime Sie können genau rekonstruieren, was der Markt zu einem bestimmten Zeitpunkt wusste und Strategie-Einträge simulieren, ohne Aussicht zu haben.
• Automatisieren mit einem Zeitplaner Verpackung run_daily_strategy() Typische Makrosignale müssen nur nach wichtigen Datenveröffentlichungen aktualisiert werden, so dass tägliche oder sogar wöchentliche Aktualisierungen für mittelfristige Positionen ausreichen.
• Verbindung mit einer Broker-API die Signal- und Losgrößen-Ausgänge sind makler-agnostisch. direction Und ... lots Sie können auch die Bestellungen in Ihrer bevorzugten Ausführungsschicht (OANDA v20, Interactive Brokers TWS oder ein Papierhandelssimulator) vermarkten.

Zusammenfassung

Makro-Signale sind keine Dekoration für ein Algo sie sind die Kante.

1. Abrufen von Leitzinsen, Inflation, Beschäftigung und Anleihenrenditen aus der FXMacroData API mit einem konsistenten Muster
2. Berechnung einer zusammengesetzten Regelungspunktzahl für jede Währung unter Verwendung der Z-Score-Normalisierung und der Mischung gewichteter Indikatoren
3. Hinzufügen eines Preistrendfilters aus der FX-Spotkurshistorie, um vor dem Eintritt Makro- und technische Vereinbarung zu verlangen
4. Verwenden Sie den Release-Kalender, um das Rauschen von Datenfenstern mit hoher Wirkung mit einem einfachen Blackout-Planer zu vermeiden
5. Größe der Positionen proportional zum Signalvertrauen und Kontokapital unter Verwendung eines Modells mit festem, fractionellem Risiko
6. Alle Komponenten in einer einzigen wiederholbaren täglichen Schleife zusammenstellen

Die Datenschicht die konsequent strukturierte, zeitgestempelte Ankündigungsreihe von FXMacroData leistet die schwere Arbeit, jedes Signal auf überprüfbare Marktfakten zu stützen.