マクロシグナルによるアルゴリズム取引：完全なFX戦略ガイド

なぜマクロシグナルがFX Algo取引の基礎になっているのか

通貨レートはランダムに動かない.それらは2つの経済の相対的な魅力を反映している:どの中央銀行がより速く引き締まるか,実質利回りが高く,どのインフレ体制が購買力を悪化させ,その結果資本が流れているか.EUR/USD,AUD/JPY,またはGBP/USDのすべての主要な多週間のトレンドは,これらの基本要素のシフトに追溯できる. 既知のスケジュールで発表され,公衆データを通して観察可能なシフト.

裁量トレーダーはこれらの信号を手動で処理する.アルゴリズムトレーダーはそれらをコード化する.このガイドは,データ層としてFXMacroDataを使用して,Pythonで完全なマクロ信号駆動FX戦略を構築する.最後に,あなたは以下の作業システムを得ます.

• FXMacroData API を使って,政策金利,インフレ,雇用,債券利回り差を2つの通貨で取得する
• 方向性バイアスを識別するために複合マクロ体制スコアを計算します
• テクニカルな文脈を提供するために,FXスポットレートの履歴を抽出します.
• 高影響のリリースカレンダーイベントの周辺のシグナル更新をスケジュールする
• 位置サイズガイド付きの長/短/中立信号を発する
• 基本的リスク管理を行います.ストップ・ロスト,サイズ制限,ニュースウィンドウのブラックアウト

条件

始める 前 に,次の こと を 備える よう に し て ください.

• Python 3.9+ すべてのスニペットは標準的なタイプ注釈を使用します
• FXMacroData API キーを 登録する / サブスクリプト 口座のダッシュボードからあなたの鍵をコピー
• Python パッケージありがとうございました requestsほら pandasほら numpy

pip install requests pandas numpy

ソースファイルにハードコード認証を決して保存する

export FXMACRO_API_KEY="YOUR_API_KEY"

取引の例は以下の通りです EUR/USD単にスワップするだけです. 交換するだけ EUR ほら USD モデル化したい通貨です

ステップ1: 基本マクロインジケーターを取得

金融政策の利率,消費者物価インフレ,労働市場の健康,国債の利回りなど,これらの指標は, マクロシステム 通貨ペアの各側について

FXMacroData は,これらのすべてのデータを一貫した 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ヶ月間の平均値と比較します.基本値がトレンドよりも高い通貨は, 強化体制傾向の下にあるものは 弱体化体制方向性偏差を示します. 偏差は,この2つの点数との間にある.

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の労働市場文脈を追加

短期的には,労働市場がレート信号を上回る. インフレが低下しているときでさえ,給与表の印刷が爆発的に増加すると,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:リスク管理とポジションのサイズを適用する

信号生成は,作業の半分に過ぎない. 体系的なリスク制御がなければ,高品質の信号源でさえ,戦略が維持できるものを上回る引き下げを生む.最低でも3つの制御が不可欠である:アカウント資本との関係で最大ポジションサイズ,ピップでのハードストップ・ロス,損失セッションの連続後に取引を一時停止する日々の損失制限.

@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,Interactive Brokers TWS,または紙取引シミュレーター) で

概要

マクロ信号は algo の装飾ではなく,エッジです.このガイドでは,

1. 政策金利,インフレ,雇用,債券の収益をFXMacroData APIから一貫したパターンで取得
2. Zスコア標準化と重度の指標の混合を用いて各通貨の複合的な制度スコアを計算する
3. 輸入前にマクロと技術的な合意を必要とする FXスポットレートの歴史から価格傾向フィルターを追加
4. リリースカレンダーを使用して,簡単なブラックアウトスケジューラーで高影響データウィンドウのノイズを回避します
5. 固定分数リスクモデルを用いたシグナル信頼性と口座資本の割合によるポジションのサイズ
6. すべてのコンポーネントを1つの繰り返す日々のループに組み込む

完全なソースは,単一のセッションで適応するのに十分なコンパクトです. データ層は,一貫して構造化され,タイムスタンプの正確な発表シリーズが,すべての信号を検証可能な市場事実に基礎を置くための重い作業です.