Cycles/cycles/Analysis/strategies.py

import pandas as pd
import numpy as np

from cycles.Analysis.boillinger_band import BollingerBands


class Strategy:

    def __init__(self, config = None, logging = None):
        if config is None:
            raise ValueError("Config must be provided.")
        self.config = config
        self.logging = logging

    def run(self, data, strategy_name):
        if strategy_name == "MarketRegimeStrategy":
            return self.MarketRegimeStrategy(data)
        else:
            if self.logging is not None:
                self.logging.warning(f"Strategy {strategy_name} not found. Using no_strategy instead.")
            return self.no_strategy(data)

    def no_strategy(self, data):
        """No strategy: returns False for both buy and sell conditions"""
        buy_condition = pd.Series([False] * len(data), index=data.index)
        sell_condition = pd.Series([False] * len(data), index=data.index)
        return buy_condition, sell_condition

    def rsi_bollinger_confirmation(self, rsi, window=14, std_mult=1.5):
        """Calculate RSI Bollinger Bands for confirmation
        
        Args:
            rsi (Series): RSI values
            window (int): Rolling window for SMA
            std_mult (float): Standard deviation multiplier
            
        Returns:
            tuple: (oversold condition, overbought condition)
        """
        valid_rsi = ~rsi.isna()
        if not valid_rsi.any():
            # Return empty Series if no valid RSI data
            return pd.Series(False, index=rsi.index), pd.Series(False, index=rsi.index)
            
        rsi_sma = rsi.rolling(window).mean()
        rsi_std = rsi.rolling(window).std()
        upper_rsi_band = rsi_sma + std_mult * rsi_std
        lower_rsi_band = rsi_sma - std_mult * rsi_std
        
        return (rsi < lower_rsi_band), (rsi > upper_rsi_band)

    def MarketRegimeStrategy(self, data):
        """Optimized Bollinger Bands + RSI Strategy for Crypto Trading (Including Sideways Markets)
        with adaptive Bollinger Bands
        
        This advanced strategy combines volatility analysis, momentum confirmation, and regime detection
        to adapt to Bitcoin's unique market conditions.
        
        Entry Conditions:
        - Trending Market (Breakout Mode):
            Buy: Price < Lower Band ∧ RSI < 50 ∧ Volume Spike (≥1.5× 20D Avg)
            Sell: Price > Upper Band ∧ RSI > 50 ∧ Volume Spike
        - Sideways Market (Mean Reversion):
            Buy: Price ≤ Lower Band ∧ RSI ≤ 40
            Sell: Price ≥ Upper Band ∧ RSI ≥ 60
            
        Enhanced with RSI Bollinger Squeeze for signal confirmation when enabled.
        """
        
        # Initialize conditions as all False
        buy_condition = pd.Series(False, index=data.index)
        sell_condition = pd.Series(False, index=data.index)
        
        # Create masks for different market regimes
        sideways_mask = data['MarketRegime'] > 0
        trending_mask = data['MarketRegime'] <= 0
        valid_data_mask = ~data['MarketRegime'].isna()  # Handle potential NaN values
        
        # Calculate volume spike (≥1.5× 20D Avg)
        if 'volume' in data.columns:
            volume_20d_avg = data['volume'].rolling(window=20).mean()
            volume_spike = data['volume'] >= 1.5 * volume_20d_avg
            
            # Additional volume contraction filter for sideways markets
            volume_30d_avg = data['volume'].rolling(window=30).mean()
            volume_contraction = data['volume'] < 0.7 * volume_30d_avg
        else:
            # If volume data is not available, assume no volume spike
            volume_spike = pd.Series(False, index=data.index)
            volume_contraction = pd.Series(False, index=data.index)
            if self.logging is not None:
                self.logging.warning("Volume data not available. Volume conditions will not be triggered.")
        
        # Calculate RSI Bollinger Squeeze confirmation
        if 'RSI' in data.columns:
            oversold_rsi, overbought_rsi = self.rsi_bollinger_confirmation(data['RSI'])
        else:
            oversold_rsi = pd.Series(False, index=data.index)
            overbought_rsi = pd.Series(False, index=data.index)
            if self.logging is not None:
                self.logging.warning("RSI data not available. RSI Bollinger Squeeze will not be triggered.")
        
        # Calculate conditions for sideways market (Mean Reversion)
        if sideways_mask.any():
            sideways_buy = (data['close'] <= data['LowerBand']) & (data['RSI'] <= 40)
            sideways_sell = (data['close'] >= data['UpperBand']) & (data['RSI'] >= 60)
            
            # Add enhanced confirmation for sideways markets
            if self.config.get("SqueezeStrategy", False):
                sideways_buy = sideways_buy & oversold_rsi & volume_contraction
                sideways_sell = sideways_sell & overbought_rsi & volume_contraction
            
            # Apply only where market is sideways and data is valid
            buy_condition = buy_condition | (sideways_buy & sideways_mask & valid_data_mask)
            sell_condition = sell_condition | (sideways_sell & sideways_mask & valid_data_mask)
        
        # Calculate conditions for trending market (Breakout Mode)
        if trending_mask.any():
            trending_buy = (data['close'] < data['LowerBand']) & (data['RSI'] < 50) & volume_spike
            trending_sell = (data['close'] > data['UpperBand']) & (data['RSI'] > 50) & volume_spike
            
            # Add enhanced confirmation for trending markets
            if self.config.get("SqueezeStrategy", False):
                trending_buy = trending_buy & oversold_rsi
                trending_sell = trending_sell & overbought_rsi
            
            # Apply only where market is trending and data is valid
            buy_condition = buy_condition | (trending_buy & trending_mask & valid_data_mask)
            sell_condition = sell_condition | (trending_sell & trending_mask & valid_data_mask)
        
        return buy_condition, sell_condition