Web Lukes
📄

【Python】移動平均線クロスボットを作る｜pandasでテクニカル分析

公開日: 2025/12/13更新日: 2025/12/13
#Python#pandas#移動平均線#FX#自動売買#GMOコイン

前回はPythonでFX APIを叩く基本を学びました。

今回は移動平均線クロス戦略をPythonで実装します。MQL4で作ったEAと同じロジックを、Pythonで再現してみましょう。

今回作るもの

  • ローソク足データを取得
  • pandasで移動平均を計算
  • ゴールデンクロス/デッドクロスを検出
  • 自動で売買注文を出す

追加ライブラリのインストール

Bash
pip install pandas numpy

ローソク足データの取得

GMOコインのPublic APIでローソク足を取得します。

Python
import requests
import pandas as pd
from datetime import datetime, timedelta


def get_klines(symbol="USD_JPY", interval="1hour", count=100):
    """
    ローソク足データを取得


    Args:
        symbol: 通貨ペア
        interval: 時間足（"1min", "5min", "15min", "30min", "1hour", "4hour", "1day"）
        count: 取得本数
    """
    url = "https://forex-api.coin.z.com/public/v1/klines"
    params = {
        "symbol": symbol,
        "interval": interval,
        "date": datetime.now().strftime("%Y%m%d")
    }


    response = requests.get(url, params=params)
    data = response.json()


    if data["status"] == 0:
        klines = data["data"]


        # DataFrameに変換
        df = pd.DataFrame(klines)
        df["openTime"] = pd.to_datetime(df["openTime"])
        df["open"] = df["open"].astype(float)
        df["high"] = df["high"].astype(float)
        df["low"] = df["low"].astype(float)
        df["close"] = df["close"].astype(float)


        # 最新のcount本だけ取得
        df = df.tail(count).reset_index(drop=True)


        return df
    else:
        print(f"エラー: {data}")
        return None


# テスト
df = get_klines()
print(df.tail())

出力例：

Code
openTime     open     high      low    close
95 2025-12-13 06:00:00+00:00  149.450  149.520  149.400  149.510
96 2025-12-13 07:00:00+00:00  149.510  149.550  149.480  149.520
97 2025-12-13 08:00:00+00:00  149.520  149.600  149.500  149.580
98 2025-12-13 09:00:00+00:00  149.580  149.620  149.550  149.600
99 2025-12-13 10:00:00+00:00  149.600  149.650  149.590  149.640

移動平均の計算

pandasを使えば、移動平均の計算は1行です。

Python
def calculate_ma(df, fast_period=5, slow_period=20):
    """
    移動平均を計算してDataFrameに追加
    """
    df["ma_fast"] = df["close"].rolling(window=fast_period).mean()
    df["ma_slow"] = df["close"].rolling(window=slow_period).mean()
    return df


# 計算実行
df = calculate_ma(df)
print(df[["close", "ma_fast", "ma_slow"]].tail())

出力例：

Code
close   ma_fast   ma_slow
95  149.510  149.4700  149.4200
96  149.520  149.4900  149.4350
97  149.580  149.5100  149.4500
98  149.600  149.5440  149.4700
99  149.640  149.5700  149.4900

MQL4ではiMA()関数を使いましたが、Pythonではrolling().mean()で同じことができます。

クロス検出ロジック

ゴールデンクロス/デッドクロスを検出します。

Python
def detect_cross(df):
    """
    クロスを検出


    Returns:
        "golden_cross": ゴールデンクロス発生
        "dead_cross": デッドクロス発生
        None: クロスなし
    """
    if len(df) < 2:
        return None


    # 現在と1本前のMA値
    current_fast = df["ma_fast"].iloc[-1]
    current_slow = df["ma_slow"].iloc[-1]
    prev_fast = df["ma_fast"].iloc[-2]
    prev_slow = df["ma_slow"].iloc[-2]


    # NaNチェック
    if pd.isna(current_fast) or pd.isna(prev_fast):
        return None


    # ゴールデンクロス: 短期MAが長期MAを下から上に突き抜けた
    if prev_fast < prev_slow and current_fast > current_slow:
        return "golden_cross"


    # デッドクロス: 短期MAが長期MAを上から下に突き抜けた
    if prev_fast > prev_slow and current_fast < current_slow:
        return "dead_cross"


    return None


# テスト
signal = detect_cross(df)
print(f"シグナル: {signal}")

完成版：MA Cross Bot

すべてを組み合わせた完成版です。

Python
import os
import time
import hmac
import hashlib
import json
import requests
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv


load_dotenv()


# === 設定 ===
API_KEY = os.environ.get("GMO_FX_API_KEY")
API_SECRET = os.environ.get("GMO_FX_API_SECRET")
SYMBOL = "USD_JPY"
ORDER_SIZE = 100
FAST_MA_PERIOD = 5
SLOW_MA_PERIOD = 20
CHECK_INTERVAL = 60  # 秒


# === APIヘルパー関数 ===
def create_signature(timestamp, method, path, body=""):
    text = timestamp + method + path + body
    return hmac.new(
        bytes(API_SECRET.encode("ascii")),
        bytes(text.encode("ascii")),
        hashlib.sha256
    ).hexdigest()


def get_timestamp():
    return str(int(time.time() * 1000))


def private_request(method, path, body=None):
    """Private APIリクエスト"""
    timestamp = get_timestamp()
    body_str = "" if body is None else json.dumps(body)


    headers = {
        "API-KEY": API_KEY,
        "API-TIMESTAMP": timestamp,
        "API-SIGN": create_signature(timestamp, method, path, body_str),
        "Content-Type": "application/json"
    }


    url = "https://forex-api.coin.z.com/private" + path


    if method == "GET":
        response = requests.get(url, headers=headers)
    elif method == "POST":
        response = requests.post(url, headers=headers, data=body_str)


    return response.json()


# === データ取得 ===
def get_klines(symbol, interval="1hour"):
    """ローソク足データを取得"""
    url = "https://forex-api.coin.z.com/public/v1/klines"
    params = {
        "symbol": symbol,
        "interval": interval,
        "date": datetime.now().strftime("%Y%m%d")
    }


    response = requests.get(url, params=params)
    data = response.json()


    if data["status"] == 0:
        df = pd.DataFrame(data["data"])
        df["openTime"] = pd.to_datetime(df["openTime"])
        df["close"] = df["close"].astype(float)
        return df
    return None


# === テクニカル分析 ===
def calculate_ma(df, fast_period, slow_period):
    """移動平均を計算"""
    df["ma_fast"] = df["close"].rolling(window=fast_period).mean()
    df["ma_slow"] = df["close"].rolling(window=slow_period).mean()
    return df


def detect_cross(df):
    """クロスを検出"""
    if len(df) < 2:
        return None


    current_fast = df["ma_fast"].iloc[-1]
    current_slow = df["ma_slow"].iloc[-1]
    prev_fast = df["ma_fast"].iloc[-2]
    prev_slow = df["ma_slow"].iloc[-2]


    if pd.isna(current_fast) or pd.isna(prev_fast):
        return None


    if prev_fast < prev_slow and current_fast > current_slow:
        return "golden_cross"
    if prev_fast > prev_slow and current_fast < current_slow:
        return "dead_cross"


    return None


# === 取引 ===
def get_positions():
    """保有ポジションを取得"""
    result = private_request("GET", f"/v1/openPositions?symbol={SYMBOL}")
    if result["status"] == 0:
        return result["data"]["list"]
    return []


def place_order(side):
    """成行注文を出す"""
    body = {
        "symbol": SYMBOL,
        "side": side,
        "executionType": "MARKET",
        "size": str(ORDER_SIZE),
        "timeInForce": "FAK"
    }
    result = private_request("POST", "/v1/order", body)


    if result["status"] == 0:
        print(f"✅ 注文成功: {side} {ORDER_SIZE} {SYMBOL}")
        return True
    else:
        print(f"❌ 注文エラー: {result}")
        return False


def close_all_positions(side):
    """全ポジションを決済"""
    positions = get_positions()
    for pos in positions:
        if (pos["side"] == "BUY" and side == "SELL") or \
           (pos["side"] == "SELL" and side == "BUY"):
            body = {
                "symbol": SYMBOL,
                "side": side,
                "executionType": "MARKET",
                "size": pos["size"],
                "positionId": pos["positionId"],
                "timeInForce": "FAK"
            }
            private_request("POST", "/v1/closeOrder", body)
            print(f"✅ 決済: {pos['side']} {pos['size']}")


# === メインループ ===
def run_bot():
    print("=" * 50)
    print(f"🤖 Python MA Cross Bot 起動")
    print(f"   通貨ペア: {SYMBOL}")
    print(f"   短期MA: {FAST_MA_PERIOD} / 長期MA: {SLOW_MA_PERIOD}")
    print("=" * 50)


    current_position = None  # "LONG", "SHORT", or None


    while True:
        try:
            # ローソク足データを取得
            df = get_klines(SYMBOL)
            if df is None:
                print("データ取得失敗。リトライ...")
                time.sleep(10)
                continue


            # 移動平均を計算
            df = calculate_ma(df, FAST_MA_PERIOD, SLOW_MA_PERIOD)


            # クロスを検出
            signal = detect_cross(df)


            # 現在の状態を表示
            latest = df.iloc[-1]
            print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                  f"Close: {latest['close']:.3f} | "
                  f"MA{FAST_MA_PERIOD}: {latest['ma_fast']:.3f} | "
                  f"MA{SLOW_MA_PERIOD}: {latest['ma_slow']:.3f} | "
                  f"Signal: {signal or '-'}")


            # シグナルに応じて売買
            if signal == "golden_cross" and current_position != "LONG":
                print("🔔 ゴールデンクロス検出！")
                if current_position == "SHORT":
                    close_all_positions("BUY")
                place_order("BUY")
                current_position = "LONG"


            elif signal == "dead_cross" and current_position != "SHORT":
                print("🔔 デッドクロス検出！")
                if current_position == "LONG":
                    close_all_positions("SELL")
                place_order("SELL")
                current_position = "SHORT"


        except Exception as e:
            print(f"エラー: {e}")


        time.sleep(CHECK_INTERVAL)


if __name__ == "__main__":
    run_bot()

MQL4 vs Python 比較

同じ移動平均線クロス戦略を、MQL4とPythonで実装しました。

MA値の取得

MQL4:

MQL4
double fastMA = iMA(NULL, 0, 5, 0, MODE_SMA, PRICE_CLOSE, 0);

Python:

Python
df["ma_fast"] = df["close"].rolling(window=5).mean()

クロス検出

MQL4:

MQL4
if(fastMA_prev < slowMA_prev && fastMA_current > slowMA_current)

Python:

Python
if prev_fast < prev_slow and current_fast > current_slow:

注文

MQL4:

MQL4
OrderSend(Symbol(), OP_BUY, 0.01, Ask, 3, sl, tp, "comment", magic, 0, clrGreen);

Python:

Python
private_request("POST", "/v1/order", {"symbol": "USD_JPY", "side": "BUY", ...})

ロジックは同じ、書き方が違うだけ！

Pythonならではの拡張

データ分析

Python
# 過去100本分のリターンを分析
df["return"] = df["close"].pct_change()
print(f"平均リターン: {df['return'].mean():.4%}")
print(f"標準偏差: {df['return'].std():.4%}")
print(f"シャープレシオ: {df['return'].mean() / df['return'].std():.2f}")

複数時間足の分析

Python
# 1時間足と日足の両方を分析
df_1h = get_klines(SYMBOL, "1hour")
df_1d = get_klines(SYMBOL, "1day")


# 日足が上昇トレンドの時だけ、1時間足でエントリー
if df_1d["close"].iloc[-1] > df_1d["ma_slow"].iloc[-1]:
    # 1時間足のシグナルに従う
    ...

Slack通知

Python
def send_slack(message):
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    requests.post(webhook_url, json={"text": message})


# シグナル発生時に通知
if signal:
    send_slack(f"🔔 {signal} 検出！ {SYMBOL}")

次のステップ

移動平均線クロスボットができました！

次はRSIを使った逆張りボットをPythonで作ります。

📚 次に読む記事: Python RSI逆張りボットを作ろう

Pythonならpandas-taという便利なライブラリで、RSIはもちろんMACD、ボリンジャーバンドなど、あらゆるテクニカル指標が簡単に計算できます！

🎁おすすめツール・サービス

FX自動売買VPSなら50%OFF

キャンペーンで月1,790円〜。MT4/MT5プリインストール済み、自動復旧機能付き。14日間無料お試し。

シンクラウドデスクトップ for FX
ゴトー日EAを探しているならソースコード付

4年で資金4.7倍のバックテスト実績。ソースコード付きで自由にカスタマイズ可能。

ゴトー日EA Ultimate版（3,980円）

※上記リンクにはプロモーション（広告）が含まれています

🔮
MEMBERSHIP

量子相場謀略｜ホソノP

ブログでは書けない本音のトレード戦略とEA開発の裏側を公開。 データに基づいた検証で、再現性のある手法だけをお届けします。

✓EAソースコード公開
✓バックテスト生データ共有
✓リアルタイム相場分析
✓メンバー限定記事
メンバーシップに参加する※ noteアカウントで簡単登録
ポストする@hosono_p をフォロー

📚あわせて読みたい

📄

【Python FXボット入門】MT4を使わない自動売買の始め方

Pythonで直接FX取引する自動売買ボットの作り方。MT4に依存しない、自由度の高いシステム開発の第一歩。GMOコインやOANDAのAPIを使った実践的な入門ガイド。

FX自動売買2025/12/13
📄

【Python】RSI逆張りボットを作る｜pandas-taでテクニカル指標計算

Pythonとpandas-taを使ってRSI逆張り戦略のFXボットを作ります。RSIが30以下で買い、70以上で売り。テクニカル指標の計算が1行で完結！

FX自動売買2025/12/13
📄

【Python】損切り・リスク管理の自動化｜トレーリングストップの実装

FXボットのリスク管理を自動化する方法を解説。固定損切り、トレーリングストップ、ブレイクイーブン、ATRベースの動的損切りまで、Pythonで実装します。

FX自動売買2025/12/13
📄

🚫【検証】ゴトー日にテクニカル指標を混ぜるな！16フィルター検証で判明した衝撃の事実

「RSIが売られすぎだから買い！」「高ボラだからエントリー！」...それ、逆効果です。96回のゴトー日トレードに16種類のテクニカルフィルターを適用した結果、9個が悪化。アノマリーはアノマリーのまま使え！

FX自動売買2025/12/15
📄

【コピペOK】移動平均線クロスEAの作り方｜ゴールデンクロス・デッドクロスで自動売買

移動平均線のゴールデンクロス・デッドクロスで自動売買するEAを作ります。コード全文公開、コピペですぐ動く！初心者でも理解できるよう1行ずつ解説します。

FX自動売買2025/12/13
📄

🎯【衝撃】トレンドフォローは勝てる！1万本のデータが暴いた「勝率58%の秘密」

「ゴールデンクロスで買ったのに負けた...」その悩み、データで解決します！1万本の足を分析して判明した驚きの事実。120分後の勝率58.9%、そして分布の歪みが示す「優位性の正体」とは？

FX自動売買2025/12/13