ビットコインの「値動きの勢い」作戦！ うまくいった？ 徹底チェック

"""
Volatility System Trading Signal
ボラティリティ拡大・縮小を利用した戦略
"""
import pandas as pd
import numpy as np


def calculate_volatility_system_signals(df: pd.DataFrame,
                                       short_period: int = 6,
                                       long_period: int = 100,
                                       threshold: float = 0.5) -> pd.DataFrame:
    """
    Volatility System戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    short_period : int
        短期ボラティリティ期間（デフォルト: 6）
    long_period : int
        長期ボラティリティ期間（デフォルト: 100）
    threshold : float
        ボラティリティ比率閾値（デフォルト: 0.5）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # Historical Volatility計算
    df['returns'] = np.log(df['close'] / df['close'].shift(1))
    df['hv_short'] = df['returns'].rolling(window=short_period).std() * np.sqrt(252)
    df['hv_long'] = df['returns'].rolling(window=long_period).std() * np.sqrt(252)
    
    # ボラティリティ比率
    df['vol_ratio'] = df['hv_short'] / df['hv_long']
    
    # Volatility Cone（ボラティリティの分布）
    percentiles = [10, 25, 50, 75, 90]
    for p in percentiles:
        df[f'vol_p{p}'] = df['hv_long'].rolling(window=252).quantile(p/100)
    
    # 現在のボラティリティ位置
    df['vol_rank'] = df['hv_short'].rolling(window=252).rank(pct=True)
    
    # ATRベースのボラティリティ
    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)
    df['atr'] = df['true_range'].rolling(window=short_period).mean()
    df['atr_long'] = df['true_range'].rolling(window=long_period).mean()
    df['atr_ratio'] = df['atr'] / df['atr_long']
    
    # ボリンジャーバンドによるスクイーズ検出
    df['bb_middle'] = df['close'].rolling(window=20).mean()
    df['bb_std'] = df['close'].rolling(window=20).std()
    df['bb_upper'] = df['bb_middle'] + 2 * df['bb_std']
    df['bb_lower'] = df['bb_middle'] - 2 * df['bb_std']
    df['bb_width'] = df['bb_upper'] - df['bb_lower']
    df['bb_width_ma'] = df['bb_width'].rolling(window=long_period).mean()
    df['squeeze'] = df['bb_width'] < df['bb_width_ma'] * threshold
    
        # シグナル初期化
    df['signal'] = 0
    df['is_buy'] = False
    df['is_sell'] = False
    
    # 現在のポジション状態を追跡
    position = 0
    
    for i in range(long_period, len(df)):
        # 買いシグナル（ボラティリティ拡大＋価格上昇）
        if position <= 0 and df['vol_ratio'].iloc[i] > threshold and df['close'].iloc[i] > df['bb_middle'].iloc[i]:
            df.loc[df.index[i], 'is_buy'] = True
            df.loc[df.index[i], 'signal'] = 1
            position = 1
            
        # 売りシグナル（ボラティリティ拡大＋価格下落）  
        elif position >= 0 and df['vol_ratio'].iloc[i] > threshold and df['close'].iloc[i] < df['bb_middle'].iloc[i]:
            df.loc[df.index[i], 'is_sell'] = True
            df.loc[df.index[i], 'signal'] = -1
            position = -1
        else:
            # ポジション維持
            df.loc[df.index[i], 'signal'] = position
    
    # NaN値を0で埋める
    df['signal'] = df['signal'].fillna(0)
    
    # エグジットシグナル
    df['exit_signal'] = 0
    
    # ボラティリティ縮小でエグジット
    df.loc[(df['signal'] == 1) & (df['vol_ratio'] < threshold), 'exit_signal'] = -1
    df.loc[(df['signal'] == -1) & (df['vol_ratio'] < threshold), 'exit_signal'] = 1
    
    return df


def get_strategy_name() -> str:
    """戦略名を返す"""
    return "Volatility System"


def get_strategy_description() -> str:
    """戦略の説明を返す"""
    return "ボラティリティの拡大・縮小サイクルを利用した戦略"