🕐【時間軸の魔法】日足で勝率70%！スキャルで負ける人がスイングで勝てる理由

- Python 3.11
- MetaTrader5 (MT5)
- pandas, numpy, matplotlib, scipy
- 通貨ペア: USDJPY
- 時間足: 1時間足 + 日足
- データ期間: 約1年分（365日）
- スイング検出期間: 5本（1時間足=5時間、日足=5日）

"""
スイングトレードの検証
スイングポイント検出後の値幅分布を分析
複数の時間軸で比較
"""

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, timedelta

# ===== 設定 =====
SYMBOL = "USDJPY"

# 分析する時間足
TIMEFRAMES = {
    'M5': (mt5.TIMEFRAME_M5, 50000, [1, 5, 15, 30, 60]),          # 5分足
    'H1': (mt5.TIMEFRAME_H1, 8760, [1, 4, 24, 72, 168]),          # 1時間足（時間単位）
    'D1': (mt5.TIMEFRAME_D1, 365, [1, 3, 5, 10, 20])              # 日足（日単位）
}

SWING_PERIOD = 5  # スイング検出期間


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 or len(rates) < 100:
        print(f"データ取得失敗 (timeframe={timeframe})")
        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 detect_swing_points(df, period):
    """
    スイングハイ・スイングローを検出
    period本の中で最高値/最安値ならスイングポイント
    """
    df['swing_high'] = False
    df['swing_low'] = False

    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')] = True

        # スイングロー検出
        if df['low'].iloc[i] == df['low'].iloc[i-period:i+period+1].min():
            df.iloc[i, df.columns.get_loc('swing_low')] = True

    high_count = df['swing_high'].sum()
    low_count = df['swing_low'].sum()
    print(f"スイングハイ検出: {high_count}個")
    print(f"スイングロー検出: {low_count}個")

    return df


def calculate_future_returns(df, horizons, timeframe_name):
    """シグナル発生後のリターンを計算"""
    for period in horizons:
        # 時間足によって単位が異なる
        if timeframe_name == 'M5':
            bars_ahead = period  # 分単位 → そのまま本数
        elif timeframe_name == 'H1':
            bars_ahead = period  # 時間単位 → そのまま本数
        else:  # D1
            bars_ahead = period  # 日単位 → そのまま本数

        future_close = df['close'].shift(-bars_ahead)

        # リターン計算（pips換算）
        df[f'return_{period}'] = (future_close - df['close']) * 100

    return df


def analyze_swing_returns(df, signal_col, horizons, signal_name):
    """スイングポイント後のリターン分布を分析"""
    signal_df = df[df[signal_col] == True].copy()

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

    results = {'name': signal_name, 'sample_size': len(signal_df)}

    for period in horizons:
        col = f'return_{period}'
        returns = signal_df[col].dropna()

        if len(returns) < 10:
            continue

        results[period] = {
            '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),
            'sharpe': returns.mean() / returns.std() if returns.std() > 0 else 0,
            'data': returns.values
        }

    return results


def plot_timeframe_comparison(all_results):
    """時間軸別の比較グラフ"""
    fig, axes = plt.subplots(2, 3, figsize=(15, 10))

    timeframe_colors = {'M5': 'coral', 'H1': 'steelblue', 'D1': 'forestgreen'}

    for col, (tf_name, results) in enumerate(all_results.items()):
        if results is None:
            continue

        color = timeframe_colors[tf_name]
        horizons = TIMEFRAMES[tf_name][2]

        # スイングハイ後（売りシグナル）
        ax_high = axes[0, col]
        if 'swing_high' in results and results['swing_high']:
            data = results['swing_high']
            last_horizon = horizons[-1]
            if last_horizon in data:
                returns = -data[last_horizon]['data']  # 売りなので符号反転
                ax_high.hist(returns, bins=30, density=True, alpha=0.7,
                           color=color, edgecolor='black')

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

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

                win_rate = (returns > 0).mean() * 100
                ax_high.set_title(f'{tf_name} スイングハイ後\n(最終期間後)\n平均={returns.mean():.2f}, 勝率={win_rate:.1f}%')

        ax_high.set_xlabel('リターン (pips)')
        ax_high.set_ylabel('密度')

        # スイングロー後（買いシグナル）
        ax_low = axes[1, col]
        if 'swing_low' in results and results['swing_low']:
            data = results['swing_low']
            last_horizon = horizons[-1]
            if last_horizon in data:
                returns = data[last_horizon]['data']  # 買いなのでそのまま
                ax_low.hist(returns, bins=30, density=True, alpha=0.7,
                          color=color, edgecolor='black')

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

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

                win_rate = (returns > 0).mean() * 100
                ax_low.set_title(f'{tf_name} スイングロー後\n(最終期間後)\n平均={returns.mean():.2f}, 勝率={win_rate:.1f}%')

        ax_low.set_xlabel('リターン (pips)')
        ax_low.set_ylabel('密度')

    plt.suptitle('時間軸別 スイングポイント後のリターン分布', fontsize=14, fontweight='bold')
    plt.tight_layout()
    return fig


def print_comprehensive_table(all_results):
    """包括的な比較テーブル"""
    print("\n" + "="*100)
    print("📊 時間軸別 スイングトレード パフォーマンス比較")
    print("="*100)

    for tf_name, results in all_results.items():
        if results is None:
            continue

        horizons = TIMEFRAMES[tf_name][2]
        unit = 'min' if tf_name == 'M5' else ('h' if tf_name == 'H1' else 'd')

        print(f"\n{'='*50}")
        print(f"【{tf_name}】時間足")
        print(f"{'='*50}")

        # スイングロー（買い）
        if 'swing_low' in results and results['swing_low']:
            data = results['swing_low']
            print(f"\n🔵 スイングロー後（買いシグナル） n={data['sample_size']}")

            header = "          | "
            for h in horizons:
                header += f" {h:>4}{unit} |"
            print(header)
            print("-" * (12 + 9 * len(horizons)))

            # 平均リターン
            row = "平均(pips)| "
            for h in horizons:
                if h in data:
                    row += f" {data[h]['mean']:>+6.1f} |"
                else:
                    row += f"    -   |"
            print(row)

            # 勝率
            row = "勝率      | "
            for h in horizons:
                if h in data:
                    row += f" {data[h]['positive_rate']:>5.1f}% |"
                else:
                    row += f"    -   |"
            print(row)

            # シャープレシオ
            row = "Sharpe    | "
            for h in horizons:
                if h in data:
                    row += f" {data[h]['sharpe']:>+5.2f} |"
                else:
                    row += f"    -   |"
            print(row)

        # スイングハイ（売り）
        if 'swing_high' in results and results['swing_high']:
            data = results['swing_high']
            print(f"\n🔴 スイングハイ後（売りシグナル） n={data['sample_size']}")

            header = "          | "
            for h in horizons:
                header += f" {h:>4}{unit} |"
            print(header)
            print("-" * (12 + 9 * len(horizons)))

            # 平均リターン（売りなので符号反転）
            row = "平均(pips)| "
            for h in horizons:
                if h in data:
                    row += f" {-data[h]['mean']:>+6.1f} |"
                else:
                    row += f"    -   |"
            print(row)

            # 勝率（売りなので100から引く）
            row = "勝率      | "
            for h in horizons:
                if h in data:
                    row += f" {100-data[h]['positive_rate']:>5.1f}% |"
                else:
                    row += f"    -   |"
            print(row)

            # シャープレシオ（売りなので符号反転）
            row = "Sharpe    | "
            for h in horizons:
                if h in data:
                    row += f" {-data[h]['sharpe']:>+5.2f} |"
                else:
                    row += f"    -   |"
            print(row)


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

    try:
        all_results = {}

        for tf_name, (timeframe, num_bars, horizons) in TIMEFRAMES.items():
            print(f"\n{'#'*60}")
            print(f"# {tf_name} 時間足の分析")
            print(f"{'#'*60}")

            # データ取得
            df = get_ohlc_data(SYMBOL, timeframe, num_bars)
            if df is None:
                all_results[tf_name] = None
                continue

            # スイングポイント検出
            df = detect_swing_points(df, SWING_PERIOD)

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

            # スイングロー（買いシグナル）分析
            swing_low_results = analyze_swing_returns(
                df, 'swing_low', horizons, f'{tf_name} スイングロー'
            )

            # スイングハイ（売りシグナル）分析
            swing_high_results = analyze_swing_returns(
                df, 'swing_high', horizons, f'{tf_name} スイングハイ'
            )

            all_results[tf_name] = {
                'swing_low': swing_low_results,
                'swing_high': swing_high_results
            }

        # 比較テーブル表示
        print_comprehensive_table(all_results)

        # グラフ保存
        fig = plot_timeframe_comparison(all_results)
        fig.savefig('swing_trade_timeframe_comparison.png', dpi=150, bbox_inches='tight')
        print("\n💾 swing_trade_timeframe_comparison.png を保存しました")

        plt.show()

    finally:
        mt5.shutdown()


if __name__ == "__main__":
    main()

# 推奨設定
TIMEFRAME = mt5.TIMEFRAME_H1
SWING_PERIOD = 5  # 5時間でスイング検出
HOLD_PERIOD = 24 * 3  # 3日（72時間）保有
STOP_LOSS = 50  # pips
TAKE_PROFIT = 150  # pips（3:1）

# 推奨設定
TIMEFRAME = mt5.TIMEFRAME_D1
SWING_PERIOD = 5  # 5日でスイング検出
HOLD_PERIOD = 10  # 10日保有
STOP_LOSS = 100  # pips
TAKE_PROFIT = 300  # pips（3:1）