Algotrabalho no Kraken com FXMacroData EUR/USD Macro Signals

Por que a macro divergência do EUR/USD impulsiona a criptomoeda no Kraken

O Kraken ocupa uma posição distinta no cenário de câmbio de criptomoedas: é muito utilizado por traders europeus, aceita depósitos diretos em EUR e cita muitos pares contra EUR, bem como USD. Isso significa que as forças macro que governam o EUR/USD divergência da política do BCE versus Fed, diferenças de inflação da Zona Euro versus EUA, e os spreads de rendimento real criam ventos favoráveis e contrários direcionais que se mostram claramente em pares listados no Kraken como XBT/USD e XB T/EUR.

Quando o Fed aperta enquanto o BCE mantém, o dólar se fortalece, os rendimentos reais aumentam e os ativos de risco (incluindo criptomoedas) geralmente enfrentam pressão de venda. Quando o BCE e o Fed estão fora de sincronia na direção oposta O aumento do BCE em uma pausa do Fed, por exemplo A força do EUR e os rendimento reais comprimidos dos EUA tendem a apoiar o posicionamento de risco em BTC. Estes regimes ocorrem ao longo de semanas e meses e são totalmente observáveis com antecedência através dos endpoints do indicador FXMacroData.

Este guia mostra como criar um bot de negociação algorítmica baseado em sinais macro para XBT/USD no Kraken.

• Extrair sinais macro do EUR e do USD do FXMacroData (taxa de juro, IPC, inflação básica e EUR/USD spot)
• Calcula uma pontuação de divergência EUR/USD para classificar o regime macro
• Agendar as janelas de execução em torno das datas de publicação do BCE e da Fed através do calendário de publicações FXMacroData
• Envia ordens de mercado e limite no Kraken através da API REST oficial
• Aplica medidas simples de dimensão das posições e controlo do risco

Requisitos prévios

Antes de começar, certifique-se de ter pronto o seguinte:

• Python 3.9+ todos os trechos usam anotações de tipo padrão
• Chave da API do FXMacroData Inscreva-se em / subscrever e copiar a sua chave do painel da conta
• Conta Kraken com um saldo de USD ou EUR gerar chaves API no painel de Kraken sob Segurança → API Com Criar e modificar ordens E ... Fundos de consulta permissões
• Pacotes Python- Não . requests- Não . krakenex- Não . pandas- Não . schedule

pip install requests krakenex pandas schedule

Armazenar todas as credenciais como variáveis de ambiente nunca chaves de código rígido:

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

Passo 1: Obter os sinais macro EUR e USD

O modelo de divergência utiliza quatro séries: Taxa de juro do BCE- O quê ? Taxa de política monetária- Não . Índice de preços de consumo da zona do euroE ... IPC dos EUAJuntos, descrevem a posição de cada banco central no seu ciclo de aceleração ou flexibilização e se os diferenciais de inflação favorecem a força do EUR ou do 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']}")

O ... announcement_datetime campo carrega o carimbo de tempo de lançamento de segundo nível você usará no Passo 4 para pausar a negociação em torno de eventos de alto impacto e retomar imediatamente após o fechamento da janela de lancamento.

Passo 2: Calcular o escore de macro-divergência do EUR/USD

Em vez de negociar com um único indicador, a pontuação de divergência sintetiza todas as quatro séries em um número direcional entre -1 (regime forte de USD risco para BTC) e +1 (regimo fraco de USD / risco para o 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")

Passo 3: Conecte-se ao Kraken

O ... krakenex A biblioteca envolve a API REST do Kraken com uma simples query_public E ... query_private Os endpoints privados (colocação de ordens, balanço de consulta) exigem o seu par de chaves 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}")

Step 4: Schedule Around ECB and Fed Releases

As macros com alto impacto decisões de taxa do BCE, declarações do FOMC, impressões do IPC da zona do euro normalmente injetam uma forte volatilidade nos mercados de câmbio e criptomoedas. A abordagem mais segura é interromper qualquer atividade de ordem aberta na janela de 30 minutos em torno de cada lançamento e reavaliar a pontuação macro imediatamente após, para que você negocie no novo regime em vez do ruído pré-lançamento.

O FXMacroData's Calendário de lançamento A partir daí, a data e hora de anúncio programada para cada par de moedas/indicadores são apresentadas, facilitando a criação de um cronograma de apagão:

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

Passo 5: Coloque ordens macro-dirigidas no Kraken

Com a pontuação macro e a verificação de apagão no local, a lógica da ordem é direta. Quando a pontuar cruza o limiar longo e não há posição aberta, envie uma compra de limite. Quando o escore inverter negativo ou a posição atingir uma meta de lucro, envie um limite de venda. O tamanho da posição é expresso como uma fração do saldo disponível em dólares, limitado a um 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'})")

Passo 6: Montar o Loop Bot Completo

O último passo conecta tudo em um loop programado que é executado uma vez por hora. schedule biblioteca mantém este leve sem exigir uma fila de tarefas 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

Passo 7: Adicionar guardas de stop-loss e take-profit

A pontuação do regime diz-lhe quando abrir e quando sair em um pivô macro, mas isso pode levar dias.

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

Quando ? open_long O preço de enchimento é registado no bot_state.jsonA cada tique, chama. check_price_guards antes da avaliação do resultado macro se retornar True, pular o resto do tick uma vez que a posição já foi fechada a um nível de preço.

Extensão da Estratégia

Uma vez que o bot principal esteja funcionando de forma confiável, várias extensões valem a pena considerar:

• Adicionar a inflação básica do EUR o IPC de base do EUR E ... PCE em USD A análise dos preços dos produtos e serviços de base revela a pressão subjacente aos preços; a sua inserção no índice junto do índice principal dos preços do produto melhora a classificação dos regimes em pontos de virada.
• Negociação de margem Kraken Kraken suporta até 5x alavancagem no XBT/USD com margem; adicione o "leverage": "2:1" Parâmetro para AddOrder As chamadas para amplificar sinais de regime positivo (apenas adequadas com um stop-loss proporcionalmente mais restrito).
• Rotação de vários pares repetição da mesma lógica de regime para o ETH/USD (XETHZUSD), utilizando a mesma pontuação macro; rotar o capital no par que mostre maior dinâmica quando ambos estiverem em regime de risco.
• Superposição de divisas intradiárias extrair dados spot do EUR/USD do Índice ponderado pelo comércio do EUR ponto final e utilizar o momento intradiário como filtro de entrada a curto prazo num regime macro positivo.
• Livro de pedidos WebSocket substituir o ciclo de votação programado pelo feed WebSocket do Kraken para atualizações de preços em tempo real, reduzindo a latência de minutos para milissegundos para refinamento de entrada e saída.

Resumo e passos seguintes

Agora você tem um bot de negociação Python completo que traduz a macro divergência EUR/USD em sinais longos/flat acionáveis no Kraken.

1. Indicador FXMacroData puxa Taxas do BCE e da Fed, IPC EUR e USD, EUR/USD spot, tudo através do /announcements/ E ... /forex/ pontos finais com marcas de tempo de segundo nível
2. Pontuação de divergência um composto ponderado que mapeia múltiplos indicadores num único sinal direcional
3. Planejador de apagão pausa a actividade de ordens em torno das janelas de liberação do BCE e da Fed utilizando o Calendário de lançamento do FXMacroData
4. Gestão de pedidos Kraken limitar as compras e vendas com apenas sinalizações de correio para minimizar as taxas
5. Guardas de preços níveis de stop-loss e take-profit que protejam a posição entre os pivots do regime macro

Como passo natural, explore. PCE em USD E ... IPC de base do EUR para afiar os componentes de inflação da pontuação, ou estender o bot para negociar o par EUR/USD diretamente no Kraken usando os mesmos sinais de regime com uma posição FX spot.

A referência completa da API e todas as combinações de moedas/indicadores disponíveis estão disponíveis em /api-referênciaPara obter a sua chave API, visite / subscrever- Não .