🔬【16手法を総当たり】ブレイクアウト自動探索で判明！Zigzag最強、でも3日間レンジにも光が...！？

- Python 3.11
- MetaTrader5 (MT5)
- pandas, numpy, matplotlib, seaborn, scipy
- 通貨ペア: USDJPY
- 時間足: 5分足（日足データも併用）
- データ期間: 約2.5ヶ月分（15,000本）
- 決済: 時間クローズ（5分、15分、30分、60分、120分後）

Donchian(10) = N足(10本) → 勝率49.1%
Donchian(20) = N足(20本) → 勝率48.1%
Donchian(50) = N足(50本) → 勝率47.2%

# Zigzag（スイングハイ・スイングロー）の検出
def detect_zigzag(df, period=20):
    swing_high = df['high'] == df['high'].rolling(period*2+1, center=True).max()
    swing_low = df['low'] == df['low'].rolling(period*2+1, center=True).min()
    return swing_high, swing_low

# 3日間の日足高値安値を取得
daily_high_3days = df_daily['high'].shift(1).rolling(3).max()
daily_low_3days = df_daily['low'].shift(1).rolling(3).min()

"""
ブレイクアウト自動探索装置
複数のブレイクアウト定義を自動検証し、最適な手法を発見する

検証するブレイクアウト定義:
1. N足高値安値（N=10, 20, 50）
2. Zigzag（ピボットの高値安値）
3. 日足高値安値
4. ATRバンド（終値±ATR×係数）
5. ドンチャンチャネル

決済: 時間クローズ（5分、15分、30分、60分、120分後）
"""

import MetaTrader5 as mt5
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg')  # GUIなしで画像保存
import seaborn as sns
from scipy import stats
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

# 日本語フォント設定
plt.rcParams['font.family'] = ['MS Gothic', 'Hiragino Sans', 'IPAexGothic', 'sans-serif']
plt.rcParams['axes.unicode_minus'] = False

# ===== 設定 =====
SYMBOL = "USDJPYm"  # ブローカーによってサフィックスが異なる
TIMEFRAME_M5 = mt5.TIMEFRAME_M5  # 5分足
TIMEFRAME_D1 = mt5.TIMEFRAME_D1  # 日足
NUM_BARS = 15000  # 取得する5分足の数
ATR_PERIOD = 14

# 分析する時間枠（分）- 決済タイミング
TIME_HORIZONS = [5, 15, 30, 60, 120]

# ブレイクアウト定義のパラメータ
N_PERIODS = [10, 20, 50]  # N足高値安値のN
ZIGZAG_PERIOD = 20  # ピボット検出期間
ATR_MULTIPLIERS = [1.0, 1.5, 2.0]  # ATRバンドの係数
DONCHIAN_PERIODS = [10, 20, 50]  # ドンチャンチャネル期間


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データを取得"""
    symbol_info = mt5.symbol_info(symbol)
    if symbol_info is None:
        print(f"Symbol {symbol} not found, trying with suffix...")
        for suffix in ['', '.i', '-m', '_SB']:
            test_symbol = symbol + suffix
            symbol_info = mt5.symbol_info(test_symbol)
            if symbol_info is not None:
                symbol = test_symbol
                print(f"Found symbol: {symbol}")
                break

    if symbol_info is None:
        print(f"Symbol not found. Error: {mt5.last_error()}")
        symbols = mt5.symbols_get()
        if symbols:
            jpy_symbols = [s.name for s in symbols if 'JPY' in s.name][:10]
            print(f"Available JPY symbols: {jpy_symbols}")
        return None

    if not symbol_info.visible:
        if not mt5.symbol_select(symbol, True):
            print(f"Failed to select symbol {symbol}")
            return None

    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, num_bars)
    if rates is None or len(rates) == 0:
        print(f"Data fetch failed. Error: {mt5.last_error()}")
        return None

    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    df.set_index('time', inplace=True)
    print(f"Data: {len(df)} bars ({df.index[0]} to {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 calculate_future_returns(df, horizons_minutes, timeframe_minutes=5):
    """シグナル発生後のリターンを計算（pips単位）"""
    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 detect_n_bar_breakout(df, n_period):
    """N足高値安値ブレイクアウトを検出"""
    df[f'highest_{n_period}'] = df['high'].rolling(window=n_period).max().shift(1)
    df[f'lowest_{n_period}'] = df['low'].rolling(window=n_period).min().shift(1)

    breakout_up = df['close'] > df[f'highest_{n_period}']
    breakout_down = df['close'] < df[f'lowest_{n_period}']

    signal_up = breakout_up & ~breakout_up.shift(1).fillna(False)
    signal_down = breakout_down & ~breakout_down.shift(1).fillna(False)
    return signal_up, signal_down


def detect_zigzag_breakout(df, period):
    """Zigzag（ピボット高値安値）ブレイクアウトを検出"""
    df['swing_high'] = np.nan
    df['swing_low'] = np.nan

    for i in range(period, len(df) - period):
        if df['high'].iloc[i] == df['high'].iloc[i-period:i+period+1].max():
            df.iloc[i, df.columns.get_loc('swing_high')] = df['high'].iloc[i]
        if df['low'].iloc[i] == df['low'].iloc[i-period:i+period+1].min():
            df.iloc[i, df.columns.get_loc('swing_low')] = df['low'].iloc[i]

    df['last_swing_high'] = df['swing_high'].ffill()
    df['last_swing_low'] = df['swing_low'].ffill()

    breakout_up = df['close'] > df['last_swing_high']
    breakout_down = df['close'] < df['last_swing_low']

    signal_up = breakout_up & ~breakout_up.shift(1).fillna(False)
    signal_down = breakout_down & ~breakout_down.shift(1).fillna(False)
    return signal_up, signal_down


def detect_daily_breakout(df_m5, df_d1, n_days=1):
    """
    日足高値安値ブレイクアウトを検出
    n_days: 前日から何日分の高値安値を使うか
      - 1 = 前日のみの高値安値
      - 2 = 前日〜2日前の高値安値（2日間のレンジ）
      - 3 = 前日〜3日前の高値安値（3日間のレンジ）
    ※当日の高値安値は日中に更新されるため、参照基準として使えない
    """
    df_d1_copy = pd.DataFrame()
    df_d1_copy['ref_high'] = df_d1['high'].shift(1).rolling(window=n_days, min_periods=1).max()
    df_d1_copy['ref_low'] = df_d1['low'].shift(1).rolling(window=n_days, min_periods=1).min()
    df_d1_copy['date'] = df_d1.index.date

    df_m5['date'] = df_m5.index.date
    df_merged = df_m5.merge(df_d1_copy, on='date', how='left')
    df_merged.index = df_m5.index

    breakout_up = df_m5['close'] > df_merged['ref_high']
    breakout_down = df_m5['close'] < df_merged['ref_low']

    signal_up = breakout_up & ~breakout_up.shift(1).fillna(False)
    signal_down = breakout_down & ~breakout_down.shift(1).fillna(False)
    return signal_up, signal_down


def detect_atr_band_breakout(df, multiplier):
    """ATRバンドブレイクアウトを検出"""
    df[f'atr_upper_{multiplier}'] = df['close'].shift(1) + df['atr'] * multiplier
    df[f'atr_lower_{multiplier}'] = df['close'].shift(1) - df['atr'] * multiplier

    breakout_up = df['close'] > df[f'atr_upper_{multiplier}']
    breakout_down = df['close'] < df[f'atr_lower_{multiplier}']

    signal_up = breakout_up & ~breakout_up.shift(1).fillna(False)
    signal_down = breakout_down & ~breakout_down.shift(1).fillna(False)
    return signal_up, signal_down


def detect_donchian_breakout(df, period):
    """ドンチャンチャネルブレイクアウトを検出"""
    df[f'donchian_high_{period}'] = df['high'].rolling(window=period).max().shift(1)
    df[f'donchian_low_{period}'] = df['low'].rolling(window=period).min().shift(1)

    breakout_up = df['close'] > df[f'donchian_high_{period}']
    breakout_down = df['close'] < df[f'donchian_low_{period}']

    signal_up = breakout_up & ~breakout_up.shift(1).fillna(False)
    signal_down = breakout_down & ~breakout_down.shift(1).fillna(False)
    return signal_up, signal_down


def analyze_signals(df, signal_up, signal_down, method_name):
    """シグナルの成績を分析"""
    results = {'method': method_name, 'up': {}, 'down': {}}

    up_df = df[signal_up].copy()
    if len(up_df) >= 10:
        for minutes in TIME_HORIZONS:
            col = f'return_{minutes}m'
            returns = up_df[col].dropna()
            if len(returns) >= 5:
                results['up'][minutes] = {
                    'count': len(returns),
                    'mean': returns.mean(),
                    'std': returns.std(),
                    'win_rate': (returns > 0).mean() * 100,
                    'sharpe': returns.mean() / returns.std() if returns.std() > 0 else 0,
                    'skewness': stats.skew(returns) if len(returns) >= 3 else 0,
                }

    down_df = df[signal_down].copy()
    if len(down_df) >= 10:
        for minutes in TIME_HORIZONS:
            col = f'return_{minutes}m'
            returns = -down_df[col].dropna()  # 売りなので符号反転
            if len(returns) >= 5:
                results['down'][minutes] = {
                    'count': len(returns),
                    'mean': returns.mean(),
                    'std': returns.std(),
                    'win_rate': (returns > 0).mean() * 100,
                    'sharpe': returns.mean() / returns.std() if returns.std() > 0 else 0,
                    'skewness': stats.skew(returns) if len(returns) >= 3 else 0,
                }
    return results


def create_summary_table(all_results):
    """全結果をサマリーテーブルに変換"""
    rows = []
    for result in all_results:
        method = result['method']
        for direction in ['up', 'down']:
            for minutes in TIME_HORIZONS:
                if minutes in result[direction]:
                    data = result[direction][minutes]
                    rows.append({
                        'method': method,
                        'direction': '買い' if direction == 'up' else '売り',
                        'hold_time': f'{minutes}分',
                        'count': data['count'],
                        'win_rate': data['win_rate'],
                        'mean_return': data['mean'],
                        'sharpe': data['sharpe'],
                        'skewness': data['skewness'],
                    })
    return pd.DataFrame(rows)


def main():
    """メイン処理"""
    print("[START] Breakout Explorer starting...")

    if not initialize_mt5():
        return None, None

    try:
        print("\n[DOWNLOAD] Getting data...")
        df_m5 = get_ohlc_data(SYMBOL, TIMEFRAME_M5, NUM_BARS)
        df_d1 = get_ohlc_data(SYMBOL, TIMEFRAME_D1, 100)

        if df_m5 is None or df_d1 is None:
            return None, None

        df_m5['atr'] = calculate_atr(df_m5, ATR_PERIOD)
        df_m5 = calculate_future_returns(df_m5, TIME_HORIZONS)

        all_results = []

        # 1. N足高値安値ブレイクアウト
        print("\n[1] Testing N-bar High/Low Breakout...")
        for n in N_PERIODS:
            signal_up, signal_down = detect_n_bar_breakout(df_m5, n)
            results = analyze_signals(df_m5, signal_up, signal_down, f'N足({n}本)')
            all_results.append(results)

        # 2. Zigzag（ピボット）ブレイクアウト
        print("\n[2] Testing Zigzag (Pivot) Breakout...")
        signal_up, signal_down = detect_zigzag_breakout(df_m5, ZIGZAG_PERIOD)
        results = analyze_signals(df_m5, signal_up, signal_down, f'Zigzag({ZIGZAG_PERIOD})')
        all_results.append(results)

        # 3. 日足高値安値ブレイクアウト（1〜6日間レンジ）
        print("\n[3] Testing Daily High/Low Breakout (N-day range)...")
        for n_days in range(1, 7):
            day_label = f"{n_days}日間"
            signal_up, signal_down = detect_daily_breakout(df_m5, df_d1, n_days)
            results = analyze_signals(df_m5, signal_up, signal_down, f'日足({day_label})')
            all_results.append(results)

        # 4. ATRバンドブレイクアウト
        print("\n[4] Testing ATR Band Breakout...")
        for mult in ATR_MULTIPLIERS:
            signal_up, signal_down = detect_atr_band_breakout(df_m5, mult)
            results = analyze_signals(df_m5, signal_up, signal_down, f'ATR×{mult}')
            all_results.append(results)

        # 5. ドンチャンチャネルブレイクアウト
        print("\n[5] Testing Donchian Channel Breakout...")
        for period in DONCHIAN_PERIODS:
            signal_up, signal_down = detect_donchian_breakout(df_m5, period)
            results = analyze_signals(df_m5, signal_up, signal_down, f'Donchian({period})')
            all_results.append(results)

        summary_df = create_summary_table(all_results)
        print("\n[DONE] Analysis complete!")

        return all_results, summary_df

    finally:
        mt5.shutdown()


if __name__ == "__main__":
    results, summary = main()
    if summary is not None:
        summary.to_csv('breakout_summary.csv', index=False, encoding='utf-8-sig')
        print("[SAVE] breakout_summary.csv saved")