ビットコインの値段の波を読もう！ジグザグ・リバーサル作戦のひみつ

#!/usr/bin/env python3
"""
ジグザグリバーサル戦略
価格の山と谷を検出して転換点を特定
"""
import pandas as pd
import numpy as np


def calculate_zigzag_signals(df: pd.DataFrame, deviation: float = 1.0) -> pd.DataFrame:
    """
    ジグザグリバーサルシグナルを生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    deviation : float
        転換判定の最小変動率（％）（デフォルト: 5.0）
    
    Returns:
    --------
    pd.DataFrame
        シグナル列が追加されたDataFrame
    """
    df = df.copy()
    
    # ジグザグポイントの初期化
    df['zigzag'] = np.nan
    df['pivot_type'] = ''  # 'high' or 'low'
    
    # 最初のポイントを設定
    df.loc[df.index[0], 'zigzag'] = df.loc[df.index[0], 'close']
    last_pivot_idx = 0
    last_pivot_val = df.loc[df.index[0], 'close']
    last_pivot_type = None
    
    for i in range(1, len(df)):
        current_idx = df.index[i]
        current_high = df.loc[current_idx, 'high']
        current_low = df.loc[current_idx, 'low']
        
        # 変動率計算
        high_change = ((current_high - last_pivot_val) / last_pivot_val) * 100
        low_change = ((current_low - last_pivot_val) / last_pivot_val) * 100
        
        if last_pivot_type is None:
            # 最初の転換点を決定
            if high_change >= deviation:
                df.loc[current_idx, 'zigzag'] = current_high
                df.loc[current_idx, 'pivot_type'] = 'high'
                last_pivot_idx = i
                last_pivot_val = current_high
                last_pivot_type = 'high'
            elif low_change <= -deviation:
                df.loc[current_idx, 'zigzag'] = current_low
                df.loc[current_idx, 'pivot_type'] = 'low'
                last_pivot_idx = i
                last_pivot_val = current_low
                last_pivot_type = 'low'
        
        elif last_pivot_type == 'high':
            # 前回が高値の場合、安値転換を探す
            if low_change <= -deviation:
                df.loc[current_idx, 'zigzag'] = current_low
                df.loc[current_idx, 'pivot_type'] = 'low'
                last_pivot_idx = i
                last_pivot_val = current_low
                last_pivot_type = 'low'
            # より高い高値が出てもピボットは更新しない（リーク防止）
            # elif current_high > last_pivot_val:
            #     pass
        
        else:  # last_pivot_type == 'low'
            # 前回が安値の場合、高値転換を探す
            if high_change >= deviation:
                df.loc[current_idx, 'zigzag'] = current_high
                df.loc[current_idx, 'pivot_type'] = 'high'
                last_pivot_idx = i
                last_pivot_val = current_high
                last_pivot_type = 'high'
            # より低い安値が出てもピボットは更新しない（リーク防止）
            # elif current_low < last_pivot_val:
            #     pass
    
    # シグナル生成（ピボットポイントの次のバーで逆張り）
    # ピボットが確定してから1本遅れてシグナルを出す
    df['is_buy'] = df['pivot_type'].shift(1) == 'low'
    df['is_sell'] = df['pivot_type'].shift(1) == 'high'
    
    # NaN値をFalseに置換
    df['is_buy'] = df['is_buy'].fillna(False)
    df['is_sell'] = df['is_sell'].fillna(False)
    
    print(f"ジグザグリバーサル: 偏差={deviation}%")
    print(f"買いシグナル数: {df['is_buy'].sum()}")
    print(f"売りシグナル数: {df['is_sell'].sum()}")
    
    return df