Negociação Algorítmica com Sinais Macroeconômicos: Um Guia Completo de Estratégia FX

Por que os sinais macro são a base do FX Algo Trading

As taxas de câmbio não se movem aleatoriamente. Eles refletem a atratividade relativa de duas economias: qual banco central está apertando mais rápido, onde os rendimentos reais são mais altos, qual regime de inflação está deteriorando o poder de compra e onde o capital está fluindo como consequência.

Os traders discricionários processam esses sinais manualmente. Os trader algoritmicos os codificam. Este guia percorre a construção de uma estratégia FX completa orientada por macro-sinal em Python usando FXMacroData como camada de dados. No final, você terá um sistema de trabalho que:

• Obtém taxas de juro, inflação, emprego e diferenças de rendimento de títulos para duas moedas através da API FXMacroData
• Calcula uma pontuação composta do macro regime para identificar o viés direcional
• Retira o histórico da taxa de câmbio a vista para fornecer contexto técnico
• Agendar atualizações de sinais em torno de eventos de calendário de lançamento de alto impacto
• Emite sinais longos/cortos/neutrais com orientação do tamanho da posição
• Aplica controles de risco básicos: stop-loss, limites de tamanho e apagões de janela de notícias

Requisitos

Antes de começar, certifique-se de ter o seguinte:

• Python 3.9+ todos os trechos usam anotações de tipo padrão
• Chave da API do FXMacroData Inscreva-se em / subscrever e copiar a sua chave do painel da conta
• Pacotes Python- Não . requests- Não . pandas- Não . numpy

pip install requests pandas numpy

Armazenar sua chave de API como uma variável de ambiente nunca credenciais de código rígido em arquivos de origem:

export FXMACRO_API_KEY="YOUR_API_KEY"

Os exemplos abaixo são comerciais EUR/USD, mas o mesmo padrão aplica-se a qualquer par disponível no FXMacroData. EUR E ... USD para as moedas que você quer modelar.

Passo 1: Obter os indicadores macro principais

Quatro famílias de indicadores conduzem a maioria dos movimentos estruturais das taxas de câmbio: as taxas de juro dos bancos centrais, a inflação dos preços ao consumidor, a saúde do mercado de trabalho e os rendimentos dos títulos do governo. regime macro para cada lado de um par de moedas.

O FXMacroData fornece todos estes dados através de um endpoint REST consistente: GET /api/v1/announcements/{currency}/{indicator}Cada observação contém um date- Não . val (valor do indicador), e um announcement_datetime Precisos até o segundo para que você sempre saiba exatamente quando o mercado descobriu.

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")

O ... Taxa de política E ... Rendimento das obrigações a 10 anos A dimensão das taxas de juro é capturada directamente. inflação A Comissão Europeia e o Conselho de Ministros da União Europeia (CE) Lista de pagamentos não agrícolas E ... desemprego A evolução do mercado de trabalho no lado do dólar americano.

Passo 2: Calcular uma pontuação do macro regime

A avaliação do regime colapsa vários indicadores em um único sinal direcional. A abordagem aqui é deliberadamente simples: para cada moeda, compare a última taxa de política, taxa de inflação e rendimento de títulos com suas próprias médias de 12 meses. regime de reforço; uma abaixo da sua tendência está em um regime de enfraquecimentoO spread entre os dois resultados dá-lhe o viés direcional do par.

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}")

Passo 3: Adicionar contexto do mercado de trabalho para USD

Para os pares de USD especificamente, o mercado de trabalho muitas vezes anula o sinal de taxa no curto prazo. Uma impressão de folha de pagamento pode empurrar o Fed para pausar cortes mesmo quando a inflação está caindo; um salto surpresa no desemprego pode acelerar as expectativas de alívio. Incluindo um componente de emprego aguda a pontuação do USD em torno de janelas de dados de alto impacto.

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}")

Passo 4: Extrair o histórico da taxa de câmbio spot

Os escores do regime macro fornecem convicção direcional, mas o tempo de entrada ainda se beneficia de um filtro baseado no preço. Ponto final de câmbio fornece taxas de fechamento diárias que você pode usar para calcular o contexto de preço básico.

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}")

Passo 5: Inscreva-se no calendário de lançamentos

O tempo mais perigoso para manter uma posição aberta de FX é os 15 minutos em torno de um grande lançamento programado. decisões de taxa de política, impressões do IPC e dados de folha de pagamento todos carregam o potencial de 3080 lacunas de pip. um algo bem disciplinado evita entrar em novas posições nessas janelas e pode fechar ou cobrir as existentes.

O ponto final do calendário de lançamento do FXMacroData retorna cada lançamentos programados próximos com o nome do indicador e a data de anúncio esperada, tornando mais fácil criar um agendador de apagão:

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)}")

Passo 6: Gerencie um sinal vivo

Com as pontuações macro, o contexto da taxa de câmbio, e uma verificação de apagão de calendário, você pode montá-los em uma única função de sinal que produz uma direção, confiança e tamanho de posição recomendado sob demanda.

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}")

Passo 7: Aplicar os controles de risco e as posições de tamanho

A geração de sinal é apenas metade do trabalho. Sem controles de risco sistemáticos, mesmo uma fonte de sinal de alta qualidade produzirá drawdowns que excedem o que a estratégia pode sustentar. Três controles são essenciais no mínimo: um tamanho máximo da posição em relação ao patrimônio da conta, um stop-loss duro em pips e um limite de perda diário que suspende a negociação após uma série de sessões perdedoras.

@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}")

Passo 8: Colocando tudo em conjunto O ciclo da estratégia diária

O passo final é montar tudo em um loop de execução que é executado uma vez por dia, atualiza todos os dados macro, avalia o sinal, verifica o calendário de lançamento e emite uma recomendação de ordem.

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()

Próximos passos: alargar a estratégia

A estrutura acima é intencionalmente magra para que você possa rastrear cada decisão dos dados brutos até a saída final.

• Adicionar mais moedas se estenderem para GBP, AUD, JPY ou CAD utilizando os mesmos endpoints do indicador. Taxa de juro da GBP E ... A inflação do AUD As séries seguem o mesmo contrato de dados.
• Adicionar dados de posicionamento COT posicionamento de grandes especuladores do relatório CFTC COT é um filtro de sentimento útil. Quando o regime macro diz longo USD, mas os longs especulativos já são extremos, o risco/recompensa de uma nova entrada é menor.
• Reverso em relação aos dados de anúncio históricos porque cada observação do FXMacroData carrega um announcement_datetime com precisão ao segundo, pode reconstruir exatamente o que o mercado sabia em qualquer momento e simular entradas de estratégia sem viés de visão.
• Automatizar com um agendador - Envolva . run_daily_strategy() Os sinais macro típicos só precisam ser atualizados após grandes lançamentos de dados, portanto, atualizações diárias ou mesmo semanais são suficientes para posições de médio prazo.
• Conectar-se a uma API de corretor as saídas de sinal e tamanho de lote são agnósticas para corretores. direction E ... lots para executar ordens de mercado na sua camada de execução preferida (OANDA v20, Interactive Brokers TWS ou um simulador de negociação em papel).

Resumo

Os sinais macro não são decoração para um algo são a borda.

1. Obter taxas de juros, inflação, emprego e rendimentos de títulos da API FXMacroData com um padrão consistente
2. Calcular uma pontuação do regime composto para cada moeda utilizando a normalização da pontuações Z e a combinação de indicadores ponderados
3. Adicionar um filtro de tendência de preços do histórico de taxas de câmbio a vista para exigir acordo macro e técnico antes de entrar
4. Use o calendário de lançamento para evitar o ruído de janelas de dados de alto impacto com um simples agendador de apagão
5. O valor das posições proporcional à confiança do sinal e ao património da conta utilizando um modelo de risco fixo fracionário
6. Montar todos os componentes em um único loop diário repetível

A camada de dados A série de anúncios de marca de tempo e precisão da FXMacroData faz o trabalho pesado de manter cada sinal baseado em fatos de mercado verificáveis.