😱【衝撃】ダブルトップは勝率58%！でもダブルボトムは51%...売りの方が強い理由

山1        山2
       ∧          ∧
      / \        / \
     /   \      /   \
    /     \    /     \
   /       ----       \
  /      ネックライン    ↓ ブレイク！（売りシグナル）

\      ネックライン    ↑ ブレイク！（買いシグナル）
   \       ----       /
    \     /    \     /
     \   /      \   /
      \ /        \ /
       ∨          ∨
      谷1        谷2

- Python 3.11
- MetaTrader5 (MT5)
- pandas, numpy, matplotlib, scipy
- 通貨ペア: USDJPY
- 時間足: 5分足
- データ期間: 約5ヶ月分（30,000本）
- スイングポイント判定: 左右10本
- 価格許容誤差: ATR × 1.5以内
- パターン形成期間: 15〜150本

# ダブルトップ売り戦略
if double_top_detected and current_atr > atr_median:
    # ネックラインブレイクで売りエントリー
    execute_sell()

"""
ダブルトップ/ダブルボトム パターン検証
「本当に勝てるのか？」をデータで完全解明！

検証項目:
1. ダブルトップ/ダブルボトムの自動検出
2. ネックラインブレイク後のリターン分布
3. パターン形成期間別の勝率
4. ボラティリティ（ATR）別の勝率
5. 価格乖離率別（2つの山/谷の近さ）の勝率
"""

import MetaTrader5 as mt5
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg')
import seaborn as sns
from scipy import stats
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

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

# ===== 設定 =====
SYMBOL = "USDJPY"
TIMEFRAME = mt5.TIMEFRAME_M5
NUM_BARS = 30000  # 約3.5ヶ月分

# パターン検出パラメータ
SWING_PERIOD = 10           # スイングポイント判定期間（左右N本）
PRICE_TOLERANCE_ATR = 1.5   # 価格許容誤差（ATRの倍数）
MIN_BARS_BETWEEN = 15       # 2つの山/谷の最小間隔
MAX_BARS_BETWEEN = 150      # 2つの山/谷の最大間隔
ATR_PERIOD = 14

# 分析する時間枠（分）
TIME_HORIZONS = [30, 60, 120, 240]


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データを取得"""
    # シンボル名のバリエーションを試す
    symbols_to_try = [symbol, symbol + 'm', symbol + '.i', symbol + '-m']

    for sym in symbols_to_try:
        symbol_info = mt5.symbol_info(sym)
        if symbol_info is not None:
            symbol = sym
            break

    if symbol_info is None:
        print(f"シンボルが見つかりません: {symbol}")
        return None

    if not symbol_info.visible:
        mt5.symbol_select(symbol, True)

    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, num_bars)
    if rates is None or len(rates) == 0:
        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計算"""
    high, low, close = df['high'], df['low'], df['close'].shift(1)
    tr = pd.concat([high - low, abs(high - close), abs(low - close)], axis=1).max(axis=1)
    return tr.rolling(window=period).mean()


def detect_swing_points(df, period):
    """スイングハイ・スイングローを検出"""
    swing_high = pd.Series(index=df.index, dtype=float)
    swing_low = pd.Series(index=df.index, dtype=float)

    for i in range(period, len(df) - period):
        # スイングハイ: 左右N本より高い
        left_max = df['high'].iloc[i-period:i].max()
        right_max = df['high'].iloc[i+1:i+period+1].max()
        if df['high'].iloc[i] > left_max and df['high'].iloc[i] > right_max:
            swing_high.iloc[i] = df['high'].iloc[i]

        # スイングロー: 左右N本より低い
        left_min = df['low'].iloc[i-period:i].min()
        right_min = df['low'].iloc[i+1:i+period+1].min()
        if df['low'].iloc[i] < left_min and df['low'].iloc[i] < right_min:
            swing_low.iloc[i] = df['low'].iloc[i]

    return swing_high, swing_low


def detect_double_top(df, swing_high, atr, tolerance_atr, min_bars, max_bars):
    """
    ダブルトップを検出
    条件:
    1. 2つのスイングハイが近い価格（ATR × tolerance以内）
    2. 適切な間隔（min_bars〜max_bars）
    3. ネックライン（2つの山の間の最安値）をブレイク
    """
    patterns = []
    swing_indices = swing_high.dropna().index.tolist()

    for i, first_idx in enumerate(swing_indices[:-1]):
        first_high = swing_high[first_idx]
        first_pos = df.index.get_loc(first_idx)

        for second_idx in swing_indices[i+1:]:
            second_pos = df.index.get_loc(second_idx)
            bars_between = second_pos - first_pos

            # 間隔チェック
            if bars_between < min_bars or bars_between > max_bars:
                continue

            second_high = swing_high[second_idx]
            current_atr = atr.iloc[second_pos] if not pd.isna(atr.iloc[second_pos]) else atr.iloc[first_pos]

            if pd.isna(current_atr):
                continue

            # 価格許容誤差チェック
            price_diff = abs(first_high - second_high)
            if price_diff > current_atr * tolerance_atr:
                continue

            # ネックライン（2つの山の間の最安値）を計算
            between_slice = df.iloc[first_pos:second_pos+1]
            neckline = between_slice['low'].min()

            # ネックラインブレイクを探す
            for j in range(second_pos + 1, min(second_pos + max_bars, len(df))):
                if df['close'].iloc[j] < neckline:
                    # ブレイク検出！
                    price_diff_pct = (price_diff / first_high) * 100
                    patterns.append({
                        'time': df.index[j],
                        'first_high': first_high,
                        'second_high': second_high,
                        'neckline': neckline,
                        'signal_idx': j,
                        'pattern_bars': bars_between,
                        'price_diff_pct': price_diff_pct,
                        'atr': current_atr,
                        'pattern_height': ((first_high + second_high) / 2) - neckline
                    })
                    break

    return pd.DataFrame(patterns)


def detect_double_bottom(df, swing_low, atr, tolerance_atr, min_bars, max_bars):
    """
    ダブルボトムを検出
    条件:
    1. 2つのスイングローが近い価格（ATR × tolerance以内）
    2. 適切な間隔（min_bars〜max_bars）
    3. ネックライン（2つの谷の間の最高値）をブレイク
    """
    patterns = []
    swing_indices = swing_low.dropna().index.tolist()

    for i, first_idx in enumerate(swing_indices[:-1]):
        first_low = swing_low[first_idx]
        first_pos = df.index.get_loc(first_idx)

        for second_idx in swing_indices[i+1:]:
            second_pos = df.index.get_loc(second_idx)
            bars_between = second_pos - first_pos

            # 間隔チェック
            if bars_between < min_bars or bars_between > max_bars:
                continue

            second_low = swing_low[second_idx]
            current_atr = atr.iloc[second_pos] if not pd.isna(atr.iloc[second_pos]) else atr.iloc[first_pos]

            if pd.isna(current_atr):
                continue

            # 価格許容誤差チェック
            price_diff = abs(first_low - second_low)
            if price_diff > current_atr * tolerance_atr:
                continue

            # ネックライン（2つの谷の間の最高値）を計算
            between_slice = df.iloc[first_pos:second_pos+1]
            neckline = between_slice['high'].max()

            # ネックラインブレイクを探す
            for j in range(second_pos + 1, min(second_pos + max_bars, len(df))):
                if df['close'].iloc[j] > neckline:
                    # ブレイク検出！
                    price_diff_pct = (price_diff / first_low) * 100
                    patterns.append({
                        'time': df.index[j],
                        'first_low': first_low,
                        'second_low': second_low,
                        'neckline': neckline,
                        'signal_idx': j,
                        'pattern_bars': bars_between,
                        'price_diff_pct': price_diff_pct,
                        'atr': current_atr,
                        'pattern_height': neckline - ((first_low + second_low) / 2)
                    })
                    break

    return pd.DataFrame(patterns)


def calculate_returns(df, patterns_df, horizons, is_short=False):
    """パターン発生後のリターンを計算"""
    if len(patterns_df) == 0:
        return patterns_df

    patterns_df = patterns_df.copy()

    for minutes in horizons:
        bars_ahead = minutes // 5  # 5分足なので
        returns = []

        for idx, row in patterns_df.iterrows():
            signal_idx = row['signal_idx']
            future_idx = signal_idx + bars_ahead

            if future_idx < len(df):
                entry_price = df['close'].iloc[signal_idx]
                exit_price = df['close'].iloc[future_idx]

                if is_short:  # ダブルトップは売り
                    ret = (entry_price - exit_price) * 100  # pips
                else:  # ダブルボトムは買い
                    ret = (exit_price - entry_price) * 100  # pips
                returns.append(ret)
            else:
                returns.append(np.nan)

        patterns_df[f'return_{minutes}m'] = returns

    return patterns_df


def analyze_by_volatility(patterns_df, atr_series, horizons):
    """ボラティリティ別分析"""
    if len(patterns_df) == 0:
        return None, None

    atr_median = patterns_df['atr'].median()

    high_vol = patterns_df[patterns_df['atr'] > atr_median]
    low_vol = patterns_df[patterns_df['atr'] <= atr_median]

    results = {}
    for label, data in [('high', high_vol), ('low', low_vol), ('all', patterns_df)]:
        if len(data) < 5:
            continue

        results[label] = {}
        for minutes in horizons:
            col = f'return_{minutes}m'
            if col not in data.columns:
                continue

            returns = data[col].dropna()
            if len(returns) < 5:
                continue

            results[label][minutes] = {
                'count': len(returns),
                'win_rate': (returns > 0).mean() * 100,
                'mean_return': returns.mean(),
                'median_return': returns.median(),
                'std': returns.std(),
            }

    return results, atr_median


def main():
    """メイン処理"""
    print("[START] ダブルトップ/ダブルボトム検証を開始...")

    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)

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

        # ダブルトップ検出
        top_patterns = detect_double_top(
            df, swing_high, df['atr'],
            PRICE_TOLERANCE_ATR, MIN_BARS_BETWEEN, MAX_BARS_BETWEEN
        )

        # ダブルボトム検出
        bottom_patterns = detect_double_bottom(
            df, swing_low, df['atr'],
            PRICE_TOLERANCE_ATR, MIN_BARS_BETWEEN, MAX_BARS_BETWEEN
        )

        # リターン計算
        top_patterns = calculate_returns(df, top_patterns, TIME_HORIZONS, is_short=True)
        bottom_patterns = calculate_returns(df, bottom_patterns, TIME_HORIZONS, is_short=False)

        # ボラティリティ別分析
        top_vol_results, _ = analyze_by_volatility(top_patterns, df['atr'], TIME_HORIZONS)
        bottom_vol_results, _ = analyze_by_volatility(bottom_patterns, df['atr'], TIME_HORIZONS)

        # 結果出力
        print(f"\nダブルトップ: {len(top_patterns)}パターン")
        print(f"ダブルボトム: {len(bottom_patterns)}パターン")

        for label, results in [('ダブルトップ', top_vol_results), ('ダブルボトム', bottom_vol_results)]:
            if results and 'all' in results:
                print(f"\n{label} - 全データ:")
                for minutes in TIME_HORIZONS:
                    if minutes in results['all']:
                        r = results['all'][minutes]
                        print(f"  {minutes}分: 勝率{r['win_rate']:.1f}%, 平均{r['mean_return']:.2f}pips")

    finally:
        mt5.shutdown()


if __name__ == "__main__":
    main()