Trading algorithmique de Bitcoin sur Coinbase avec les signaux macro de FXMacroData

Pourquoi les données macro alimentent le Bitcoin

Bitcoin se négocie moins comme un stock technologique et plus comme un actif macro. Il ne paie pas de dividendes, ne rapporte aucun bénéfice et n'a pas de modèle de flux de trésorerie pour ancrer sa valeur. Au lieu de cela, BTC se déplace avec les conditions de liquidité mondiale, les attentes de taux d'intérêt réels et la force du dollar les mêmes forces qui entraînent EUR/USD et AUD/JPY. Cela signifie que la boîte à outils macro conçue pour les traders FX s'applique directement à Bitcoin.

Lorsque la Fed assouplit, la liquidité du dollar inonde les marchés de risque et BTC a tendance à mener la charge. Lorsque l'IPC surprend à la hausse, les récits de dépréciation monétaire poussent le capital vers des actifs dures. Lorsque la haussière de l'inflation de seuil d'équilibre, les rendements réels se compriment et l'or se relève aux côtés de BTC. Chacun de ces signaux est observable à l'avance via les points d'extrémité de l"indicateur de FXMacroData timestamped à la seconde et disponible via une API REST propre.

Ce guide construit un robot de trading algorithmique basé sur des signaux macro pour le BTC-USD sur Coinbase Advanced Trade est une société de commerceÀ la fin, vous aurez une stratégie Python qui fonctionne:

• Extrait des données sur le taux directeur, l'inflation, l"inflation de rentabilité et l'or à partir de FXMacroData
• Les combine en une note composite de macro-régime
• Planifier l'exécution autour des macros à impact élevé via le calendrier de sortie FXMacroData
• Envoie des ordres de marché BTC-USD sur Coinbase en utilisant le service officiel coinbase-advanced-py Le SDK

Pré-requis

Avant de commencer, assurez- vous d'avoir les choses suivantes prêtes:

• Python 3.10+ tous les extraits utilisent une syntaxe de saisie moderne
• Clé de l'API FXMacroData inscrivez-vous à / souscrivez et saisissez votre clé du tableau de bord du compte
• Compte de commerce avancé de Coinbase avec un solde en USD créer un Clé de négociation de l'API du cloud (nom de la clé API + clé privée CE) dans le Plateforme de développement Je suis avec la console . Commerce et commerce permissions activées
• Paquets PythonJe suis désolé . requestsJe suis désolé . coinbase-advanced-pyJe suis désolé . pandasJe suis désolé . schedule

pip install requests coinbase-advanced-py pandas schedule

Stockez vos informations d'identification sous forme de variables d'environnement ne jamais coder des secrets dans des fichiers source:

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-----"

Étape 1: Récupérer les signaux macro à partir de FXMacroData

Quatre macro-série ancrent le modèle de régime BTC: Taux directeur en USDJe suis désolé . Inflation par IPC- Je ne sais pas . Taux d'inflation de rupture de rentabilité à 10 ans, et prix au comptant de l'orEnsemble, ils décrivent l'environnement de liquidité, le régime d'inflation et le sentiment de demande d'actifs dures.

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}")

Chaque paramètre retourne les données par ordre de priorité: les plus récentes. val Il tient le titre et ... announcement_datetime porte le timestamp de sortie de deuxième niveau utile pour la planification visée à l'étape 4. val est le prix au comptant.

Étape 2: Construire un score macro composite

Au lieu de réagir à un seul indicateur, un score composite synthétise les quatre signaux en un nombre directionnel compris entre -1 (risque-off, BTC baissier) et +1 (riske-on, BTC haussier).

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)

Étape 3: Connectez-vous à l'API de commerce avancé de Coinbase

Le fonctionnaire . coinbase-advanced-py La bibliothèque enveloppe l'API REST de Coinbase et gère l'authentification JWT automatiquement. RESTClient en utilisant votre nom de clé de négociation de l'API Cloud et la clé privée CE associée.

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}")

Étape 4: Planifier autour des événements de sortie de macro

L'une des fonctionnalités les plus puissantes de FXMacroData pour le trading d'algo est le point final du calendrier de sortie.

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()}")

Avec l'horodatage de sortie exact, vous pouvez planifier une mise à jour du signal post-sortie permettant à la réaction initiale du prix de se régler avant de réévaluer et de négocier:

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)

Étape 5: Taille des positions et envoyer des ordres sur Coinbase

Les ordres de Coinbase Advanced Trade fonctionnent différemment de Binance: acheter Les ordres précisent un quote_size (montant à dépenser en dollars américains), tandis que vendre Les ordres précisent un base_size La fonction de dimensionnement ci-dessous évalue l'allocation en USD avec la grandeur absolue du score macro une conviction plus élevée justifie une allocation plus importante, plafonnée à un maximum configurable.

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

Étape 6: assembler la boucle de stratégie complète

Maintenant , rassemblez toutes les pièces en une seule . run_strategy() fonction. À chaque appel, il récupère de nouvelles données macro, recalcule le score, détecte un changement de régime et entre ou sort en conséquence.

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)

Étape 7: Gestion des risques et considérations opérationnelles

Une stratégie de signaux macro négocie rarement les changements de régime entraînés par les impressions FOMC, IPC et NFP produisent généralement 610 signaux exploitables par an. Cette faible fréquence est par conception: vous positionnez pour des mouvements directionnels de plusieurs semaines, pas du bruit intraday. Mais chaque position comporte un risque significatif, donc des contrôles disciplinés sont essentiels.

Tous les horodatages de FXMacroData y compris announcement_datetime dans les réponses des indicateurs et les dates de sortie du point final du calendrier sont UTC. Gardez votre logique de planification en UTC et convertissez uniquement à des fins d'affichage. Cela élimine l'ambiguïté de l'heure d'été autour des sorties de données américaines, qui est une source commune de bogues de planificateur.

Élargissement de la stratégie

Le cadre est modulaire, et plusieurs extensions naturelles méritent d'être explorées:

• Ajouter des données de positionnement COT Le point de fin CFTC COT de FXMacroData fournit un positionnement spéculatif hebdomadaire sur les contrats à terme en USD. /cot/usd et ajouter un terme de positionnement net au score macro.
• Indice de liquidité multivalutaire incorporer les signaux macro de l'EUR, du JPY et de la GBP aux côtés du USD. Lorsque plusieurs banques centrales du G10 assurent simultanément un assouplissement, les conditions de liquidité mondiales sont les plus favorables pour les actifs à risque comme le BTC.
• Signal de vitesse de rupture au lieu du niveau de rebond, utiliser le taux de variation de 4 semaines.
• Exécution des ordres de limite remplacer market_order_buy- Je suis désolé .market_order_sell avec limit_order_gtc_buy- Je suis désolé .limit_order_gtc_sell Pour éviter de payer le plein spread. client.get_best_bid_ask et de s'asseoir une tique à l'intérieur de la meilleure offre / demande.
• Clustering du calendrier de sortie demande à calendrier de sortie pour tous les événements USD un mois et identifier les fenêtres où plusieurs émissions à fort impact se regroupent dans les 48 heures.

Résumé

Vous avez maintenant un robot de trading Bitcoin fonctionnel basé sur des signaux macro connecté à Coinbase Advanced Trade. La stratégie lit le taux directeur USD, l'IPC, l"inflation de rentabilité et l'or de FXMacroData pour construire un score composite, se planifie autour d'événements d'annonce macro du monde réel et soumet des ordres de marché BTC-USD proportionnels à la conviction du régime.

L'approche macro-échange peu fréquemment et avec une grande conviction , ce qui est précisément ce qui sépare les stratégies basées sur le régime des systèmes techniques qui chassent le bruit.