FXMacroData EUR/USD 거시 신호를 활용한 Kraken 알고 트레이딩

왜 EUR/USD 거시적 이차가 크래켄에 암호화폐를 유도하는가

크라켄은 암호화폐 거래소 지형에서 뚜렷한 위치를 차지하고 있습니다. 유럽 거래자들이 많이 사용하고 있으며, 직접적인 유로 예금을 받아내고, 유로와 USD에 비해 많은 쌍을 코팅합니다. 즉, EUR/USD를 지배하는 거시적 힘들, ECB 대 FED 정책의 분차, 유로의 지역 대 미국 인플레이션 차이는, 그리고 실제 수익률 스프레드는 크라킨에 상장된 XBT/USD와 XBt/EUR와 같은 쌍에서 명확하게 나타나는 방향적인 후풍과 역풍을 만듭니다.

중앙은행이 지분을 보유하는 동안 FED이 강화되면 달러가 강해지고, 실수익이 상승하며, 위험 자산 (암호 포함) 은 일반적으로 판매 압력을 받는다. ECB와 FED이 반대 방향으로 동기화되지 않을 때, 예를 들어, ECB가 FED의 휴식기에 들어간 경우, EUR의 강도와 압축된 미국 실수율은 BTC의 위험 입지를 지원하는 경향이 있다. 이러한 체제는 몇 주와 몇 달 동안 진행되며 FXMacroData의 지표 최종 지점으로 미리 완전히 관찰 될 수 있다.

이 가이드는 크라켄에서 XBT/USD를 위한 매크로 신호가 구동되는 알고리즘 거래 봇을 구축하는 과정을 안내합니다.

• FXMacroData에서 EUR 및 USD 매크로 신호를 가져옵니다 (정책율, CPI, 핵심 인플레이션 및 EUR/USD 스팟)
• 거시적 체제를 분류하기 위해 EUR/USD 분차 점수를 계산합니다.
• FXMacroData 발매 달력을 통해 ECB 및 Fed 발매 날짜에 따라 실행 창을 스케줄합니다.
• 공식 REST API를 통해 Kraken에 시장 및 제한 명령을 제출합니다.
• 간단한 포지션 크기와 위험 통제를 적용합니다.

필수 조건

시작 하기 전 다음 과 같은 것 들 을 준비 해 두십시오.

• 파이썬 3.9+ 모든 단편들은 표준 타입 주석을 사용합니다
• FXMacroData API 키 등록하세요 / 가입 그리고 계정 대시보드에서 키를 복사
• 크라켄 계좌 크라켄 대시보드에서 API 키를 생성합니다. 보안 → API 랑 주문을 생성 및 수정 그리고 쿼리 펀드 허가
• 파이썬 패키지 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 매크로 신호를 가져오기

이 분산 모델은 네 개의 시리즈를 사용합니다. ECB 정책금리, 그 미국 연방준비제도 유로존 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차 레벨의 릴리스 타임 스탬프를 가지고 있습니다. 당신은 4단계에서 사용해서 높은 영향력 있는 이벤트 주변에서 거래를 일시 중지하고 릴리즈 창이 닫히면 즉시 재개할 수 있습니다.

단계 2: EUR/USD 매크로 디버전스 점수를 계산합니다.

단일 지표에서 거래하기보다는 분산 점수는 모든 네 개의 시리즈를 -1 (강 USD 체제 BTC의 위험-제도) 와 +1 (미국 달러 / 위험-상 체제 약자 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 라이브러리는 크래켄의 REST API를 간단하게 엮어줍니다. 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 분 창에서 모든 오픈 오더 활동을 중단하고 즉시 거시 점수를 재평가하는 것입니다. 따라서 출시 전 소음보다는 새로운 체제를 거래합니다.

FXMacroData의 발매 일정은 각 통화/지표 쌍의 다음 예정된 발표 날짜와 시간을 표시하여 블랙아웃 스케줄러를 쉽게 만들 수 있습니다.

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: 전체 봇 루프를 조립합니다

마지막 단계는 모든 것을 시간당 한 번 실행되는 일정 루프로 연결합니다. 매 틱마다 매크로 데이터를 갱신하고, 블랙아웃 창을 확인하고, 분리가 점수를 재 계산하고 신호를 실행합니다. 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/USD 현시 데이터를 거래 중인 EUR 지수 엔드포인트 및 내일 모멘텀을 긍정적인 거시적 체제 내에서 단기 입시 필터로 사용한다.
• 웹소켓 주문서 실시간 가격 업데이트를 위해 예정된 투표 루프를 크라켄의 웹소켓 피드로 대체하여 입출출 정리를 위해 지연 시간을 분에서 밀리초로 줄입니다.

요약 및 다음 단계

이제 당신은 EUR/USD 매크로 디버전스를 크라켄에서 실행 가능한 긴/평면 신호로 변환하는 완전한 파이썬 거래 봇을 가지고 있습니다. 주요 구성 요소는:

1. FXMacroData 표시기가 당겨집니다. ECB 및 FED 금리, EUR 및 USD CPI, EUR/USD 스팟, 모두 /announcements/ 그리고 /forex/ 2급 시간표가 있는 최종점
2. 분산점 하나의 방향 신호에 여러 지표를 매핑하는 가중된 복합
3. 블랙아웃 스케줄러 ECB 및 Fed의 릴리스 윈도우 주변에서 주문 활동을 중지합니다. FXMacroData 발매 달력
4. 크라켄 주문 관리 수수료를 최소화하기 위해 포스트 전용 플래그로 구매 및 판매를 제한합니다.
5. 가격 보호자 매크로 레지엄 피워드 사이의 위치를 보호하는 스톱 로스 및 영업 취득 수준

다음 단계로 자연스럽게 탐험을 해보세요. 미국 달러 PCE 그리고 EUR 기본 CPI 점수의 인플레이션 구성 요소를 선명하게 하거나, 또는 봇을 확장하여 스팟 FX 포지션과 동일한 레지엄 신호를 사용하여 EUR/USD 쌍을 크라켄에 직접 거래합니다.

전체 API 참조 및 모든 사용 가능한 통화/지표 조합은 /api- 참조API 키를 받으려면 / 가입- 그래요