Perdagangan Algoritmik dengan Sinyal Makro: Panduan Strategi FX Lengkap

Mengapa Makro Sinyal adalah dasar dari FX Algo Trading

Nilai tukar tidak bergerak secara acak. Mereka mencerminkan daya tarik relatif dari dua ekonomi: bank sentral mana yang lebih cepat, dimana hasil riil lebih tinggi, mana rezim inflasi yang memperburuk daya beli, dan di mana modal mengalir sebagai akibatnya. Setiap tren multi-minggu utama di EUR/USD, AUD/JPY, atau GBP/USD dapat ditelusuri kembali ke pergeseran dalam dasar-dasar ini pergeser yang diumumkan pada jadwal yang diketahui dan diamati melalui data publik.

Para trader discretionary memproses sinyal ini secara manual. Para trader algoritmik mengkodikannya. Panduan ini berjalan melalui membangun strategi FX yang didorong sinyal makro lengkap di Python menggunakan FXMacroData sebagai lapisan data. Pada akhirnya Anda akan memiliki sistem kerja yang:

• Mengambil suku bunga kebijakan, inflasi, pekerjaan, dan diferensial imbal hasil obligasi untuk dua mata uang melalui FXMacroData API
• Menghitung skor makro rezim komposit untuk mengidentifikasi bias arah
• Mengambil riwayat nilai spot FX untuk memberikan konteks teknis
• Jadwal pembaruan sinyal di sekitar acara kalender rilis berdampak tinggi
• Mengeluarkan sinyal panjang/pendek/netral dengan panduan ukuran posisi
• Menggunakan pengendalian risiko dasar: stop-loss, batas ukuran, dan pemadaman jendela berita

Persyaratan

Sebelum memulai, pastikan Anda memiliki hal-hal berikut:

• Python 3.9+ semua cuplikan menggunakan anotasi tipe standar
• Kunci API FXMacroData daftar di /langganan dan menyalin kunci Anda dari dasbor akun
• Paket PythonAku tidak tahu. requestsAku akan pergi. pandasAku akan pergi. numpy

pip install requests pandas numpy

Simpan kunci API Anda sebagai variabel lingkungan tidak pernah hard-code kredensial dalam file sumber:

export FXMACRO_API_KEY="YOUR_API_KEY"

Contoh di bawah ini perdagangan EUR/USD, namun pola yang sama berlaku untuk setiap pasangan yang tersedia di FXMacroData. EUR Dan USD untuk mata uang yang ingin Anda model.

Langkah 1: Dapatkan Indikator Makro Inti

Empat kelompok indikator mendorong sebagian besar pergerakan nilai tukar struktural: suku bunga kebijakan bank sentral, inflasi harga konsumen, kesehatan pasar tenaga kerja, dan imbal hasil obligasi pemerintah. sistem makro untuk setiap sisi dari pasangan mata uang.

FXMacroData menyediakan semua ini melalui titik akhir REST yang konsisten: GET /api/v1/announcements/{currency}/{indicator}Setiap pengamatan berisi dateAku akan pergi. val (nilai indikator), dan announcement_datetime akurat ke detik sehingga Anda selalu tahu persis kapan pasar tahu.

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

- Apa? suku bunga kebijakan Dan Hasil obligasi 10 tahun . . . dan . inflasi Seri ini mengungkapkan apakah bank sentral harus memperketat lebih lanjut atau memiliki ruang untuk melonggarkan. Gaji non-pertanian Dan pengangguran menyempurnakan gambaran pasar tenaga kerja untuk sisi USD.

Langkah 2: Menghitung skor makro

Sebuah skor rezim runtuh beberapa indikator ke dalam satu sinyal arah. Pendekatan di sini sengaja sederhana: untuk setiap mata uang, bandingkan tingkat kebijakan terbaru, tingkat inflasi, dan imbal hasil obligasi dengan rata-rata 12 bulan mereka sendiri. Mata uang yang mendasarnya di atas tren mereka adalah dalam kondisi yang sangat rendah. sistem penguatan; satu di bawah tren adalah di rezim melemahkanPerbedaan antara kedua skor memberikan bias arah pasangan.

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

Langkah 3: Tambahkan Konteks Pasar Tenaga Kerja untuk USD

Untuk pasangan USD secara khusus, pasar tenaga kerja sering mengesampingkan sinyal suku bunga dalam jangka pendek. cetakan gaji yang meledak dapat mendorong Fed untuk menghentikan pemotongan bahkan ketika inflasi menurun; lompatan kejutan dalam pengangguran dapat mempercepat harapan pelonggaran.

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

Langkah 4: Tarik riwayat nilai spot FX

Skor makro memberikan keyakinan arah, tetapi waktu masuk masih mendapat manfaat dari filter berbasis harga. Memasuki EUR/USD pendek dalam rejim USD yang kuat ketika pasangan 3 standar deviasi di atas rata-rata 50 hari adalah profil risiko yang berbeda dari memasuki ketika sudah tren lebih rendah. titik akhir forex memberikan harga penutupan harian yang dapat Anda gunakan untuk menghitung konteks harga dasar.

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

Langkah 5: Langganan Kalender Rilis

The most dangerous time to hold an open FX position is the 15 minutes around a major scheduled release. Policy rate decisions, CPI prints, and payroll data all carry the potential for 30–80 pip gaps. A well-disciplined algo avoids entering new positions in these windows and can optionally close or hedge existing ones.

Titik akhir kalender rilis FXMacroData mengembalikan setiap rilis terjadwal mendatang dengan nama indikator dan tanggal pengumuman yang diharapkan, sehingga mudah untuk membangun penjadwal pemadaman:

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

Langkah 6: Membuat sinyal langsung

Dengan skor makro, konteks spot rate, dan pemeriksaan blackout yang sadar kalender, Anda dapat mengumpulkan mereka menjadi satu fungsi sinyal yang menghasilkan arah, kepercayaan, dan ukuran posisi yang direkomendasikan sesuai permintaan.

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

Langkah 7: Menggunakan Kontrol Risiko dan Ukuran Posisi

Generasi sinyal hanya setengah dari pekerjaan. Tanpa kontrol risiko yang sistematis, bahkan sumber sinyal berkualitas tinggi akan menghasilkan penarikan yang melebihi apa yang dapat ditanggung strategi. Tiga kontrol sangat penting minimal: ukuran posisi maksimum relatif terhadap ekuitas akun, stop-loss keras dalam pips, dan batas kerugian harian yang menghentikan perdagangan setelah berturut-turut sesi kehilangan.

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

Langkah 8: Menggabungkan Semua Daur Strategi Harian

Langkah terakhir mengumpulkan semuanya ke dalam loop eksekusi yang berjalan sekali sehari, memperbarui semua data makro, mengevaluasi sinyal, memeriksa kalender rilis, dan mengeluarkan rekomendasi pesanan.

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

Langkah Selanjutnya: Memperluas Strategi

Kerangka kerja di atas sengaja ramping sehingga Anda dapat melacak setiap keputusan dari data mentah hingga output akhir.

• Tambahkan lebih banyak mata uang memperluas ke GBP, AUD, JPY, atau CAD menggunakan titik akhir indikator yang sama. Nilai mata uang dasar GBP Dan Inflasi AUD seri mengikuti kontrak data yang sama.
• Tambahkan data posisi COT posisi spekulan besar dari laporan CFTC COT adalah filter sentimen yang berguna. Ketika rezim makro mengatakan USD panjang tetapi panjang spekulatif sudah ekstrem, risiko / imbalan entri baru lebih rendah. FXMacroData menyediakan data COT melalui API yang sama.
• Backtest terhadap data pengumuman historis karena setiap observasi FXMacroData membawa announcement_datetime dengan akurasi detik, Anda dapat merekonstruksi persis apa yang diketahui pasar pada setiap titik waktu dan mensimulasikan entri strategi tanpa bias lookahead.
• Otomatis dengan penjadwal bungkus run_daily_strategy() sinyal makro khas hanya perlu diperbarui setelah rilis data utama, sehingga pembaruan harian atau bahkan mingguan cukup untuk posisi jangka menengah.
• Hubungkan ke API broker output sinyal dan ukuran lot tidak ada broker. direction Dan lots untuk memasarkan pesanan di lapisan eksekusi pilihan Anda (OANDA v20, Interactive Brokers TWS, atau simulator perdagangan kertas).

Ringkasan

Makro sinyal tidak dekorasi untuk algo mereka adalah tepi.

1. Dapatkan suku bunga kebijakan, inflasi, pekerjaan, dan hasil obligasi dari FXMacroData API dengan pola yang konsisten
2. Menghitung skor rezim komposit untuk setiap mata uang menggunakan normalisasi Z-score dan pencampuran indikator tertimbang
3. Tambahkan filter tren harga dari riwayat nilai spot FX untuk memerlukan kesepakatan makro dan teknis sebelum masuk
4. Gunakan kalender rilis untuk menghindari kebisingan jendela data berdampak tinggi dengan penjadwal pemadaman sederhana
5. Ukuran posisi proporsional dengan kepercayaan sinyal dan ekuitas akun menggunakan model risiko tetap-fractional
6. Mengumpulkan semua komponen ke dalam loop harian yang bisa diulang

Sumber lengkap cukup kompak untuk beradaptasi dalam satu sesi. lapisan data FXMacroData konsisten terstruktur, timestamp-precise seri pengumuman melakukan angkat berat menjaga setiap sinyal berlandaskan fakta pasar yang dapat diverifikasi.