Algo-Trading on Kraken with FXMacroData EUR/USD Macro Signals

Why EUR/USD Macro Divergence Drives Crypto on Kraken

Kraken occupies a distinct position in the crypto exchange landscape: it is heavily used by European traders, accepts direct EUR deposits, and quotes many pairs against EUR as well as USD. That means the macro forces that govern EUR/USD — ECB versus Fed policy divergence, Eurozone versus US inflation differentials, and real yield spreads — create directional tailwinds and headwinds that show up clearly in Kraken-listed pairs like XBT/USD and XBT/EUR.

When the Fed tightens while the ECB holds, the dollar strengthens, real yields rise, and risk assets (crypto included) typically face selling pressure. When the ECB and Fed are out of sync in the opposite direction — ECB hiking into a Fed pause, for example — EUR strength and compressed US real yields tend to support risk-on positioning in BTC. These regimes play out over weeks and months, and they are fully observable in advance through FXMacroData's indicator endpoints.

This guide walks through building a macro-signal-driven algorithmic trading bot for XBT/USD on Kraken. By the end you will have a working Python strategy that:

• Pulls EUR and USD macro signals from FXMacroData (policy rate, CPI, core inflation, and EUR/USD spot)
• Computes a EUR/USD divergence score to classify the macro regime
• Schedules execution windows around ECB and Fed release dates via the FXMacroData release calendar
• Submits market and limit orders on Kraken via the official REST API
• Applies simple position-sizing and risk controls

Prerequisites

Before starting, make sure you have the following ready:

• Python 3.9+ — all snippets use standard type annotations
• FXMacroData API key — sign up at /subscribe and copy your key from the account dashboard
• Kraken account with a funded USD or EUR balance — generate API keys in the Kraken dashboard under Security → API with Create & Modify Orders and Query Funds permissions
• Python packages: requests, krakenex, pandas, schedule

pip install requests krakenex pandas schedule

Store all credentials as environment variables — never hard-code keys:

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

Step 1: Fetch EUR and USD Macro Signals

The divergence model uses four series: the ECB policy rate, the Fed policy rate, Eurozone CPI, and US CPI. Together they describe where each central bank stands in its hiking or easing cycle and whether inflation differentials favour EUR or USD strength.

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

The announcement_datetime field carries the second-level release timestamp — you will use it in Step 4 to pause trading around high-impact events and resume immediately after the release window closes.

Step 2: Compute the EUR/USD Macro Divergence Score

Rather than trading off a single indicator, the divergence score synthesises all four series into a directional number between -1 (strong USD regime — risk-off for BTC) and +1 (weak USD / risk-on regime — supportive for BTC). Positive scores suggest the conditions for a long; negative scores suggest staying flat or short.

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

Step 3: Connect to Kraken

The krakenex library wraps Kraken's REST API with straightforward query_public and query_private calls. Private endpoints (placing orders, querying balance) require your API key pair.

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

Step 4: Schedule Around ECB and Fed Releases

High-impact macro releases — ECB rate decisions, FOMC statements, Eurozone CPI prints — typically inject sharp volatility into both FX and crypto markets. The safest approach is to halt any open order activity in the 30-minute window around each release and re-evaluate the macro score immediately after, so you trade on the new regime rather than the pre-release noise.

FXMacroData's release calendar surfaces the next scheduled announcement datetime for each currency/indicator pair, making it straightforward to build a blackout scheduler:

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

Step 5: Place Macro-Driven Orders on Kraken

With the macro score and the blackout check in place, the order logic is straightforward. When the score crosses above the long threshold and there is no open position, submit a limit buy. When the score flips negative or the position has hit a profit target, submit a limit sell. Position size is expressed as a fraction of available USD balance, capped at a hard 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'})")

Step 6: Assemble the Full Bot Loop

The final step wires everything into a scheduled loop that runs once per hour. At each tick it refreshes macro data, checks for blackout windows, recomputes the divergence score, and executes the signal. The schedule library keeps this lightweight without requiring a full task queue.

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

Step 7: Add Stop-Loss and Take-Profit Guards

The regime score tells you when to open and when to exit on a macro pivot, but that can take days. In the interim, you need price-level guards so a sharp liquidation move does not wipe the position before the exit signal fires.

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

When open_long succeeds, record the fill price into bot_state.json. At each tick, call check_price_guards before the macro score evaluation — if it returns True, skip the rest of the tick since the position has already been closed at a price level.

Extending the Strategy

Once the core bot is running reliably, several extensions are worth considering:

• Add EUR core inflation — the EUR core CPI and USD PCE reveal underlying price pressure; plugging these into the score alongside headline CPI improves regime classification at turning points.
• Kraken margin trading — Kraken supports up to 5× leverage on XBT/USD with margin; add the "leverage": "2:1" parameter to AddOrder calls to amplify positive-regime signals (only appropriate with a proportionally tighter stop-loss).
• Multi-pair rotation — repeat the same regime logic for ETH/USD (XETHZUSD), using the same macro score; rotate capital into whichever pair shows stronger momentum when both are in a risk-on regime.
• Intraday FX overlay — pull EUR/USD spot data from the EUR trade-weighted index endpoint and use intraday momentum as a short-term entry filter within a positive macro regime.
• WebSocket order book — replace the scheduled polling loop with Kraken's WebSocket feed for real-time price updates, reducing latency from minutes to milliseconds for entry and exit refinement.

Summary and Next Steps

You now have a complete Python trading bot that translates EUR/USD macro divergence into actionable long/flat signals on Kraken. The key components are:

1. FXMacroData indicator pulls — ECB and Fed rates, EUR and USD CPI, EUR/USD spot, all via the /announcements/ and /forex/ endpoints with second-level timestamps
2. Divergence score — a weighted composite that maps multiple indicators onto a single directional signal
3. Blackout scheduler — pauses order activity around ECB and Fed release windows using the FXMacroData release calendar
4. Kraken order management — limit buys and sells with post-only flags to minimise fees
5. Price guards — stop-loss and take-profit levels that protect the position between macro regime pivots

As a natural next step, explore USD PCE and EUR core CPI to sharpen the inflation components of the score, or extend the bot to trade the EUR/USD pair directly on Kraken using the same regime signals with a spot FX position.

Full API reference and all available currency/indicator combinations are at /api-reference. To get your API key, visit /subscribe.