MT5×PythonでFXドル円の自動売買を作る！移動平均クロス戦略のバックテストから実装まで【入門】

ドル円は毎日動きがあって、子供が生まれる前は毎朝チャートを確認していたんですが、最近は全くできていません。。。先週も朝起きたら想定外の円高になっていて、ポジションが含み損のまま放置されていました。「せめて自動でエントリー・決済の判断だけでもしてくれないか」と思い、MT5とPythonを組み合わせた自動売買の勉強を始めました。

なぜMT5×Pythonなのか
環境準備
MT5からドル円のデータを取得する
移動平均クロス戦略を実装する
簡易バックテストで戦略を検証する
まとめ

なぜMT5×Pythonなのか

FXの自動売買といえばMT4のEA（Expert Advisor）が有名ですが、PythonでMT5を操作できるMetaTrader5ライブラリが公式に提供されています。MQL5（EA言語）を覚えなくてもPythonから直接MT5を操作できるので、すでにPythonを使っている人には入りやすいです。

ただし注意点：MT5はWindowsアプリです。MacやLinuxで動かすにはワークアラウンドが必要なので、Windowsユーザー向けの内容になります。

環境準備

まずMT5をインストールして口座に接続します（XMやOANDAなど国内外の対応FX業者）。次にPythonライブラリをインストールします。

pip install MetaTrader5 pandas numpy matplotlib

MT5からドル円のデータを取得する

import MetaTrader5 as mt5
import pandas as pd
from datetime import datetime, timezone

# MT5初期化・接続
if not mt5.initialize():
    print(f"MT5初期化失敗: {mt5.last_error()}")
    quit()

print(f"MT5バージョン: {mt5.version()}")

# ドル円（USDJPY）の1時間足データを過去500本取得
rates = mt5.copy_rates_from_pos("USDJPY", mt5.TIMEFRAME_H1, 0, 500)

if rates is None:
    print(f"データ取得失敗: {mt5.last_error()}")
    mt5.shutdown()
    quit()

# DataFrameに変換
df = pd.DataFrame(rates)
df["time"] = pd.to_datetime(df["time"], unit="s", utc=True)
df = df.set_index("time")

print(df[["open", "high", "low", "close", "tick_volume"]].tail(5))
mt5.shutdown()

移動平均クロス戦略を実装する

まずはシンプルな戦略から始めます。短期EMA（20期間）が長期EMA（50期間）を上抜けたら買い、下抜けたら売りという古典的なクロス戦略です。シンプルすぎると思われるかもしれませんが、ドル円の1時間足ではトレンドが出やすい時間帯に効果的です。

import MetaTrader5 as mt5
import pandas as pd
import numpy as np

def get_usdjpy_data(bars: int = 500) -> pd.DataFrame:
    """ドル円1時間足データ取得"""
    if not mt5.initialize():
        raise RuntimeError(f"MT5初期化失敗: {mt5.last_error()}")
    
    rates = mt5.copy_rates_from_pos("USDJPY", mt5.TIMEFRAME_H1, 0, bars)
    mt5.shutdown()
    
    df = pd.DataFrame(rates)
    df["time"] = pd.to_datetime(df["time"], unit="s", utc=True)
    df = df.set_index("time")
    return df

def add_ema_signals(df: pd.DataFrame, fast: int = 20, slow: int = 50) -> pd.DataFrame:
    """EMAクロスのシグナルを追加"""
    df = df.copy()
    df[f"EMA{fast}"] = df["close"].ewm(span=fast, adjust=False).mean()
    df[f"EMA{slow}"] = df["close"].ewm(span=slow, adjust=False).mean()
    
    df["prev_fast"] = df[f"EMA{fast}"].shift(1)
    df["prev_slow"] = df[f"EMA{slow}"].shift(1)
    
    # ゴールデンクロス（買い）
    df["buy"] = (
        (df["prev_fast"] < df["prev_slow"]) &
        (df[f"EMA{fast}"] > df[f"EMA{slow}"])
    )
    
    # デッドクロス（売り）
    df["sell"] = (
        (df["prev_fast"] > df["prev_slow"]) &
        (df[f"EMA{fast}"] < df[f"EMA{slow}"])
    )
    
    return df

# データ取得とシグナル追加
df = get_usdjpy_data(500)
df = add_ema_signals(df)

# 最新シグナル確認
latest = df.iloc[-1]
print("=== 最新ドル円シグナル ===")
print(f"現在値: {latest['close']:.3f}")
print(f"EMA20: {latest['EMA20']:.3f}")
print(f"EMA50: {latest['EMA50']:.3f}")
if latest["buy"]:
    print("🟢 買いシグナル（ゴールデンクロス）")
elif latest["sell"]:
    print("🔴 売りシグナル（デッドクロス）")
else:
    print("⬜ シグナルなし")

簡易バックテストで戦略を検証する

実際にエントリーする前に必ずバックテストをしましょう。過去データでどのくらい機能しているかを確認します。

def simple_backtest(df: pd.DataFrame, lot_size: float = 0.1) -> dict:
    """
    シンプルなバックテスト（固定ロット、次足始値でエントリー）
    """
    position = 0  # 1: 買い保有, -1: 売り保有, 0: ノーポジ
    entry_price = 0.0
    trades = []
    
    for i in range(1, len(df)):
        row = df.iloc[i]
        prev = df.iloc[i-1]
        
        # 買いシグナルでエントリー
        if prev["buy"] and position == 0:
            position = 1
            entry_price = row["open"]
        
        # 売りシグナルで決済 or 逆エントリー
        elif prev["sell"] and position == 1:
            profit = (row["open"] - entry_price) * 100 * lot_size  # pip換算
            trades.append({"type": "buy", "profit_jpy": profit * 100})
            position = 0
    
    if not trades:
        return {"total_trades": 0, "total_profit": 0}
    
    trade_df = pd.DataFrame(trades)
    return {
        "total_trades": len(trade_df),
        "win_rate": (trade_df["profit_jpy"] > 0).mean() * 100,
        "total_profit": trade_df["profit_jpy"].sum(),
        "avg_profit": trade_df["profit_jpy"].mean(),
    }

result = simple_backtest(df)
print("=== バックテスト結果（過去500本1時間足） ===")
for k, v in result.items():
    print(f"{k}: {v:.2f}" if isinstance(v, float) else f"{k}: {v}")