import pandas as pd import numpy as np from cycles.Analysis.boillinger_band import BollingerBands from cycles.Analysis.rsi import RSI from cycles.utils.data_utils import aggregate_to_daily 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. Returns: DataFrame: A unified DataFrame containing original data, BB, RSI, and signals. """ data = aggregate_to_daily(data) # Calculate Bollinger Bands bb_calculator = BollingerBands(config=self.config) # Ensure we are working with a copy to avoid modifying the original DataFrame upstream data_bb = bb_calculator.calculate(data.copy()) # Calculate RSI rsi_calculator = RSI(config=self.config) # Use the original data's copy for RSI calculation as well, to maintain index integrity data_with_rsi = rsi_calculator.calculate(data.copy(), price_column='close') # Combine BB and RSI data into a single DataFrame for signal generation # Ensure indices are aligned; they should be as both are from data.copy() if 'RSI' in data_with_rsi.columns: data_bb['RSI'] = data_with_rsi['RSI'] else: # If RSI wasn't calculated (e.g., not enough data), create a dummy column with NaNs # to prevent errors later, though signals won't be generated. data_bb['RSI'] = pd.Series(index=data_bb.index, dtype=float) if self.logging: self.logging.warning("RSI column not found or not calculated. Signals relying on RSI may not be generated.") # Initialize conditions as all False buy_condition = pd.Series(False, index=data_bb.index) sell_condition = pd.Series(False, index=data_bb.index) # Create masks for different market regimes # MarketRegime is expected to be in data_bb from BollingerBands calculation sideways_mask = data_bb['MarketRegime'] > 0 trending_mask = data_bb['MarketRegime'] <= 0 valid_data_mask = ~data_bb['MarketRegime'].isna() # Handle potential NaN values # Calculate volume spike (≥1.5× 20D Avg) # 'volume' column should be present in the input 'data', and thus in 'data_bb' if 'volume' in data_bb.columns: volume_20d_avg = data_bb['volume'].rolling(window=20).mean() volume_spike = data_bb['volume'] >= 1.5 * volume_20d_avg # Additional volume contraction filter for sideways markets volume_30d_avg = data_bb['volume'].rolling(window=30).mean() volume_contraction = data_bb['volume'] < 0.7 * volume_30d_avg else: # If volume data is not available, assume no volume spike volume_spike = pd.Series(False, index=data_bb.index) volume_contraction = pd.Series(False, index=data_bb.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 # RSI column is now part of data_bb if 'RSI' in data_bb.columns and not data_bb['RSI'].isna().all(): oversold_rsi, overbought_rsi = self.rsi_bollinger_confirmation(data_bb['RSI']) else: oversold_rsi = pd.Series(False, index=data_bb.index) overbought_rsi = pd.Series(False, index=data_bb.index) if self.logging is not None and ('RSI' not in data_bb.columns or data_bb['RSI'].isna().all()): self.logging.warning("RSI data not available or all NaN. RSI Bollinger Squeeze will not be triggered.") # Calculate conditions for sideways market (Mean Reversion) if sideways_mask.any(): sideways_buy = (data_bb['close'] <= data_bb['LowerBand']) & (data_bb['RSI'] <= 40) sideways_sell = (data_bb['close'] >= data_bb['UpperBand']) & (data_bb['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_bb['close'] < data_bb['LowerBand']) & (data_bb['RSI'] < 50) & volume_spike trending_sell = (data_bb['close'] > data_bb['UpperBand']) & (data_bb['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) # Add buy/sell conditions as columns to the DataFrame data_bb['BuySignal'] = buy_condition data_bb['SellSignal'] = sell_condition return data_bb