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

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

比特币不是一家公司.它不支付股息,不报告收益,并且没有收入倍数来定估值.相反,比特币像市场中最敏感的风险资产一样交易,这是对全球流动性状况,美元强度和实际利率预期的杆押注.这意味着FX交易者使用的相同宏观工具包用于EUR/USD或AUD/JPY的定位直接适用于比特幣.

美国联邦储备委员会 (Fed) 降息时,美元流动性扩大,风险资产升,比特币通常引领收费.当CPI惊上升时,通货膨胀对冲叙述再次出现.当实际收益率崩时,货币值的叙述将资本推向硬资产替代品.这些信号中的每一个都可以通过FXMacroData的指标终点预先观察到,时间印到秒,并可通过清洁的REST API.

本指南介绍了在Binance上构建一个基于宏观信号的算法交易机器人.

• 从FXMacroData中提取实时宏观信号 (政策利率,通货膨胀,破产率,黄金)
• 将它们结合成一个复合的宏观分数
• 通过发布日历围绕高影响力发布进行执行计划
• 通过官方 Python SDK 在 Binance 上提交市场和限制订单

预先要求

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

• Python 3.9+ 所有片段使用标准的打字语法
• 汇率数据API键 登记在 订阅 拿出你的钥匙从仪表板上
• 币安账户 在Binance仪表板中创建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制度模型的核心: 美元政策利率没有人知道. 消费者指数通胀没有 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 带有第二级发布时间,可用于调度 (在步骤4中涵盖).

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

复合分数不仅仅是对单个指标进行反应,还将所有四个信号合成到 -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)

步骤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:围绕宏发布事件进行计划

对于算法交易,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

步骤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)

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

宏观信号策略的交易频率比纯技术系统低得多. 入口和出口通常发生在每年68个主要指标发布左右. 这种低频率是一种特征,而不是一个 bug:您正在定位为多周的政权变化,而非日内噪音. 然而,这也意味着每个位置每次交易都承担着更多的风险,使纪律风险管理不可谈判.

另外一个考虑点:BTC24/7交易,但宏观事件是计划的.FOMC,CPI和NFP发布都发生在美国市场时间.您的调度器应该了解时区. announcement_datetime 字段总是UTC, 这使得这很简单.

扩大战略

这种框架是故意模块化的.下面我们来探索一些自然扩展:

• 添加COT定位数据 FXMacroData的CFTC COT终点提供每周美元期货投机定位. 极端短美元定位历来是BTC的尾风. /cot/usd 并将一个净定位项添加到宏观分数中.
• 多货币分数 结合欧元,日元和英宏观信号,构建全球流动性评分.当多个G10央行同时处于宽松模式时,BTC风险条件最强.
• 破产平衡趋势速度 调整率的水平,而不是调整水平,使用四周的变化率.调整点的急剧上升是比单单调整的水平更及时的早期信号.
• 发布日历集成 查询一个 发布日程 对于所有一个月外的美元事件,按冲击重量分组,并确定在48小时内多次高冲击释放集成的窗口 这些是值得定位的时间段.

总结

现在您有一个运行宏观信号驱动的比特币交易机器人,它直接将FXMacroData指标数据连接到Binance执行.该策略读取美元政策利率,CPI,平衡通胀和黄金,构建复合宏观分数,围绕现实世界公告事件进行安排,并根据政权的信念比例进行BTC/USDT订单的尺寸和提交.

接下来的文章将对此框架进行对比测试,并使用到2015年的真实FXMacroData时间序列校准分数权重.