宏观信号算法交易：一份完整的FX策略指南

为什么宏观信号是外汇交易的基础

汇率不会随机移动.它们反映了两个经济体的相对吸引力:哪个央行收紧得更快,实际收益率较高,哪个通胀制度正在恶化购买力,以及资本流动的结果.每一个主要的多周趋势在欧元/美元,澳元/日元或英/美元可以追溯到这些基本面的变化 变化,这些变化是在已知的时间表上宣布的,可以通过公共数据观察到.

根据专业的要求,交易员可以手动处理这些信号.算法交易员将它们编码.本指南将通过Python中使用FXMacroData作为数据层构建一个完整的宏观信号驱动的外汇策略.到最后,您将有一个工作系统:

• 通过FXMacroData API获取两种货币的政策利率,通胀,就业和债券收益率差异
• 计算一个复合的宏观模式评分来识别方向偏差
• 查看汇率现货价格历史以提供技术背景
• 计划高影响力发布日历事件的信号更新
• 发出长/短/中性信号,并引导位置大小
• 应用基本风险控制:停止损失,尺寸限制,新闻窗口停电

预先要求

在开始之前,请确保有以下措施:

• Python 3.9+ 所有片段使用标准类型注释
• 汇率数据API键 登记在 订阅 并从帐户仪表板复制您的密钥
• Python 包没有人知道. requests没有人知道. pandas没有人知道. numpy

pip install requests pandas numpy

存储您的API密钥作为环境变量 永远不要在源文件中硬代码凭证:

export FXMACRO_API_KEY="YOUR_API_KEY"

下面的例子是贸易. 欧元/美元,但同样的模式适用于任何在FXMacroData中可用的对. EUR 现在我 USD 对于您想要模拟的货币.

步骤1:获取核心宏观指标

四个指标家族推动了大部分结构性外汇变动:央行政策利率,消费者价格通胀,劳动力市场健康状况和政府债券收益率. 宏观制度 对于货币对的每一边.

通过一个一致的REST终端点提供所有这些: GET /api/v1/announcements/{currency}/{indicator}每个观察都包含一个 date没有 val (指标的值),以及 announcement_datetime 准确到秒,所以你总是知道市场什么时候发现.

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:为美元添加劳动力市场背景

对于美元货币对,就业市场通常会在短期内取代利率信号. 爆发的工资表格打印可能会迫使美联储在通胀下降时暂停削减;失业率的意外跳跃可能会加速缓解预期. 包括就业组件,会在高影响数据窗口周围加快美元的分数.

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:查询外汇现货汇率历史

宏观模式的分数提供了方向的确定性,但进入时机仍然受益于基于价格的过器.当该货币对比在50天平均值以上3个标准偏差时,在强的美元模式中进入EUR/USD的短期,与当它已经走向下跌时进入的风险配置不同. 外汇终点 提供每日关闭率,可以用来计算基本价格背景.

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:订阅发布日历

持有开放外汇仓位最危险的时间是计划发布的主要交易周围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()

下一步:扩大战略

上面的框架是故意的精简,这样您就可以追踪从原始数据到最终输出的每一个决定. 一旦您对历史数据验证了逻辑,几个自然扩展可以提高信号质量和执行强度:

• 添加更多的货币 扩展到英,澳元,日元或CAD,使用相同的指标终点. 英政策利率 现在我 澳元通货膨胀 系列遵循相同的数据合同.
• 添加COT定位数据 从CFTC COT报告中获得的大投机者定位是一个有用的情绪过器.当宏观制度说美元长期但投机长期已经极端时,新入口的风险/回报较低.FXMacroData通过相同的API提供COT数据.
• 根据历史公告数据进行反向测试 因为每一个FXMacroData的观察都带有 announcement_datetime 您可以重建市场在任何时间点所知道的情况,
• 通过调度器自动化 包装 run_daily_strategy() 在cron工作或云函数中.典型的宏观信号只需要在主要数据发布后更新,因此每天甚至每周更新足以适用于中期仓位.
• 连接到一个经纪人API 信号和批量大小输出是无经纪人. direction 现在我 lots 在您喜欢的执行层 (OANDA v20,互动经纪人TWS或纸质交易模拟器) 中进行交易.

总结

宏信号不是一个算法的装饰,而是边缘.

1. 从FXMacroData API中获取政策利率,通胀,就业和债券收益率,并以一致的模式
2. 计算每个货币的复合制度分数,使用Z分数正常化和加权指标混合
3. 添加从外汇现货汇率历史的价格趋势过器,在进入之前需要宏观和技术协议
4. 使用发布日历以避免使用简单的黑幕调度器的高影响力数据窗口的噪音
5. 根据固定分数风险模型,对信号信任和账户资本的比例,对位规模
6. 组装所有组件成一个可重复的每日循环

整个源是足够紧的,可以在一次会议中进行调整.数据层 FXMacroData的一致结构化,时间精确的公告系列 负责将每个信号都基于可验证的市场事实.