Algo-Trading auf Kraken mit FXMacroData EUR/USD Makrosignalen

Warum die Makro-Divergenz zwischen EUR/USD Krypto auf Kraken treibt

Kraken nimmt eine ausgeprägte Position in der Krypto-Börsenlandschaft ein: Es wird stark von europäischen Händlern verwendet, akzeptiert direkte EUR-Einzahlungen und notiert viele Paare gegenüber EUR sowie USD. Das bedeutet, dass die Makro-Kräfte, die EUR/USD EZB gegenüber Fed-Politikdivergenz, Eurozone gegenüber US-Inflationdifferenzen und reale Rendite-Spreads regeln richtungsweisende Rückenwinde und Gegenwinde erzeugen, die sich in den von Kraken notierten Paaren wie XBT/USD und XB T/EUR deutlich zeigen.

Wenn die Fed sich verschärft, während die EZB hält, stärkt sich der Dollar, die realen Renditen steigen und Risikovermögenswerte (einschließlich Krypto) stehen in der Regel unter Verkaufsdruck. Wenn die EZV und die Fed in entgegengesetzte Richtung nicht synchron sind ECB in eine Fed-Pause hinein, zum Beispiel EUR-Stärke und komprimierte US-Realrenditen neigen dazu, die Risikopositionierung in BTC zu unterstützen. Diese Regime finden über Wochen und Monate hinweg statt und sind im Voraus durch die Indikatorenendpunkte von FXMacroData vollständig beobachtbar.

Dieser Leitfaden führt Sie durch den Aufbau eines makrosignaldriven algorithmischen Handelsbots für XBT/USD auf Kraken. Am Ende haben Sie eine funktionierende Python-Strategie, die:

• Zieht Makro-Signale für EUR und USD aus FXMacroData (Policy-Rate, CPI, Kerninflation und EUR/USD-Spot)
• Berechnet einen EUR/USD-Divergenzwert zur Einstufung des Makroreserves
• Zeitplanung der Ausführungszeiten um die Veröffentlichungsdaten der EZB und der Fed über den Veröffentlichtungskalender FXMacroData
• Übermittelt Markt- und Limit-Orders auf Kraken über die offizielle REST-API
• Anwendung einfacher Positionsgrößen und Risikokontrollen

Voraussetzungen

Bevor Sie anfangen, stellen Sie sicher, daß Sie Folgendes bereit haben:

• 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
• Kraken-Konto mit einem fündig gestellten USD- oder EUR-Guthaben API-Schlüssel im Kraken-Dashboard unter Sicherheit → API Mit Erstellen und Ändern von Bestellungen Und ... Abfragefonds Genehmigungen
• Python-Pakete- Ich weiß . requests- Ich weiß . krakenex- Ich weiß . pandas- Ich weiß . schedule

pip install requests krakenex pandas schedule

Alle Anmeldeinformationen als Umgebungsvariablen speichern niemals Hardcode-Schlüssel:

export FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY"
export KRAKEN_API_KEY="YOUR_KRAKEN_API_KEY"
export KRAKEN_PRIVATE_KEY="YOUR_KRAKEN_PRIVATE_KEY"

Schritt 1: Abrufen von Makrosignalen für EUR und USD

Das Divergenzmodell verwendet vier Reihen: Leitzins der EZB- Das ist ... Zinssatz der Fed- Ich weiß . CPI der EurozoneUnd ... US-amerikanischer KPIZusammen beschreiben sie, wo sich jede Zentralbank in ihrem Zyklus der Anhebung oder Lockerung befindet und ob die Inflationsdifferenzen die Stärke des EUR oder des USD begünstigen.

import os
import requests

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


def get_series(path: str, start: str = "2023-01-01") -> list[dict]:
    """Fetch a time-series from FXMacroData."""
    resp = requests.get(
        f"{BASE_URL}{path}",
        params={"api_key": FXMACRO_KEY, "start": start},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["data"]


# Macro inputs
ecb_rate   = get_series("/announcements/eur/policy_rate")
fed_rate   = get_series("/announcements/usd/policy_rate")
eur_cpi    = get_series("/announcements/eur/inflation")
usd_cpi    = get_series("/announcements/usd/inflation")

# Forex confirmation: EUR/USD spot trend
eurusd     = get_series("/forex/eur/usd", start="2024-01-01")

# Each item: {"date": "2025-04-08", "val": 4.0, "announcement_datetime": "2025-04-08T12:15:00Z"}
print(f"ECB rate:  {ecb_rate[0]['val']}%")
print(f"Fed rate:  {fed_rate[0]['val']}%")
print(f"EUR CPI:   {eur_cpi[0]['val']}%")
print(f"USD CPI:   {usd_cpi[0]['val']}%")
print(f"EUR/USD:   {eurusd[0]['val']}")

Die ... announcement_datetime Das Feld enthält den Zeitstempel für die Freigabe der zweiten Ebene Sie verwenden ihn in Schritt 4 um den Handel bei hochrangigen Ereignissen anzuhalten und sofort nach Schließen des Freigabefensters wieder aufzunehmen.

Schritt 2: Berechnung des Makro-Divergenz-Score des EUR/USD

Anstatt einen einzigen Indikator zu handeln, synthetisiert der Divergenz-Score alle vier Reihen in eine Richtungszahl zwischen -1 (starkes USD-Regime Risiko-Off für BTC) und +1 (schwaches USD / Risiko-On-Regeme unterstützend für BTC). Positive Scores deuten auf die Bedingungen für eine lange; negative Scores legen nahe, flach oder kurz zu bleiben.

def divergence_score(
    ecb_rate_pct: float,
    fed_rate_pct: float,
    eur_cpi_pct: float,
    usd_cpi_pct: float,
    eurusd_rate: float,
    *,
    eurusd_neutral: float = 1.08,
) -> float:
    """
    Returns a composite EUR/USD macro divergence score in [-1, +1].

    Positive  → USD relatively weak, risk-on environment, BTC bullish.
    Negative  → USD relatively strong, risk-off environment, BTC bearish.
    """
    score = 0.0

    # Component 1: rate spread (ECB rate − Fed rate)
    # Positive spread → ECB tighter than Fed → EUR-supportive → +score
    rate_spread = ecb_rate_pct - fed_rate_pct
    score += max(-1.0, min(1.0, rate_spread / 2.5)) * 0.35

    # Component 2: inflation differential (EUR CPI − USD CPI)
    # Higher EUR inflation → ECB forced to stay hawkish → EUR-supportive
    infl_diff = eur_cpi_pct - usd_cpi_pct
    score += max(-1.0, min(1.0, infl_diff / 3.0)) * 0.25

    # Component 3: Fed hawkishness drag
    # High Fed rate in absolute terms → USD strength → negative for BTC
    fed_drag = (fed_rate_pct - 3.0) / 3.0   # neutral at 3 %
    score -= max(-1.0, min(1.0, fed_drag)) * 0.20

    # Component 4: EUR/USD spot vs neutral (1.08)
    # EUR/USD above neutral → dollar relatively weak → +score
    fx_signal = (eurusd_rate - eurusd_neutral) / 0.08
    score += max(-1.0, min(1.0, fx_signal)) * 0.20

    return max(-1.0, min(1.0, score))


score = divergence_score(
    ecb_rate_pct=ecb_rate[0]["val"],
    fed_rate_pct=fed_rate[0]["val"],
    eur_cpi_pct=eur_cpi[0]["val"],
    usd_cpi_pct=usd_cpi[0]["val"],
    eurusd_rate=eurusd[0]["val"],
)

print(f"Divergence score: {score:+.3f}")
if score >= 0.25:
    print("Regime: RISK-ON → consider long XBT/USD")
elif score <= -0.25:
    print("Regime: RISK-OFF → consider flat / short")
else:
    print("Regime: NEUTRAL → no directional edge")

Schritt 3: Verbindung mit Kraken

Die ... krakenex Die Bibliothek wickelt die REST-API von Kraken mit einfachen query_public Und ... query_private Private Endpunkte (Orders, Abfrage) benötigen Ihr API-Schlüsselpaar.

import krakenex

kraken = krakenex.API(
    key=os.environ["KRAKEN_API_KEY"],
    secret=os.environ["KRAKEN_PRIVATE_KEY"],
)


def get_balance() -> dict[str, float]:
    """Return current account balance as {asset: amount}."""
    result = kraken.query_private("Balance")
    if result.get("error"):
        raise RuntimeError(f"Kraken balance error: {result['error']}")
    return {k: float(v) for k, v in result["result"].items()}


def get_xbt_price() -> float:
    """Return latest BTC/USD mid-price from Kraken ticker."""
    result = kraken.query_public("Ticker", {"pair": "XBTUSD"})
    if result.get("error"):
        raise RuntimeError(f"Kraken ticker error: {result['error']}")
    ticker = result["result"]["XXBTZUSD"]
    bid = float(ticker["b"][0])
    ask = float(ticker["a"][0])
    return (bid + ask) / 2.0


balance = get_balance()
usd_balance = balance.get("ZUSD", 0.0)
xbt_balance = balance.get("XXBT", 0.0)
xbt_price   = get_xbt_price()

print(f"USD balance: ${usd_balance:,.2f}")
print(f"XBT balance: {xbt_balance:.6f} BTC")
print(f"XBT/USD price: ${xbt_price:,.2f}")

Schritt 4: Zeitplan für die Veröffentlichungen der EZB und der Fed

Die wichtigsten Merkmale der Markteinführung sind die Auswirkungen der Marktreform auf die Marktlage und die Auswirkungen auf die Marktreformen.

- Das ist FXMacroData. Veröffentlichungskalender Die Daten werden in einem anderen Format als dem der Datenbank angezeigt.

from datetime import datetime, timezone, timedelta


def get_next_releases(currencies: list[str], indicators: list[str]) -> list[datetime]:
    """
    Fetch upcoming announcement datetimes for the given
    currency/indicator combinations via FXMacroData.
    """
    upcoming: list[datetime] = []
    for currency in currencies:
        for indicator in indicators:
            try:
                # Fetch recent releases; announcement_datetime on future entries
                # will be greater than now, while past entries will be skipped below.
                data = get_series(f"/announcements/{currency}/{indicator}", start="2024-01-01")
                for item in data[:6]:   # scan the six most-recent entries for future datetimes
                    dt_str = item.get("announcement_datetime")
                    if dt_str:
                        dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
                        if dt > datetime.now(tz=timezone.utc):
                            upcoming.append(dt)
                            break
            except (requests.RequestException, KeyError, ValueError) as exc:
                log.warning("Skipping %s/%s in release calendar: %s", currency, indicator, exc)
    return sorted(upcoming)


def in_blackout_window(
    release_times: list[datetime],
    buffer_minutes: int = 30,
) -> bool:
    """Return True if we are within buffer_minutes of any scheduled release."""
    now = datetime.now(tz=timezone.utc)
    for rt in release_times:
        if abs((rt - now).total_seconds()) < buffer_minutes * 60:
            return True
    return False


# Watch ECB and Fed policy rates plus both CPI prints
WATCH_CURRENCIES  = ["eur", "usd"]
WATCH_INDICATORS  = ["policy_rate", "inflation", "core_inflation"]

release_times = get_next_releases(WATCH_CURRENCIES, WATCH_INDICATORS)
print("Upcoming release windows:")
for rt in release_times:
    print(f"  {rt.strftime('%Y-%m-%d %H:%M UTC')}")

if in_blackout_window(release_times):
    print("⚠  BLACKOUT — halting order activity")
else:
    print("✓  Clear to trade")

Schritt 5: Platzieren Sie Makro-gesteuerte Aufträge auf Kraken

Mit dem Makro-Score und der Blackout-Check ist die Orderlogik einfach. Wenn der Score über die lange Schwelle geht und es keine offene Position gibt, senden Sie einen Limit-Kauf ein. Wenn die Punktzahl negativ ist oder die Position ein Gewinnziel erreicht hat, senden sie einen Limitverkauf ein. Die Positionsgröße wird als Bruchteil des verfügbaren USD-Guthabens ausgedrückt, begrenzt auf ein hartes Maximum.

import math


# ── Risk parameters ──────────────────────────────────────────────────────────
LONG_THRESHOLD   = 0.25    # score above this → open long
FLAT_THRESHOLD   = -0.10   # score below this → close long / stay flat
MAX_POSITION_USD = 2_000   # absolute cap per trade
RISK_FRACTION    = 0.10    # 10 % of USD balance per signal
LIMIT_SLIP_BPS   = 5       # place limit 5 bps above mid to improve fill rate
MIN_XBT_POSITION = 0.0001  # Kraken minimum order size for XBT


def open_long(usd_amount: float, xbt_price: float) -> dict:
    """Submit a limit buy order for XBT/USD on Kraken."""
    volume = round(usd_amount / xbt_price, 5)
    limit_price = round(xbt_price * (1 + LIMIT_SLIP_BPS / 10_000), 2)

    result = kraken.query_private("AddOrder", {
        "pair":      "XBTUSD",
        "type":      "buy",
        "ordertype": "limit",
        "price":     str(limit_price),
        "volume":    str(volume),
        "oflags":    "post",          # post-only: never pays taker fee
    })
    if result.get("error"):
        raise RuntimeError(f"Kraken order error: {result['error']}")
    return result["result"]


def close_long(xbt_volume: float, xbt_price: float) -> dict:
    """Submit a limit sell order to close an existing long."""
    limit_price = round(xbt_price * (1 - LIMIT_SLIP_BPS / 10_000), 2)
    result = kraken.query_private("AddOrder", {
        "pair":      "XBTUSD",
        "type":      "sell",
        "ordertype": "limit",
        "price":     str(limit_price),
        "volume":    str(round(xbt_volume, 5)),
        "oflags":    "post",
    })
    if result.get("error"):
        raise RuntimeError(f"Kraken order error: {result['error']}")
    return result["result"]


def run_signal(score: float) -> None:
    """Execute the macro signal: open, hold, or close a XBT/USD position."""
    balance   = get_balance()
    usd_bal   = balance.get("ZUSD", 0.0)
    xbt_bal   = balance.get("XXBT", 0.0)
    xbt_price = get_xbt_price()

    has_position = xbt_bal >= MIN_XBT_POSITION  # treat dust as no position

    if score >= LONG_THRESHOLD and not has_position:
        usd_to_deploy = min(usd_bal * RISK_FRACTION, MAX_POSITION_USD)
        if usd_to_deploy < 10:
            print("Insufficient USD balance for trade.")
            return
        result = open_long(usd_to_deploy, xbt_price)
        print(f"Long opened — txid: {result.get('txid')}, "
              f"volume: {result.get('descr', {}).get('order')}")

    elif score <= FLAT_THRESHOLD and has_position:
        result = close_long(xbt_bal, xbt_price)
        print(f"Long closed — txid: {result.get('txid')}")

    else:
        print(f"Score {score:+.3f} — no action (position={'open' if has_position else 'flat'})")

Schritt 6: Zusammenstellen der vollständigen Bot-Schleife

Der letzte Schritt vernetzt alles in eine geplante Schleife, die einmal pro Stunde läuft. Bei jedem Ticke aktualisiert es Makrodaten, prüft nach Ausfallfenstern, berechnet den Divergenzwert neu und führt das Signal aus. schedule Die Bibliothek hält dieses Leichtgewicht ohne eine vollständige Aufgabenwarteschlange.

import schedule
import time
import logging

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


def bot_tick() -> None:
    """Single iteration of the macro bot."""
    log.info("── Macro bot tick ──────────────────────────────────────────")

    # 1. Fetch fresh macro data
    try:
        ecb   = get_series("/announcements/eur/policy_rate")[0]["val"]
        fed   = get_series("/announcements/usd/policy_rate")[0]["val"]
        ecpi  = get_series("/announcements/eur/inflation")[0]["val"]
        ucpi  = get_series("/announcements/usd/inflation")[0]["val"]
        fx    = get_series("/forex/eur/usd")[0]["val"]
    except Exception as exc:
        log.error("Failed to fetch macro data: %s", exc)
        return

    score = divergence_score(ecb, fed, ecpi, ucpi, fx)
    log.info("ECB %.2f%%  Fed %.2f%%  EUR CPI %.1f%%  USD CPI %.1f%%  EUR/USD %.4f",
             ecb, fed, ecpi, ucpi, fx)
    log.info("Divergence score: %+.3f", score)

    # 2. Check release calendar blackout
    try:
        releases = get_next_releases(WATCH_CURRENCIES, WATCH_INDICATORS)
    except Exception as exc:
        log.warning("Release calendar fetch failed: %s — proceeding without blackout", exc)
        releases = []

    if in_blackout_window(releases, buffer_minutes=30):
        log.warning("BLACKOUT window active — skipping order activity")
        return

    # 3. Execute signal
    try:
        run_signal(score)
    except Exception as exc:
        log.error("Order execution failed: %s", exc)


# Run immediately on start, then every hour
bot_tick()
schedule.every(1).hours.do(bot_tick)

log.info("Bot running — press Ctrl+C to stop")
while True:
    schedule.run_pending()
    time.sleep(30)  # poll every 30 s so scheduled hourly tasks fire on time

Schritt 7: Hinzufügen von Stop-Loss- und Take-Profit-Guardiers

Der Regime-Score sagt Ihnen, wann Sie auf einem Makro-Pivot öffnen und wann Sie aussteigen müssen, aber das kann Tage dauern.

STOP_LOSS_PCT   = 0.05    # exit if price drops 5 % below entry
TAKE_PROFIT_PCT = 0.12    # exit if price rises 12 % above entry

# Store entry price in a lightweight state file
import json, pathlib

STATE_FILE = pathlib.Path("bot_state.json")


def load_state() -> dict:
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text())
        except (json.JSONDecodeError, OSError):
            log.warning("bot_state.json is corrupt or unreadable — resetting state")
    return {"entry_price": None}


def save_state(state: dict) -> None:
    STATE_FILE.write_text(json.dumps(state))


def check_price_guards(xbt_price: float, xbt_bal: float) -> bool:
    """
    Returns True if a stop or take-profit was triggered (position closed).
    Call this before evaluating the macro score so price guards take priority.
    """
    state = load_state()
    entry = state.get("entry_price")
    if entry is None or xbt_bal < 0.0001:
        return False

    pnl_pct = (xbt_price - entry) / entry

    if pnl_pct <= -STOP_LOSS_PCT:
        log.warning("Stop-loss triggered at %.2f%% loss — closing position", pnl_pct * 100)
        close_long(xbt_bal, xbt_price)
        save_state({"entry_price": None})
        return True

    if pnl_pct >= TAKE_PROFIT_PCT:
        log.info("Take-profit triggered at %.2f%% gain — closing position", pnl_pct * 100)
        close_long(xbt_bal, xbt_price)
        save_state({"entry_price": None})
        return True

    return False

Wann ? open_long Erfolgreich, wird der Füllpreis in bot_state.jsonBei jedem Ticken rufen Sie an. check_price_guards Vor der Bewertung der Makroscore wenn sie zurückkehrt True, überspringen Sie den Rest des Ticks, da die Position bereits auf einem Preisniveau geschlossen wurde.

Erweiterung der Strategie

Sobald der Kernbot zuverlässig läuft, sind mehrere Erweiterungen in Betracht zu ziehen:

• Zusätzliche Kerninflation im EUR die EUR-Kern-CPI Und ... USD PCE Die Ergebnisse der Untersuchung zeigen, dass die Preise in den letzten Jahren im Vergleich zum Vorjahr im Vergleich zu den letzten Jahrzehnten im Vergleich zur letzten Jahreshälfte im Vergleich mit den letzten Monaten im Vorjahr um etwa 10% gestiegen sind.
• Handelsmargen von Kraken Kraken unterstützt eine Hebelwirkung von bis zu 5x auf XBT/USD mit Marge; fügen Sie die "leverage": "2:1" Parameter auf AddOrder Aufrufe zur Verstärkung von Positivregime-Signalien (nur bei einem verhältnismäßig engeren Stop-Loss geeignet).
• Mehrpaar-Rotation die gleiche Regelmethode für ETH/USD wiederholen (XETHZUSDDie Banken können die Risiken der Banken in einem anderen Mitgliedstaat als dem EU-Mitgliedstaat festlegen.
• Überlagerung von Tageswechselkurs die EUR/USD-Spotdaten aus dem Handelsgewichteter Index EUR Endpunkt und Nutzung der Intraday-Dynamik als kurzfristiger Einstiegsfilter innerhalb eines positiven Makrorahmens.
• WebSocket-Bestellbuch die geplante Umfrageschleife durch den WebSocket-Feed von Kraken für Echtzeit-Preisaktualisierungen ersetzen, wodurch die Latenzzeit für die Ein- und Ausstiegsveredelung von Minuten auf Millisekunden reduziert wird.

Zusammenfassung und weitere Schritte

Sie haben jetzt einen vollständigen Python-Handelsbot, der die EUR/USD Makro-Divergenz in umsetzbare Long/Flat-Signale auf Kraken übersetzt.

1. Der FXMacroData-Indikator zieht Zinssätze der EZB und der Fed, EUR und USD-PI, EUR/USD-Spot, alles über die /announcements/ Und ... /forex/ Endpunkte mit Zeitstempeln der zweiten Ebene
2. Divergenz-Score ein gewichtetes Komposit, das mehrere Indikatoren auf ein einziges Richtungssignal abbildet
3. Blackout-Planer Pausiert die Auftragsaktivität in den Zeitfenstern der Veröffentlichung durch die EZB und die Fed unter Verwendung der FXMacroData-Veröffentlichungskalender
4. Kraken-Auftragsverwaltung Beschränkung der Kauf- und Verkaufszahlen mit Post-only-Flaggen, um die Gebühren zu minimieren
5. Preisschützer Stop-Loss- und Take-Profit-Level, die die Position zwischen den Makrorechtspivoten schützen

Als natürlichen nächsten Schritt, erkunden Sie USD PCE Und ... EUR-Kern-CPI Die in Artikel 2 Absatz 1 Buchstabe b genannten Anwendungen werden in den folgenden Fällen angewendet:

Die vollständige API-Referenz und alle verfügbaren Währungs-/Indikatorkombinationen finden Sie unter /api-VerweisUm Ihren API-Schlüssel zu erhalten, besuchen Sie /abonnieren- Ich weiß .