🌟【黄金比の奇跡】61.8%で勝率69.8%！フィボナッチの秘密をデータで解明

- Python 3.11
- MetaTrader5 (MT5)
- pandas, numpy, matplotlib, scipy
- 通貨ペア: USDJPY
- 時間足: 5分足
- データ期間: 約35日分（10,000本）
- スイング検出期間: 20本
- フィボナッチレベル: 38.2%, 50%, 61.8%

"""
押し目買い・戻り売り（フィボナッチリトレースメント）の検証
各レベルへのタッチ後の値幅分布を分析
"""

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  # 取得する足の数
SWING_PERIOD = 20  # スイング検出期間

# フィボナッチレベル（リトレースメント率）
FIBO_LEVELS = [0.382, 0.500, 0.618]

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

# タッチ判定の許容範囲（pips）
TOUCH_TOLERANCE = 3


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 detect_swing_points(df, period):
    """
    スイングハイ・スイングローを検出
    period本の中で最高値/最安値ならスイングポイント
    """
    df['swing_high'] = False
    df['swing_low'] = False
    df['swing_high_price'] = np.nan
    df['swing_low_price'] = 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')] = True
            df.iloc[i, df.columns.get_loc('swing_high_price')] = 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')] = True
            df.iloc[i, df.columns.get_loc('swing_low_price')] = df['low'].iloc[i]

    # スイングポイントを前方に伝播（最新のスイングを常に参照できるように）
    df['last_swing_high'] = df['swing_high_price'].ffill()
    df['last_swing_low'] = df['swing_low_price'].ffill()

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

    return df


def calculate_fibo_levels(df):
    """フィボナッチレベルを計算"""
    # 上昇トレンド用（スイングローからスイングハイへ）
    df['fibo_range'] = df['last_swing_high'] - df['last_swing_low']

    for level in FIBO_LEVELS:
        level_name = f'fibo_{int(level*1000)}'
        # リトレースメントレベル = 高値 - (高値-安値) × レベル
        df[level_name] = df['last_swing_high'] - df['fibo_range'] * level

    return df


def detect_fibo_touches(df, tolerance_pips=3):
    """フィボナッチレベルへのタッチを検出"""
    tolerance = tolerance_pips * 0.01  # pipsを価格単位に変換（USDJPYの場合）

    for level in FIBO_LEVELS:
        level_name = f'fibo_{int(level*1000)}'
        touch_col = f'touch_{int(level*1000)}'

        # 現在の足の安値がフィボレベル付近にタッチ（上から下へ）
        df[touch_col] = (
            (df['low'] <= df[level_name] + tolerance) &
            (df['low'] >= df[level_name] - tolerance) &
            (df['close'].shift(1) > df[level_name])  # 前の足はレベルより上
        )

        touch_count = df[touch_col].sum()
        print(f"フィボ{int(level*100)}%タッチ: {touch_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

        # 未来の価格との差分（pips換算）
        future_close = df['close'].shift(-bars_ahead)
        df[f'return_{minutes}m'] = (future_close - df['close']) * 100  # pips

    return df


def analyze_fibo_returns(df, level, horizons):
    """特定のフィボレベルへのタッチ後のリターン分布を分析"""
    touch_col = f'touch_{int(level*1000)}'
    signal_df = df[df[touch_col] == True].copy()

    if len(signal_df) < 10:
        print(f"フィボ{int(level*100)}%: サンプル数不足 ({len(signal_df)})")
        return None

    results = {}
    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_fibo_comparison(all_results):
    """全フィボレベルの比較グラフ"""
    fig, axes = plt.subplots(len(FIBO_LEVELS), len(TIME_HORIZONS),
                             figsize=(18, 12))

    colors = {'382': 'steelblue', '500': 'forestgreen', '618': 'coral'}

    for i, level in enumerate(FIBO_LEVELS):
        level_key = int(level * 1000)
        results = all_results.get(level_key)

        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']

            # ヒストグラム
            color = colors[str(level_key)]
            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)

            # 0と平均のライン
            ax.axvline(x=0, color='gray', linestyle='--', alpha=0.5)
            ax.axvline(x=returns.mean(), color='darkgreen', linestyle='-', lw=2)

            title = f'{int(level*100)}% / {minutes}分後'
            subtitle = f'n={data["count"]}, μ={data["mean"]:.2f}, 勝率{data["positive_rate"]:.1f}%'
            ax.set_title(f'{title}\n{subtitle}', fontsize=10)

            if i == len(FIBO_LEVELS) - 1:
                ax.set_xlabel('リターン (pips)')
            if j == 0:
                ax.set_ylabel('密度')

    plt.suptitle('フィボナッチレベル別 タッチ後のリターン分布',
                fontsize=14, fontweight='bold')
    plt.tight_layout()
    return fig


def print_comparison_table(all_results):
    """比較テーブルを表示"""
    print(f"\n{'='*80}")
    print("📊 フィボナッチレベル別 パフォーマンス比較")
    print(f"{'='*80}")

    # ヘッダー
    header = "レベル  | "
    for minutes in TIME_HORIZONS:
        header += f" {minutes}分後    |"
    print(header)
    print("-" * 80)

    # 各レベルの結果
    for level in FIBO_LEVELS:
        level_key = int(level * 1000)
        results = all_results.get(level_key)

        if results is None:
            continue

        # 平均リターン行
        row = f"{int(level*100):>5.1f}% | "
        for minutes in TIME_HORIZONS:
            if minutes in results:
                mean = results[minutes]['mean']
                row += f" {mean:>+6.2f}pips |"
            else:
                row += f"    -     |"
        print(row)

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

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

        print("-" * 80)


def main():
    """メイン処理"""
    # MT5初期化
    if not initialize_mt5():
        return

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

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

        # フィボナッチレベル計算
        df = calculate_fibo_levels(df)

        # フィボナッチタッチ検出
        df = detect_fibo_touches(df, TOUCH_TOLERANCE)

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

        # 各レベルを分析
        all_results = {}
        for level in FIBO_LEVELS:
            level_key = int(level * 1000)
            results = analyze_fibo_returns(df, level, TIME_HORIZONS)
            if results:
                all_results[level_key] = results

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

        # 比較グラフ作成
        if all_results:
            fig = plot_fibo_comparison(all_results)
            fig.savefig('fibonacci_analysis.png', dpi=150, bbox_inches='tight')
            print("\n💾 fibonacci_analysis.png を保存しました")
            plt.show()

    finally:
        mt5.shutdown()


if __name__ == "__main__":
    main()

pip install MetaTrader5 pandas numpy matplotlib scipy

# 通貨ペアを変更
SYMBOL = "EURUSD"

# 時間足を変更（1時間足）
TIMEFRAME = mt5.TIMEFRAME_H1

# スイング検出期間を変更
SWING_PERIOD = 50

# フィボナッチレベルを追加
FIBO_LEVELS = [0.236, 0.382, 0.500, 0.618, 0.786]