FXMacroData 거시 신호를 활용한 Coinbase 비트코인 알고리즘 트레이딩

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

비트코인은 기술 주식과 덜 비슷하고 거시적 자산과 더 비슷하게 거래됩니다. 이산화나 수익을 보고하지 않으며 가치를 고정시키는 현금 흐름 모델이 없습니다. 대신 BTC는 글로벌 유동성 조건, 실제 금리 기대, 달러 강도와 함께 움직입니다. EUR/USD와 AUD/JPY를 움직이는 것과 같은 힘입니다. 즉 FX 트레이더를 위해 구축된 거시 도구 키트는 비트코인에 직접 적용됩니다.

미국 연방준비은행이 완화할 때 달러 유동성은 리스크 시장을 홍수하고 BTC는 비용을 주도하는 경향이 있다. CPI가 상승에 놀라면 통화 부진 서술은 자본을 하드 자산으로 밀어 넣는다. 평형 인플레이션이 상승할 때 실수익은 압축되고 BTC와 함께 금이 상승한다. 이 신호들 각각은 FXMacroData의 지표 엔드포인트 을 통해 미리 관찰할 수 있으며, 순수한 REST API를 통해 사용할 수 있다.

이 가이드는 BTC-USD를 위한 거시 신호가 구동되는 알고리즘 거래 봇을 구축합니다. 코인베이스 고급 무역. 끝으로 당신은 작동하는 파이썬 전략을 가지고 있습니다:

• FXMacroData에서 USD 정책금리, 인플레이션, 동점점 인플레이션을 가져옵니다.
• 복합적인 매크로 레지엄 점수를 합쳐
• FXMacroData 릴리스 캘린더를 통해 높은 영향력 있는 매크로 릴리스를 중심으로 실행할 예정
• 코인베이스에서 BTC-USD 시장 주문을 공식으로 제출합니다. coinbase-advanced-py SDK

필수 조건

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

• 파이썬 3.10+ 모든 스니펙트들은 현대적인 타이핑 문법을 사용합니다
• FXMacroData API 키 등록하세요 / 가입 그리고 계정 대시보드에서 열쇠를 가져
• 코인베이스 고급 거래 계좌 을 만들어 클라우드 API 거래 키 (API 키 이름 + EC 사설 키) 개발자 플랫폼 콘솔과 함께 무역 권한이 활성화
• 파이썬 패키지 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 체제 모델을 어 둡니다. 미국 달러 정책금리 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에 포함 된 스케줄링에 유용합니다. 상품의 경우, val 즉석 가격입니다.

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

단일 지표에 반응하기보다는 복합 점수는 모든 네 개의 신호를 -1 (위험-아프, BTC 하락) 과 +1 (위기-온, BTC 상승) 사이의 하나의 방향 숫자로 합성합니다. 각 구성 요소는 현재 판독이 역사적으로 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: 코인베이스 고급 무역 API에 연결

공무원 coinbase-advanced-py 라이브러리는 코인베이스의 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: 크기 위치 및 코인베이스에서 주문을 제출

코인베이스 고급 거래 주문은 바이낸스와 다르게 작동합니다: 구매 명령에 명시된 quote_size (USD 지출 금액) 팔아 명령에 명시된 base_size (BTC 판매 금액) 아래 크기의 함수는 매크로 점수의 절대 크기와 USD 할당을 스케일합니다.

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 인쇄물에 의해 주도되는 체제 변경이 드물게 거래됩니다. 일반적으로 1 년에 610 개의 실행 가능한 신호를 생산합니다. 그 낮은 빈도는 설계에 의해됩니다: 당신은 하루 내 소음이 아닌 여러 주 방향 움직임에 위치하고 있습니다. 그러나 각 위치에는 의미있는 위험이 있습니다. 따라서 규율적인 통제가 필수적입니다.

FXMacroData의 모든 시간표는 announcement_datetime 일기 종료점 에서 표시자 응답 및 출시 날짜 시간은 UTC입니다. 일정 논리를 전체적으로 UTC로 유지하고 표시 목적으로만 변환하십시오. 이것은 US 데이터 릴리스 주변의 낮 절약 모호성을 제거합니다. 이것은 스케줄러 버그의 일반적인 원천입니다.

전략의 확장

프레임워크는 모듈형입니다. 다음으로 탐구 할 가치가있는 여러 가지 자연 확장:

• COT 위치 데이터를 추가합니다. FXMacroData의 CFTC COT 엔드포인트는 주간 USD 선물에 대한 투기적 포지셔닝을 제공합니다. 극한 USD 짧은 포지시셔닝은 역사적으로 BTC에 대한 바람이 불었습니다. /cot/usd 매크로 점수에 순위 위치 용어를 추가합니다.
• 다화폐 유동성 지수 USD와 함께 EUR, JPY 및 GBP 거시 신호를 통합합니다. 여러 G10 중앙 은행이 동시에 완화 할 때 BTC와 같은 위험 자산에 대한 글로벌 유동성 조건이 가장 유리한 것입니다.
• 동점 속도 신호 부진률의 수준보다는 4주간 변화율을 사용하십시오. 부진율의 급격한 상승은 수준 자체보다 더 일찍 그리고 더 적절한 신호입니다.
• 제한 명령 실행 교체 market_order_buy/market_order_sell 랑 limit_order_gtc_buy/limit_order_gtc_sell 전체 구매자 스프레드를 지불하지 않도록 client.get_best_bid_ask 그리고 가장 좋은 가격/가상 가격 안에 한 틱을 넣습니다.
• 발매 일정에 대한 그룹화 에 문의 발매 일정은 모든 USD 이벤트에 한 달 거리를 두고 48시간 이내에 여러 개의 고효과 방출이 집합되는 창을 식별합니다.

요약

이제 코인베이스 고급 무역에 연결된 작동하는 거시 신호가 구동되는 비트코인 거래 봇이 있습니다. 전략은 FXMacroData에서 USD 정책율, CPI, 동점 인플레이션 및 금을 읽고 복합 체제 점수를 구성하고 실제 세계 거시 발표 이벤트에 따라 스케줄을 설정하고 체제 확신에 비례하여 BTC-USD 시장 주문을 제출합니다.

거시적 접근 방식은 드물게 거래되고 높은 확신으로 거래됩니다. 이것은 정권 기반 전략과 소음을 쫓는 기술적 시스템을 구분하는 것입니다. 다음 논리적 단계는 여러 FED 주기에 걸쳐 점수 가중을 캘리브레이트하고 마이너 다운을 측정하기 위해 역사적인 FXMacroData 시간 시리즈에 대한이 프레임워크를 역 테스트하는 것입니다..