FXMacroDataのマクロシグナルを活用したCoinbaseでのビットコインのアルゴリズム取引

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

ビットコインは技術株ではなくマクロ資産のように取引される.配当を支払わず,収益を報告せず,価値を固定するためのキャッシュフローモデルも持っていない.代わりに,BTCは世界的な流動性条件,実質金利予想,ドル強度と動いている.EUR/USDとAUD/JPYを動かす同じ力.つまり,FXトレーダーのために構築されたマクロツールキットは直接ビットコインに適用される.

ドル流動性は市場をリスクに押しつぶし,BTCはリードする傾向にある.CPIが上昇するときに,通貨の値下げナラティブは資本をハード資産に押し込む.ブレイクイブンインフレが上昇すると,実質利回りは圧縮し,ビットコインと共にゴールドが上昇します.これらの信号の1つ"つがFXMacroDataの指標エンドポイントを介して事前に観察できます.

このガイドでは,BTC-USDのマクロシグナル駆動アルゴリズム取引ボットを構築しています. コインベース アドバンスト・トレード実行可能な Python 戦略が表示されます.

• FXMacroDataからドル政策金利,インフレ,ブレイクブーンインフレ,金に関するデータを抽出
• 複合マクロシステムスコアに組み合わせる
• FXMacroData リリース カレンダーを通じて高影響マクロリリース周辺の実行をスケジュールする
• 公式のCoinbaseでBTC-USD市場オーダーを送信します coinbase-advanced-py SDK

条件

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

• Python 3.10+ すべてのスニペットは現代的なタイプ構文を使用します
• FXMacroData API キーを 登録する / サブスクリプト 口座のダッシュボードから鍵を拾って
• コインベース アドバンスト・トレードアカウント 資金調達済みのドル残高で 創出する クラウド API トレーディング キー キーに入力する 開発者プラットフォーム コンソールと 貿易 許可が有効
• Python パッケージありがとうございました requestsほら coinbase-advanced-pyほら pandasほら schedule

pip install requests coinbase-advanced-py pandas schedule

ソースファイルにハードコードの秘密を保存する

export FXMACRO_API_KEY="YOUR_FXMACRODATA_KEY"
export COINBASE_API_KEY_NAME="organizations/ORG_ID/apiKeys/KEY_ID"
export COINBASE_PRIVATE_KEY="-----BEGIN EC PRIVATE KEY-----
MHQCAQEEIBs...
-----END EC PRIVATE KEY-----"

ステップ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")

# data[0] is always the most recent reading
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}")

各エンドポイントは,最も最近のデータから順番を順序付けします.マクロ指標では, val タイトルを握っている announcement_datetime ステップ4で取り扱われるスケジューリングに役立つ2次レベルのリリースタイムスタンプを載せています.商品については, val ポイント価格です

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

複合スコアは,単一の指標に反応する代わりに,4つの信号を1つの方向的な数字に合成します. -1 (リスクオフ,BTCベアシス) と +1 (リスクオン,BTSバイシス).各コンポーネントは,現在の読み方が歴史的にBTCを支持するか反対するかによって重み付けされた項目を寄与します.

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) ──────────────────────────────
    # Below 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) ────────────────────────────────
    # 2–4% moderate inflation → neutral/slightly bullish
    # >6% high inflation → debasement narrative → bullish
    # <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 re-anchoring inflation expectations → 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.6000  (bullish regime)

ステップ3: Coinbase アドバンスト・トレード API に接続する

官員 coinbase-advanced-py データベースはCoinbaseの REST APIを包み込み JWT認証を自動的に処理します RESTClient クラウド API トレーディング キー名と関連ECプライベート キーを使用します

import os
import uuid
from coinbase.rest import RESTClient

CB_KEY_NAME    = os.environ["COINBASE_API_KEY_NAME"]
CB_PRIVATE_KEY = os.environ["COINBASE_PRIVATE_KEY"]

client = RESTClient(api_key=CB_KEY_NAME, api_secret=CB_PRIVATE_KEY)

# Verify connectivity and fetch current BTC-USD price
product = client.get_best_bid_ask(product_ids=["BTC-USD"])
bids = product["pricebooks"][0]["bids"]
asks = product["pricebooks"][0]["asks"]
btc_price = (float(bids[0]["price"]) + float(asks[0]["price"])) / 2
print(f"BTC-USD mid-price: ${btc_price:,.2f}")

# Available USD balance
accounts = client.get_accounts()
usd_balance = 0.0
btc_balance = 0.0
for acct in accounts["accounts"]:
    if acct["currency"] == "USD":
        usd_balance = float(acct["available_balance"]["value"])
    if acct["currency"] == "BTC":
        btc_balance = float(acct["available_balance"]["value"])

print(f"Available USD: ${usd_balance:,.2f}")
print(f"Available BTC: {btc_balance:.6f}")

ステップ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 events in ascending date order.
    """
    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


# Example: find the next FOMC policy rate decision
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: {next_fomc.isoformat()}  ({delta.days}d {delta.seconds // 3600}h away)")

# Example: find the next CPI print
next_cpi = get_next_release("usd", "inflation")
if next_cpi:
    print(f"Next CPI:  {next_cpi.isoformat()}")

発売時の正確なタイムスタンプで,発売後の信号更新をスケジュールできます. 開始価格反応が再評価と取引の前に決着をつけることができます.

def on_macro_release():
    """Called shortly after a scheduled macro release fires."""
    print("Macro release detected — 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 absorb before entry.
    """
    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.every().day.at(fire_str).do(on_macro_release).tag(
        f"{currency}_{indicator}"
    )
    print(f"Scheduled 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: Coinbase でのサイズ・ポジションとオーダー送信

コインベース・アドバンスト・トレード・オーダーはバイナンスとは違う 買える 命令は 指定する quote_size 費用は,ドルで支払われる. 売る 命令は 指定する base_size (BTCの販売額) 下のサイズ機能は,マクロスコアの絶対値でドル配分をスケールします. 高い確信は,設定可能な最大限で制限されたより大きな配分を要求します.

import math


def compute_usd_allocation(
    score: float,
    usd_balance: float,
    max_position_pct: float = 0.20,
    min_threshold: float = 0.30,
) -> float:
    """
    Returns the USD amount to deploy for a BUY order.
    Scales between 0 and max_position_pct of available USD balance.
    Returns 0.0 if |score| < min_threshold (noise-filtered).
    Minimum order size on Coinbase Advanced Trade is $1 USD.
    """
    if abs(score) < min_threshold:
        return 0.0
    conviction = (abs(score) - min_threshold) / (1.0 - min_threshold)
    usd_to_trade = usd_balance * max_position_pct * conviction
    return max(1.0, round(usd_to_trade, 2))


def place_buy(usd_amount: float) -> dict | None:
    """Submit a market BUY order for a given USD amount."""
    if usd_amount <= 0:
        print("USD amount zero — no buy order placed.")
        return None
    order_id = str(uuid.uuid4())
    try:
        result = client.market_order_buy(
            client_order_id=order_id,
            product_id="BTC-USD",
            quote_size=str(usd_amount),
        )
        print(f"BUY order submitted: ${usd_amount:.2f} USD  (order_id={order_id})")
        return result
    except Exception as exc:
        print(f"Coinbase buy error: {exc}")
        return None


def place_sell(btc_amount: float) -> dict | None:
    """Submit a market SELL order for a given BTC amount."""
    # Minimum BTC lot size on Coinbase Advanced Trade is 0.000001 BTC
    btc_amount = math.floor(btc_amount * 1_000_000) / 1_000_000
    if btc_amount <= 0:
        print("BTC amount zero — no sell order placed.")
        return None
    order_id = str(uuid.uuid4())
    try:
        result = client.market_order_sell(
            client_order_id=order_id,
            product_id="BTC-USD",
            base_size=str(btc_amount),
        )
        print(f"SELL order submitted: {btc_amount:.6f} BTC  (order_id={order_id})")
        return result
    except Exception as exc:
        print(f"Coinbase sell error: {exc}")
        return None

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

片付けました run_strategy() 機能.各呼び出しで,新しいマクロデータを取得し,スコアを再計算し,体制変更を検知し,それに応じて入力または終了する.状態はJSONファイルに持続されるため,位置追跡を失うことなくプロセスが再起動を生き残ることができます.

import json
import pathlib

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


def load_state() -> dict:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"position_btc": 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. Refresh 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)
    prev  = state["last_score"]
    print(f"Macro score: {score:+.4f}  (prev: {prev:+.4f})")

    # ── 3. Fetch current Coinbase balances ─────────────────────────────
    accounts  = client.get_accounts()
    usd_avail = 0.0
    btc_avail = 0.0
    for acct in accounts["accounts"]:
        if acct["currency"] == "USD":
            usd_avail = float(acct["available_balance"]["value"])
        if acct["currency"] == "BTC":
            btc_avail = float(acct["available_balance"]["value"])

    # ── 4. Regime change logic ─────────────────────────────────────────
    is_bullish = score >= 0.30
    was_bullish = prev >= 0.30
    is_neutral  = abs(score) < 0.30
    is_bearish  = score < -0.30

    if is_bullish and not was_bullish:
        # New bullish regime — enter long via USD allocation
        usd_to_use = compute_usd_allocation(score, usd_avail)
        place_buy(usd_to_use)

    elif is_neutral and was_bullish and btc_avail > 0.000001:
        # Regime turned neutral — exit position
        place_sell(btc_avail)
        print("Regime neutral — exiting BTC position.")

    elif is_bearish and btc_avail > 0.000001:
        # Hard exit on bearish macro signal
        place_sell(btc_avail)
        print("BEARISH regime — full exit.")

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


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

while True:
    schedule.run_pending()
    time.sleep(10)

ステップ7 リスク管理と運用上の考慮

マクロシグナル戦略は,FOMC,CPI,NFPの印刷によって動かすレジム変化が,通常,年間610の実行可能なシグナルを生産する.その低周波は設計によるものです.あなたは日中の騒音ではなく,複数週間の方向動向のためにポジションを設定しています.しかし,各ポジションには有意義なリスクがあります.したがって,規律的な制御は不可欠です.

FXMacroDataのすべてのタイムスタンプは, announcement_datetime 暦のエンドポイントから表示応答とリリース日付は UTCである.スケジューリングロジックをUTCで保存し,表示目的のためにのみ変換する.これは,スケジュラーバグの一般的な源である米国データリリースに関する日光節約の曖昧さを排除する.

戦略の拡大

フレームワークはモジュール式で,次のいくつかの自然拡張を調査する価値があります.

• COTの位置情報を追加する FXMacroDataのCFTC COTエンドポイントは,毎週USD先物での投機的なポジションを提供しています. 極度のショートUSDポジションは,歴史的にBTCにとって後風です. /cot/usd マクロスコアに 位置付け項目を追加します
• 複数の通貨の流動性指数 ユーロ,JPY,GBPのマクロシグナルをUSDに併用する.複数のG10中央銀行が同時に緩和している場合,グローバル流動性条件はBTCのようなリスク資産にとって最も有利である.
• 速度のブレイクブランス信号 ブルックベンインフレのレベルではなく,4週間の変動率を使用します.ブルックベンの急上昇はレベル単独よりも早く,よりタイミングよく信号です.
• 制限命令の実行 置き換える market_order_buy/ わかったmarket_order_sell 付き合って limit_order_gtc_buy/ わかったlimit_order_gtc_sell オーダーブックを入手してください client.get_best_bid_ask 売り上げの最高値に1つ入る
• リリースカレンダークラスタリング 問い合わせ リリースカレンダー 複数の高影響放出が48時間以内に集まる窓を特定します.これらの密度の高い窓は位置付けの最も高い確信期間のものです.

概要

コインベース・アドバンスト・トレードに接続された動作するマクロシグナル駆動のビットコイン取引ボットが現在あります. 戦略は,複合的なレジムのスコアを構築するために,FXMacroDataからUSD政策レート,CPI,ブレイクイーブインフレ,および金を読み,現実世界のマクロ発表イベントの周りにスケジュールし,レジム信念に比例してBTC-USD市場オーダーを提出します.

マクロアプローチは,頻繁に取引され,高い確信を持って取引されます. これは,騒音を追いかける技術システムから体制ベースの戦略を区別するものです. 次の論理的なステップは,このフレームワークを歴史的なFXMacroData時間系列に対してバックテストし,複数のFedサイクルを通じてスコア重さを校正し,引き下げを測定することです.