ビットコインの「勢い」で売り買い！「ネットボリューム戦略」をわかりやすく解説

"""
Net Volume Trading Signal
ネットボリューム指標によるトレンド判定
"""
import pandas as pd
import numpy as np


def calculate_net_volume_signals(df: pd.DataFrame,
                              period: int = 14) -> pd.DataFrame:
    """
    Net Volume戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        計算期間（デフォルト: 14）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # 価格変化
    df['price_change'] = df['close'] - df['close'].shift(1)
    
    # ネットボリューム
    df['net_volume'] = np.where(df['price_change'] > 0, df['volume'], 
                                np.where(df['price_change'] < 0, -df['volume'], 0))
    
    # 累積ネットボリューム
    df['cumulative_net_volume'] = df['net_volume'].cumsum()
    
    # ネットボリュームのMA
    df['nv_ma'] = df['cumulative_net_volume'].rolling(window=period).mean()
    
    # シグナル初期化
    df['signal'] = 0
    df['is_buy'] = False
    df['is_sell'] = False
    
    position = 0
    
    for i in range(period + 1, len(df)):
        # 買いシグナル
        if position <= 0 and df['cumulative_net_volume'].iloc[i] > df['nv_ma'].iloc[i] and \
           df['cumulative_net_volume'].iloc[i-1] <= df['nv_ma'].iloc[i-1]:
            df.loc[df.index[i], 'is_buy'] = True
            df.loc[df.index[i], 'signal'] = 1
            position = 1
            
        # 売りシグナル
        elif position >= 0 and df['cumulative_net_volume'].iloc[i] < df['nv_ma'].iloc[i] and \
             df['cumulative_net_volume'].iloc[i-1] >= df['nv_ma'].iloc[i-1]:
            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
    
    df['signal'] = df['signal'].fillna(0)
    
    return df


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


def get_strategy_description() -> str:
    """戦略の説明を返す"""
    return "ネットボリューム指標によるトレンド判定戦略"