ビットコインの値動きの「勢い」をつかまえる！ASI作戦のやさしい解説

"""
Accumulated Swing Index Trading Signal
ASIを使用したトレンド追従戦略
"""
import pandas as pd
import numpy as np


def calculate_accumulated_swing_index_signals(df: pd.DataFrame,
                                            limit_move: float = 0.5) -> pd.DataFrame:
    """
    Accumulated Swing Index戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    limit_move : float
        制限価格動き（デフォルト: 0.5）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # Swing Index計算
    df['c1'] = df['close'].shift(1)
    df['o1'] = df['open'].shift(1)
    df['h1'] = df['high'].shift(1)
    df['l1'] = df['low'].shift(1)
    
    df['k'] = np.maximum(df['high'] - df['c1'], df['c1'] - df['low'])
    
    # R計算
    df['r'] = 0.0
    df['hl'] = np.abs(df['high'] - df['c1'])
    df['lc'] = np.abs(df['low'] - df['c1'])
    df['hc'] = np.abs(df['high'] - df['low'])
    
    # 条件分岐
    mask1 = (df['hl'] >= df['lc']) & (df['hl'] >= df['hc'])
    mask2 = (df['lc'] >= df['hl']) & (df['lc'] >= df['hc'])
    mask3 = (df['hc'] >= df['hl']) & (df['hc'] >= df['lc'])
    
    df.loc[mask1, 'r'] = df.loc[mask1, 'hl'] - 0.5 * df.loc[mask1, 'lc'] + 0.25 * (df.loc[mask1, 'c1'] - df.loc[mask1, 'o1'])
    df.loc[mask2, 'r'] = df.loc[mask2, 'lc'] - 0.5 * df.loc[mask2, 'hl'] + 0.25 * (df.loc[mask2, 'c1'] - df.loc[mask2, 'o1'])
    df.loc[mask3, 'r'] = df.loc[mask3, 'hc'] + 0.25 * (df.loc[mask3, 'c1'] - df.loc[mask3, 'o1'])
    
    # Swing Index
    df['si'] = 50 * ((df['close'] - df['c1']) + 0.5 * (df['close'] - df['open']) + 0.25 * (df['c1'] - df['o1'])) / (df['r'] + 0.0001) * (df['k'] / limit_move)
    df['asi'] = df['si'].cumsum()
    
    # ASIトレンド
    df['asi_ma'] = df['asi'].rolling(window=10).mean()
    
    # シグナル初期化
    df['signal'] = 0
    df['is_buy'] = False
    df['is_sell'] = False
    
    position = 0
    
    for i in range(10, len(df)):
        # 買いシグナル
        if position <= 0 and df['asi'].iloc[i] > df['asi_ma'].iloc[i] and df['asi'].iloc[i] > df['asi'].iloc[i-1]:
            df.loc[df.index[i], 'is_buy'] = True
            df.loc[df.index[i], 'signal'] = 1
            position = 1
            
        # 売りシグナル
        elif position >= 0 and df['asi'].iloc[i] < df['asi_ma'].iloc[i] and df['asi'].iloc[i] < df['asi'].iloc[i-1]:
            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
    
    df['signal'] = df['signal'].fillna(0)
    
    return df


def get_strategy_name() -> str:
    """戦略名を返す"""
    return "Accumulated Swing Index"


def get_strategy_description() -> str:
    """戦略の説明を返す"""
    return "ASIを使用したトレンド追従戦略"