FXMacroData 거시 신호를 활용한 바이낸스 비트코인 알고 트레이딩

왜 매크로 데이터가 비트코인을 주도하는 걸까요?

비트코인은 회사가 아닙니다. 배당금을 지급하지 않으며, 수익을 보고하지 않으며 평가에 ancorate 할 수익 배수가 없습니다. 대신, BTC는 시장에서 가장 위험 민감한 자산처럼 거래됩니다. 글로벌 유동성 조건, 달러 강도 및 실제 금리 기대에 대한 레버리지 베팅입니다. 즉, FX 거래자가 EUR/USD 또는 AUD/JPY에 위치하는 데 사용하는 동일한 거시 도구 키트가 비트코인에 직접 적용됩니다.

미국 연방준비제도 (Fed) 가 금리를 감축할 때 달러 유동성이 확대되고 위험 자산이 BTC로 상승하면 일반적으로 전세를 주도한다. CPI가 상승에 놀라면 인플레이션 헤지 내러티브가 다시 등장한다. 실제 수익률이 붕괴할 때 통화 부진에 대한 내러티브가 자본을 하드 자산 대안으로 밀어 넣는다. 이 신호들 각각은 FXMacroData의 지표 엔드포인트를 통해 사전에 관찰되며, 초에 시간표가 찍혀 있으며 깨끗한 REST API를 통해 사용할 수 있다.

이 가이드는 바이낸스에서 BTC/USDT를 위한 매크로 신호가 구동되는 알고리즘 거래 봇을 구축하는 과정을 안내합니다.

• FXMacroData에서 실시간 매크로 신호를 가져옵니다 (정책율, 인플레이션, 손익분기율, 금)
• 복합 매크로 점수를 만드는
• 릴리스 캘린더를 통해 높은 영향력 있는 릴리스를 실행할 예정
• 공식 파이썬 SDK를 통해 바이낸스에 시장 및 제한 주문을 제출합니다.

필수 조건

시작 하기 전 다음 과 같은 것 들 을 준비 해 두십시오.

• 파이썬 3.9+ 모든 스니펙트들은 표준 타이핑 문법을 사용합니다
• FXMacroData API 키 등록하세요 / 가입 그리고 대시보드에서 열쇠를 가져와
• 바이낸스 계정 펀딩된 USDT 잔액으로 바이낸스 대시보드에서 API 키를 생성합니다 스팟 및 마진 거래를 가능하게 합니다. 확인
• 파이썬 패키지 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 체제 모델의 중심입니다: 미국 달러 정책금리 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 두 번째 레벨의 출시 시간표가 스케줄링에 유용합니다 (단계 4에 포함됩니다).

단계 2: 복합 매크로 점수를 작성

단일 지표에 반응하기보다는 복합 점수는 모든 네 개의 신호를 -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: 바이낸스와 연결하여 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)

단계 7: 위험 관리 및 운영 고려 사항

매크로 신호 전략은 순수 기술 시스템보다 훨씬 낮은 거래 빈도를 가지고 있습니다. 입출출은 일반적으로 1 년 6~8 개의 주요 지표 출시 정도에서 발생합니다. 그 낮은 빈도는 버그가 아닌 특징입니다. 당신은 일내 소음이 아닌 여러 주 동안의 체제 변화에 위치하고 있습니다. 그러나 그것은 또한 각 포지션이 거래 당 더 많은 위험을 가지고 있다는 것을 의미합니다. 규율적인 위험 관리가 협상 불가능합니다.

한 가지 추가 고려 사항: BTC는 24/7 거래되지만 거시 이벤트가 예정되어 있습니다. FOMC, CPI 및 NFP 발표는 모두 미국 시장 시간대에 발생합니다. 스케줄러는 시간대 인식이 있어야합니다. announcement_datetime 필드는 항상 UTC입니다.

전략의 확장

이 프레임워크는 의도적으로 모듈화되어 있습니다. 다음은 자연적인 확장입니다.

• COT 위치 데이터를 추가합니다. FXMacroData의 CFTC COT 엔드포인트는 주간 USD 선물에 대한 투기적 포지셔닝을 제공합니다. 극한 USD 짧은 포지시셔닝은 역사적으로 BTC의 후풍입니다. /cot/usd 매크로 점수에 순위 위치 용어를 추가합니다.
• 다화폐 점수 글로벌 유동성 점수를 구성하기 위해 EUR, JPY 및 GBP 거시 신호를 통합합니다. 여러 G10 중앙 은행이 동시에 완화 모드에있을 때 BTC 위험 조건이 가장 강합니다.
• 동점 추세 속도 부진률의 수준보다는 4주간 변화율을 사용하십시오. 부진율의 급격한 상승은 수준 자체보다 더 적절한 초기 신호입니다.
• 릴리스 캘린더 통합 에 문의 발매 일정은 한 달 거리에 있는 모든 USD 이벤트에 대해, 영향을 미치는 무게에 따라 그룹화하고, 48시간 이내에 여러 개의 큰 영향을 미치는 방출이 그룹화되는 창문을 식별합니다.

요약

이제 FXMacroData 지표 데이터를 직접 바이낸스 실행에 연결하는 작동하는 매크로 신호가 구동되는 비트코인 거래 봇이 있습니다. 전략은 복합 매크로의 점수를 구성하기 위해 USD 정책율, CPI, 동점 인플레이션 및 금을 읽고 실제 세계 발표 이벤트에 따라 스케줄을 설정하고, 크기와 BTC/USDT 주문을 제출합니다.

이 시리즈의 다음 기사는 이 프레임워크를 역사적 데이터에 대한 역 테스트와 실제 FXMacroData 시간 시리즈를 사용하여 점수 가중치를 캘리브레이트하는 것을 다루고 있습니다.