市場の『気分』を読んでみよう！BNB/USDTの売り買い作戦をわかりやすく解説します

"""
Market Meanness Index Trading Signal Generator
市場の平均回帰性を測定する指標
"""
import pandas as pd
import numpy as np


def calculate_mmi_signals(df: pd.DataFrame,
                         period: int = 200,
                         lookback: int = 20) -> pd.DataFrame:
    """
    Market Meanness Index戦略のシグナル生成
    
    Parameters:
    -----------
    df : pd.DataFrame
        OHLCVデータ
    period : int
        MMI計算期間（デフォルト: 200）
    lookback : int
        中央値計算期間（デフォルト: 20）
    
    Returns:
    --------
    pd.DataFrame
        シグナルが追加されたDataFrame
    """
    df = df.copy()
    
    # 中央値計算
    df['median'] = df['close'].rolling(window=lookback).median()
    
    # 価格が中央値より上か下か
    df['above_median'] = (df['close'] > df['median']).astype(int)
    df['below_median'] = (df['close'] < df['median']).astype(int)
    
    # 連続性のカウント
    df['consecutive'] = 0
    current_streak = 0
    prev_above = None
    
    for i in range(len(df)):
        if df.iloc[i]['above_median'] == 1:
            if prev_above == True:
                current_streak += 1
            else:
                current_streak = 1
            prev_above = True
        elif df.iloc[i]['below_median'] == 1:
            if prev_above == False:
                current_streak += 1
            else:
                current_streak = 1
            prev_above = False
        df.loc[df.index[i], 'consecutive'] = current_streak
    
    # MMI計算（平均回帰の度合い）
    df['moves'] = (df['close'] > df['close'].shift(1)).astype(int)
    df['nl'] = df['moves'].rolling(window=period).sum()
    df['mmi'] = 100 * (period - df['nl']) / period
    df['mmi'] = df['mmi'].fillna(50)  # 初期値を50に設定
    
    # MMIのレベル（閾値を緩和）
    df['mmi_high'] = df['mmi'] > 60  # 高い平均回帰性
    df['mmi_low'] = df['mmi'] < 40   # トレンド性が強い
    
    # 価格のボリンジャーバンド
    df['price_ma'] = df['close'].rolling(window=lookback).mean()
    df['price_std'] = df['close'].rolling(window=lookback).std()
    df['bb_upper'] = df['price_ma'] + 2 * df['price_std']
    df['bb_lower'] = df['price_ma'] - 2 * df['price_std']
    
    # シグナル生成
    df['is_buy'] = (
        (df['mmi_high'] & (df['close'] < df['bb_lower'])) |  # 平均回帰環境で売られ過ぎ
        (df['mmi_low'] & (df['close'] > df['price_ma']) & (df['close'].shift(1) <= df['price_ma'].shift(1)))  # トレンド環境でMAクロス
    ) & df['mmi'].notna()
    df['is_sell'] = (
        (df['mmi_high'] & (df['close'] > df['bb_upper'])) |  # 平均回帰環境で買われ過ぎ
        (df['mmi_low'] & (df['close'] < df['price_ma']) & (df['close'].shift(1) >= df['price_ma'].shift(1)))  # トレンド環境でMAクロス
    ) & df['mmi'].notna()
    
    # 不要カラム削除
    df.drop(['median', 'above_median', 'below_median', 'consecutive', 'moves', 'nl',
             'mmi_high', 'mmi_low', 'price_ma', 'price_std', 'bb_upper', 'bb_lower'], 
            axis=1, inplace=True, errors='ignore')
    
    return df