【中学生にもわかる！】仮想通貨の売り買い作戦「ガッピー」の結果と学んだこと

"""
Guppy Multiple Moving Average Trading Signal Generator
12本のEMAを使用したトレンド判定戦略
"""
import pandas as pd
import numpy as np


def calculate_guppy_signals(df: pd.DataFrame) -> pd.DataFrame:
    """
    Guppy MMA戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # 短期EMAグループ（トレーダー）
    short_periods = [3, 5, 8, 10, 12, 15]
    # 長期EMAグループ（投資家）
    long_periods = [30, 35, 40, 45, 50, 60]
    
    # 各EMAを計算
    for period in short_periods:
        df[f'ema_short_{period}'] = df['close'].ewm(span=period, adjust=False).mean()
    
    for period in long_periods:
        df[f'ema_long_{period}'] = df['close'].ewm(span=period, adjust=False).mean()
    
    # 短期グループと長期グループの平均
    short_cols = [f'ema_short_{p}' for p in short_periods]
    long_cols = [f'ema_long_{p}' for p in long_periods]
    
    df['short_avg'] = df[short_cols].mean(axis=1)
    df['long_avg'] = df[long_cols].mean(axis=1)
    
    # グループ間の距離（トレンド強度）
    df['separation'] = df['short_avg'] - df['long_avg']
    df['separation_pct'] = (df['separation'] / df['long_avg']) * 100
    
    # 短期グループの収束度（圧縮度）
    df['short_max'] = df[short_cols].max(axis=1)
    df['short_min'] = df[short_cols].min(axis=1)
    df['short_compression'] = (df['short_max'] - df['short_min']) / df['short_avg']
    
    # シグナル生成
    df['separation_prev'] = df['separation'].shift(1)
    df['is_buy'] = (
        (df['separation'] > 0) & 
        (df['separation_prev'] <= 0) & 
        (df['short_compression'] < 0.02) &  # 短期EMAが収束
        df['separation'].notna()
    )
    df['is_sell'] = (
        (df['separation'] < 0) & 
        (df['separation_prev'] >= 0) & 
        (df['short_compression'] < 0.02) &  # 短期EMAが収束
        df['separation'].notna()
    )
    
    # 不要カラム削除
    drop_cols = short_cols + long_cols + ['short_max', 'short_min', 'separation_prev']
    df.drop(drop_cols, axis=1, inplace=True, errors='ignore')
    
    return df