Algo-Trading en Kraken con las Señales Macro EUR/USD de FXMacroData

Por qué la macrodivergencia del EUR/USD impulsa a las criptomonedas en Kraken

Kraken ocupa una posición distinta en el panorama de los cripto intercambios: es muy utilizado por los operadores europeos, acepta depósitos directos en euros y cotiza muchos pares frente al euro y al dólar estadounidense. Eso significa que las fuerzas macro que rigen el EUR/USD la divergencia de la política del BCE frente a la Fed, los diferenciales de inflación de la eurozona frente a los EE.UU., y los diferenciale de rendimiento real crean vientos de cola y contral que se muestran claramente en los pares listados en Kraken como XBT/USD y XB T/EUR.

Cuando la Fed se endurece mientras el BCE mantiene, el dólar se fortalece, los rendimientos reales aumentan y los activos de riesgo (incluidas las criptomonedas) generalmente enfrentan presión de venta. Cuando el BCE y la Fed están fuera de sincronización en la dirección opuesta El aumento del BCE en una pausa de la Fed, por ejemplo La fortaleza del EUR y los rendimentos reales comprimidos de los EE.UU. tienden a apoyar el posicionamiento de riesgo en BTC. Estos regímenes se desarrollan durante semanas y meses, y son totalmente observables de antemano a través de los puntos finales del indicador de FXMacroData.

Esta guía recorre la construcción de un bot de trading algorítmico basado en señales macro para XBT/USD en Kraken.

• Extrae las señales macro del EUR y del USD de FXMacroData (tipo de interés, IPC, inflación básica y spot del EUR/USD)
• Calcula una puntuación de la divergencia EUR/USD para clasificar el régimen macro
• Programar las ventanas de ejecución en torno a las fechas de publicación del BCE y de la Fed a través del calendario de publicación de FXMacroData
• Envía órdenes de mercado y límite en Kraken a través de la API oficial REST
• Aplica controles simples de tamaño de las posiciones y de riesgo

Los requisitos previos

Antes de comenzar, asegúrese de tener listos los siguientes:

• Python 3.9+ todos los fragmentos utilizan anotaciones de tipo estándar
• La clave de la API de FXMacroData inscribirse en / suscribirse y copia tu clave desde el panel de la cuenta
• Cuenta Kraken con un saldo en USD o EUR generar claves API en el panel de Kraken bajo Seguridad → API con Crear y modificar órdenes ¿ Qué ? Fondos de consulta las autorizaciones
• Paquetes de Python¿ Qué ? requests¿ Qué ? krakenex¿ Qué ? pandas¿ Qué ? schedule

pip install requests krakenex pandas schedule

Almacenar todas las credenciales como variables de entorno nunca claves de código duro:

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

Paso 1: Obtener las señales macro del EUR y del USD

El modelo de divergencia utiliza cuatro series: Tipo de interés de la BCE, el Tasa de política de la Fed¿ Qué ? Indicador de consumo de la zona euro, y Indicador de precios de los Estados UnidosEn conjunto, describen la posición de cada banco central en su ciclo de alza o de flexibilización y si los diferenciales de inflación favorecen la fortaleza del euro o del dólar.

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

El announcement_datetime El campo contiene la marca de tiempo de liberación de segundo nivel lo utilizará en el paso 4 para pausar el comercio alrededor de eventos de alto impacto y reanudar inmediatamente después de que se cierre la ventana de liberaciones.

Paso 2: Calcular el puntaje de la macrodivergencia del EUR/USD

En lugar de operar con un solo indicador, la puntuación de divergencia sintetiza las cuatro series en un número direccional entre -1 (regimen fuerte de USD riesgo para BTC) y +1 (regime débil de USD / riesgo de apoyo para 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")

Paso 3: Conecta con Kraken

El krakenex La biblioteca envuelve la API REST de Kraken con una sencilla query_public ¿ Qué ? query_private Los puntos finales privados (orden de colocación, balance de consulta) requieren su par de claves 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}")

Paso 4: Calendario de las emisiones del BCE y de la Fed

Las publicaciones macro de alto impacto las decisiones de tasas del BCE, las declaraciones del FOMC, las impresiones del IPC de la zona euro generalmente inyectan una fuerte volatilidad en los mercados de divisas y criptomonedas.

Es de FXMacroData. calendario de liberación aparece la próxima fecha y hora de anuncio programado para cada par de divisas/indicadores, lo que facilita la creación de un cronograma de apagones:

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

Paso 5: Poner órdenes macro-conducidas en Kraken

Con el puntaje macro y la verificación de apagón en su lugar, la lógica de orden es sencilla. Cuando el puntajes cruza por encima del umbral largo y no hay posición abierta, enviar un límite de compra. Cuando la puntuación vuelve negativo o la posición ha alcanzado un objetivo de ganancia, enviar una venta límite. El tamaño de la posición se expresa como una fracción del saldo disponible en dólares, limitado a un máximo duro.

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

Paso 6: Ensambla el bucle completo de bot

El paso final conecta todo a un bucle programado que se ejecuta una vez por hora. En cada tick actualiza los datos de macro, verifica las ventanas de apagón, recalcula la puntuación de divergencia y ejecuta la señal. schedule La biblioteca mantiene este peso ligero sin requerir una cola de tareas completa.

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

Paso 7: Añadir protectores de pérdida y ganancia

El puntaje del régimen le dice cuándo abrir y cuándo salir en un pivote macro, pero eso puede llevar días.

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

¿ Cuándo ? open_long Si tiene éxito, registre el precio de llenado en bot_state.jsonA cada tick, llama check_price_guards antes de la evaluación de la puntuación macro si se devuelve True, omita el resto del tick ya que la posición ya ha sido cerrada a un nivel de precio.

Ampliación de la estrategia

Una vez que el bot principal está funcionando de manera confiable, varias extensiones vale la pena considerar:

• Se añade la inflación básica del EUR el Indicador de precios de compra de base del EUR ¿ Qué ? PCE en dólares estadounidenses El índice de precios de los precios de las mercancías es el índice más alto de los mercados de la Unión Europea.
• Negociación de margen de Kraken Kraken admite hasta 5x de apalancamiento en XBT/USD con margen; añadir el "leverage": "2:1" Parámetro a AddOrder las llamadas para amplificar las señales de régimen positivo (sólo apropiadas con un stop-loss proporcionalmente más ajustado).
• Rotación de varios pares repita la misma lógica de régimen para ETH/USD (XETHZUSD), utilizando el mismo puntaje macro; girar el capital en el par que muestre mayor impulso cuando ambos estén en un régimen de riesgo.
• Superposición de divisas intradiarias extraer datos al contado del EUR/USD de la Indice ponderado por el comercio en euros punto final y utilizar el impulso intradiario como filtro de entrada a corto plazo dentro de un régimen macro positivo.
• Carpeta de pedidos de WebSocket reemplazar el ciclo de votación programado con el feed WebSocket de Kraken para actualizaciones de precios en tiempo real, reduciendo la latencia de minutos a milisegundos para el refinamiento de entrada y salida.

Resumen y pasos siguientes

Ahora tiene un bot de trading Python completo que traduce la macro divergencia EUR/USD en señales largas/planas procesables en Kraken.

1. El indicador FXMacroData se mueve Los tipos del BCE y de la Fed, el IPC del EUR y del USD, el spot del EUR/USD, todo a través del /announcements/ ¿ Qué ? /forex/ puntos finales con marcas de tiempo de segundo nivel
2. Puntuación de la divergencia un compuesto ponderado que asigna múltiples indicadores a una sola señal direccional
3. Programador de apagones pausa la actividad de las órdenes en el marco de las ventanas de liberación del BCE y de la Fed utilizando el Calendario de lanzamiento de FXMacroData
4. Gestión de pedidos de Kraken limitar las compras y ventas con las señales de correo únicamente para minimizar las tarifas
5. Guardianes de precios niveles de stop-loss y take-profit que protejan la posición entre los pivotes del régimen macro

Como siguiente paso natural, explorar. PCE en dólares estadounidenses ¿ Qué ? Indicador de precios de compra de base del EUR para agudizar los componentes de inflación de la puntuación, o extender el bot para operar el par EUR/USD directamente en Kraken utilizando las mismas señales de régimen con una posición FX al contado.

La referencia completa de la API y todas las combinaciones de divisas/indicadores disponibles se encuentran en /api-referenciaPara obtener su clave API, visite / suscribirse- ¿ Qué ?