التداول الخوارزمي على Kraken باستخدام إشارات الاقتصاد الكلي لزوج EUR/USD من FXMacroData

لماذا يُدفع الاختلاف الكلي بين اليورو والدولار إلى التداول في العملات الرقمية على كراكين

يحتل كراكين مكانة متميزة في مشهد تبادل العملات الرقمية: يتم استخدامه بشكل كبير من قبل التجار الأوروبيين ، ويقبل الودائع المباشرة باليورو ، ويتعرض العديد من الأزواج مقابل اليورو بالإضافة إلى الدولار الأمريكي. وهذا يعني أن القوى الكلية التي تحكم EUR / USD ECB مقابل تباين سياسة الاحتياطي الفيدرالي ، ومنطقة اليورو مقابل فروق التضخم الأمريكية ، وتفاضلات العائد الحقيقي تخلق رياحًا مؤدية ورائدًا تظهر بوضوح في الأزوج المدرجة في كراكن مثل XBT / USD و XB T / EUR.

عندما يشد الاحتياطي الفيدرالي بينما يحتفظ البنك المركزي الأوروبي، يرتفع الدولار، ويرتفع العائدات الحقيقية، ويتعرض أصول المخاطر (بما في ذلك العملات المشفرة) عادة لضغوط البيع. عندما يكون البنك الأوروبى والاحتياط الفيدرالية خارج التزامن في الاتجاه المعاكس، على سبيل المثال، فإن قوة اليورو وارتفاع العائادات الحقيقية الأمريكية المضغطة يميلان إلى دعم وضع المخاطر في BTC. هذه الأنظمة تتطور على مدى أسابيع وأشهر، ويمكن ملاحظتها بشكل كامل مقدما من خلال نقاط نهاية مؤشر FXMacroData.

هذا الدليل يمر عبر بناء روبوت تداول خوارزمي مدفوع بإشارات كبيرة لـ XBT / USD على كراكين. بحلول النهاية سيكون لديك استراتيجية بايثون تعمل:

• يسحب إشارات اليورو والدولار الأمريكي الكلية من FXMacroData (سعر السعر السياسي، مؤشر أسعار المستهلك، التضخم الأساسي، و EUR/USD الفوري)
• يحسب درجة تباين EUR/USD لتصنيف النظام الكلي
• يحدد مواعيد التنفيذ حول تواريخ إصدار البنك المركزي الأوروبي والاحتياطي الفيدرالي عبر تقويم إصدر FXMacroData
• يقدم أوامر السوق والحد على كراكين عبر واجهة برمجة التطبيقات الرسمية REST
• تطبيق قياسات بسيطة لحجم الموقف ومراقبة المخاطر

الشروط المسبقة

قبل البدء، تأكد من أن لديك ما يلي جاهز:

• بايثون 3.9+ جميع المقتطفات تستخدم تعليقات نوعية قياسية
• مفتاح FXMacroData API التسجيل في / اشترك ونسخ مفتاحك من لوحة حسابك
• حساب كراكين مع رصيد تمويله بالدولار أو باليورو توليد مفاتيح واجهة برمجة التطبيقات في لوحة Kraken تحت الأمن → واجهة برمجة التطبيقات مع إنشاء وتعديل الطلبات و صناديق الاستفسار الإذن
• حزم بايثون. 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: احضر إشارات اليورو والدولار الأمريكي

يستخدم نموذج التباين أربع سلسلات: سعر سعر العملة، ال سعر الفائدة- لا مؤشر أسعار المستهلك في منطقة اليوروو مؤشر أسعار المستهلكين الأمريكيويشرحون معاً مكان كل بنك مركزي في دورة التراجع أو التيسير وما إذا كانت فروق التضخم تفضل قوة اليورو أو الدولار الأمريكي.

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 يحمل هذا المجال الطابع الزمني للطابق الثاني للإصدار ستستخدمه في الخطوة 4 لوقف التداول حول الأحداث ذات التأثير العالي واستئنافها مباشرة بعد إغلاق نافذة الإصدار.

الخطوة 2: حساب درجة الاختلاف الكلي للعملة الأوربية والدولار الأمريكي

بدلاً من التداول من مؤشر واحد، تقوم درجة التباين بتوليف جميع السلسلة الأربع إلى رقم اتجاهي بين -1 (نظام الدولار الأمريكي القوي خطر التخلص من BTC) و +1 (نظم الدولار الأوسع / خطر الدفع داعم لـ 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")

الخطوة الثالثة: اتصل بـ (كركن)

- ... krakenex المكتبة تغلف API REST لـ Kraken بسهولة 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: الجدول الزمني حول إصدارات البنك المركزي الأوروبي والاحتياطي الفيدرالي

الإصدارات الكلية ذات التأثير العالي قرارات سعر الفائدة من البنك المركزي الأوروبي، بيانات مجلس الفوركس الفيدرالي، مطبوعات مؤشر أسعار المستهلك في منطقة اليورو عادة ما تسبب تقلبات حادة في أسواق العملات الأجنبية والعملات الرقمية. إن أكثر الأساليب أمانًا هي وقف أي نشاط طلب مفتوح في نافذة 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: وضع أوامر مدفوعة بالماكرو على كراكين

مع وجود درجة الكلية والتحقق من التعتيم ، يكون منطق النظام بسيطًا. عندما تعبر النتيجة فوق الحد الأدنى الطويل ولا يوجد موقف مفتوح ، قم بإرسال شراء الحد. عندما تتحول النتيجه سالبة أو يصل الموقف إلى هدف الربح ، قم بتقديم بيع الحد . يتم التعبير عن حجم الموقف كجزء من الرصيد المتاح بالدولار الأمريكي ، مع قياسه بحد أقصى صعب.

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 ال مؤشر أسعار المستهلك الأساسي للدولار و الدولار الأمريكي كشف الضغوط السعرية الأساسية؛ وربطها في النتيجة جنبا إلى جنب مع مؤشر أسعار المستهلك الرئيسي يحسن تصنيف النظام في نقاط التحول.
• تداول الهامش في كراكين تدعم كراكين ما يصل إلى 5 × الرافعة المالية على XBT / USD مع الهامش ؛ أضف "leverage": "2:1" المعلم إلى AddOrder دعوات لتضخيم إشارات النظام الإيجابي (مناسبة فقط مع وقف خسارة أقوى نسبياً).
• تدوير متعدد الأزواج تكرار نفس منطق النظام لـ ETH/USD (XETHZUSD), باستخدام نفس النتيجة الكلية؛ تدوير رأس المال في أي زوج يظهر قوة دفع أقوى عندما كلاهما في نظام المخاطر.
• التداولات الفوركسية خلال اليوم استخدم بيانات فورية لـ EUR/USD من مؤشر EUR المرجح في التجارة نقطة النهاية واستخدام الزخم داخل اليوم كمرشح دخول قصير الأجل في نظام ماكرو إيجابي.
• كتاب طلبات WebSocket استبدال حلقة الاستطلاع المجدولة بتغذية WebSocket في كراكين لتحديثات الأسعار في الوقت الحقيقي، مما يقلل من فترة الكمون من دقائق إلى ميلي ثانية لتحسين الدخول والخروج.

ملخص وخطوات لاحقة

لديك الآن روبوت تداول كامل من Python يترجم تباين اليورو/الدولار الأمريكي إلى إشارات طويلة/سطحة قابلة للتنفيذ على كراكين. المكونات الرئيسية هي:

1. مؤشر FXMacroData يسحب أسعار البنك المركزي الأوروبي والاحتياطي الفيدرالي، مؤشر أسعار السلع في اليورو والدولار الأمريكي، سعر الفورية لليورو والدولاري، كل ذلك عبر /announcements/ و /forex/ نقاط النهاية مع طوابع زمنية من المستوى الثاني
2. درجة التباين مركب معدل يضع مؤشرات متعددة على إشارة اتجاه واحدة
3. جدولة التوقف توقف نشاط الطلبات حول فترات إصدار البنك المركزي الأوروبي والاحتياطي الفيدرالي باستخدام تقويم إصدار FXMacroData
4. إدارة طلبات كراكين الحد من عمليات الشراء والبيع باستخدام علامات البريد فقط لتقليل الرسوم
5. حراس الأسعار مستويات وقف الخسارة والاستفادة التي تحمي الموقف بين محاور النظام الكلي

كخطوة طبيعية، استكشاف الدولار الأمريكي و مؤشر أسعار المستهلك الأساسي للدولار لتشديد مكونات التضخم في النتيجة، أو توسيع الروبوت لتداول زوج اليورو/دولار مباشرة على كراكين باستخدام نفس إشارات النظام مع موقف الفوركس الفوري.

المرجع الكامل لـ API وجميع مجموعات العملات / المؤشرات المتاحة في /أب-الإشارةللحصول على مفتاح API الخاص بك، زيارة / اشترك. .