FXMacroDataのマクロシグナルでBinanceにてビットコインをアルゴリズム取引

マクロデータがビットコインを動かしている理由

ビットコインは会社ではない.配当を払ったり,収益を報告したり,評価を固定するための収益倍数を持っていない.代わりに,BTCは市場の最もリスクに敏感な資産のように取引される.グローバル流動性条件,ドル強度,実質金利予想に対するレバレッジの賭け.つまり,FXトレーダーがEUR/USDまたはAUD/JPYのポジションに使用する同じマクロツールキットは,ビットコインに直接適用される.

ドル流動性は拡大し,リスク資産が上昇する.CPIが上昇すると,インフレヘッジナラティブが再出現する.実質的収益が崩壊すると,通貨の下落についてのナラティブは資本をハード資産の代替品に押し込む.これらの信号のすべてがFXMacroDataの指標エンドポイントを通じて事前に観察され,秒にタイムスタンプされ,クリーンなREST APIで利用可能である.

このガイドでは,BinanceでBTC/USDTのマクロ信号駆動アルゴリズム取引ボットを構築します.

• FXMacroDataからリアルタイムマクロ信号を抽出します (レート,インフレ,ブレイクイーンのレート)
• 複合マクロスコアに組み合わせます
• リリースカレンダーで高影響リリース周辺の実行をスケジュールする
• 公式 Python SDK を使ってBinance に市場と制限オーダーを送信します

条件

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

• Python 3.9+ すべてのスニペットは標準的なタイプ構文を使用します
• FXMacroData API キーを 登録する / サブスクリプト 鍵をダッシュボードから
• バイナンスアカウント 資金調達済みのUSDT残高で バイナンスダッシュボードで API キーを作成します スポット・マーージン・トレーディングを可能にする チェック
• Python パッケージありがとうございました requestsほら python-binanceほら pandasほら schedule

pip install requests python-binance pandas schedule

硬式コード化ではなく環境変数として API キーを保存します:

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

ステップ1: FXMacroData からマクロ信号を取得する

基本的には,BTCのモデルには4つのマクロシリーズが中心です. 政策金利ドルほら CPIインフレほら 10年間のブレイク・ペインインフレ率ほら 黄金の現金価格流動性環境,インフレ体制,そして硬貨への逃亡の感情を 共に記述しています

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 スケジューリングに役立つ第2レベルのリリースタイムスタンプを搭載している (ステップ4で説明されている).

ステップ2: マクロスコアを複合的に作成

複合スコアは,単一の指標に反応する代わりに,4つの信号を1つの方向的な数字に合成します. -1 (完全にリスクオフ) と +1 (完全にリスクをオフ) の間で.各コンポーネントは,Bitcoinの読み上げが上昇するか下落するかによって重さを追加または減算します.

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)

ステップ3: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: マクロリリースイベントのスケジュール

取引のアルゴリズム化には, リリースカレンダーエンドポイント固定タイマーで指標を調査する代わりに 任意の指標の正確なリリース時間をクエリして 新しいデータが到着したときに正確に論理を起動できます

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

ステップ 6: 戦略のループを 完全に組み立て

片付けました 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 リスク管理と運用上の考慮

マクロシグナル戦略は純粋な技術システムよりも取引頻度がはるかに低い.入口と出口は通常,1年に6~8つの主要な指標リリースで発生する.その低周波は機能であり,バグではありません.あなたは日中騒音ではなく,複数週間の体制変化のためにポジションを設定しています.しかし,それはまた,各ポジションが取引ごとにより多くのリスクを持ち,規律的なリスク管理を交渉不可にすることを意味します.

追加的な考慮事項: BTCは24/7取引していますが,マクロイベントはスケジュールされています. FOMC,CPI,NFPのリリースはすべて米国の市場時間中に発生します.スケジューラーはタイムゾーンに配慮して,UTCを全体的に使用し,エンドユーザーに表示するときにのみ変換する必要があります. FXMacroDataの announcement_datetime 簡単なことです. 簡単な場合は,

戦略の拡大

このフレームワークは意図的にモジュール化されています.次の探索のための自然拡張は以下です:

• COTの位置情報を追加する FXMacroDataのCFTC COTエンドポイントは,USD先物で週間投機的なポジションを提示します. 極短USDポジションは,歴史的に BTCの後押しです. /cot/usd マクロスコアに 位置付け項目を追加します
• 複数の通貨のスコア グローバル流動性スコアを構築するためにEUR,JPY,GBPのマクロ信号を組み込む.複数のG10中央銀行が同時に緩和モードにある場合,BTCリスク条件が最も強い.
• 破損率傾向速度 ブルックベンインフレのレベルではなく,4週間の変動率を使用します.ブルックベンの急上昇はレベル単独よりもより適切な早期信号です.
• リリースカレンダー統合 問い合わせ リリースカレンダー 影響重量によってグループ化し,48時間以内に複数の高影響のリリースがグループ化される窓を特定します.

概要

今,あなたは,FXMacroData指標データを直接Binanceの実行に接続する,マクロシグナル駆動のビットコイン取引ボットを持っています. 戦略は,複合マクロスコアを構築するために,USD政策レート,CPI,ブレイクイーンのインフレ,および金を読み,現実世界の発表イベントの周りにスケジュールし,体制の信念に比例してBTC/USDT注文をサイズし提出します.

このシリーズの次の記事は,このフレームワークを歴史的データに対してバックテストし,実際のFXMacroDataの時間列を 2015年まで遡る方法でスコア重さを校正することをカバーします.