ビットコインで挑戦！「かしこいチャネル」作戦をわかりやすく解説します

"""
Adaptive Channel Index Trading Signal Generator
市場状況に応じて動的に調整されるチャネル
"""
import pandas as pd
import numpy as np


def calculate_adaptive_channel_signals(df: pd.DataFrame,
                                      period: int = 20,
                                      multiplier: float = 2.0) -> pd.DataFrame:
    """
    Adaptive Channel Index戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        基本期間（デフォルト: 20）
    multiplier : float
        チャネル幅の倍率（デフォルト: 2.0）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # ハイローチャネル
    df['highest'] = df['high'].rolling(window=period).max()
    df['lowest'] = df['low'].rolling(window=period).min()
    df['hl_mid'] = (df['highest'] + df['lowest']) / 2
    
    # ATRベースの適応幅
    df['h_l'] = df['high'] - df['low']
    df['h_c'] = np.abs(df['high'] - df['close'].shift(1))
    df['l_c'] = np.abs(df['low'] - df['close'].shift(1))
    df['true_range'] = df[['h_l', 'h_c', 'l_c']].max(axis=1)
    df['atr'] = df['true_range'].rolling(window=period).mean()
    
    # 適応係数（ボラティリティ比率）
    df['atr_ratio'] = df['atr'] / df['atr'].rolling(window=period * 2).mean()
    df['adaptive_mult'] = multiplier * df['atr_ratio']
    
    # 適応チャネル
    df['channel_upper'] = df['hl_mid'] + df['adaptive_mult'] * df['atr']
    df['channel_lower'] = df['hl_mid'] - df['adaptive_mult'] * df['atr']
    
    # チャネル内での位置（ACI）
    df['aci'] = (df['close'] - df['channel_lower']) / (df['channel_upper'] - df['channel_lower'] + 0.0001)
    
    # ACIのスムージング
    df['aci_smooth'] = df['aci'].rolling(window=3).mean()
    
    # シグナル生成
    df['aci_prev'] = df['aci_smooth'].shift(1)
    df['is_buy'] = (
        ((df['aci_smooth'] > 0.2) & (df['aci_prev'] <= 0.2)) |  # 下部チャネルからの反発
        ((df['close'] > df['channel_lower']) & (df['close'].shift(1) <= df['channel_lower'].shift(1)))
    ) & df['aci_smooth'].notna()
    df['is_sell'] = (
        ((df['aci_smooth'] < 0.8) & (df['aci_prev'] >= 0.8)) |  # 上部チャネルからの反落
        ((df['close'] < df['channel_upper']) & (df['close'].shift(1) >= df['channel_upper'].shift(1)))
    ) & df['aci_smooth'].notna()
    
    # 不要カラム削除
    df.drop(['highest', 'lowest', 'hl_mid', 'h_l', 'h_c', 'l_c', 'true_range',
             'atr_ratio', 'adaptive_mult', 'aci_smooth', 'aci_prev'], 
            axis=1, inplace=True, errors='ignore')
    
    return df