Trading Algorítmico con Señales Macro: Una Guía Completa de Estrategia FX

Por qué las señales macro son la base del comercio de FX Algo

Las tasas de cambio no se mueven al azar. Reflejan el atractivo relativo de dos economías: qué banco central se está ajustando más rápido, dónde los rendimientos reales son más altos, qué régimen de inflación está deteriorando el poder adquisitivo y dónde el capital fluye como consecuencia.

Los operadores discrecionales procesan estas señales manualmente. Los operadores algorítmicos las codifican. Esta guía recorre la construcción de una estrategia completa de FX impulsada por macro señales en Python utilizando FXMacroData como capa de datos. Al final, tendrá un sistema de trabajo que:

• Obtiene las tasas de interés, la inflación, el empleo y los diferencial de rendimiento de los bonos para dos monedas a través de la API FXMacroData
• Calcula una puntuación compuesta del régimen macro para identificar el sesgo direccional
• Se extrae el historial de las tasas de cambio al contado para proporcionar el contexto técnico
• Programa actualizaciones de señales en torno a eventos de calendario de liberación de alto impacto
• Emite señales largas/cortas/neutrales con orientación de tamaño de posición
• Aplica controles de riesgo básicos: stop-loss, límites de tamaño y apagones de ventanas de noticias

Los requisitos previos

Antes de empezar, asegúrese de tener las siguientes cosas:

• Python 3.9+ todos los fragmentos utilizan anotaciones de tipo estándar
• La clave de la API de FXMacroData inscribirse en / suscribirse y copia tu clave desde el panel de la cuenta
• Paquetes de Python¿ Qué ? requests¿ Qué ? pandas¿ Qué ? numpy

pip install requests pandas numpy

Almacene su clave de API como una variable de entorno nunca credenciales de código duro en archivos fuente:

export FXMACRO_API_KEY="YOUR_API_KEY"

Los ejemplos a continuación se refieren a la comercialización . El valor de la moneda de referencia, pero el mismo patrón se aplica a cualquier par disponible en FXMacroData. EUR ¿ Qué ? USD para las monedas que desea modelar.

Paso 1: Obtener los indicadores macro básicos

En el mercado de trabajo, la tasa de cambio de los tipos de cambio se sitúa en torno a la tasa del cambio de divisas de la moneda de la Unión Europea. régimen macro para cada lado de un par de divisas.

FXMacroData proporciona todos estos datos a través de un punto final REST consistente: GET /api/v1/announcements/{currency}/{indicator}Cada observación contiene un date¿ Qué ? val (valor del indicador), y un announcement_datetime exacto a la segunda así que siempre sabes exactamente cuando el mercado se enteró.

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

El tipo de interés ¿ Qué ? Rendimiento de los bonos a 10 años La tasa de interés es la dimensión de la tasa de la inflación La serie de medidas de reducción de los depósitos revela si un banco central tendrá que seguir ajustando o si tiene margen para flexibilizar. Las nóminas no agrícolas ¿ Qué ? el desempleo En el caso de los Estados miembros, el mercado de trabajo se ha convertido en un mercado de la economía.

Paso 2: Calcular el puntaje de un régimen macro

Un sistema de puntuación colapsa varios indicadores en una sola señal direccional. El enfoque aquí es deliberadamente simple: para cada moneda, comparar la última tasa de política, tasa de inflación y rendimiento de bonos con sus propios promedios de 12 meses. el régimen de refuerzo; una por debajo de su tendencia está en un régimen de debilitamientoEl diferencial entre las dos puntuaciones te da el sesgo direccional del 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}")

Paso 3: Añadir contexto del mercado laboral para el dólar estadounidense

En el caso de los pares de dólares, el mercado laboral a menudo anula la señal de tasa a corto plazo. Una impresión de nómina de desplome puede empujar a la Fed a pausar los recortes incluso cuando la inflación está cayendo; un salto sorpresa en el desempleo puede acelerar las expectativas de alivio. Incluir un componente de empleo agudiza la puntuación del dólar en torno a ventanas de datos 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}")

Paso 4: Extraer el historial de las tasas de cambio al contado

Los puntajes del régimen macro proporcionan una convicción direccional, pero el momento de entrada todavía se beneficia de un filtro basado en el precio. punto final de cambio proporciona tasas de cierre diarias que se pueden utilizar para calcular el contexto de precios básicos.

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

Paso 5: Suscríbete al calendario de lanzamientos

El momento más peligroso para mantener una posición de divisas abierta es los 15 minutos alrededor de una publicación programada importante. Las decisiones de tasas de política, las impresiones del IPC y los datos de nómina tienen el potencial de 3080 lagunas de pip. Un algoritmo bien disciplinado evita ingresar nuevas posiciones en estas ventanas y puede cerrar o cubrir las existentes.

El punto final del calendario de lanzamiento de FXMacroData devuelve cada próxima versión programada con su nombre de indicador y fecha de anuncio esperada, lo que facilita la creación de un programador de apagones:

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

Paso 6: Generar una señal en vivo

Con las puntuaciones macro, el contexto de la tasa spot y una verificación de apagón basada en el calendario, puede reunirlas en una sola función de señal que produce una dirección, confianza y tamaño de posición recomendado a pedido.

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

Paso 7: Aplicar controles de riesgos y posiciones de tamaño

La generación de señales es solo la mitad del trabajo. Sin controles de riesgo sistemáticos, incluso una fuente de señal de alta calidad producirá reducciones que exceden lo que la estrategia puede sostener. Tres controles son esenciales como mínimo: un tamaño máximo de posición en relación con el patrimonio de la cuenta, un stop-loss duro en pips y un límite de pérdida diaria que detiene la negociación después de una serie de sesiones 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}")

Paso 8: ponerlo todo en conjunto El ciclo de estrategia diaria

El paso final ensambla todo en un bucle de ejecución que se ejecuta una vez al día, actualiza todos los datos macro, evalúa la señal, verifica el calendario de lanzamiento y emite una recomendación de pedido.

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 pasos: Ampliación de la estrategia

El marco anterior es intencionalmente ligero para que pueda rastrear cada decisión desde los datos en bruto hasta la salida final.

• Añadir más monedas se extienden a GBP, AUD, JPY o CAD utilizando los mismos puntos finales del indicador. Tasa de referencia de la libra esterlina ¿ Qué ? Inflación en dólares australianos Las series siguen el mismo contrato de datos.
• Añadir datos de posicionamiento de COT el posicionamiento de grandes especuladores del informe de la CFTC COT es un filtro de sentimiento útil. Cuando el régimen macro dice largo USD pero los largos especulativos ya son extremos, el riesgo/recompensación de una nueva entrada es menor. FXMacroData proporciona datos de COT a través de la misma API.
• Prueba de retroceso con respecto a los datos de los anuncios históricos porque cada observación de FXMacroData tiene un announcement_datetime Con una precisión de segundo, puedes reconstruir exactamente lo que el mercado sabía en cualquier momento y simular entradas de estrategia sin sesgo de lookhead.
• Automatizar con un programador Envuelto run_daily_strategy() Las señales macro típicas solo necesitan actualizarse después de las principales publicaciones de datos, por lo que las actualizaciones diarias o incluso semanales son suficientes para las posiciones a mediano plazo.
• Conectar a una API de corredor las salidas de señal y tamaño de lote son independientes de los intermediarios. Mapa direction ¿ Qué ? lots para realizar órdenes de mercado en la capa de ejecución preferida (OANDA v20, Interactive Brokers TWS o un simulador de negociación en papel).

Resumen de las actividades

Las señales macro no son decoración para un algo son el borde.

1. Obtener tasas de interés, inflación, empleo y rendimientos de bonos de la API FXMacroData con un patrón consistente
2. Calcular una puntuación compuesta del régimen para cada moneda utilizando la normalización del puntaje Z y la combinación de indicadores ponderados
3. Añadir un filtro de tendencia de precios del historial de tipos de cambio al contado para requerir un acuerdo macro y técnico antes de ingresar
4. Utilice el calendario de lanzamiento para evitar el ruido de las ventanas de datos de alto impacto con un simple programador de apagones
5. En el caso de las posiciones de riesgo de riesgo fijo, el valor de las mismas se determinará por el valor razonable de las operaciones de riesgo.
6. Reúne todos los componentes en un solo bucle diario repetible

La fuente completa es lo suficientemente compacta como para adaptarse en una sola sesión. La capa de datos La serie de anuncios de marca horaria y precisa de FXMacroData hace el trabajo pesado de mantener cada señal basada en hechos de mercado verificables.