ビットコインの値動きをキャッチ！「プライスチャネル」で投資に挑戦してみよう

"""
Price Channel Trading Signal
価格チャネルブレイクアウト戦略
"""
import pandas as pd
import numpy as np


def calculate_price_channel_signals(df: pd.DataFrame,
                                   upper_period: int = 20,
                                   lower_period: int = 20,
                                   exit_period: int = 10) -> pd.DataFrame:
    """
    Price Channel戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    upper_period : int
        上部チャネル期間（デフォルト: 20）
    lower_period : int
        下部チャネル期間（デフォルト: 20）
    exit_period : int
        エグジット用チャネル期間（デフォルト: 10）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # エントリー用チャネル
    df['upper_channel'] = df['high'].shift(1).rolling(window=upper_period).max()
    df['lower_channel'] = df['low'].shift(1).rolling(window=lower_period).min()
    df['middle_channel'] = (df['upper_channel'] + df['lower_channel']) / 2
    
    # エグジット用チャネル（より短期）
    df['upper_exit'] = df['high'].shift(1).rolling(window=exit_period).max()
    df['lower_exit'] = df['low'].shift(1).rolling(window=exit_period).min()
    
    # チャネル幅（ボラティリティ指標）
    df['channel_width'] = df['upper_channel'] - df['lower_channel']
    df['channel_width_ma'] = df['channel_width'].rolling(window=upper_period).mean()
    
    # シグナル初期化
    df['signal'] = 0
    df['is_buy'] = False
    df['is_sell'] = False
    
    # 現在のポジション状態を追跡
    position = 0
    
    for i in range(max(upper_period, lower_period), len(df)):
        # 買いシグナル（上部チャネルブレイクアウト）
        if position <= 0 and df['close'].iloc[i] > df['upper_channel'].iloc[i]:
            df.loc[df.index[i], 'is_buy'] = True
            df.loc[df.index[i], 'signal'] = 1
            position = 1
            
        # 売りシグナル（下部チャネルブレイクアウト）  
        elif position >= 0 and df['close'].iloc[i] < df['lower_channel'].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)
    
    return df


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


def get_strategy_description() -> str:
    """戦略の説明を返す"""
    return "価格チャネルのブレイクアウトを狙うトレンドフォロー戦略"