Algoroading di Kraken dengan FXMacroData EUR/USD Makro Signals

Mengapa Divergensi Makro EUR/USD Mendorong Crypto di Kraken

Kraken menempati posisi yang berbeda dalam lanskap pertukaran kripto: banyak digunakan oleh pedagang Eropa, menerima deposito EUR langsung, dan mengutip banyak pasangan terhadap EUR serta USD. Itu berarti kekuatan makro yang mengatur divergensi kebijakan ECB terhadap ECB, zona euro terhadap US, dan penyebaran laba bersih real, menciptakan angin belakang dan angin kebalikan yang terlihat jelas dalam pasangan yang terdaftar di Kraken seperti XBT / USD dan XB T / EUR.

Ketika Fed memperketat sementara ECB memegang, dolar menguat, imbal hasil riil naik, dan aset berisiko (termasuk crypto) biasanya menghadapi tekanan penjualan. Ketika ECB dan Fed tidak sinkron ke arah yang berlawanan ECB naik ke jeda Fed, misalnya kekuatan EUR dan imbal balik riil AS yang terkompresi cenderung mendukung posisi risiko di BTC. Rezim ini berlangsung selama berminggu-minggu dan berbulan-bulan, dan mereka sepenuhnya dapat diamati sebelumnya melalui titik akhir indikator FXMacroData.

Panduan ini berjalan melalui membangun robot perdagangan algoritmik yang didorong oleh sinyal makro untuk XBT / USD di Kraken.

• Mendapatkan sinyal makro EUR dan USD dari FXMacroData (tingkat kebijakan, CPI, inflasi inti, dan spot EUR/USD)
• Menghitung skor divergensi EUR/USD untuk mengklasifikasikan sistem makro
• Jadwal jendela eksekusi di sekitar ECB dan Fed tanggal rilis melalui kalender rilis FXMacroData
• Mengirim pasar dan batas pesanan pada Kraken melalui resmi REST API
• Menggunakan ukuran posisi sederhana dan kontrol risiko

Persyaratan

Sebelum memulai, pastikan Anda siapkan:

• Python 3.9+ semua cuplikan menggunakan anotasi tipe standar
• Kunci API FXMacroData daftar di /langganan dan menyalin kunci Anda dari dasbor akun
• Akun Kraken dengan saldo USD atau EUR yang didanai menghasilkan kunci API di dasbor Kraken di bawah Keamanan → API dengan Buat & Ubah Pesenan Dan Dana Pertanyaan izin
• Paket PythonAku tidak tahu. requestsAku akan pergi. krakenexAku akan pergi. pandasAku akan pergi. schedule

pip install requests krakenex pandas schedule

Simpan semua kredensial sebagai variabel lingkungan tidak pernah kunci hard-code:

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

Langkah 1: Dapatkan Sinyal Makro EUR dan USD

Model divergensi menggunakan empat seri: Nilai tukar mata uang, yang Tingkat kebijakan FedAku akan pergi. Indeks harga konsumen zona euro, dan US CPIBersama-sama mereka menggambarkan di mana setiap bank sentral berada dalam siklus kenaikan atau pelonggaran dan apakah perbedaan inflasi mendukung kekuatan EUR atau 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']}")

- Apa? announcement_datetime bidang membawa timestamp rilis tingkat dua Anda akan menggunakannya di Langkah 4 untuk menghentikan perdagangan sekitar peristiwa berdampak tinggi dan melanjutkan segera setelah jendela rilis ditutup.

Langkah 2: Menghitung Skor Divergensi Makro EUR/USD

Daripada berdagang dari satu indikator, skor divergensi mensintesis semua empat seri menjadi angka arah antara -1 (rezim USD yang kuat risiko-off untuk BTC) dan +1 (rejim USD / risiko-on yang lemah mendukung untuk 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")

Langkah 3: Hubungkan ke Kraken

- Apa? krakenex library membungkus REST API Kraken dengan mudah query_public Dan query_private panggilan. private endpoint (memasukkan pesanan, saldo kueri) membutuhkan pasangan kunci API Anda.

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

Langkah 4: Jadwal Sekitar ECB dan Fed Rilis

Rilis makro berdampak tinggi keputusan suku bunga ECB, pernyataan FOMC, cetakan CPI zona euro biasanya menyuntikkan volatilitas tajam ke pasar FX dan crypto. Pendekatan paling aman adalah menghentikan aktivitas order terbuka dalam jendela 30 menit di sekitar setiap rilis dan mengevaluasi kembali skor makro segera setelahnya, sehingga Anda berdagang pada rezim baru daripada kebisingan pra-pembukaan.

FXMacroData's Kalender rilis muncul tanggal dan waktu pengumuman yang dijadwalkan berikutnya untuk setiap pasangan mata uang/indikator, sehingga mudah untuk membangun penjadwal pemadaman:

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

Langkah 5: Tempatkan pesanan yang dikendalikan makro di Kraken

Dengan skor makro dan pemeriksaan pemadaman di tempat, logika pesanan sederhana. Ketika skor melintasi ambang panjang dan tidak ada posisi terbuka, kirimkan pembelian batas. Ketika nilai berbalik negatif atau posisi telah mencapai target keuntungan, kirimkeun penjualan batas. Ukuran posisi dinyatakan sebagai sebagian dari saldo USD yang tersedia, dibatasi pada maksimum keras.

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

Langkah 6: Mengumpulkan Bot Full Loop

Langkah terakhir menghubungkan semuanya ke loop terjadwal yang berjalan sekali per jam. Pada setiap tanda itu memperbarui data makro, memeriksa jendela blackout, menghitung kembali skor divergensi, dan mengeksekusi sinyal. schedule library ini tetap ringan tanpa membutuhkan antrian tugas penuh.

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

Langkah 7: Tambahkan Stop-Loss dan Take-Profit Guards

skor regime memberitahu Anda kapan untuk membuka dan kapan untuk keluar pada pivot makro, tapi itu bisa memakan waktu berhari-hari. sementara itu, Anda perlu penjaga tingkat harga sehingga langkah likuidasi tajam tidak menghapus posisi sebelum sinyal keluar menyala.

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

Kapan? open_long berhasil, catat harga isi ke dalam bot_state.json. Pada setiap tik, panggilan check_price_guards sebelum evaluasi skor makro jika kembali True, lewatkan sisa tanda tanda karena posisi sudah ditutup pada tingkat harga.

Memperluas Strategi

Setelah bot inti berjalan dengan andal, beberapa ekstensi layak dipertimbangkan:

• Tambahkan inflasi inti EUR yang Indeks harga konsumen inti EUR Dan USD PCE menunjukkan tekanan harga yang mendasari; menghubungkan ini ke dalam skor bersama dengan indeks harga konsumen utama meningkatkan klasifikasi sistem pada titik balik.
• Perdagangan margin Kraken Kraken mendukung leverage hingga 5x pada XBT/USD dengan margin; tambahkan "leverage": "2:1" Parameter ke AddOrder panggilan untuk memperkuat sinyal mode positif (hanya sesuai dengan stop-loss yang proporsional lebih ketat).
• Rotasi multi-pasangan mengulangi logika rezim yang sama untuk ETH/USD (XETHZUSD), menggunakan skor makro yang sama; putar modal ke pasangan mana pun yang menunjukkan momentum yang lebih kuat ketika keduanya berada dalam rezim risiko.
• Overlay FX intraday menarik data spot EUR/USD dari Indeks tertimbang perdagangan EUR titik akhir dan menggunakan momentum intraday sebagai filter masuk jangka pendek dalam rezim makro positif.
• Buku pesanan WebSocket menggantikan lingkaran pemungutan suara yang dijadwalkan dengan umpan WebSocket Kraken untuk pembaruan harga real-time, mengurangi latensi dari menit menjadi milidetik untuk penyempurnaan masuk dan keluar.

Ringkasan dan Langkah Selanjutnya

Anda sekarang memiliki robot perdagangan Python lengkap yang menerjemahkan divergensi makro EUR/USD menjadi sinyal panjang/datar yang dapat ditindaklanjuti di Kraken.

1. Indikator FXMacroData menarik Nilai ECB dan Fed, EUR dan USD CPI, EUR/USD spot, semuanya melalui /announcements/ Dan /forex/ titik akhir dengan time stamp tingkat dua
2. Skor perbedaan komposit tertimbang yang memetakan beberapa indikator ke dalam satu sinyal arah
3. Penjadwal pemadaman menghentikan aktivitas order di sekitar jendela rilis ECB dan Fed menggunakan Kalender rilis FXMacroData
4. Manajemen pesanan Kraken membatasi pembelian dan penjualan dengan bendera hanya untuk meminimalkan biaya
5. Penjaga harga Stop loss dan level take profit yang melindungi posisi antara pivot makro regime

Sebagai langkah selanjutnya, eksplorasi USD PCE Dan Indeks harga konsumen inti EUR untuk mempertajam komponen inflasi dari skor, atau memperluas bot untuk memperdagangkan pasangan EUR/USD langsung di Kraken menggunakan sinyal regime yang sama dengan posisi FX spot.

Referensi API lengkap dan semua kombinasi mata uang/indikator yang tersedia ada di /api-referensiUntuk mendapatkan kunci API Anda, kunjungi /langgananAku tidak tahu.