ビットコインの値段が大きく動いた！元に戻る力を利用した取引の作戦

#!/usr/bin/env python3
"""
VWAP回帰戦略
価格がVWAPから乖離したら回帰を狙う
"""
import pandas as pd
import numpy as np


def calculate_vwap_signals(df: pd.DataFrame, deviation_threshold: float = 2.0) -> pd.DataFrame:
    """
    VWAP回帰シグナルを生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    deviation_threshold : float
        乖離率の閾値（％）（デフォルト: 2.0）
    
    Returns:
    --------
    pd.DataFrame
        シグナル列が追加されたDataFrame
    """
    df = df.copy()
    
    # 日付でグループ化（日次VWAPを計算）
    df['date'] = pd.to_datetime(df.index).date
    
    # 各日のVWAPを計算
    df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
    df['pv'] = df['typical_price'] * df['volume']
    
    # 累積値を計算（日次でリセット）
    df['cumulative_pv'] = df.groupby('date')['pv'].cumsum()
    df['cumulative_volume'] = df.groupby('date')['volume'].cumsum()
    
    # VWAP計算
    df['vwap'] = df['cumulative_pv'] / df['cumulative_volume']
    
    # VWAPからの乖離率
    df['vwap_deviation'] = ((df['close'] - df['vwap']) / df['vwap']) * 100
    
    # 標準偏差バンド（オプション）
    df['vwap_std'] = df.groupby('date')['typical_price'].transform(lambda x: x.expanding().std())
    df['upper_band'] = df['vwap'] + (df['vwap_std'] * 2)
    df['lower_band'] = df['vwap'] - (df['vwap_std'] * 2)
    
    # シグナル生成（乖離からの回帰狙い）
    df['is_buy'] = (
        (df['vwap_deviation'] < -deviation_threshold) &  # VWAPより下に乖離
        (df['vwap_deviation'].shift(1) >= -deviation_threshold)  # 閾値を下回った瞬間
    )
    
    df['is_sell'] = (
        (df['vwap_deviation'] > deviation_threshold) &  # VWAPより上に乖離
        (df['vwap_deviation'].shift(1) <= deviation_threshold)  # 閾値を上回った瞬間
    )
    
    # NaN値をFalseに置換
    df['is_buy'] = df['is_buy'].fillna(False)
    df['is_sell'] = df['is_sell'].fillna(False)
    
    # デバッグ情報
    print(f"VWAP乖離閾値: {deviation_threshold}%")
    print(f"買いシグナル数: {df['is_buy'].sum()}")
    print(f"売りシグナル数: {df['is_sell'].sum()}")
    
    return df