ビットコイン値段の「波」に乗る作戦！ DPOサイクル作戦の結果と反省点

#!/usr/bin/env python3
"""
DPO（Detrended Price Oscillator）サイクル戦略
トレンドを除去して価格サイクルを検出
"""
import pandas as pd
import numpy as np


def calculate_dpo_signals(df: pd.DataFrame, dpo_period: int = 20) -> pd.DataFrame:
    """
    DPOサイクルシグナルを生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    dpo_period : int
        DPO計算期間（デフォルト: 20）
    
    Returns:
    --------
    pd.DataFrame
        シグナル列が追加されたDataFrame
    """
    df = df.copy()
    
    # 移動平均計算（中心化）
    shift_period = int(dpo_period / 2) + 1
    df['dpo_ma'] = df['close'].rolling(window=dpo_period).mean().shift(-shift_period)
    
    # DPO計算（現在価格 - シフトされた移動平均）
    df['dpo'] = df['close'] - df['dpo_ma'].shift(shift_period)
    
    # DPOの移動平均（スムージング）
    df['dpo_smooth'] = df['dpo'].rolling(window=3).mean()
    
    # サイクルの山と谷を検出（未来データを参照しない）
    # 前2本と比較して判定
    df['dpo_peak'] = (df['dpo_smooth'].shift(1) > df['dpo_smooth'].shift(2)) & (df['dpo_smooth'] < df['dpo_smooth'].shift(1))
    df['dpo_trough'] = (df['dpo_smooth'].shift(1) < df['dpo_smooth'].shift(2)) & (df['dpo_smooth'] > df['dpo_smooth'].shift(1))
    
    # シグナル生成（サイクルの谷で買い、山で売り）
    df['is_buy'] = df['dpo_trough'] & (df['dpo_smooth'] < 0)
    df['is_sell'] = df['dpo_peak'] & (df['dpo_smooth'] > 0)
    
    # NaN値をFalseに置換
    df['is_buy'] = df['is_buy'].fillna(False)
    df['is_sell'] = df['is_sell'].fillna(False)
    
    print(f"DPOサイクル: 期間={dpo_period}")
    print(f"買いシグナル数: {df['is_buy'].sum()}")
    print(f"売りシグナル数: {df['is_sell'].sum()}")
    
    return df