ビットコインが急に動く時をねらえ！「取引量ブレイクアウト」作戦

#!/usr/bin/env python3
"""
出来高ブレイクアウト戦略
平均出来高の倍数以上でブレイクアウト確認
"""
import pandas as pd
import numpy as np


def calculate_volume_breakout_signals(df: pd.DataFrame, volume_period: int = 20, 
                                     volume_multiplier: float = 1.5, price_threshold: float = 0.002) -> pd.DataFrame:
    """
    出来高ブレイクアウトシグナルを生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    volume_period : int
        出来高移動平均期間（デフォルト: 20）
    volume_multiplier : float
        出来高倍率（デフォルト: 2.0）
    price_threshold : float
        価格変動閾値（デフォルト: 0.01 = 1%）
    
    Returns:
    --------
    pd.DataFrame
        シグナル列が追加されたDataFrame
    """
    df = df.copy()
    
    # 出来高移動平均
    df['volume_ma'] = df['volume'].rolling(window=volume_period).mean()
    
    # 出来高比率
    df['volume_ratio'] = df['volume'] / df['volume_ma']
    
    # 価格変動率
    df['price_change'] = (df['close'] - df['close'].shift(1)) / df['close'].shift(1)
    
    # シグナル生成
    df['is_buy'] = (
        (df['volume_ratio'] > volume_multiplier) &  # 出来高急増
        (df['price_change'] > price_threshold)  # 価格上昇
    )
    
    df['is_sell'] = (
        (df['volume_ratio'] > volume_multiplier) &  # 出来高急増
        (df['price_change'] < -price_threshold)  # 価格下降
    )
    
    # NaN値をFalseに置換
    df['is_buy'] = df['is_buy'].fillna(False)
    df['is_sell'] = df['is_sell'].fillna(False)
    
    print(f"出来高ブレイクアウト: 期間={volume_period}, 倍率={volume_multiplier}")
    print(f"買いシグナル数: {df['is_buy'].sum()}")
    print(f"売りシグナル数: {df['is_sell'].sum()}")
    
    return df