ビットコインの未来の動きを予測？「線形回帰チャネル」で取引にチャレンジしてみよう！

"""
Linear Regression Channel Trading Signal
線形回帰チャネルを使用した統計的戦略
"""
import pandas as pd
import numpy as np


def calculate_linear_regression_channel_signals(df: pd.DataFrame,
                                               period: int = 100,
                                               deviation: float = 2.0) -> pd.DataFrame:
    """
    Linear Regression Channel戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        回帰期間（デフォルト: 100）
    deviation : float
        標準偏差の倍数（デフォルト: 2.0）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # 線形回帰の計算
    df['lr_line'] = pd.Series(index=df.index, dtype=float)
    df['lr_slope'] = pd.Series(index=df.index, dtype=float)
    df['lr_upper'] = pd.Series(index=df.index, dtype=float)
    df['lr_lower'] = pd.Series(index=df.index, dtype=float)
    df['r_squared'] = pd.Series(index=df.index, dtype=float)
    
    for i in range(period - 1, len(df)):
        # 期間内のデータ
        y = df['close'].iloc[i - period + 1:i + 1].values
        x = np.arange(period)
        
        # 線形回帰（numpy使用）
        A = np.vstack([x, np.ones(len(x))]).T
        m, c = np.linalg.lstsq(A, y, rcond=None)[0]
        slope, intercept = m, c
        
        # 相関係数
        correlation = np.corrcoef(x, y)[0, 1]
        r_value = correlation
        
        # 回帰線の値
        regression_line = slope * x + intercept
        current_value = slope * (period - 1) + intercept
        
        # 残差の標準偏差
        residuals = y - regression_line
        std_dev = np.std(residuals)
        
        # 値を保存
        df.loc[df.index[i], 'lr_line'] = current_value
        df.loc[df.index[i], 'lr_slope'] = slope
        df.loc[df.index[i], 'lr_upper'] = current_value + deviation * std_dev
        df.loc[df.index[i], 'lr_lower'] = current_value - deviation * std_dev
        df.loc[df.index[i], 'r_squared'] = r_value ** 2
    
    # 短期の線形回帰（エントリータイミング用）
    short_period = int(period / 4)
    df['lr_short'] = pd.Series(index=df.index, dtype=float)
    df['lr_short_slope'] = pd.Series(index=df.index, dtype=float)
    
    for i in range(short_period - 1, len(df)):
        y = df['close'].iloc[i - short_period + 1:i + 1].values
        x = np.arange(short_period)
        A = np.vstack([x, np.ones(len(x))]).T
        m, c = np.linalg.lstsq(A, y, rcond=None)[0]
        slope, intercept = m, c
        current_value = slope * (short_period - 1) + intercept
        df.loc[df.index[i], 'lr_short'] = current_value
        df.loc[df.index[i], 'lr_short_slope'] = slope
    
        # シグナル初期化
    df['signal'] = 0
    df['is_buy'] = False
    df['is_sell'] = False
    
    # 現在のポジション状態を追跡
    position = 0
    
    for i in range(period, len(df)):
        # 買いシグナル（下部チャネルからの反発）
        if position <= 0 and df['close'].iloc[i] <= df['lr_lower'].iloc[i] and df['lr_slope'].iloc[i] > 0:
            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['lr_upper'].iloc[i] and df['lr_slope'].iloc[i] < 0:
            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)
    
    # エグジットシグナル
    df['exit_signal'] = 0
    
    # 回帰線タッチでエグジット
    for i in range(period, len(df)):
        # ロングポジションのエグジット
        if df['signal'].iloc[i] == 1:
            if df['close'].iloc[i] >= df['lr_line'].iloc[i] or df['close'].iloc[i] <= df['lr_lower'].iloc[i] - df['lr_lower'].iloc[i] * 0.02:
                df.loc[df.index[i], 'exit_signal'] = -1
        
        # ショートポジションのエグジット
        if df['signal'].iloc[i] == -1:
            if df['close'].iloc[i] <= df['lr_line'].iloc[i] or df['close'].iloc[i] >= df['lr_upper'].iloc[i] + df['lr_upper'].iloc[i] * 0.02:
                df.loc[df.index[i], 'exit_signal'] = 1
    
    return df


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


def get_strategy_description() -> str:
    """戦略の説明を返す"""
    return "線形回帰チャネルの上下限でのリバーサルを狙う統計的戦略"