値段の『いきおい』で判断！仮想通貨の売り買いタイミングを見つける方法

"""
Kase Peak Oscillator Trading Signal Generator
トレンドの転換点を検出する高度なオシレーター
"""
import pandas as pd
import numpy as np


def calculate_kase_signals(df: pd.DataFrame,
                          period: int = 30,
                          smooth: int = 3) -> pd.DataFrame:
    """
    Kase Peak Oscillator戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        ルックバック期間（デフォルト: 30）
    smooth : int
        平滑化期間（デフォルト: 3）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # True Range計算
    df['h_l'] = df['high'] - df['low']
    df['h_c'] = np.abs(df['high'] - df['close'].shift(1))
    df['l_c'] = np.abs(df['low'] - df['close'].shift(1))
    df['true_range'] = df[['h_l', 'h_c', 'l_c']].max(axis=1)
    
    # Average True Range
    df['atr'] = df['true_range'].rolling(window=period).mean()
    
    # 上昇・下降レンジ
    df['up_move'] = df['high'] - df['high'].shift(1)
    df['down_move'] = df['low'].shift(1) - df['low']
    
    # 正規化された動き
    df['up_norm'] = df['up_move'] / (df['atr'] + 0.0001)
    df['down_norm'] = df['down_move'] / (df['atr'] + 0.0001)
    
    # KPO計算
    df['kpo_raw'] = df['up_norm'].rolling(window=period).sum() - df['down_norm'].rolling(window=period).sum()
    df['kpo'] = df['kpo_raw'].ewm(span=smooth, adjust=False).mean()
    
    # 標準偏差バンド
    df['kpo_std'] = df['kpo'].rolling(window=period).std()
    df['kpo_upper'] = df['kpo_std'] * 1.5
    df['kpo_lower'] = -df['kpo_std'] * 1.5
    
    # シグナル生成
    df['kpo_prev'] = df['kpo'].shift(1)
    df['is_buy'] = (
        (df['kpo'] > df['kpo_lower']) & 
        (df['kpo_prev'] <= df['kpo_lower']) & 
        df['kpo'].notna()
    )
    df['is_sell'] = (
        (df['kpo'] < df['kpo_upper']) & 
        (df['kpo_prev'] >= df['kpo_upper']) & 
        df['kpo'].notna()
    )
    
    # 不要カラム削除
    df.drop(['h_l', 'h_c', 'l_c', 'true_range', 'up_move', 'down_move',
             'up_norm', 'down_norm', 'kpo_raw', 'kpo_prev'], axis=1, inplace=True, errors='ignore')
    
    return df