التداول الخوارزمي للبيتكوين على Binance باستخدام إشارات الماكرو من FXMacroData

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

بيتكوين ليست شركة. لا تدفع أرباحاً، ولا تبلغ عن أرباح، وليس لديها مضاعفات إيرادات لتثبيت التقييم. بدلاً من ذلك، تتداول بيتكوین مثل الأصول الأكثر حساسية للمخاطر في السوق رهان ممول على ظروف السيولة العالمية وقوة الدولار وتوقعات أسعار الفائدة الحقيقية. وهذا يعني أن مجموعة الأدوات الكلية نفسها التي يستخدمها تجار الفوركس لتحديد المواقع في EUR / USD أو AUD / JPY تنطبق مباشرة على بيتكويت.

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

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

• يسحب إشارات الكليّة في الوقت الحقيقي من FXMacroData (سعر السياسة، التضخم، معدل التعادل، الذهب)
• يجمعها في درجة جامعة
• جدولة تنفيذ حول الإصدارات ذات التأثير العالي عبر تقويم الإصلاح
• يقدم طلبات السوق والحد على Binance عبر Python SDK الرسمي

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

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

• بايثون 3.9+ جميع المقتطفات تستخدم نحو المطبوعات القياسية
• مفتاح FXMacroData API التسجيل في / اشترك وخذ مفتاحك من لوحة القيادة
• حساب باينانس مع رصيد USDT ممول إنشاء مفاتيح API في لوحة Binance مع تمكين تداول الفورية والهامش تم التحقق
• حزم بايثون. requests- لا python-binance- لا pandas- لا schedule

pip install requests python-binance pandas schedule

تخزين مفاتيح واجهة برمجة التطبيقات الخاصة بك كمتغيرات البيئة بدلا من ترميزها صلبة:

export FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY"
export BINANCE_API_KEY="YOUR_BINANCE_KEY"
export BINANCE_SECRET_KEY="YOUR_BINANCE_SECRET"

الخطوة 1: احضر إشارات الماكرو من FXMacroData

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

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 = "2024-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
policy_rate   = get_series("/announcements/usd/policy_rate")
cpi           = get_series("/announcements/usd/inflation")
breakeven_10y = get_series("/announcements/usd/breakeven_inflation_rate")
gold          = get_series("/commodities/gold")

# Each item: {"date": "2025-04-08", "val": 5.25, "announcement_datetime": "..."}
print(f"Policy rate (latest): {policy_rate[0]['val']}%")
print(f"CPI (latest):          {cpi[0]['val']}%")
print(f"10Y breakeven:         {breakeven_10y[0]['val']}%")
print(f"Gold spot:             ${gold[0]['val']:.2f}")

كل نقطة نهاية تعيد قائمة مرتبة من الأخيرة إلى الأقدم data[0] دائماً ما تكون آخر قراءات. val؛ بالنسبة للمؤشرات الكلية ، val يحمل الرقم الرئيسي و announcement_datetime يحمل الطابع الزمني للإصدار من المستوى الثاني المفيد لجدولة (المغطاة في الخطوة 4).

الخطوة الثانية: قم ببناء درجة مكرو مركبة

بدلاً من الاستجابة لمؤشر واحد، تقوم النتيجة المركبة بتوليف جميع الإشارات الأربع إلى رقم اتجاهي واحد بين -1 (خطر كامل) و +1 (خاطر كامل). يضيف كل عنصر أو يطرح الوزن بناءً على ما إذا كانت القراءة صاعدة أو هبوطية لبيتكوين.

def macro_score(
    policy_rate_pct: float,
    cpi_pct: float,
    breakeven_pct: float,
    gold_usd: float,
    *,
    gold_baseline: float = 1900.0,
) -> float:
    """
    Returns a composite macro score in [-1, +1].

    Positive = risk-on / BTC bullish environment.
    Negative = risk-off / BTC bearish environment.
    """
    score = 0.0

    # ── Policy rate regime (weight 0.35) ──────────────────────────
    # Sub-4.5% = accommodative → bullish; above 5.5% = restrictive → bearish
    if policy_rate_pct < 4.5:
        score += 0.35
    elif policy_rate_pct <= 5.5:
        score += 0.35 * (5.5 - policy_rate_pct) / 1.0
    else:
        score -= 0.20

    # ── Inflation regime (weight 0.25) ────────────────────────────
    # CPI 2–4%: moderate inflation → neutral/slightly bullish
    # CPI > 6%: high inflation → monetary debasement narrative → bullish
    # CPI < 1.5%: deflationary risk → bearish
    if cpi_pct > 6.0:
        score += 0.25
    elif cpi_pct >= 2.0:
        score += 0.10
    else:
        score -= 0.15

    # ── Breakeven inflation (weight 0.20) ─────────────────────────
    # Rising breakevens signal inflation expectations are re-anchoring → bullish
    if breakeven_pct >= 2.5:
        score += 0.20
    elif breakeven_pct >= 2.0:
        score += 0.10
    else:
        score -= 0.10

    # ── Gold trend (weight 0.20) ──────────────────────────────────
    # Gold above baseline confirms hard-asset demand → bullish
    gold_ratio = (gold_usd - gold_baseline) / gold_baseline
    score += 0.20 * max(-1.0, min(1.0, gold_ratio * 5))

    return round(max(-1.0, min(1.0, score)), 4)


score = macro_score(
    policy_rate_pct=policy_rate[0]["val"],
    cpi_pct=cpi[0]["val"],
    breakeven_pct=breakeven_10y[0]["val"],
    gold_usd=gold[0]["val"],
)
print(f"Composite macro score: {score:+.4f}")
# e.g. → +0.6500  (bullish regime)

الخطوة الثالثة: اتصل بـ Binance واستلم سعر BTC

مع إشارة الكلية جاهزة، اتصل بـ (بينانس) باستخدام الرسمية python-binance دائماً احصل على السعر الفوري الحالي قبل وضع الطلب لتجنب القيم المرجعية القديمة.

import os
from binance.client import Client

BINANCE_KEY    = os.environ["BINANCE_API_KEY"]
BINANCE_SECRET = os.environ["BINANCE_SECRET_KEY"]

client = Client(BINANCE_KEY, BINANCE_SECRET)

# Verify connectivity
status = client.get_system_status()
print(f"Binance status: {status['msg']}")  # → "normal"

# Latest BTC/USDT price
ticker = client.get_symbol_ticker(symbol="BTCUSDT")
btc_price = float(ticker["price"])
print(f"BTC/USDT spot: ${btc_price:,.2f}")

# Current USDT balance
account = client.get_account()
usdt_balance = next(
    (float(b["free"]) for b in account["balances"] if b["asset"] == "USDT"),
    0.0,
)
print(f"Available USDT: ${usdt_balance:,.2f}")

الخطوة 4: جدولة حول أحداث الإصدار الكلي

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

import datetime
import schedule
import time


def get_next_release(currency: str, indicator: str) -> datetime.datetime | None:
    """
    Returns the next scheduled release datetime for an indicator.
    The calendar endpoint returns upcoming release dates ordered ascending.
    """
    resp = requests.get(
        f"{BASE_URL}/calendar/{currency}",
        params={"api_key": FXMACRO_KEY},
        timeout=10,
    )
    resp.raise_for_status()
    events = resp.json().get("data", [])

    now_utc = datetime.datetime.now(tz=datetime.timezone.utc)
    for event in events:
        if event.get("indicator") != indicator:
            continue
        release_str = event.get("release_datetime") or event.get("date")
        if not release_str:
            continue
        release_dt = datetime.datetime.fromisoformat(release_str.replace("Z", "+00:00"))
        if release_dt > now_utc:
            return release_dt
    return None


# Find when the next FOMC policy rate decision is scheduled
next_fomc = get_next_release("usd", "policy_rate")
if next_fomc:
    delta = next_fomc - datetime.datetime.now(tz=datetime.timezone.utc)
    print(f"Next FOMC release: {next_fomc.isoformat()} ({delta.days}d {delta.seconds // 3600}h away)")
else:
    print("No upcoming policy_rate event found in calendar.")


# Find when the next CPI release is scheduled
next_cpi = get_next_release("usd", "inflation")
if next_cpi:
    print(f"Next CPI release:  {next_cpi.isoformat()}")

مع وجود الطابع الزمني الدقيق للإصدار، يمكنك تحديد موعد لتحديث إشارة ما بعد الإصدار مما يسمح للسوق بمستحوذ على الطباعة لبضع دقائق قبل إعادة التسويق وإعادة التداول:

def on_macro_release():
    """Called shortly after a scheduled macro release."""
    print("Macro release fired — refreshing signals...")
    run_strategy()


def schedule_next_release(currency: str, indicator: str, delay_seconds: int = 90):
    """
    Schedules the strategy to run 'delay_seconds' after the next release.
    A 90-second delay lets the initial market reaction settle slightly.
    """
    release_dt = get_next_release(currency, indicator)
    if not release_dt:
        return

    fire_at = release_dt + datetime.timedelta(seconds=delay_seconds)
    fire_str = fire_at.strftime("%H:%M:%S")  # schedule library uses HH:MM:SS
    schedule.every().day.at(fire_str).do(on_macro_release).tag(f"{currency}_{indicator}")
    print(f"Scheduled signal refresh at {fire_str} UTC after {currency.upper()} {indicator}")


schedule_next_release("usd", "policy_rate", delay_seconds=90)
schedule_next_release("usd", "inflation", delay_seconds=60)

الخطوة 5: حجم المواقع وتقديم الطلبات

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

from binance.enums import SIDE_BUY, SIDE_SELL, ORDER_TYPE_MARKET, ORDER_TYPE_LIMIT
from binance.exceptions import BinanceAPIException
import math


def compute_quantity(
    score: float,
    usdt_balance: float,
    btc_price: float,
    max_position_pct: float = 0.20,
) -> float:
    """
    Scale position size between 0 and max_position_pct of USDT balance.
    Only trade when |score| > 0.30 to avoid noise-driven entries.
    Returns BTC quantity rounded to Binance's minimum lot size (0.00001 BTC).
    """
    if abs(score) < 0.30:
        return 0.0

    conviction = (abs(score) - 0.30) / 0.70          # 0.0 → 1.0
    usdt_to_trade = usdt_balance * max_position_pct * conviction
    btc_qty = usdt_to_trade / btc_price
    return math.floor(btc_qty * 100_000) / 100_000    # 5 decimal places


def place_order(side: str, quantity: float, btc_price: float) -> dict | None:
    """
    Submit a market order. For limit orders, pass a price to get_order_book
    and sit 0.1% inside the spread.
    """
    if quantity <= 0.0:
        print("Quantity zero — no order placed.")
        return None

    try:
        order = client.order_market(
            symbol="BTCUSDT",
            side=side,
            quantity=quantity,
        )
        print(f"Order placed: {side} {quantity:.5f} BTC @ ~${btc_price:,.2f}")
        return order
    except BinanceAPIException as exc:
        print(f"Binance order error: {exc.message}")
        return None

الخطوة السادسة: تجميع حلقة الاستراتيجية الكاملة

الآن ، قم بتجميع كل قطعة إلى قطعة واحدة run_strategy() وظيفة تحصل على بيانات ماكرو جديدة، وتحسب النتيجة، وتتحقق من تغيير النظام، وتفتح أو تغلق موقف BTC وفقا لذلك.

import json
import pathlib

STATE_FILE = pathlib.Path("strategy_state.json")


def load_state() -> dict:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"position": 0.0, "last_score": 0.0}


def save_state(state: dict) -> None:
    STATE_FILE.write_text(json.dumps(state, indent=2))


def run_strategy() -> None:
    state = load_state()

    # ── 1. Fetch fresh macro data ──────────────────────────────────
    policy_rate_val   = get_series("/announcements/usd/policy_rate")[0]["val"]
    cpi_val           = get_series("/announcements/usd/inflation")[0]["val"]
    breakeven_val     = get_series("/announcements/usd/breakeven_inflation_rate")[0]["val"]
    gold_val          = get_series("/commodities/gold")[0]["val"]

    # ── 2. Compute macro score ─────────────────────────────────────
    score = macro_score(policy_rate_val, cpi_val, breakeven_val, gold_val)
    print(f"Macro score: {score:+.4f}  (prev: {state['last_score']:+.4f})")

    # ── 3. Fetch current Binance position and price ───────────────
    account   = client.get_account()
    btc_held  = float(next(b["free"] for b in account["balances"] if b["asset"] == "BTC"))
    usdt_held = float(next(b["free"] for b in account["balances"] if b["asset"] == "USDT"))
    btc_price = float(client.get_symbol_ticker(symbol="BTCUSDT")["price"])

    # ── 4. Regime change logic ─────────────────────────────────────
    prev_score    = state["last_score"]
    was_long      = prev_score >= 0.30
    is_long_now   = score >= 0.30
    was_flat      = abs(prev_score) < 0.30
    is_flat_now   = abs(score) < 0.30

    if is_long_now and (was_flat or prev_score < 0):
        # Enter or increase long
        qty = compute_quantity(score, usdt_held, btc_price)
        place_order(SIDE_BUY, qty, btc_price)

    elif is_flat_now and was_long and btc_held > 0.0001:
        # Exit long — macro regime has turned neutral
        exit_qty = math.floor(btc_held * 100_000) / 100_000
        place_order(SIDE_SELL, exit_qty, btc_price)

    elif score < -0.30 and btc_held > 0.0001:
        # Hard exit on bearish signal
        exit_qty = math.floor(btc_held * 100_000) / 100_000
        place_order(SIDE_SELL, exit_qty, btc_price)
        print("BEARISH regime — full exit.")

    # ── 5. Persist state ───────────────────────────────────────────
    state["last_score"] = score
    save_state(state)


# ── Run once immediately, then on each scheduled release ──────────────
run_strategy()

# Keep scheduler alive
while True:
    schedule.run_pending()
    time.sleep(10)

الخطوة 7: إدارة المخاطر والاعتبارات التشغيلية

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

ملاحظة إضافية: يتم تداول BTC على مدار الساعة ولكن الأحداث الكلية مقررة. تقع جميع إصدارات FOMC و CPI و NFP خلال ساعات السوق الأمريكية. يجب أن يكون جدولةك على دراية بالزمنية استخدم UTC طوال الوقت وتحويل فقط عند عرضها للمستخدمين النهائيين. announcement_datetime المجال هو دائما UTC، مما يجعل هذا واضحا.

توسيع الاستراتيجية

هذا الإطار هو وحدات عمدا. وهنا هي التمديدات الطبيعية لاستكشاف التالي:

• إضافة بيانات تحديد الموقع COT توفر نقطة نهاية CFTC COT من FXMacroData تحديد المواقع المضاربة الأسبوعية في العقود الآجلة للدولار الأمريكي. الموقع القصير للغاية للدولار هو تاريخياً رياحًا مؤاتية لـ BTC. سحبها مع /cot/usd وأضاف مصطلح تحديد الموقع الصافي إلى درجة الكلية.
• النتيجة المتعددة العملات دمج إشارات اليورو والين الياباني والجنيه الإسترليني الكبرى لبناء درجة السيولة العالمية. عندما يكون العديد من البنوك المركزية في مجموعة العشرة في وضع التيسير في وقت واحد، تكون ظروف مخاطر BTC أقوى.
• سرعة الاتجاه عند التعادل بدلاً من مستوى التضخم عند التساوي، استخدم معدل التغير الذي يبلغ أربعة أسابيع. يعد الارتفاع الحاد في مستويات التسوية إشارة مبكرة أكثر توفراً من المستوى وحده.
• دمج تقويم الإصدارات سألت عن تقويم الإصدار لجميع أحداث الدولار الأمريكي في شهر واحد، تجميعها حسب وزن التأثير، وتحديد النوافذ التي تتجمع فيها إطلاقات متعددة ذات تأثير كبير خلال 48 ساعة هذه هي الفترات التي تستحق التوجه.

ملخص

لديك الآن روبوت تداول بيتكوين يعمل على إشارة ماكرو مدفوعة بالبوت الذي يربط بيانات مؤشر FXMacroData مباشرة بتنفيذ Binance. تقرأ الاستراتيجية سعر سعر السياسة بالدولار الأمريكي، ومعدل أسعار المستهلك، وتضخم التعادل، والذهب لبناء درجة جامعة، وتجدول نفسها حول أحداث الإعلان في العالم الحقيقي، وحجم وتقديم أوامر BTC/USDT متناسبة مع إدانة النظام.

المقالة التالية في هذه السلسلة تغطي اختبار هذا الإطار ضد البيانات التاريخية وتعديل وزنات النتيجة باستخدام سلسلة زمنية حقيقية من FXMacroData يعود إلى عام 2015.