FXMacroDataのEUR/USDマクロシグナルを活用したKrakenでのアルゴリズム取引

なぜEUR/USDマクロダイバージェンスが,クレーンで暗号化を推進するのか

クラケンには暗号取引の格局で明確な位置があります.欧州のトレーダーによって広く使用され,直接ユーロ入金を受け入れ,EURとUSDの両方に多くのペアを配当しています.これは,EUR/USDを支配するマクロ力 (ECB対FED政策の差異,ユーロ圏対米国インフレ差差,実利差) がXBT/USDやXB T/EURのようなクラケンに上場したペアに明確に現れる方向的な後風と逆風を生み出します.

ユーロは,BTCの投資のコストを削減する傾向にある.ECBが持てる間,Fedが引き締まる時,ドルは強くなって,実質利回りが上昇し,リスク資産 (暗号を含む) は通常販売圧力に直面する.ECBとFedは反対方向に同期していない場合,例えば,ECBがFedの休止期間に上昇すると,EURの強さと圧縮された米国の実質利得はBTCでのリスクポジショニングをサポートする傾向がある.これらのレジミンは数週間や数ヶ月間進行し,FXMacroDataの指標エンドポイントを通じて事前に完全に観察できる.

このガイドでは,XBT/USDのマクロシグナル駆動アルゴリズム取引ボットを Kraken で構築します. 最終的には,動作する Python 戦略が表示されます.

• FXMacroDataからEURとUSDのマクロシグナルを抽出する (レート,CPI,コアインフレ,EUR/USDスポット)
• マクロシステム分類のためにEUR/USDの差点を計算する
• FXMacroData のリリースカレンダーを通じて ECB と Fed のリリース日付の周りの実行窓をスケジュールする
• 公式のREST API を介して,クラケンに市場と制限オーダーを送信します.
• 簡単なポジションサイズとリスク管理を行います

条件

始める前に,次のものを用意してごらん.

• Python 3.9+ すべてのスニペットは標準的なタイプ注釈を使用します
• FXMacroData API キーを 登録する / サブスクリプト 口座のダッシュボードからあなたの鍵をコピー
• クレーケン口座 資金調達されたUSDまたはEUR残高で Krakenのダッシュボードで APIキーを生成します セキュリティ → API 付き合って 注文を作成&変更 ほら 資金提供について 許可
• Python パッケージありがとうございました requestsほら krakenexほら pandasほら schedule

pip install requests krakenex pandas schedule

すべての認証情報を環境変数として保存する 硬コードキーには決して:

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

ステップ1: EUR と USD マクロ信号 を 取得

差異モデルでは4つの列を使用します. 政策金利ほら 政策金利ほら ユーロ圏CPIほら アメリカCPI共同で,各中央銀行の上昇または緩和サイクルにおける位置と,インフレ差がEURまたは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']}")

ほら announcement_datetime 公開窓が閉まる直後に再開します. 公開の時間スタンプは,第2レベルのリリースタイムスタンプを表示します.

ステップ2:EUR/USDマクロディバージェンスのスコアを計算する

ダイバージェンスのスコアは,単一の指標で取引するのではなく,すべての4つのシリーズを -1 (強いUSDレジム BTCのリスクオフ) と +1 (弱いUSD/リスクオフレジーム Bitcoinのサポート) の間の方向的な数字に合成する.ポジティブなスコアは長期間の条件を示唆する.ネガティブなスコアでは,フラットまたはショートにとどまることを示唆する

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

ステップ3: クレーケンに接続

ほら krakenex 簡単にRESTAPIを包み込むことができます query_public ほら query_private プライベートエンドポイント (注文の提出,リクエストの残高) は 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}")

ステップ4: ECB と FED の リリース の 周りの スケジュール

高影響マクロリリース ECBのレート決定,FOMCの発表,ユーロ圏CPIの発表 は,通常,FXと暗号市場の両方に急激な変動を誘発する.最も安全なアプローチは,各リリース周辺の30分間のオープンオーダー活動を停止し,マクロスコアを直後に再評価することです.

金融機関が リリースカレンダー 各通貨/指標ペアの次の予定発表日時が表示され,ブラックアウトスケジュラーを作成することは簡単になります.

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

ステップ5:クラケンにマクロ・ドライブオーダーを

マクロスコアとブラックアウトチェックが設定されているため,オーダーロジックは簡単です.スコアがロングスロージックを超え,オープンポジションがない場合は,リミット購入を提出します.スコアはマイナスに転がり,ポジションが利益目標に達した場合,リリミートセールを提出してください.ポジションサイズは,ハード最大で上限を設定した,利用可能なUSD残高の割当として表現されます.

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

ステップ 6: ボットループを完全に組み立て

テストの最終段階は,すべてのものを1時間ごとに実行されるスケジュールループに接続します.各点点でマクロデータをリフレッシュし,ブラックアウトウィンドウを確認し,ディバージェンススコアを再計算し,信号を実行します. schedule 作業のキューを完了させる必要がないので,

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

ステップ7:ストップ・ロストとテイク・プロフィート・ガードを追加

マクロピボットでいつ開くか,いつ退くかを指示します. しかし,それは数日かかるかもしれません. その間,急激な清算が出口信号が発射される前にポジションを消さないように価格レベルガードが必要です.

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

いつ open_long 記入する bot_state.json鳴るたびに 呼び出せ check_price_guards 返信される場合 Trueポジションが価格レベルで既に閉ざされているので,残りのティックのスキップします.

戦略の拡大

核心ボットが信頼性のある状態で動作すると,いくつかの拡張機能が考慮されるべきです:

• EUR の基本インフレを足す EURコアCPI ほら ドルPCE 価格圧力を明らかにします.これらの指標をCPIのスコアに組み込むと,ターニングポイントでの制度分類が改善されます.
• クレーケン・マーージン・トレーディング XBT/USDの利回りが5倍まで,マージンでサポートします. "leverage": "2:1" パラメータを AddOrder ポジティブ・モード信号の増幅の呼びかけ (比例的により厳しいストップ・ロスの場合のみ適切)
• 多対回転 ETH/USDのレジムの論理を繰り返す (XETHZUSDリスクの上昇状態にある場合,資本をどちらのペアに回転させます.
• 内部FXの重複 ユーロ/ドルのスポットデータを EUR取引重量指数 終点点と,ポジティブなマクロ体制内の短期入口フィルターとして日中の動力を使用する.
• WebSocket 注文簿 予定された投票ループを,リアルタイム価格更新のための Kraken の WebSocket フィードに置き換えて,エントリーと出口精製のために,リテンシーを数分からミリ秒に削減します.

概要 と 次 の ステップ

キーコンポーネントは以下の通りです. キーパーは,Pythonのプロファイルで,

1. FXMacroData インジケーターが引く ECBとFedのレート,EURとUSDCPI,EUR/USDスポット,すべて /announcements/ ほら /forex/ 2次レベルのタイムスタンプを持つエンドポイント
2. 差点点数 複数のインジケーターを単一の方向信号にマッピングする重量化された複合材料
3. ブラックアウトスケジューラー 発行期間中に注文活動を停止する FXMacroData のリリース カレンダー
4. クラクン オーダー管理 料金を最小限に抑えるため,ポストのみのフラグで購入と販売を制限する
5. 価格ガード マクロレジムのピボット間のポジションを保護するストップ・ロストとテイク・プロフィートのレベル

探求する ドルPCE ほら EURコアCPI ポイントのインフレ要素を鋭くしたり,ボットを拡張して,即時FXポジションと同じレジム信号を使用して,EUR/USDペアを直接Krakenで取引します.

完全なAPI参照と,利用可能な通貨/指標の組み合わせは, /api参照APIキーを取得するには, / サブスクリプトわかった