使用 FXMacroData 宏观信号在 Coinbase 上算法交易比特币

为什么宏观数据驱动比特币

比特币的交易不像技术股票,更像宏观资产.它不支付股息,不报告收益,没有现金流模型来固其价值.相反,比特币与全球流动性条件,实际利率预期和美元强度相适应.

美国联邦储备委员会放宽了货币市场的收支,美元流动性淹没了市场,BTC倾向于引领收支.当CPI惊上升时,货币值叙述将资本推向硬资产.当平衡通胀上升的时候,实际收益率压缩,黄金与BTC相比上.这些信号中的每一个都可以通过FXMacroData的指标终点时间到秒进行预测,并通过清洁的REST API可用.

本指南为BTC-USD建立了一个基于宏观信号的算法交易机器人. 代币基地高级贸易. 到最后,你将有一个可行的Python策略:

• 从FXMacroData获取美元政策利率,通胀,平衡通胀和黄金数据
• 将它们结合成一个综合的宏观制度分数
• 通过FXMacroData发布日历围绕高影响力宏发布进行执行
• 在Coinbase上提交BTC-USD市场订单,使用官方的 coinbase-advanced-py 技术支持

预先要求

在开始之前,请确保您准备好以下内容:

• Python 3.10+ 所有片段使用现代打字语法
• 汇率数据API键 登记在 订阅 拿出您的钥匙从帐户仪表板
• 代币基地高级交易账户 创建一个 云API交易密钥 关键字名字 + EC私钥) 在 开发者平台 控制台与 贸易 允许
• 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制度模型: 美元政策利率没有人知道. 消费者指数通胀没有 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所述的计划. val 是现货价格.

步骤2:建立一个复合宏观分数

复合分数不仅仅是对单个指标进行反应,还将所有四个信号合成到 -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:围绕宏发布事件进行计划

对于ALGO交易,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 下面的尺寸函数将美元配置与宏观评分的绝对大小进行尺度,更高的信念意味着更大的配置,限于可配置的最大值.

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)

第七步:风险管理和运营考虑

宏观信号策略 trading不经常 由于FOMC,CPI和NFP的压力导致的制度变化,通常每年产生610个可操作的信号.这种低频率是设计的:您正在定位多周方向动作,而不是日内噪音.但每个位置都带有有意义的风险,因此有纪律的控制是必不可少的.

所有时间从FXMacroData 包括 announcement_datetime 在指标响应和发布日期时间从日历终点是UTC.保持您的调度逻辑在UTC整个和转换仅用于显示目的. 这消除了美国数据发布周围的节日模糊性,这是一个常见的调节器错误来源.

扩大战略

框架是模块化的. 值得我们进一步探讨的几种自然延伸:

• 添加COT定位数据 FXMacroData的CFTC COT终点提供每周美元期货投机定位. 极端短美元定位历来一直是BTC的尾风. 拉它在 /cot/usd 并将一个净定位项添加到宏观分数中.
• 多货币流动性指数 结合EUR,JPY和英的宏观信号,同时与美元.当多个G10央行同时放宽时,全球流动性条件对比特币等风险资产最有利.
• 破局速度信号 调整率的水平,而不是调整水平,使用四周的变化率.调整点的急剧上升是比单单调整的水平更早,更及时的信号.
• 限制命令执行 取代 market_order_buy没有.market_order_sell 随着 limit_order_gtc_buy没有.limit_order_gtc_sell 为了避免支付全额收购者差距. client.get_best_bid_ask 并且在最好的买卖/买卖中坐下来.
• 发布日历集群 查询一个 发布日程 对于所有美元事件,一个月外,并确定在48小时内多次高影响性释放聚合的窗口.这些密集的窗户是定位的最高定位期.

总结

现在您有一个与Coinbase高级贸易连接的可操作的宏观信号驱动的比特币交易机器人.该策略从FXMacroData读取美元政策利率,CPI,平衡通胀率和黄金,构建复合的制度评分,围绕现实世界的宏观公告事件进行调度,并根据制度信念提交BTC-USD市场订单.

宏观方法的交易不常见,而且具有很高的信心,这正是区分基于制度的策略和追逐噪音的技术系统的原因.下一步是对历史FXMacroData时间序列进行后台测试,以校准分数权重并测量多个美联储周期的收益率.