ビットコインの値動きのナゾに挑戦！『DPO』で未来を予想する方法

"""
Detrended Price Oscillator Trading Signal Generator
価格からトレンドを除去してサイクルを検出
"""
import pandas as pd
import numpy as np


def calculate_detrended_signals(df: pd.DataFrame,
                               period: int = 20,
                               ma_type: str = 'sma') -> pd.DataFrame:
    """
    Detrended Price Oscillator戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        移動平均期間（デフォルト: 20）
    ma_type : str
        移動平均タイプ（'sma' or 'ema'、デフォルト: 'sma'）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # シフト期間
    shift_period = period // 2 + 1
    
    # 移動平均計算
    if ma_type == 'ema':
        df['ma'] = df['close'].ewm(span=period, adjust=False).mean()
    else:
        df['ma'] = df['close'].rolling(window=period).mean()
    
    # シフトした移動平均
    df['ma_shifted'] = df['ma'].shift(-shift_period)
    
    # DPO計算
    df['dpo'] = df['close'] - df['ma_shifted'].shift(shift_period)
    
    # DPOの移動平均（シグナルライン）
    df['dpo_signal'] = df['dpo'].rolling(window=3).mean()
    
    # DPOのピークとボトム検出（未来データを参照しない）
    # 前2本と比較して判定
    df['dpo_peak'] = (df['dpo'].shift(1) > df['dpo'].shift(2)) & (df['dpo'] < df['dpo'].shift(1))
    df['dpo_bottom'] = (df['dpo'].shift(1) < df['dpo'].shift(2)) & (df['dpo'] > df['dpo'].shift(1))
    
    # サイクル長推定
    df['cycle_count'] = 0
    peak_positions = df[df['dpo_peak']].index
    if len(peak_positions) > 1:
        cycle_lengths = [peak_positions[i] - peak_positions[i-1] for i in range(1, len(peak_positions))]
        if cycle_lengths:
            avg_cycle = np.mean(cycle_lengths)
            df['cycle_length'] = avg_cycle
    
    # ゼロラインクロス
    df['dpo_prev'] = df['dpo'].shift(1)
    df['zero_cross_up'] = (df['dpo'] > 0) & (df['dpo_prev'] <= 0)
    df['zero_cross_down'] = (df['dpo'] < 0) & (df['dpo_prev'] >= 0)
    
    # シグナル生成
    df['is_buy'] = (
        df['zero_cross_up'] |
        (df['dpo_bottom'] & (df['dpo'] < -np.std(df['dpo'].dropna()) * 0.5))
    ) & df['dpo'].notna()
    df['is_sell'] = (
        df['zero_cross_down'] |
        (df['dpo_peak'] & (df['dpo'] > np.std(df['dpo'].dropna()) * 0.5))
    ) & df['dpo'].notna()
    
    # 不要カラム削除
    df.drop(['ma', 'ma_shifted', 'dpo_signal', 'dpo_peak', 'dpo_bottom',
             'cycle_count', 'dpo_prev', 'zero_cross_up', 'zero_cross_down'], 
            axis=1, inplace=True, errors='ignore')
    if 'cycle_length' in df.columns:
        df.drop(['cycle_length'], axis=1, inplace=True, errors='ignore')
    
    return df