【Python】ドル円 移動平均ゴールデンクロス自動売買シグナル実装|1時間足で自動アラート化

Uncategorized

子供が寝た後のドル円チャート確認、それすら無理になったので自動化した

ドル円は毎日動きがあるから、以前は子供が寝てから30分だけチャートを確認する習慣があったんですが、最近それすらできなくて。。。夜中に「あ、今日ドル円どうだったっけ」と思っても疲れて画面開けない日が続きました。

もう開き直って「Pythonにゴールデンクロスを検知させてアラートを出してもらおう」と思い立ちました。自分で毎日見なくてもシグナルが出たときだけ通知が来ればいい。。。という完全な省力化作戦です。この仕組みを作る過程で移動平均クロス戦略の実装をちゃんと理解できたので、コードと一緒にまとめます。

ゴールデンクロス・デッドクロスとは

移動平均線を2本用意して、短期線が長期線を下から上に突き抜けたときが「ゴールデンクロス(買いシグナル)」、上から下に突き抜けたときが「デッドクロス(売りシグナル)」です。

FXドル円ではよく使われる組み合わせは以下のとおりです:

  • 短期スキャルピング:5日 ÷ 20日
  • スイングトレード:25日 ÷ 75日
  • トレンドフォロー:50日 ÷ 200日

今回は1時間足データを使った短〜中期のアプローチで実装します。

PythonでドルYen移動平均クロス検知を実装する

環境準備

pip install yfinance pandas numpy

データ取得と移動平均の計算

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def get_usdjpy_data(period="60d", interval="1h"):
    """
    ドル円の価格データを取得する
    interval: 1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo
    """
    ticker = yf.Ticker("USDJPY=X")
    df = ticker.history(period=period, interval=interval)
    df = df[["Open", "High", "Low", "Close", "Volume"]].copy()
    df.index = df.index.tz_localize(None)  # タイムゾーン除去
    return df

df = get_usdjpy_data(period="60d", interval="1h")
print(f"取得件数: {len(df)}本  最新: {df.index[-1]}  終値: {df['Close'].iloc[-1]:.3f}")

移動平均の計算とシグナル生成

def add_sma_signals(df, short_win=25, long_win=75):
    """移動平均線とクロスシグナルを追加"""
    d = df.copy()
    d[f"SMA_{short_win}"] = d["Close"].rolling(short_win).mean()
    d[f"SMA_{long_win}"]  = d["Close"].rolling(long_win).mean()

    # ポジション:1=ロング、-1=ショート
    d["position"] = np.where(
        d[f"SMA_{short_win}"] > d[f"SMA_{long_win}"], 1, -1
    )

    # シグナル:前の足からポジションが変わった瞬間を検出
    d["signal"] = d["position"].diff()
    # +2: ゴールデンクロス(売→買)
    # -2: デッドクロス(買→売)

    return d

df = add_sma_signals(df, short_win=25, long_win=75)

# 直近のシグナルを確認
recent_signals = df[df["signal"].abs() >= 2].tail(5)
print("\n直近のクロスシグナル:")
print(recent_signals[["Close", "SMA_25", "SMA_75", "signal"]])

最新シグナルの状態確認

def check_latest_signal(df):
    """最新のシグナル状況をわかりやすく出力"""
    latest = df.iloc[-1]
    prev   = df.iloc[-2]

    short_col = [c for c in df.columns if c.startswith("SMA_") and int(c.split("_")[1]) < 50][0]
    long_col  = [c for c in df.columns if c.startswith("SMA_") and int(c.split("_")[1]) >= 50][0]

    print("=" * 45)
    print(f"【ドル円 移動平均シグナル確認】")
    print(f"時刻     : {df.index[-1].strftime('%Y-%m-%d %H:%M')}")
    print(f"現在値   : {latest['Close']:.3f} 円")
    print(f"{short_col:9}: {latest[short_col]:.3f}  (前足: {prev[short_col]:.3f})")
    print(f"{long_col:9}: {latest[long_col]:.3f}  (前足: {prev[long_col]:.3f})")
    print("-" * 45)

    if latest["signal"] >= 2:
        print("🟢 ゴールデンクロス発生! → 買いシグナル")
    elif latest["signal"] <= -2:
        print("🔴 デッドクロス発生! → 売りシグナル")
    elif latest["position"] == 1:
        diff = latest[short_col] - latest[long_col]
        print(f"📊 現在ポジション: ロング(短期 > 長期 差={diff:+.3f})")
    else:
        diff = latest[short_col] - latest[long_col]
        print(f"📊 現在ポジション: ショート(短期 < 長期 差={diff:+.3f})")

    print("=" * 45)

check_latest_signal(df)

簡易バックテストで戦略を確認

def simple_backtest(df):
    """クロス戦略の簡易バックテスト"""
    d = df.dropna().copy()
    d["ret"]   = d["Close"].pct_change()
    d["strat"] = d["position"].shift(1) * d["ret"]

    total_ret  = (1 + d["strat"]).prod() - 1
    bh_ret     = (d["Close"].iloc[-1] / d["Close"].iloc[0]) - 1
    sharpe     = d["strat"].mean() / d["strat"].std() * np.sqrt(252 * 24)  # 1時間足

    print(f"\n=== バックテスト結果 ===")
    print(f"期間       : {d.index[0].date()} 〜 {d.index[-1].date()}")
    print(f"戦略リターン: {total_ret:.2%}")
    print(f"BH リターン : {bh_ret:.2%}")
    print(f"シャープ比 : {sharpe:.2f}")

simple_backtest(df)

スケジューラで定期実行(自動アラート化)

import schedule
import time

def run_check():
    """定期実行する関数"""
    df = get_usdjpy_data(period="60d", interval="1h")
    df = add_sma_signals(df)
    check_latest_signal(df)

    # シグナル発生時にログ記録
    latest = df.iloc[-1]
    if abs(latest["signal"]) >= 2:
        signal_type = "GOLDEN_CROSS" if latest["signal"] > 0 else "DEAD_CROSS"
        with open("usdjpy_signals.log", "a") as f:
            f.write(f"{df.index[-1]},{signal_type},{latest['Close']:.3f}\n")
        print(f"シグナルをログに記録しました: {signal_type}")

# 1時間ごとに実行
schedule.every(1).hours.do(run_check)

print("監視開始。Ctrl+Cで停止。")
run_check()  # 起動時に即時実行

while True:
    schedule.run_pending()
    time.sleep(60)

まとめ:まずシグナルを「見える化」することから始めてみてください

移動平均クロス戦略そのものは古典的で、単体では勝てない場合も多いです。でも「自分がチャートを見られない時間帯に何が起きているか」を記録する仕組みとして使うだけでも価値があります。

僕はこのスクリプトをラズパイで動かして、シグナルが出たら翌朝確認するようにしました。100%信じるのではなく「判断のきっかけ」として使うのがいいと思います。次のステップとしては、RSIやBBとの組み合わせでフィルタリングを試してみようと思っています。まず動かしてみてから考えるスタイルで、一緒に改善していきましょう。

タイトルとURLをコピーしました