거시 신호를 활용한 알고리즘 트레이딩: FX 전략 완벽 가이드

왜 매크로 신호가 FX 알고 트레이딩의 기초가 되는가

외환율은 무작위로 움직이지 않습니다. 그들은 두 경제의 상대적 매력을 반영합니다: 어느 중앙은행이 더 빨리 강화되고, 실제 수익률이 높고, 어떤 인플레이션 체제가 구매력을 악화시키고, 결과적으로 자본이 흐르는 곳. EUR/USD, AUD/JPY 또는 GBP/USD의 모든 주요 수 주 트렌드는 이러한 기본 변동으로 거슬러 올라갈 수 있습니다. 알려진 일정으로 발표되고 공공 데이터를 통해 관찰 가능한 변동.

자율적인 거래자는 이 신호를 수동으로 처리합니다. 알고리즘적 거래자는 그것들을 코딩합니다. 이 가이드는 FXMacroData를 데이터 레이어로 사용하여 파이썬에서 완전한 매크로 신호가 구동되는 FX 전략을 구축하는 과정을 안내합니다.

• FXMacroData API를 통해 두 화폐의 정책금, 인플레이션, 고용 및 채권 수익률 차이를 가져옵니다.
• 방향 편향을 식별하기 위해 복합 매크로 레지엄 점수를 계산합니다.
• 기술적인 맥락을 제공하기 위해 FX 스팟 리스크 역사
• 높은 영향력 발매 달력 이벤트에 대한 신호 업데이트를 스케줄
• 위치 크기의 안내와 함께 긴/단기/중립 신호를 발사
• 기본 리스크 제어: 중지 손실, 크기 제한, 뉴스 창 블랙아웃을 적용

필수 조건

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

• 파이썬 3.9+ 모든 단편들은 표준 타입 주석을 사용합니다
• FXMacroData API 키 등록하세요 / 가입 그리고 계정 대시보드에서 키를 복사
• 파이썬 패키지 requests pandas numpy

pip install requests pandas numpy

환경 변수로 API 키를 저장합니다.

export FXMACRO_API_KEY="YOUR_API_KEY"

아래의 예들은 EUR/USD, 하지만 같은 패턴은 FXMacroData에서 사용할 수 있는 모든 쌍에 적용됩니다. EUR 그리고 USD 모델링을 원하는 통화에 대해

단계 1: 핵심 매크로 표시기를 가져오기

네 가지 지표가 주요한 구조적 외환 변동을 주도합니다. 중앙은행 정책금리, 소비자 가격 인플레이션, 노동 시장 건강 및 정부 채권 수익률. 거시적 제도 통화 쌍의 각 쪽에 대해

FXMacroData는 일관된 REST 엔드포인트를 통해 이 모든 것을 제공합니다: GET /api/v1/announcements/{currency}/{indicator}각 관찰은 date, a val (지표의 값) announcement_datetime 2분의 1까지 정확해서 시장이 언제 알게 되었는지 항상 정확히 알 수 있습니다.

import os
import requests
from datetime import date, timedelta

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY  = os.environ["FXMACRO_API_KEY"]


def fetch_indicator(currency: str, indicator: str, lookback_days: int = 400) -> list[dict]:
    """Fetch the most recent observations for a macro indicator."""
    start = (date.today() - timedelta(days=lookback_days)).isoformat()
    resp = requests.get(
        f"{BASE_URL}/announcements/{currency}/{indicator}",
        params={"api_key": API_KEY, "start": start},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])


# Pull the four core series for both sides of EUR/USD
usd_rate      = fetch_indicator("usd", "policy_rate")
eur_rate      = fetch_indicator("eur", "policy_rate")
usd_inflation = fetch_indicator("usd", "inflation")
eur_inflation = fetch_indicator("eur", "inflation")
usd_nfp       = fetch_indicator("usd", "non_farm_payrolls")
usd_unemp     = fetch_indicator("usd", "unemployment")
usd_yield     = fetch_indicator("usd", "gov_bond_10y")
eur_yield     = fetch_indicator("eur", "gov_bond_10y")

- 정책금리 그리고 10년 채권 수익률 이 은 이자율의 차원을 직접적으로 파악합니다. 인플레이션 이 일련의 자료들은 중앙은행이 더 긴축해야 할 것인지 아니면 감축할 여지가 있는지 여부를 보여줍니다. 농부 외의 임금 그리고 실업 미국 달러 측면의 노동 시장 그림을 완성합니다.

단계 2: 매크로 레지엄 점수를 계산합니다

체제 점수는 여러 지표를 하나의 방향 신호로 붕괴시킵니다. 여기서 접근법은 의도적으로 간단합니다: 각 통화에 대해 최신 정책 금리, 인플레이션율 및 채권 수익률을 자신의 12 개월 평균과 비교하십시오. 기본 지수가 트렌드보다 높은 통화는 강화제도트렌드 아래로 내려가는 것은 약화 체제두 점수 사이의 스프레드는 쌍의 방향 편향을 보여줍니다.

import pandas as pd
import numpy as np


def latest_val(series: list[dict]) -> float | None:
    """Return the most recent value from a sorted indicator series."""
    if not series:
        return None
    return series[-1]["val"]


def rolling_zscore(series: list[dict], window: int = 12) -> float | None:
    """Z-score of the latest value relative to the last `window` observations."""
    vals = [r["val"] for r in series if r.get("val") is not None]
    if len(vals) < 2:
        return None
    arr = np.array(vals[-window:], dtype=float)
    mu, sigma = arr.mean(), arr.std()
    if sigma == 0:
        return 0.0
    return float((arr[-1] - mu) / sigma)


def macro_score(
    policy_rate: list[dict],
    inflation:   list[dict],
    bond_yield:  list[dict],
) -> float:
    """
    Composite macro score for one currency.
    Positive → strengthening macro backdrop.
    Negative → weakening macro backdrop.
    """
    weights = {"policy_rate": 0.40, "inflation": 0.30, "bond_yield": 0.30}
    scores = {
        "policy_rate": rolling_zscore(policy_rate),
        "inflation":   rolling_zscore(inflation),
        "bond_yield":  rolling_zscore(bond_yield),
    }
    total, weight_sum = 0.0, 0.0
    for key, w in weights.items():
        z = scores[key]
        if z is not None:
            total      += w * z
            weight_sum += w
    return total / weight_sum if weight_sum > 0 else 0.0


usd_score = macro_score(usd_rate, usd_inflation, usd_yield)
eur_score = macro_score(eur_rate, eur_inflation, eur_yield)

# Positive → USD macro stronger → bias SHORT EUR/USD
# Negative → EUR macro stronger → bias LONG EUR/USD
regime_spread = usd_score - eur_score
print(f"USD macro score: {usd_score:+.3f}")
print(f"EUR macro score: {eur_score:+.3f}")
print(f"Regime spread (USD − EUR): {regime_spread:+.3f}")

단계 3: USD에 대한 노동 시장 맥락을 추가하십시오.

특히 USD 쌍의 경우, 노동 시장은 단기적으로 금리 신호를 대체한다. 파업 인턴 롤 인쇄는 인플레이션이 떨어질 때에도 FED을 중단하도록 밀어낼 수 있다. 실업의 갑작스러운 점프는 완화 기대를 가속화시킬 수 있다 . 고용 구성 요소를 포함하면 높은 영향력 데이터 창을 중심으로 USD 점수가 날카롭다.

def employment_score(nfp: list[dict], unemployment: list[dict]) -> float:
    """
    Labour market contribution to the USD score.
    Positive NFP momentum + falling unemployment → bullish.
    """
    nfp_z   = rolling_zscore(nfp)
    unemp_z = rolling_zscore(unemployment)

    if nfp_z is None and unemp_z is None:
        return 0.0
    score = 0.0
    count = 0
    if nfp_z is not None:
        score += 0.60 * nfp_z   # NFP gets more weight
        count += 1
    if unemp_z is not None:
        # Unemployment is inverse: a rising z-score is bearish for USD
        score -= 0.40 * unemp_z
        count += 1
    return score


usd_employment = employment_score(usd_nfp, usd_unemp)

# Rebuild USD score including labour market
usd_score_full = (
    0.35 * (rolling_zscore(usd_rate) or 0.0) +
    0.25 * (rolling_zscore(usd_inflation) or 0.0) +
    0.25 * (rolling_zscore(usd_yield) or 0.0) +
    0.15 * usd_employment
)

regime_spread_full = usd_score_full - eur_score
print(f"USD score (with labour): {usd_score_full:+.3f}")
print(f"Regime spread (full):    {regime_spread_full:+.3f}")

단계 4: FX 스팟 리스크 역사

매크로 레지엄 스코어는 방향성을 결정하지만, 엔트리 타이밍은 여전히 가격 기반 필터에서 이익을 얻습니다. 이 쌍이 50 일 평균보다 3 표준편차가 높을 때 강한 USD 레지먼트에서 EUR/USD 코트에 진입하는 것은 이미 추세가 낮을 때 진입한 것과 다른 위험 프로필입니다. FXMacroData 외환 종점 기본 가격 컨텍스트를 계산하는 데 사용할 수 있는 일일 폐쇄율을 제공합니다.

def fetch_spot_rates(base: str, quote: str, lookback_days: int = 200) -> pd.Series:
    """Fetch FX spot rate history and return as a date-indexed Series."""
    start = (date.today() - timedelta(days=lookback_days)).isoformat()
    resp = requests.get(
        f"{BASE_URL}/forex/{base}/{quote}",
        params={"api_key": API_KEY, "start": start},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json().get("data", [])
    if not data:
        return pd.Series(dtype=float)
    df = pd.DataFrame(data).set_index("date").sort_index()
    return df["close"].astype(float)


spot = fetch_spot_rates("EUR", "USD")
sma50  = spot.rolling(50).mean()
sma200 = spot.rolling(200).mean()

latest_price  = spot.iloc[-1]
latest_sma50  = sma50.iloc[-1]
latest_sma200 = sma200.iloc[-1]

# Simple trend filter: is price above or below key moving averages?
price_trend = "bullish" if latest_price > latest_sma50 > latest_sma200 else (
              "bearish" if latest_price < latest_sma50 < latest_sma200 else "mixed")

print(f"EUR/USD latest: {latest_price:.5f}  SMA50: {latest_sma50:.5f}  SMA200: {latest_sma200:.5f}")
print(f"Price trend: {price_trend}")

단계 5: 출시 달력에 가입하세요

오픈 FX 포지션을 보유하는 가장 위험한 시간은 주요 예정된 출시 주위 15 분입니다. 정책 금리 결정, CPI 인쇄 및 임금 기록 데이터는 모두 3080 파이프 격차의 잠재력을 가지고 있습니다. 잘 규율 된 알고는 이러한 창에 새로운 포지션에 진입하는 것을 피하고 선택적으로 기존 포지션은 닫거나 헤지 할 수 있습니다.

FXMacroData 릴리스 캘린더 엔드포인트는 모든 다가오는 예정된 릴리스를 표시자 이름과 예상 발표 날짜와 함께 반환하여 블랙아웃 스케줄러를 직접적으로 만들 수 있습니다.

from datetime import datetime, timezone

def fetch_upcoming_releases(currency: str, days_ahead: int = 14) -> list[dict]:
    """Return scheduled macro releases for a currency over the next N days."""
    resp = requests.get(
        f"{BASE_URL}/calendar/{currency}",
        params={"api_key": API_KEY},
        timeout=10,
    )
    resp.raise_for_status()
    events = resp.json().get("data", [])
    cutoff = datetime.now(timezone.utc) + timedelta(days=days_ahead)
    upcoming = []
    for evt in events:
        dt_str = evt.get("announcement_datetime") or evt.get("date")
        if not dt_str:
            continue
        try:
            evt_dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
        except ValueError:
            continue
        if datetime.now(timezone.utc) <= evt_dt <= cutoff:
            upcoming.append(evt)
    return upcoming


HIGH_IMPACT = {"policy_rate", "inflation", "non_farm_payrolls", "unemployment", "gdp"}
BLACKOUT_MINUTES = 20  # minutes before/after release to block new entries


def is_in_blackout_window(releases: list[dict], now: datetime | None = None) -> bool:
    """Return True if the current moment falls inside any high-impact release window."""
    if now is None:
        now = datetime.now(timezone.utc)
    window = timedelta(minutes=BLACKOUT_MINUTES)
    for evt in releases:
        if evt.get("indicator") not in HIGH_IMPACT:
            continue
        dt_str = evt.get("announcement_datetime") or evt.get("date")
        if not dt_str:
            continue
        try:
            evt_dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
        except ValueError:
            continue
        if abs(now - evt_dt) <= window:
            return True
    return False


usd_releases = fetch_upcoming_releases("usd")
eur_releases = fetch_upcoming_releases("eur")
all_releases = usd_releases + eur_releases

print(f"Upcoming high-impact releases (next 14 days): {len(all_releases)}")
print(f"Currently in blackout window: {is_in_blackout_window(all_releases)}")

단계 6: 실시간 신호 를 생성 한다

매크로 스코어, 스팟 리트 컨텍스트, 그리고 캘린더에 기반한 블랙아웃 체크를 통해, 당신은 그들을 하나의 신호 기능으로 집어넣을 수 있습니다. 이는 방향, 신뢰성, 그리고 요구에 따라 권장된 포지션 크기를 생성합니다.

from dataclasses import dataclass
from typing import Literal


@dataclass
class Signal:
    direction: Literal["long", "short", "neutral"]
    confidence: float        # 0.0 → 1.0
    regime_spread: float     # positive → USD stronger
    price_trend: str
    in_blackout: bool
    reason: str


REGIME_THRESHOLD = 0.45   # minimum spread magnitude to take a position
TREND_CONFIRMATION = True  # require price trend to agree with regime signal


def generate_signal(
    regime_spread: float,
    price_trend:   str,
    releases:      list[dict],
) -> Signal:
    """
    Combine macro regime spread and price trend into a trade signal.

    regime_spread > 0  → USD stronger → short EUR/USD (quote currency up)
    regime_spread < 0  → EUR stronger → long  EUR/USD (base currency up)
    """
    in_blackout = is_in_blackout_window(releases)

    if in_blackout:
        return Signal("neutral", 0.0, regime_spread, price_trend, True,
                      "Blackout window: high-impact release imminent.")

    magnitude = abs(regime_spread)

    if magnitude < REGIME_THRESHOLD:
        return Signal("neutral", 0.0, regime_spread, price_trend, False,
                      f"Regime spread {regime_spread:+.3f} below threshold {REGIME_THRESHOLD}.")

    # Determine raw macro direction
    macro_dir = "short" if regime_spread > 0 else "long"

    # Price trend confirmation
    if TREND_CONFIRMATION:
        if macro_dir == "short" and price_trend == "bullish":
            return Signal("neutral", 0.20, regime_spread, price_trend, False,
                          "Macro bearish EUR/USD but price trend still bullish — wait for confirmation.")
        if macro_dir == "long" and price_trend == "bearish":
            return Signal("neutral", 0.20, regime_spread, price_trend, False,
                          "Macro bullish EUR/USD but price trend still bearish — wait for confirmation.")

    # Confidence scales with regime magnitude (capped at 0.90)
    confidence = min(0.90, magnitude / 1.5)

    return Signal(macro_dir, confidence, regime_spread, price_trend, False,
                  f"Macro {'USD' if macro_dir == 'short' else 'EUR'} outperformance confirmed by price trend.")


signal = generate_signal(regime_spread_full, price_trend, all_releases)
print(f"Signal: {signal.direction.upper()}  confidence: {signal.confidence:.0%}")
print(f"Reason: {signal.reason}")

단계 7: 위험 통제 및 규모 포지션을 적용

신호 생성은 절반의 작업에 불과합니다. 체계적인 위험 통제 없이는 고품질 신호 소스조차도 전략이 유지할 수있는 것을 초과하는 인수를 생산 할 것입니다. 최소 세 가지 통제가 필수적입니다: 계좌 자본에 대한 최대 위치 크기, 피프에서 하드 스톱-러스 및 손실 세션의 연속 후 거래를 중단하는 일일 손실 제한.

@dataclass
class RiskConfig:
    account_equity:      float = 10_000.0   # USD
    risk_per_trade_pct:  float = 1.0        # percent of equity risked per trade
    stop_loss_pips:      float = 30.0       # maximum allowed loss in pips
    pip_value_per_lot:   float = 10.0       # USD per pip per standard lot (EUR/USD)
    max_lots:            float = 2.0        # hard cap on position size
    daily_loss_limit_pct: float = 3.0       # pause trading if daily loss exceeds this


def compute_position_size(signal: Signal, config: RiskConfig) -> float:
    """
    Return lot size based on risk per trade and stop-loss.
    Scales with signal confidence — higher confidence allows up to full risk.
    """
    if signal.direction == "neutral":
        return 0.0

    risk_amount = config.account_equity * (config.risk_per_trade_pct / 100)
    # Scale risk by confidence
    adjusted_risk = risk_amount * signal.confidence
    # Max loss per lot at this stop = stop_loss_pips * pip_value_per_lot
    max_loss_per_lot = config.stop_loss_pips * config.pip_value_per_lot
    if max_loss_per_lot == 0:
        return 0.0
    raw_lots = adjusted_risk / max_loss_per_lot
    return round(min(raw_lots, config.max_lots), 2)


risk   = RiskConfig()
lots   = compute_position_size(signal, risk)
dollar_risk = lots * risk.stop_loss_pips * risk.pip_value_per_lot

print(f"Recommended position: {lots} lots  ({signal.direction.upper()} EUR/USD)")
print(f"Max risk at {risk.stop_loss_pips}-pip stop: ${dollar_risk:.2f}")

단계 8: 모든 것 을 합쳐 놓음

마지막 단계는 실행 루프로 모든 것을 집합하여 하루에 한 번 실행하고 모든 매크로 데이터를 갱신하고 신호를 평가하고 출시 달력을 확인하고 주문 추천을 발산합니다. 라이브 환경에서 출력을 브로커 API 또는 종이 거래 시스템으로 연결합니다. 여기서 브로크어 무지정성을 유지하고 콘솔에 결정을 로그인합니다.

import logging
from datetime import date, datetime, timedelta, timezone

logging.basicConfig(level=logging.INFO, format="%(asctime)s  %(message)s")
log = logging.getLogger(__name__)


def run_daily_strategy():
    """Main strategy loop — call once per trading day."""
    log.info("─── Daily macro strategy update ───")

    # 1. Fetch macro data
    log.info("Fetching macro indicators...")
    usd_rate_data  = fetch_indicator("usd", "policy_rate")
    eur_rate_data  = fetch_indicator("eur", "policy_rate")
    usd_inf_data   = fetch_indicator("usd", "inflation")
    eur_inf_data   = fetch_indicator("eur", "inflation")
    usd_nfp_data   = fetch_indicator("usd", "non_farm_payrolls")
    usd_unemp_data = fetch_indicator("usd", "unemployment")
    usd_bond_data  = fetch_indicator("usd", "gov_bond_10y")
    eur_bond_data  = fetch_indicator("eur", "gov_bond_10y")

    # 2. Compute regime scores
    usd_emp    = employment_score(usd_nfp_data, usd_unemp_data)
    usd_s      = (
        0.35 * (rolling_zscore(usd_rate_data) or 0.0) +
        0.25 * (rolling_zscore(usd_inf_data) or 0.0) +
        0.25 * (rolling_zscore(usd_bond_data) or 0.0) +
        0.15 * usd_emp
    )
    eur_s = macro_score(eur_rate_data, eur_inf_data, eur_bond_data)
    spread = usd_s - eur_s
    log.info(f"USD score: {usd_s:+.3f}  EUR score: {eur_s:+.3f}  Spread: {spread:+.3f}")

    # 3. Fetch spot rates and compute trend
    log.info("Fetching spot rates...")
    spot_series = fetch_spot_rates("EUR", "USD")
    sma50_val   = spot_series.rolling(50).mean().iloc[-1] if len(spot_series) >= 50 else None
    sma200_val  = spot_series.rolling(200).mean().iloc[-1] if len(spot_series) >= 200 else None
    last_price  = spot_series.iloc[-1]
    trend = "mixed"
    if sma50_val and sma200_val:
        trend = ("bullish" if last_price > sma50_val > sma200_val else
                 "bearish" if last_price < sma50_val < sma200_val else "mixed")
    log.info(f"EUR/USD {last_price:.5f}  trend: {trend}")

    # 4. Fetch release calendar
    log.info("Fetching release calendars...")
    releases = fetch_upcoming_releases("usd") + fetch_upcoming_releases("eur")
    log.info(f"Upcoming events: {len(releases)}")

    # 5. Generate signal
    sig  = generate_signal(spread, trend, releases)
    lots = compute_position_size(sig, RiskConfig())
    log.info(f"Signal: {sig.direction.upper()}  confidence: {sig.confidence:.0%}  lots: {lots}")
    log.info(f"Reason: {sig.reason}")

    return sig, lots


if __name__ == "__main__":
    run_daily_strategy()

다음 단계: 전략의 확장

위의 프레임워크는 의도적으로 가늘어져서 원료 데이터에서 최종 출력까지 모든 결정을 추적 할 수 있습니다. 역사 데이터에 대한 논리를 검증 한 후에는 여러 자연 확장 프로그램이 신호 품질과 실행 안정성을 향상시킵니다.

• 더 많은 화폐를 추가합니다 같은 지표 최종 지점을 사용하여 GBP, AUD, JPY 또는 CAD로 확장됩니다. 각 지표는 새로운 쌍과 새로운 차원의 기회를 추가합니다. GBP 정책금리 그리고 AUD 인플레이션 일련은 동일한 데이터 계약을 따르고 있습니다.
• COT 위치 데이터를 추가합니다. CFTC COT 보고서에서 큰 투기자의 위치화는 유용한 감정 필터입니다. 거시 정권이 USD를 길게 말하지만 투기적인 긴 지점이 이미 극단적 인 경우 새로운 입출금의 위험 / 보상률이 낮습니다. FXMacroData는 동일한 API를 통해 COT 데이터를 제공합니다.
• 과거 발표 데이터에 대한 백테스트 FXMacroData의 모든 관찰은 announcement_datetime 초에 정확하게, 시장이 어느 시점에서 알고 있던 것을 정확히 재구성하고
• 스케줄러로 자동화 팩 run_daily_strategy() 크론 작업이나 클라우드 기능에서. 일반적인 매크로 신호는 주요 데이터 발표 후에만 업데이트되어야하므로 중장기 포지션에 대해 매일 또는 심지어 주간 업데이트가 충분합니다.
• 브로커 API에 연결 신호와 롯 크기의 출력은 중개자 무지입니다. 지도 direction 그리고 lots 원하는 실행 계층 (OANDA v20, 인터랙티브 브로커 TWS, 또는 종이 거래 시뮬레이터) 에서 시장 주문을 처리합니다.

요약

매크로 신호는 알고의 장식물이 아니라 가장자리입니다. 이 가이드는 어떻게:

1. 정책금, 인플레이션, 고용 및 채권 수익률을 일관된 패턴으로 FXMacroData API에서 가져오기
2. 각 통화에 대한 복합 체제 점수를 계산하기 위해 Z 점수 정상화와 가중된 지표 혼합을 사용
3. 입출전 전에 거시적 및 기술적 합의가 필요하기 위해 FX 스팟 가격 역사에서 가격 트렌드 필터를 추가
4. 간단한 블랙아웃 스케줄러로 높은 영향 데이터 창의 소음을 피하기 위해 릴리스 캘린더를 사용
5. 고정 분수 위험 모형을 이용한 신호 신뢰도 및 계정 자금에 비례한 지점 크기
6. 모든 구성 요소를 반복 가능한 일일 루프로 조립합니다.

전체 소스는 단 한 세션에서 적응하기에 충분히 컴팩트합니다. 데이터 계층 FXMacroData의 일관성 있게 구조화되어 있으며 시간표가 정확한 발표 시리즈 는 모든 신호를 검증 가능한 시장 사실에 근거하여 유지하는 무거운 업무를 수행합니다.