⚡【ダマシの正体判明】高ボラ時ブレイクで勝率66%！ATRフィルターの威力

- Python 3.11
- MetaTrader5 (MT5)
- pandas, numpy, matplotlib, scipy
- 通貨ペア: USDJPY
- 時間足: 5分足
- データ期間: 約35日分（10,000本）
- ブレイク判定期間: 20本
- ATRフィルター: 14期間

"""
ブレイクアウト手法の検証
高値・安値ブレイク後の値幅分布を分析
ボラティリティ（ATR）別にも分析
"""

import MetaTrader5 as mt5
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy import stats
from datetime import datetime

# ===== 設定 =====
SYMBOL = "USDJPY"
TIMEFRAME = mt5.TIMEFRAME_M5  # 5分足
NUM_BARS = 10000  # 取得する足の数
BREAKOUT_PERIOD = 20  # ブレイク判定期間
ATR_PERIOD = 14  # ATR期間

# 分析する時間枠（分） - 5分足なので5の倍数
TIME_HORIZONS = [5, 15, 30, 60, 120]


def initialize_mt5():
    """MT5を初期化"""
    if not mt5.initialize():
        print("MT5の初期化に失敗")
        return False
    print(f"MT5バージョン: {mt5.version()}")
    return True


def get_ohlc_data(symbol, timeframe, num_bars):
    """OHLCデータを取得"""
    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, num_bars)
    if rates is None:
        print("データ取得失敗")
        return None

    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    df.set_index('time', inplace=True)
    print(f"取得データ: {len(df)}本 ({df.index[0]} 〜 {df.index[-1]})")
    return df


def calculate_atr(df, period):
    """ATR（Average True Range）を計算"""
    high = df['high']
    low = df['low']
    close = df['close'].shift(1)

    tr1 = high - low
    tr2 = abs(high - close)
    tr3 = abs(low - close)

    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    atr = tr.rolling(window=period).mean()

    return atr


def detect_breakouts(df, period):
    """ブレイクアウトを検出"""
    # 直近N本の高値・安値
    df['highest'] = df['high'].rolling(window=period).max().shift(1)
    df['lowest'] = df['low'].rolling(window=period).min().shift(1)

    # 上方ブレイク: 終値が直近高値を超えた
    df['breakout_up'] = df['close'] > df['highest']

    # 下方ブレイク: 終値が直近安値を割った
    df['breakout_down'] = df['close'] < df['lowest']

    # 連続ブレイクを除外（最初の1本だけカウント）
    df['breakout_up_signal'] = df['breakout_up'] & ~df['breakout_up'].shift(1).fillna(False)
    df['breakout_down_signal'] = df['breakout_down'] & ~df['breakout_down'].shift(1).fillna(False)

    up_count = df['breakout_up_signal'].sum()
    down_count = df['breakout_down_signal'].sum()
    print(f"上方ブレイク検出: {up_count}回")
    print(f"下方ブレイク検出: {down_count}回")

    return df


def calculate_future_returns(df, horizons_minutes, timeframe_minutes=5):
    """シグナル発生後のリターンを計算"""
    for minutes in horizons_minutes:
        bars_ahead = minutes // timeframe_minutes
        if bars_ahead < 1:
            bars_ahead = 1

        future_close = df['close'].shift(-bars_ahead)
        df[f'return_{minutes}m'] = (future_close - df['close']) * 100  # pips

    return df


def analyze_breakout_returns(df, signal_col, horizons, signal_name, atr_filter=None):
    """
    ブレイクアウト後のリターン分布を分析
    atr_filter: 'high', 'low', None
    """
    signal_df = df[df[signal_col] == True].copy()

    # ATRフィルター適用
    if atr_filter == 'high':
        atr_median = df['atr'].median()
        signal_df = signal_df[signal_df['atr'] > atr_median]
        signal_name += " (高ボラ)"
    elif atr_filter == 'low':
        atr_median = df['atr'].median()
        signal_df = signal_df[signal_df['atr'] <= atr_median]
        signal_name += " (低ボラ)"

    if len(signal_df) < 10:
        print(f"{signal_name}: サンプル数不足 ({len(signal_df)})")
        return None

    results = {'name': signal_name}
    for minutes in horizons:
        col = f'return_{minutes}m'
        returns = signal_df[col].dropna()

        if len(returns) < 10:
            continue

        results[minutes] = {
            'mean': returns.mean(),
            'std': returns.std(),
            'median': returns.median(),
            'count': len(returns),
            'positive_rate': (returns > 0).mean() * 100,
            'skewness': stats.skew(returns),
            'kurtosis': stats.kurtosis(returns),
            'data': returns.values
        }

    return results


def plot_breakout_analysis(up_results, down_results):
    """ブレイクアウト分析結果を可視化"""
    fig, axes = plt.subplots(2, len(TIME_HORIZONS), figsize=(18, 8))

    # 上方ブレイク
    for j, minutes in enumerate(TIME_HORIZONS):
        ax = axes[0, j]
        if minutes not in up_results:
            ax.set_visible(False)
            continue

        data = up_results[minutes]
        returns = data['data']

        ax.hist(returns, bins=30, density=True, alpha=0.7,
               color='forestgreen', edgecolor='black')

        mu, std = stats.norm.fit(returns)
        x = np.linspace(returns.min(), returns.max(), 100)
        ax.plot(x, stats.norm.pdf(x, mu, std), 'r-', lw=2)

        ax.axvline(x=0, color='gray', linestyle='--', alpha=0.5)
        ax.axvline(x=returns.mean(), color='darkgreen', linestyle='-', lw=2)

        title = f'上方ブレイク {minutes}分後'
        subtitle = f'n={data["count"]}, μ={data["mean"]:.2f}, 勝率{data["positive_rate"]:.1f}%'
        ax.set_title(f'{title}\n{subtitle}', fontsize=10)
        ax.set_xlabel('リターン (pips)')
        ax.set_ylabel('密度')

    # 下方ブレイク
    for j, minutes in enumerate(TIME_HORIZONS):
        ax = axes[1, j]
        if minutes not in down_results:
            ax.set_visible(False)
            continue

        data = down_results[minutes]
        returns = -data['data']  # 売りなので符号反転

        ax.hist(returns, bins=30, density=True, alpha=0.7,
               color='coral', edgecolor='black')

        mu, std = stats.norm.fit(returns)
        x = np.linspace(returns.min(), returns.max(), 100)
        ax.plot(x, stats.norm.pdf(x, mu, std), 'r-', lw=2)

        ax.axvline(x=0, color='gray', linestyle='--', alpha=0.5)
        ax.axvline(x=returns.mean(), color='darkred', linestyle='-', lw=2)

        title = f'下方ブレイク {minutes}分後'
        subtitle = f'n={data["count"]}, μ={-data["mean"]:.2f}, 勝率{100-data["positive_rate"]:.1f}%'
        ax.set_title(f'{title}\n{subtitle}', fontsize=10)
        ax.set_xlabel('リターン (pips)')
        ax.set_ylabel('密度')

    plt.suptitle('ブレイクアウト後のリターン分布', fontsize=14, fontweight='bold')
    plt.tight_layout()
    return fig


def plot_volatility_comparison(results_all, results_high, results_low, direction):
    """ボラティリティ別比較グラフ"""
    fig, axes = plt.subplots(3, len(TIME_HORIZONS), figsize=(18, 10))

    configs = [
        (results_all, '全データ', 'steelblue'),
        (results_high, '高ボラ', 'coral'),
        (results_low, '低ボラ', 'lightgreen')
    ]

    for i, (results, label, color) in enumerate(configs):
        if results is None:
            continue

        for j, minutes in enumerate(TIME_HORIZONS):
            ax = axes[i, j]
            if minutes not in results:
                ax.set_visible(False)
                continue

            data = results[minutes]
            returns = data['data']
            if direction == 'down':
                returns = -returns

            ax.hist(returns, bins=25, density=True, alpha=0.7,
                   color=color, edgecolor='black')

            mu, std = stats.norm.fit(returns)
            x = np.linspace(returns.min(), returns.max(), 100)
            ax.plot(x, stats.norm.pdf(x, mu, std), 'r-', lw=2)

            ax.axvline(x=0, color='gray', linestyle='--', alpha=0.5)
            ax.axvline(x=returns.mean(), color='darkgreen', linestyle='-', lw=2)

            mean_val = returns.mean()
            win_rate = (returns > 0).mean() * 100

            ax.set_title(f'{label} {minutes}分後\nμ={mean_val:.2f}, WR={win_rate:.1f}%', fontsize=9)

    direction_text = '上方' if direction == 'up' else '下方'
    plt.suptitle(f'{direction_text}ブレイク - ボラティリティ別比較', fontsize=14, fontweight='bold')
    plt.tight_layout()
    return fig


def print_comparison_table(results_all, results_high, results_low, direction):
    """比較テーブルを表示"""
    direction_text = '上方' if direction == 'up' else '下方'

    print(f"\n{'='*80}")
    print(f"📊 {direction_text}ブレイク - ボラティリティ別比較")
    print(f"{'='*80}")

    configs = [
        (results_all, '全データ'),
        (results_high, '高ボラ '),
        (results_low, '低ボラ ')
    ]

    for results, label in configs:
        if results is None:
            continue

        print(f"\n【{label}】")
        header = "          | "
        for minutes in TIME_HORIZONS:
            header += f" {minutes:>4}分後 |"
        print(header)
        print("-" * 65)

        # 平均リターン
        row = "平均(pips)| "
        for minutes in TIME_HORIZONS:
            if minutes in results:
                val = results[minutes]['mean']
                if direction == 'down':
                    val = -val
                row += f" {val:>+6.2f} |"
            else:
                row += f"    -   |"
        print(row)

        # 勝率
        row = "勝率      | "
        for minutes in TIME_HORIZONS:
            if minutes in results:
                wr = results[minutes]['positive_rate']
                if direction == 'down':
                    wr = 100 - wr
                row += f" {wr:>5.1f}% |"
            else:
                row += f"    -   |"
        print(row)

        # 歪度
        row = "歪度      | "
        for minutes in TIME_HORIZONS:
            if minutes in results:
                skew = results[minutes]['skewness']
                if direction == 'down':
                    skew = -skew
                row += f" {skew:>+5.2f} |"
            else:
                row += f"    -   |"
        print(row)


def main():
    """メイン処理"""
    if not initialize_mt5():
        return

    try:
        # データ取得
        df = get_ohlc_data(SYMBOL, TIMEFRAME, NUM_BARS)
        if df is None:
            return

        # ATR計算
        df['atr'] = calculate_atr(df, ATR_PERIOD)

        # ブレイクアウト検出
        df = detect_breakouts(df, BREAKOUT_PERIOD)

        # 将来リターン計算
        df = calculate_future_returns(df, TIME_HORIZONS)

        # ===== 上方ブレイク分析 =====
        print("\n" + "="*60)
        print("上方ブレイク分析")
        print("="*60)

        up_all = analyze_breakout_returns(df, 'breakout_up_signal', TIME_HORIZONS, '上方ブレイク')
        up_high = analyze_breakout_returns(df, 'breakout_up_signal', TIME_HORIZONS, '上方ブレイク', 'high')
        up_low = analyze_breakout_returns(df, 'breakout_up_signal', TIME_HORIZONS, '上方ブレイク', 'low')

        print_comparison_table(up_all, up_high, up_low, 'up')

        # ===== 下方ブレイク分析 =====
        print("\n" + "="*60)
        print("下方ブレイク分析")
        print("="*60)

        down_all = analyze_breakout_returns(df, 'breakout_down_signal', TIME_HORIZONS, '下方ブレイク')
        down_high = analyze_breakout_returns(df, 'breakout_down_signal', TIME_HORIZONS, '下方ブレイク', 'high')
        down_low = analyze_breakout_returns(df, 'breakout_down_signal', TIME_HORIZONS, '下方ブレイク', 'low')

        print_comparison_table(down_all, down_high, down_low, 'down')

        # グラフ保存
        if up_all and down_all:
            fig1 = plot_breakout_analysis(up_all, down_all)
            fig1.savefig('breakout_distribution.png', dpi=150, bbox_inches='tight')
            print("\n💾 breakout_distribution.png を保存しました")

        if up_all and up_high and up_low:
            fig2 = plot_volatility_comparison(up_all, up_high, up_low, 'up')
            fig2.savefig('breakout_up_volatility.png', dpi=150, bbox_inches='tight')
            print("💾 breakout_up_volatility.png を保存しました")

        if down_all and down_high and down_low:
            fig3 = plot_volatility_comparison(down_all, down_high, down_low, 'down')
            fig3.savefig('breakout_down_volatility.png', dpi=150, bbox_inches='tight')
            print("💾 breakout_down_volatility.png を保存しました")

        plt.show()

    finally:
        mt5.shutdown()


if __name__ == "__main__":
    main()

# ATRが中央値以上の時だけエントリー
if current_atr > atr_median:
    # ブレイクアウトシグナルを有効とする
    execute_trade()

// ATRフィルター付きブレイクアウト判定
input int BreakoutPeriod = 20;
input int ATRPeriod = 14;

void OnTick() {
    double highest = iHigh(NULL, 0, iHighest(NULL, 0, MODE_HIGH, BreakoutPeriod, 1));
    double lowest = iLow(NULL, 0, iLowest(NULL, 0, MODE_LOW, BreakoutPeriod, 1));
    double atr = iATR(NULL, 0, ATRPeriod, 0);

    // ATRの中央値を計算（簡易版：過去100本の平均）
    double atr_sum = 0;
    for(int i = 0; i < 100; i++) {
        atr_sum += iATR(NULL, 0, ATRPeriod, i);
    }
    double atr_avg = atr_sum / 100;

    // 高ボラ時のみブレイクアウトを有効に
    if(atr < atr_avg) {
        Print("低ボラティリティ - ブレイクアウト無効");
        return;
    }

    // 上方ブレイク
    if(Close[0] > highest) {
        Print("高ボラ時の上方ブレイク検出！");
        // エントリー処理
    }

    // 下方ブレイク
    if(Close[0] < lowest) {
        Print("高ボラ時の下方ブレイク検出！");
        // エントリー処理
    }
}

def is_valid_breakout(df, idx, atr_threshold='median'):
    """有効なブレイクアウトかどうか判定"""
    current_atr = df['atr'].iloc[idx]

    if atr_threshold == 'median':
        threshold = df['atr'].median()
    else:
        threshold = df['atr'].quantile(0.6)  # 上位40%

    return current_atr > threshold