Trading algorithmique sur Kraken avec les signaux macro EUR/USD de FXMacroData

Pourquoi la macro-divergence de l'EUR/USD entraîne la crypto sur Kraken

Kraken occupe une position distincte dans le paysage des échanges de crypto: il est largement utilisé par les traders européens, accepte les dépôts directs en euros et cite de nombreuses paires par rapport à l'EUR ainsi qu'au USD. Cela signifie que les forces macro qui régissent l'euro/USD divergence de la politique de la BCE par rapport au FED, les différentiels d'inflation de la zone euro par rapport aux États-Unis et les écarts de rendement réels créent des vents de fond et des ventes en arrière directifs qui se manifestent clairement dans les paires cotées sur Kraken comme XBT/USD et XB T/EUR.

Lorsque la Fed resserre tandis que la BCE détient, le dollar se renforce, les rendements réels augmentent et les actifs à risque (y compris les crypto) sont généralement soumis à une pression de vente. Lorsque la BCE et la Fed sont désynchronisées dans la direction opposée, par exemple, la force de l'EUR et les rendement réels américains comprimés ont tendance à soutenir le positionnement du risque en BTC. Ces régimes se déroulent sur des semaines et des mois et sont entièrement observables à l'avance grâce aux indicateurs d'indicateur de FXMacroData.

Ce guide vous guide dans la construction d'un robot de trading algorithmique basé sur des signaux macro pour XBT/USD sur Kraken.

• Extrait des signaux macro de l'EUR et du USD à partir de FXMacroData (taux directeur, IPC, inflation de base et spot EUR/USD)
• Calcule un score de divergence EUR/USD pour classer le régime macro
• Planifie les fenêtres d'exécution autour des dates de publication de la BCE et de la Fed via le calendrier de publication FXMacroData
• Envoie des ordres de marché et de limite sur Kraken via l'API REST officielle
• Applique des mesures simples de taille des positions et de contrôle des risques

Pré-requis

Avant de commencer, assurez- vous d'avoir les choses suivantes prêtes:

• 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
• Le compte Kraken avec un solde en USD ou EUR générer des clés API dans le tableau de bord Kraken sous Sécurité → API avec Créer et modifier des commandes Je suis désolé . Fonds de requête les autorisations
• Paquets PythonJe suis désolé . requestsJe suis désolé . krakenexJe suis désolé . pandasJe suis désolé . schedule

pip install requests krakenex pandas schedule

Stockez toutes les informations d'identification comme variables d'environnement jamais des clés de code dur:

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

Étape 1: Récupérer les signaux macro EUR et USD

Le modèle de divergence utilise quatre séries: Taux de référence de la BCE- Je suis désolé . Taux de la FedJe suis désolé . Indice des prix à la consommation de la zone euro, et Indice de consommation des États-UnisEnsemble, ils décrivent la position de chaque banque centrale dans son cycle de hausse ou d'assouplissement et si les différentiels d'inflation favorisent la force de l'EUR ou du USD.

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

Le announcement_datetime Le champ contient l'horodatage de sortie de deuxième niveau que vous utiliserez à l'étape 4 pour mettre en pause les transactions autour d'événements à fort impact et reprendre immédiatement après la fermeture de la fenêtre de sorties.

Étape 2: calculer le score de la macro-divergence EUR/USD

Au lieu de négocier sur un seul indicateur, le score de divergence synthétise les quatre séries en un nombre directionnel compris entre -1 (régime USD fort risque-off pour BTC) et +1 (régimen USD faible / risque-on favorable pour BTC).

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

Étape 3: Connectez-vous à Kraken

Le krakenex La bibliothèque enveloppe l'API REST de Kraken avec une interface simple query_public Je suis désolé . query_private Les terminaux privés (commandes, solde de requête) nécessitent votre paire de clés API.

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

Étape 4: Calendrier des communiqués de la BCE et de la Fed

Les macro-déclarations à fort impact les décisions de la BCE sur les taux, les déclarations du FOMC, les impressions de l'IPC de la zone euro injectent généralement une forte volatilité dans les marchés des changes et des crypto-monnaies.

FXMacroData est calendrier de sortie Il apparaît la date et l'heure de l'annonce prévue pour chaque paire de devises/indicateurs, ce qui facilite la création d'un calendrier de black-out:

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

Étape 5: Placer des ordres macro-alimentés sur Kraken

Avec le score macro et la vérification de l'éteinte, la logique d'ordre est simple. Lorsque le score dépasse le seuil long et qu'il n'y a pas de position ouverte, soumettez un achat à limite. Lorsque la note est négative ou que la position a atteint un objectif de profit, soumettre une vente à limite . La taille de la position est exprimée en fraction du solde disponible en USD, plafonnée à un maximum dur.

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

Étape 6: assembler la boucle de bot complète

La dernière étape est de connecter tout dans une boucle programmée qui s'exécute une fois par heure. À chaque clic, il rafraîchit les données macro, vérifie les fenêtres de black-out, recalcule le score de divergence et exécute le signal. schedule La bibliothèque garde cette légèreté sans nécessiter une file d'attente complète.

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

Étape 7: Ajouter des gardiens de stop-loss et de take-profit

Le score du régime vous indique quand ouvrir et quand sortir sur un pivot macro, mais cela peut prendre des jours.

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

Quand ? open_long Si le test est un succès, enregistrer le prix de remplissage dans bot_state.jsonÀ chaque tique, appelle. check_price_guards avant l'évaluation du score macro si elle est retournée True, sautez le reste du tick puisque la position a déjà été fermée à un niveau de prix.

Élargissement de la stratégie

Une fois que le bot de base fonctionne de manière fiable, plusieurs extensions méritent d'être considérées:

• Ajouter l'inflation de base en EUR le Indice de consommation par habitant de base en euros Je suis désolé . PCE en USD Les prix des produits et services sont généralement plus élevés que les prix des services.
• Opérations de marge Kraken Kraken prend en charge un effet de levier jusqu'à 5x sur XBT/USD avec marge; ajouter le "leverage": "2:1" Paramètre à AddOrder les appels à amplifier les signaux de mode positif (uniquement appropriés avec un stop-loss proportionnellement plus serré).
• Rotation à plusieurs paires répéter la même logique pour l'ETH/USD (XETHZUSD), en utilisant le même score macro; faire pivoter le capital dans la paire qui affiche une dynamique plus forte lorsque les deux sont en régime de risque.
• Surcharge de change intraday extraire les données au comptant de l'EUR/USD à partir des Indice pondéré par les échanges en euros le point final et utiliser la dynamique intradienne comme filtre d'entrée à court terme dans un régime macro positif.
• Carnet de commandes WebSocket remplacer la boucle de vote programmée par le flux WebSocket de Kraken pour les mises à jour de prix en temps réel, réduisant la latence de minutes à millisecondes pour le raffinement d'entrée et de sortie.

Résumé et prochaines étapes

Vous avez maintenant un robot de trading Python complet qui traduit la divergence macro EUR/USD en signaux longs/plat actionnables sur Kraken.

1. L'indicateur FXMacroData est en train de se déclencher. taux de la BCE et de la Fed, IPC EUR et USD, spot EUR/USD, le tout via le /announcements/ Je suis désolé . /forex/ points de terminaison avec horodatage de deuxième niveau
2. Score de divergence un composite pondéré qui mape plusieurs indicateurs sur un seul signal directionnel
3. Planificateur de coupure met en pause l' activité des ordres autour des périodes de déblocage de la BCE et de la Fed en utilisant le Calendrier de sortie de FXMacroData
4. Gestion des commandes Kraken limiter les achats et les ventes avec des signaux de poste uniquement afin de minimiser les frais
5. Gardiens des prix niveaux de stop-loss et de take-profit protégeant la position entre les axes du régime macro

Comme étape naturelle, explorez PCE en USD Je suis désolé . Indice de consommation par habitant de base en euros pour affiner les composantes inflationnistes du score, ou étendre le bot pour négocier la paire EUR/USD directement sur Kraken en utilisant les mêmes signaux de régime avec une position FX au comptant.

La référence complète de l'API et toutes les combinaisons de devises/indicateurs disponibles sont à /pi-référencePour obtenir votre clé API, visitez / souscrivezJe suis désolé .