値段の勢いを見てチャレンジ！SOL/USDTで試す「VMA作戦」のしくみ

"""
Variable Moving Average Trading Signal Generator
ボラティリティに応じて動的に調整される移動平均
"""
import pandas as pd
import numpy as np


def calculate_vma_signals(df: pd.DataFrame,
                         period: int = 20,
                         sensitivity: float = 0.5) -> pd.DataFrame:
    """
    Variable Moving Average戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        基本期間（デフォルト: 20）
    sensitivity : float
        感度係数（デフォルト: 0.5）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # 変動率（ボラティリティ）計算
    df['price_change'] = np.abs(df['close'] - df['close'].shift(1))
    
    # 方向性指標
    df['direction'] = df['close'] - df['close'].shift(1)
    df['positive_change'] = np.where(df['direction'] > 0, df['price_change'], 0)
    df['negative_change'] = np.where(df['direction'] < 0, df['price_change'], 0)
    
    # Volatility Index（VI）
    df['pos_sum'] = df['positive_change'].rolling(window=period).sum()
    df['neg_sum'] = df['negative_change'].rolling(window=period).sum()
    df['vi'] = df['pos_sum'] / (df['pos_sum'] + df['neg_sum'] + 0.0001)
    
    # Variable Ratio（VR）
    df['vr'] = 2 * np.abs(df['vi'] - 0.5) * sensitivity
    
    # Alpha（平滑化係数）
    df['alpha'] = 2 / (period * df['vr'] + 1)
    
    # Variable Moving Average
    df['vma'] = df['close'].copy()
    for i in range(1, len(df)):
        if i >= period:
            alpha = df.iloc[i]['alpha']
            if not np.isnan(alpha):
                df.loc[df.index[i], 'vma'] = alpha * df.iloc[i]['close'] + (1 - alpha) * df.iloc[i-1]['vma']
    
    # VMA傾き（トレンド方向）
    df['vma_slope'] = df['vma'] - df['vma'].shift(1)
    df['vma_trend'] = df['vma_slope'].rolling(window=3).mean()
    
    # 価格とVMAの位置関係
    df['price_above_vma'] = df['close'] > df['vma']
    df['price_below_vma'] = df['close'] < df['vma']
    
    # シグナル生成
    df['price_above_prev'] = df['price_above_vma'].shift(1)
    df['price_below_prev'] = df['price_below_vma'].shift(1)
    df['is_buy'] = (
        (df['price_above_vma'] & (df['price_above_prev'] == False)) &  # VMAを上抜け
        (df['vma_trend'] > 0)  # VMAが上昇トレンド
    ) & df['vma'].notna()
    df['is_sell'] = (
        (df['price_below_vma'] & (df['price_below_prev'] == False)) &  # VMAを下抜け
        (df['vma_trend'] < 0)  # VMAが下降トレンド
    ) & df['vma'].notna()
    
    # 不要カラム削除
    df.drop(['price_change', 'direction', 'positive_change', 'negative_change',
             'pos_sum', 'neg_sum', 'vi', 'vr', 'alpha', 'vma_slope', 'vma_trend',
             'price_above_vma', 'price_below_vma', 'price_above_prev', 'price_below_prev'], 
            axis=1, inplace=True, errors='ignore')
    
    return df