PythonでRSI×MACD組み合わせ戦略を実装してバックテストする方法【完全コード付き】

Uncategorized

RSIとMACDを組み合わせると何が嬉しいのか？

RSI（相対力指数）とMACD（移動平均収束発散法）は、どちらも人気の高いテクニカル指標ですが、それぞれ「異なる視点」で相場を見ています。RSIは買われすぎ・売られすぎを判断するオシレーター系、MACDはトレンドの方向と勢いを判断するトレンド系です。この2つを組み合わせることで、お互いの弱点を補い、より精度の高いシグナルを生成できます。本記事では、PythonでRSI×MACD複合戦略を実装し、バックテストで検証するコードを一から解説します。

使用するライブラリのインストール

まず必要なライブラリをインストールしましょう。

pip install yfinance pandas numpy matplotlib

Step1: データ取得とRSI・MACDの計算

yfinanceでデータを取得し、RSIとMACDを計算する関数を作ります。

import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

def fetch_data(ticker: str, period: str = "2y") -> pd.DataFrame:
    """yfinanceでOHLCVデータを取得"""
    df = yf.Ticker(ticker).history(period=period)
    return df[['Open', 'High', 'Low', 'Close', 'Volume']].copy()

def calc_rsi(series: pd.Series, period: int = 14) -> pd.Series:
    """RSIを計算する"""
    delta = series.diff()
    gain = delta.clip(lower=0)
    loss = (-delta).clip(lower=0)
    avg_gain = gain.ewm(com=period - 1, min_periods=period).mean()
    avg_loss = loss.ewm(com=period - 1, min_periods=period).mean()
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

def calc_macd(series: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame:
    """MACDを計算する"""
    ema_fast = series.ewm(span=fast, adjust=False).mean()
    ema_slow = series.ewm(span=slow, adjust=False).mean()
    macd_line = ema_fast - ema_slow
    signal_line = macd_line.ewm(span=signal, adjust=False).mean()
    histogram = macd_line - signal_line
    return pd.DataFrame({
        'MACD': macd_line,
        'Signal': signal_line,
        'Histogram': histogram
    })

# データ取得と指標計算
df = fetch_data("7203.T", period="2y")
df['RSI'] = calc_rsi(df['Close'])
macd_df = calc_macd(df['Close'])
df = pd.concat([df, macd_df], axis=1)
print(df.tail())

Step2: 売買シグナルの生成ロジック

RSIとMACDを組み合わせた売買ルールを実装します。買いシグナルは「RSIが30以下（売られすぎ）かつMACDがシグナル線を上抜け」、売りシグナルは「RSIが70以上（買われすぎ）かつMACDがシグナル線を下抜け」です。

def generate_signals(df: pd.DataFrame, rsi_oversold: float = 30, rsi_overbought: float = 70) -> pd.DataFrame:
    """RSI×MACDの複合シグナルを生成する"""
    df = df.copy()
    
    # MACDクロスオーバーの検出
    df['MACD_Cross_Up'] = (df['MACD'] > df['Signal']) & (df['MACD'].shift(1) <= df['Signal'].shift(1))
    df['MACD_Cross_Down'] = (df['MACD'] < df['Signal']) & (df['MACD'].shift(1) >= df['Signal'].shift(1))
    
    # 複合シグナル
    df['Signal_Buy'] = (df['RSI'] < rsi_oversold) & df['MACD_Cross_Up']
    df['Signal_Sell'] = (df['RSI'] > rsi_overbought) & df['MACD_Cross_Down']
    
    # ポジション管理（1: ロング, 0: ホールドなし）
    df['Position'] = 0
    position = 0
    for i in range(len(df)):
        if df['Signal_Buy'].iloc[i] and position == 0:
            position = 1
        elif df['Signal_Sell'].iloc[i] and position == 1:
            position = 0
        df.iloc[i, df.columns.get_loc('Position')] = position
    
    return df

df = generate_signals(df)
print(f"買いシグナル発生回数: {df['Signal_Buy'].sum()}")
print(f"売りシグナル発生回数: {df['Signal_Sell'].sum()}")

Step3: バックテストの実装とパフォーマンス評価

生成したシグナルを使ってバックテストを実施し、各種パフォーマンス指標を計算します。

def backtest(df: pd.DataFrame, initial_capital: float = 1_000_000) -> dict:
    """バックテストを実行してパフォーマンスを評価する"""
    df = df.copy()
    df['Daily_Return'] = df['Close'].pct_change()
    df['Strategy_Return'] = df['Position'].shift(1) * df['Daily_Return']
    df['Cumulative_Market'] = (1 + df['Daily_Return']).cumprod()
    df['Cumulative_Strategy'] = (1 + df['Strategy_Return']).cumprod()
    
    # パフォーマンス指標の計算
    total_return = df['Cumulative_Strategy'].iloc[-1] - 1
    annual_return = (1 + total_return) ** (252 / len(df)) - 1
    volatility = df['Strategy_Return'].std() * np.sqrt(252)
    sharpe_ratio = annual_return / volatility if volatility > 0 else 0
    
    # 最大ドローダウン
    rolling_max = df['Cumulative_Strategy'].cummax()
    drawdown = (df['Cumulative_Strategy'] - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    
    # 勝率計算
    trade_returns = []
    in_position = False
    entry_price = 0
    for _, row in df.iterrows():
        if row['Signal_Buy'] and not in_position:
            in_position = True
            entry_price = row['Close']
        elif row['Signal_Sell'] and in_position:
            trade_returns.append((row['Close'] - entry_price) / entry_price)
            in_position = False
    win_rate = sum(1 for r in trade_returns if r > 0) / len(trade_returns) if trade_returns else 0
    
    return {
        'total_return': f"{total_return:.2%}",
        'annual_return': f"{annual_return:.2%}",
        'volatility': f"{volatility:.2%}",
        'sharpe_ratio': f"{sharpe_ratio:.2f}",
        'max_drawdown': f"{max_drawdown:.2%}",
        'win_rate': f"{win_rate:.2%}",
        'num_trades': len(trade_returns),
        'df': df
    }

result = backtest(df)
print("===== バックテスト結果 =====")
for key, value in result.items():
    if key != 'df':
        print(f"{key}: {value}")

Step4: チャートで結果を可視化する

バックテスト結果を視覚的に確認するためのチャートを作成します。

def plot_results(result: dict):
    """バックテスト結果をチャートで可視化"""
    df = result['df'].dropna()
    fig, axes = plt.subplots(4, 1, figsize=(14, 12), gridspec_kw={'height_ratios': [3, 1, 1, 2]})
    fig.patch.set_facecolor('#0d1117')
    for ax in axes:
        ax.set_facecolor('#111827')
        ax.tick_params(colors='white')
        ax.spines['bottom'].set_color('#374151')
        ax.spines['left'].set_color('#374151')

    # 株価 + 売買シグナル
    axes[0].plot(df.index, df['Close'], color='#e5e7eb', linewidth=1.5, label='株価')
    buy_signals = df[df['Signal_Buy']]
    sell_signals = df[df['Signal_Sell']]
    axes[0].scatter(buy_signals.index, buy_signals['Close'], marker='^', color='#34d399', s=100, zorder=5, label='買いシグナル')
    axes[0].scatter(sell_signals.index, sell_signals['Close'], marker='v', color='#f87171', s=100, zorder=5, label='売りシグナル')
    axes[0].set_title('株価 + 売買シグナル', color='white', fontsize=12)
    axes[0].legend(facecolor='#1f2937', labelcolor='white')

    # RSI
    axes[1].plot(df.index, df['RSI'], color='#fb923c', linewidth=1.5)
    axes[1].axhline(70, color='#f87171', linestyle='--', alpha=0.5)
    axes[1].axhline(30, color='#34d399', linestyle='--', alpha=0.5)
    axes[1].set_title('RSI(14)', color='white', fontsize=10)
    axes[1].set_ylim(0, 100)

    # MACD
    axes[2].plot(df.index, df['MACD'], color='#38bdf8', linewidth=1.5, label='MACD')
    axes[2].plot(df.index, df['Signal'], color='#f87171', linewidth=1.5, label='Signal')
    axes[2].bar(df.index, df['Histogram'], color=['#34d399' if v >= 0 else '#f87171' for v in df['Histogram']], alpha=0.6)
    axes[2].set_title('MACD', color='white', fontsize=10)
    axes[2].legend(facecolor='#1f2937', labelcolor='white')

    # 累積リターン比較
    axes[3].plot(df.index, df['Cumulative_Strategy'], color='#a78bfa', linewidth=2, label='戦略')
    axes[3].plot(df.index, df['Cumulative_Market'], color='#9ca3af', linewidth=1.5, label='Buy&Hold')
    axes[3].set_title('累積リターン比較', color='white', fontsize=10)
    axes[3].legend(facecolor='#1f2937', labelcolor='white')

    plt.tight_layout()
    plt.savefig('rsi_macd_backtest.png', dpi=150, facecolor='#0d1117')
    plt.show()

plot_results(result)

RSI×MACD戦略のパラメータチューニング

バックテスト結果が出たら、パラメータを変えて最適化を試みましょう。RSIの期間（デフォルト14）、MACDの期間（12/26/9）、RSIの過売り・過買い水準（30/70）などを変えることで、パフォーマンスが変わります。ただし、過度な最適化（カーブフィッティング）には注意が必要です。過去データに過剰適合した戦略は、未来のデータでは機能しないことが多いため、アウトオブサンプルテスト（テスト期間を分ける）も必ず実施してください。

まとめ

本記事では、PythonでRSIとMACDを組み合わせた複合テクニカル戦略を実装し、バックテストでパフォーマンスを評価する方法を解説しました。RSIが売られすぎを示し、かつMACDがゴールデンクロスするタイミングでエントリーするこの戦略は、シンプルながら実践的なアプローチです。コードをそのままコピーして動かせるので、まずはトヨタ（7203.T）や日経225（^N225）で試してみてください。戦略の改善にはAIを活用するのもおすすめです！

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