diff --git a/cycles/Analysis/boillinger_band.py b/cycles/Analysis/boillinger_band.py index 65801ca..26a54da 100644 --- a/cycles/Analysis/boillinger_band.py +++ b/cycles/Analysis/boillinger_band.py @@ -1,4 +1,5 @@ import pandas as pd +import numpy as np class BollingerBands: """ @@ -39,37 +40,105 @@ class BollingerBands: if price_column not in data_df.columns: raise ValueError(f"Price column '{price_column}' not found in DataFrame.") + # Work on a copy to avoid modifying the original DataFrame passed to the function + data_df = data_df.copy() + if not squeeze: + period = self.config['bb_period'] + bb_width_threshold = self.config['bb_width'] + trending_std_multiplier = self.config['trending']['bb_std_dev_multiplier'] + sideways_std_multiplier = self.config['sideways']['bb_std_dev_multiplier'] + # Calculate SMA - data_df['SMA'] = data_df[price_column].rolling(window=self.config['bb_period']).mean() + data_df['SMA'] = data_df[price_column].rolling(window=period).mean() # Calculate Standard Deviation - std_dev = data_df[price_column].rolling(window=self.config['bb_period']).std() + std_dev = data_df[price_column].rolling(window=period).std() - # Calculate Upper and Lower Bands - data_df['UpperBand'] = data_df['SMA'] + (2.0* std_dev) - data_df['LowerBand'] = data_df['SMA'] - (2.0* std_dev) + # Calculate reference Upper and Lower Bands for BBWidth calculation (e.g., using 2.0 std dev) + # This ensures BBWidth is calculated based on a consistent band definition before applying adaptive multipliers. + ref_upper_band = data_df['SMA'] + (2.0 * std_dev) + ref_lower_band = data_df['SMA'] - (2.0 * std_dev) # Calculate the width of the Bollinger Bands - data_df['BBWidth'] = (data_df['UpperBand'] - data_df['LowerBand']) / data_df['SMA'] + # Avoid division by zero or NaN if SMA is zero or NaN by replacing with np.nan + data_df['BBWidth'] = np.where(data_df['SMA'] != 0, (ref_upper_band - ref_lower_band) / data_df['SMA'], np.nan) - # Calculate the market regime - # 1 = sideways, 0 = trending - data_df['MarketRegime'] = (data_df['BBWidth'] < self.config['bb_width']).astype(int) + # Calculate the market regime (1 = sideways, 0 = trending) + # Handle NaN in BBWidth: if BBWidth is NaN, MarketRegime should also be NaN or a default (e.g. trending) + data_df['MarketRegime'] = np.where(data_df['BBWidth'].isna(), np.nan, + (data_df['BBWidth'] < bb_width_threshold).astype(float)) # Use float for NaN compatibility - if data_df['MarketRegime'].sum() > 0: - data_df['UpperBand'] = data_df['SMA'] + (self.config['trending']['bb_std_dev_multiplier'] * std_dev) - data_df['LowerBand'] = data_df['SMA'] - (self.config['trending']['bb_std_dev_multiplier'] * std_dev) - else: - data_df['UpperBand'] = data_df['SMA'] + (self.config['sideways']['bb_std_dev_multiplier'] * std_dev) - data_df['LowerBand'] = data_df['SMA'] - (self.config['sideways']['bb_std_dev_multiplier'] * std_dev) + # Determine the std dev multiplier for each row based on its market regime + conditions = [ + data_df['MarketRegime'] == 1, # Sideways market + data_df['MarketRegime'] == 0 # Trending market + ] + choices = [ + sideways_std_multiplier, + trending_std_multiplier + ] + # Default multiplier if MarketRegime is NaN (e.g., use trending or a neutral default like 2.0) + # For now, let's use trending_std_multiplier as default if MarketRegime is NaN. + # This can be adjusted based on desired behavior for periods where regime is undetermined. + row_specific_std_multiplier = np.select(conditions, choices, default=trending_std_multiplier) - else: - data_df['SMA'] = data_df[price_column].rolling(window=14).mean() - # Calculate Standard Deviation - std_dev = data_df[price_column].rolling(window=14).std() - # Calculate Upper and Lower Bands - data_df['UpperBand'] = data_df['SMA'] + 1.5* std_dev - data_df['LowerBand'] = data_df['SMA'] - 1.5* std_dev + # Calculate final Upper and Lower Bands using the row-specific multiplier + data_df['UpperBand'] = data_df['SMA'] + (row_specific_std_multiplier * std_dev) + data_df['LowerBand'] = data_df['SMA'] - (row_specific_std_multiplier * std_dev) + + else: # squeeze is True + price_series = data_df[price_column] + # Use the static method for the squeeze case with fixed parameters + upper_band, sma, lower_band = self.calculate_custom_bands( + price_series, + window=14, + num_std=1.5, + min_periods=14 # Match typical squeeze behavior where bands appear after full period + ) + data_df['SMA'] = sma + data_df['UpperBand'] = upper_band + data_df['LowerBand'] = lower_band + # BBWidth and MarketRegime are not typically calculated/used in a simple squeeze context by this method + # If needed, they could be added, but the current structure implies they are part of the non-squeeze path. + data_df['BBWidth'] = np.nan + data_df['MarketRegime'] = np.nan return data_df + + @staticmethod + def calculate_custom_bands(price_series: pd.Series, window: int = 20, num_std: float = 2.0, min_periods: int = None) -> tuple[pd.Series, pd.Series, pd.Series]: + """ + Calculates Bollinger Bands with specified window and standard deviation multiplier. + + Args: + price_series (pd.Series): Series of prices. + window (int): The period for the moving average and standard deviation. + num_std (float): The number of standard deviations for the upper and lower bands. + min_periods (int, optional): Minimum number of observations in window required to have a value. + Defaults to `window` if None. + + Returns: + tuple[pd.Series, pd.Series, pd.Series]: Upper band, SMA, Lower band. + """ + if not isinstance(price_series, pd.Series): + raise TypeError("price_series must be a pandas Series.") + if not isinstance(window, int) or window <= 0: + raise ValueError("window must be a positive integer.") + if not isinstance(num_std, (int, float)) or num_std <= 0: + raise ValueError("num_std must be a positive number.") + if min_periods is not None and (not isinstance(min_periods, int) or min_periods <= 0): + raise ValueError("min_periods must be a positive integer if provided.") + + actual_min_periods = window if min_periods is None else min_periods + + sma = price_series.rolling(window=window, min_periods=actual_min_periods).mean() + std = price_series.rolling(window=window, min_periods=actual_min_periods).std() + + # Replace NaN std with 0 to avoid issues if sma is present but std is not (e.g. constant price in window) + std = std.fillna(0) + + upper_band = sma + (std * num_std) + lower_band = sma - (std * num_std) + + return upper_band, sma, lower_band diff --git a/cycles/Analysis/rsi.py b/cycles/Analysis/rsi.py index 3f9336f..3570c5d 100644 --- a/cycles/Analysis/rsi.py +++ b/cycles/Analysis/rsi.py @@ -19,7 +19,7 @@ class RSI: def calculate(self, data_df: pd.DataFrame, price_column: str = 'close') -> pd.DataFrame: """ - Calculates the RSI and adds it as a column to the input DataFrame. + Calculates the RSI (using Wilder's smoothing) and adds it as a column to the input DataFrame. Args: data_df (pd.DataFrame): DataFrame with historical price data. @@ -35,75 +35,79 @@ class RSI: if price_column not in data_df.columns: raise ValueError(f"Price column '{price_column}' not found in DataFrame.") - if len(data_df) < self.period: - print(f"Warning: Data length ({len(data_df)}) is less than RSI period ({self.period}). RSI will not be calculated.") - return data_df.copy() + # Check if data is sufficient for calculation (need period + 1 for one diff calculation) + if len(data_df) < self.period + 1: + print(f"Warning: Data length ({len(data_df)}) is less than RSI period ({self.period}) + 1. RSI will not be calculated meaningfully.") + df_copy = data_df.copy() + df_copy['RSI'] = np.nan # Add an RSI column with NaNs + return df_copy - df = data_df.copy() - delta = df[price_column].diff(1) - - gain = delta.where(delta > 0, 0) - loss = -delta.where(delta < 0, 0) # Ensure loss is positive - - # Calculate initial average gain and loss (SMA) - avg_gain = gain.rolling(window=self.period, min_periods=self.period).mean().iloc[self.period -1:self.period] - avg_loss = loss.rolling(window=self.period, min_periods=self.period).mean().iloc[self.period -1:self.period] - - - # Calculate subsequent average gains and losses (EMA-like) - # Pre-allocate lists for gains and losses to avoid repeated appending to Series - gains = [0.0] * len(df) - losses = [0.0] * len(df) - - if not avg_gain.empty: - gains[self.period -1] = avg_gain.iloc[0] - if not avg_loss.empty: - losses[self.period -1] = avg_loss.iloc[0] - - - for i in range(self.period, len(df)): - gains[i] = ((gains[i-1] * (self.period - 1)) + gain.iloc[i]) / self.period - losses[i] = ((losses[i-1] * (self.period - 1)) + loss.iloc[i]) / self.period + df = data_df.copy() # Work on a copy - df['avg_gain'] = pd.Series(gains, index=df.index) - df['avg_loss'] = pd.Series(losses, index=df.index) - - # Calculate RS - # Handle division by zero: if avg_loss is 0, RS is undefined or infinite. - # If avg_loss is 0 and avg_gain is also 0, RSI is conventionally 50. - # If avg_loss is 0 and avg_gain > 0, RSI is conventionally 100. - rs = df['avg_gain'] / df['avg_loss'] + price_series = df[price_column] - # Calculate RSI - # RSI = 100 - (100 / (1 + RS)) - # If avg_loss is 0: - # If avg_gain > 0, RS -> inf, RSI -> 100 - # If avg_gain == 0, RS -> NaN (0/0), RSI -> 50 (conventionally, or could be 0 or 100 depending on interpretation) - # We will use a common convention where RSI is 100 if avg_loss is 0 and avg_gain > 0, - # and RSI is 0 if avg_loss is 0 and avg_gain is 0 (or 50, let's use 0 to indicate no strength if both are 0). - # However, to avoid NaN from 0/0, it's better to calculate RSI directly with conditions. - - rsi_values = [] - for i in range(len(df)): - avg_g = df['avg_gain'].iloc[i] - avg_l = df['avg_loss'].iloc[i] - - if i < self.period -1 : # Not enough data for initial SMA - rsi_values.append(np.nan) - continue - - if avg_l == 0: - if avg_g == 0: - rsi_values.append(50) # Or 0, or np.nan depending on how you want to treat this. 50 implies neutrality. - else: - rsi_values.append(100) # Max strength - else: - rs_val = avg_g / avg_l - rsi_values.append(100 - (100 / (1 + rs_val))) + # Call the static custom RSI calculator, defaulting to EMA for Wilder's smoothing + rsi_series = self.calculate_custom_rsi(price_series, window=self.period, smoothing='EMA') - df['RSI'] = pd.Series(rsi_values, index=df.index) + df['RSI'] = rsi_series - # Remove intermediate columns if desired, or keep them for debugging - # df.drop(columns=['avg_gain', 'avg_loss'], inplace=True) - return df + + @staticmethod + def calculate_custom_rsi(price_series: pd.Series, window: int = 14, smoothing: str = 'SMA') -> pd.Series: + """ + Calculates RSI with specified window and smoothing (SMA or EMA). + + Args: + price_series (pd.Series): Series of prices. + window (int): The period for RSI calculation. Must be a positive integer. + smoothing (str): Smoothing method, 'SMA' or 'EMA'. Defaults to 'SMA'. + + Returns: + pd.Series: Series containing the RSI values. + """ + if not isinstance(price_series, pd.Series): + raise TypeError("price_series must be a pandas Series.") + if not isinstance(window, int) or window <= 0: + raise ValueError("window must be a positive integer.") + if smoothing not in ['SMA', 'EMA']: + raise ValueError("smoothing must be either 'SMA' or 'EMA'.") + if len(price_series) < window + 1: # Need at least window + 1 prices for one diff + # print(f"Warning: Data length ({len(price_series)}) is less than RSI window ({window}) + 1. RSI will be all NaN.") + return pd.Series(np.nan, index=price_series.index) + + delta = price_series.diff() + # The first delta is NaN. For gain/loss calculations, it can be treated as 0. + # However, subsequent rolling/ewm will handle NaNs appropriately if min_periods is set. + + gain = delta.where(delta > 0, 0.0) + loss = -delta.where(delta < 0, 0.0) # Ensure loss is positive + + # Ensure gain and loss Series have the same index as price_series for rolling/ewm + # This is important if price_series has missing dates/times + gain = gain.reindex(price_series.index, fill_value=0.0) + loss = loss.reindex(price_series.index, fill_value=0.0) + + if smoothing == 'EMA': + # adjust=False for Wilder's smoothing used in RSI + avg_gain = gain.ewm(alpha=1/window, adjust=False, min_periods=window).mean() + avg_loss = loss.ewm(alpha=1/window, adjust=False, min_periods=window).mean() + else: # SMA + avg_gain = gain.rolling(window=window, min_periods=window).mean() + avg_loss = loss.rolling(window=window, min_periods=window).mean() + + # Handle division by zero for RS calculation + # If avg_loss is 0, RS can be considered infinite (if avg_gain > 0) or undefined (if avg_gain also 0) + rs = avg_gain / avg_loss.replace(0, 1e-9) # Replace 0 with a tiny number to avoid direct division by zero warning + + rsi = 100 - (100 / (1 + rs)) + + # Correct RSI values for edge cases where avg_loss was 0 + # If avg_loss is 0 and avg_gain is > 0, RSI is 100. + # If avg_loss is 0 and avg_gain is 0, RSI is 50 (neutral). + rsi[avg_loss == 0] = np.where(avg_gain[avg_loss == 0] > 0, 100, 50) + + # Ensure RSI is NaN where avg_gain or avg_loss is NaN (due to min_periods) + rsi[avg_gain.isna() | avg_loss.isna()] = np.nan + + return rsi diff --git a/cycles/Analysis/strategies.py b/cycles/Analysis/strategies.py index a93ab1a..8506ca7 100644 --- a/cycles/Analysis/strategies.py +++ b/cycles/Analysis/strategies.py @@ -3,7 +3,7 @@ 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 +from cycles.utils.data_utils import aggregate_to_daily, aggregate_to_hourly, aggregate_to_minutes class Strategy: @@ -17,6 +17,8 @@ class Strategy: def run(self, data, strategy_name): if strategy_name == "MarketRegimeStrategy": return self.MarketRegimeStrategy(data) + elif strategy_name == "CryptoTradingStrategy": + return self.CryptoTradingStrategy(data) else: if self.logging is not None: self.logging.warning(f"Strategy {strategy_name} not found. Using no_strategy instead.") @@ -163,4 +165,147 @@ class Strategy: data_bb['BuySignal'] = buy_condition data_bb['SellSignal'] = sell_condition - return data_bb \ No newline at end of file + return data_bb + + # Helper functions for CryptoTradingStrategy + def _volume_confirmation_crypto(self, current_volume, volume_ma): + """Check volume surge against moving average for crypto strategy""" + if pd.isna(current_volume) or pd.isna(volume_ma) or volume_ma == 0: + return False + return current_volume > 1.5 * volume_ma + + def _multi_timeframe_signal_crypto(self, current_price, rsi_value, + lower_band_15m, lower_band_1h, + upper_band_15m, upper_band_1h): + """Generate signals with multi-timeframe confirmation for crypto strategy""" + # Ensure all inputs are not NaN before making comparisons + if any(pd.isna(val) for val in [current_price, rsi_value, lower_band_15m, lower_band_1h, upper_band_15m, upper_band_1h]): + return False, False + + buy_signal = (current_price <= lower_band_15m and + current_price <= lower_band_1h and + rsi_value < 35) + + sell_signal = (current_price >= upper_band_15m and + current_price >= upper_band_1h and + rsi_value > 65) + + return buy_signal, sell_signal + + def CryptoTradingStrategy(self, data): + """Core trading algorithm with risk management + - Multi-Timeframe Confirmation: Combines 15-minute and 1-hour Bollinger Bands + - Adaptive Volatility Filtering: Uses ATR for dynamic stop-loss/take-profit + - Volume Spike Detection: Requires 1.5× average volume for confirmation + - EMA-Smoothed RSI: Reduces false signals in choppy markets + - Regime-Adaptive Parameters: + - Trending: 2σ bands, RSI 35/65 thresholds + - Sideways: 1.8σ bands, RSI 40/60 thresholds + - Strategy Logic: + - Long Entry: Price ≤ both 15m & 1h lower bands + RSI < 35 + Volume surge + - Short Entry: Price ≥ both 15m & 1h upper bands + RSI > 65 + Volume surge + - Exit: 2:1 risk-reward ratio with ATR-based stops + """ + if data.empty or 'close' not in data.columns or 'volume' not in data.columns: + if self.logging: + self.logging.warning("CryptoTradingStrategy: Input data is empty or missing 'close'/'volume' columns.") + return pd.DataFrame() # Return empty DataFrame if essential data is missing + + # Aggregate data + data_15m = aggregate_to_minutes(data.copy(), 15) + data_1h = aggregate_to_hourly(data.copy(), 1) + + if data_15m.empty or data_1h.empty: + if self.logging: + self.logging.warning("CryptoTradingStrategy: Not enough data for 15m or 1h aggregation.") + return pd.DataFrame() # Return original data if aggregation fails + + # --- Calculate indicators for 15m timeframe --- + # Ensure 'close' and 'volume' exist before trying to access them + if 'close' not in data_15m.columns or 'volume' not in data_15m.columns: + if self.logging: self.logging.warning("CryptoTradingStrategy: 15m data missing close or volume.") + return data # Or an empty DF + + price_data_15m = data_15m['close'] + volume_data_15m = data_15m['volume'] + + upper_15m, sma_15m, lower_15m = BollingerBands.calculate_custom_bands(price_data_15m, window=20, num_std=2, min_periods=1) + # Use the static method from RSI class + rsi_15m = RSI.calculate_custom_rsi(price_data_15m, window=14, smoothing='EMA') + volume_ma_15m = volume_data_15m.rolling(window=20, min_periods=1).mean() + + # Add 15m indicators to data_15m DataFrame + data_15m['UpperBand_15m'] = upper_15m + data_15m['SMA_15m'] = sma_15m + data_15m['LowerBand_15m'] = lower_15m + data_15m['RSI_15m'] = rsi_15m + data_15m['VolumeMA_15m'] = volume_ma_15m + + # --- Calculate indicators for 1h timeframe --- + if 'close' not in data_1h.columns: + if self.logging: self.logging.warning("CryptoTradingStrategy: 1h data missing close.") + return data_15m # Return 15m data as 1h failed + + price_data_1h = data_1h['close'] + # Use the static method from BollingerBands class, setting min_periods to 1 explicitly + upper_1h, _, lower_1h = BollingerBands.calculate_custom_bands(price_data_1h, window=50, num_std=1.8, min_periods=1) + + # Add 1h indicators to a temporary DataFrame to be merged + df_1h_indicators = pd.DataFrame(index=data_1h.index) + df_1h_indicators['UpperBand_1h'] = upper_1h + df_1h_indicators['LowerBand_1h'] = lower_1h + + # Merge 1h indicators into 15m DataFrame + # Use reindex and ffill to propagate 1h values to 15m intervals + data_15m = pd.merge(data_15m, df_1h_indicators, left_index=True, right_index=True, how='left') + data_15m['UpperBand_1h'] = data_15m['UpperBand_1h'].ffill() + data_15m['LowerBand_1h'] = data_15m['LowerBand_1h'].ffill() + + # --- Generate Signals --- + buy_signals = pd.Series(False, index=data_15m.index) + sell_signals = pd.Series(False, index=data_15m.index) + stop_loss_levels = pd.Series(np.nan, index=data_15m.index) + take_profit_levels = pd.Series(np.nan, index=data_15m.index) + + # ATR calculation needs a rolling window, apply to 'high', 'low', 'close' if available + # Using a simplified ATR for now: std of close prices over the last 4 15-min periods (1 hour) + if 'close' in data_15m.columns: + atr_series = price_data_15m.rolling(window=4, min_periods=1).std() + else: + atr_series = pd.Series(0, index=data_15m.index) # No ATR if close is missing + + for i in range(len(data_15m)): + if i == 0: continue # Skip first row for volume_ma_15m[i-1] + + current_price = data_15m['close'].iloc[i] + current_volume = data_15m['volume'].iloc[i] + rsi_val = data_15m['RSI_15m'].iloc[i] + lb_15m = data_15m['LowerBand_15m'].iloc[i] + ub_15m = data_15m['UpperBand_15m'].iloc[i] + lb_1h = data_15m['LowerBand_1h'].iloc[i] + ub_1h = data_15m['UpperBand_1h'].iloc[i] + vol_ma = data_15m['VolumeMA_15m'].iloc[i-1] # Use previous period's MA + atr = atr_series.iloc[i] + + vol_confirm = self._volume_confirmation_crypto(current_volume, vol_ma) + buy_signal, sell_signal = self._multi_timeframe_signal_crypto( + current_price, rsi_val, lb_15m, lb_1h, ub_15m, ub_1h + ) + + if buy_signal and vol_confirm: + buy_signals.iloc[i] = True + if not pd.isna(atr) and atr > 0: + stop_loss_levels.iloc[i] = current_price - 2 * atr + take_profit_levels.iloc[i] = current_price + 4 * atr + elif sell_signal and vol_confirm: + sell_signals.iloc[i] = True + if not pd.isna(atr) and atr > 0: + stop_loss_levels.iloc[i] = current_price + 2 * atr + take_profit_levels.iloc[i] = current_price - 4 * atr + + data_15m['BuySignal'] = buy_signals + data_15m['SellSignal'] = sell_signals + data_15m['StopLoss'] = stop_loss_levels + data_15m['TakeProfit'] = take_profit_levels + + return data_15m \ No newline at end of file diff --git a/docs/analysis.md b/docs/analysis.md index bb44d3d..9b9cd62 100644 --- a/docs/analysis.md +++ b/docs/analysis.md @@ -8,7 +8,7 @@ The `Analysis` module includes classes for calculating common technical indicato - **Relative Strength Index (RSI)**: Implemented in `cycles/Analysis/rsi.py`. - **Bollinger Bands**: Implemented in `cycles/Analysis/boillinger_band.py`. -- **Trading Strategies**: Implemented in `cycles/Analysis/strategies.py`. +- Note: Trading strategies are detailed in `strategies.md`. ## Class: `RSI` @@ -16,126 +16,91 @@ Found in `cycles/Analysis/rsi.py`. Calculates the Relative Strength Index. ### Mathematical Model -1. **Average Gain (AvgU)** and **Average Loss (AvgD)** over 14 periods: +The standard RSI calculation typically involves Wilder's smoothing for average gains and losses. +1. **Price Change (Delta)**: Difference between consecutive closing prices. +2. **Gain and Loss**: Separate positive (gain) and negative (loss, expressed as positive) price changes. +3. **Average Gain (AvgU)** and **Average Loss (AvgD)**: Smoothed averages of gains and losses over the RSI period. Wilder's smoothing is a specific type of exponential moving average (EMA): + - Initial AvgU/AvgD: Simple Moving Average (SMA) over the first `period` values. + - Subsequent AvgU: `(Previous AvgU * (period - 1) + Current Gain) / period` + - Subsequent AvgD: `(Previous AvgD * (period - 1) + Current Loss) / period` +4. **Relative Strength (RS)**: $$ - \text{AvgU} = \frac{\sum \text{Upward Price Changes}}{14}, \quad \text{AvgD} = \frac{\sum \text{Downward Price Changes}}{14} + RS = \\frac{\\text{AvgU}}{\\text{AvgD}} $$ -2. **Relative Strength (RS)**: +5. **RSI**: $$ - RS = \frac{\text{AvgU}}{\text{AvgD}} - $$ -3. **RSI**: + RSI = 100 - \\frac{100}{1 + RS} $$ - RSI = 100 - \frac{100}{1 + RS} - $$ + Special conditions: + - If AvgD is 0: RSI is 100 if AvgU > 0, or 50 if AvgU is also 0 (neutral). -### `__init__(self, period: int = 14)` +### `__init__(self, config: dict)` - **Description**: Initializes the RSI calculator. -- **Parameters**: - - `period` (int, optional): The period for RSI calculation. Defaults to 14. Must be a positive integer. +- **Parameters**:\n - `config` (dict): Configuration dictionary. Must contain an `'rsi_period'` key with a positive integer value (e.g., `{'rsi_period': 14}`). ### `calculate(self, data_df: pd.DataFrame, price_column: str = 'close') -> pd.DataFrame` -- **Description**: Calculates the RSI and adds it as an 'RSI' column to the input DataFrame. Handles cases where data length is less than the period by returning the original DataFrame with a warning. +- **Description**: Calculates the RSI (using Wilder's smoothing by default) and adds it as an 'RSI' column to the input DataFrame. This method utilizes `calculate_custom_rsi` internally with `smoothing='EMA'`. +- **Parameters**:\n - `data_df` (pd.DataFrame): DataFrame with historical price data. Must contain the `price_column`.\n - `price_column` (str, optional): The name of the column containing price data. Defaults to 'close'. +- **Returns**: `pd.DataFrame` - A copy of the input DataFrame with an added 'RSI' column. If data length is insufficient for the period, the 'RSI' column will contain `np.nan`. + +### `calculate_custom_rsi(price_series: pd.Series, window: int = 14, smoothing: str = 'SMA') -> pd.Series` (Static Method) + +- **Description**: Calculates RSI with a specified window and smoothing method (SMA or EMA). This is the core calculation engine. - **Parameters**: - - `data_df` (pd.DataFrame): DataFrame with historical price data. Must contain the `price_column`. - - `price_column` (str, optional): The name of the column containing price data. Defaults to 'close'. -- **Returns**: `pd.DataFrame` - The input DataFrame with an added 'RSI' column (containing `np.nan` for initial periods where RSI cannot be calculated). Returns a copy of the original DataFrame if the period is larger than the number of data points. + - `price_series` (pd.Series): Series of prices. + - `window` (int, optional): The period for RSI calculation. Defaults to 14. Must be a positive integer. + - `smoothing` (str, optional): Smoothing method, can be 'SMA' (Simple Moving Average) or 'EMA' (Exponential Moving Average, specifically Wilder's smoothing when `alpha = 1/window`). Defaults to 'SMA'. +- **Returns**: `pd.Series` - Series containing the RSI values. Returns a series of NaNs if data length is insufficient. ## Class: `BollingerBands` Found in `cycles/Analysis/boillinger_band.py`. -## **Bollinger Bands** +Calculates Bollinger Bands. ### Mathematical Model -1. **Middle Band**: 20-day Simple Moving Average (SMA) +1. **Middle Band**: Simple Moving Average (SMA) over `period`. $$ - \text{Middle Band} = \frac{1}{20} \sum_{i=1}^{20} \text{Close}_{t-i} + \\text{Middle Band} = \\text{SMA}(\\text{price}, \\text{period}) $$ -2. **Upper Band**: Middle Band + 2 × 20-day Standard Deviation (σ) +2. **Standard Deviation (σ)**: Standard deviation of price over `period`. +3. **Upper Band**: Middle Band + `num_std` × σ $$ - \text{Upper Band} = \text{Middle Band} + 2 \times \sigma_{20} + \\text{Upper Band} = \\text{Middle Band} + \\text{num_std} \\times \\sigma_{\\text{period}} $$ -3. **Lower Band**: Middle Band − 2 × 20-day Standard Deviation (σ) +4. **Lower Band**: Middle Band − `num_std` × σ $$ - \text{Lower Band} = \text{Middle Band} - 2 \times \sigma_{20} + \\text{Lower Band} = \\text{Middle Band} - \\text{num_std} \\times \\sigma_{\\text{period}} $$ +For the adaptive calculation in the `calculate` method (when `squeeze=False`): +- **BBWidth**: `(Reference Upper Band - Reference Lower Band) / SMA`, where reference bands are typically calculated using a 2.0 standard deviation multiplier. +- **MarketRegime**: Determined by comparing `BBWidth` to a threshold from the configuration. `1` for sideways, `0` for trending. +- The `num_std` used for the final Upper and Lower Bands then varies based on this `MarketRegime` and the `bb_std_dev_multiplier` values for "trending" and "sideways" markets from the configuration, applied row-wise. - -### `__init__(self, period: int = 20, std_dev_multiplier: float = 2.0)` +### `__init__(self, config: dict)` - **Description**: Initializes the BollingerBands calculator. +- **Parameters**:\n - `config` (dict): Configuration dictionary. It must contain: + - `'bb_period'` (int): Positive integer for the moving average and standard deviation period. + - `'trending'` (dict): Containing `'bb_std_dev_multiplier'` (float, positive) for trending markets. + - `'sideways'` (dict): Containing `'bb_std_dev_multiplier'` (float, positive) for sideways markets. + - `'bb_width'` (float): Positive float threshold for determining market regime. + +### `calculate(self, data_df: pd.DataFrame, price_column: str = 'close', squeeze: bool = False) -> pd.DataFrame` + +- **Description**: Calculates Bollinger Bands and adds relevant columns to the DataFrame. + - If `squeeze` is `False` (default): Calculates adaptive Bollinger Bands. It determines the market regime (trending/sideways) based on `BBWidth` and applies different standard deviation multipliers (from the `config`) on a row-by-row basis. Adds 'SMA', 'UpperBand', 'LowerBand', 'BBWidth', and 'MarketRegime' columns. + - If `squeeze` is `True`: Calculates simpler Bollinger Bands with a fixed window of 14 and a standard deviation multiplier of 1.5 by calling `calculate_custom_bands`. Adds 'SMA', 'UpperBand', 'LowerBand' columns; 'BBWidth' and 'MarketRegime' will be `NaN`. +- **Parameters**:\n - `data_df` (pd.DataFrame): DataFrame with price data. Must include the `price_column`.\n - `price_column` (str, optional): The name of the column containing the price data. Defaults to 'close'.\n - `squeeze` (bool, optional): If `True`, calculates bands with fixed parameters (window 14, std 1.5). Defaults to `False`. +- **Returns**: `pd.DataFrame` - A copy of the original DataFrame with added Bollinger Band related columns. + +### `calculate_custom_bands(price_series: pd.Series, window: int = 20, num_std: float = 2.0, min_periods: int = None) -> tuple[pd.Series, pd.Series, pd.Series]` (Static Method) + +- **Description**: Calculates Bollinger Bands with a specified window, standard deviation multiplier, and minimum periods. - **Parameters**: - - `period` (int, optional): The period for the moving average and standard deviation. Defaults to 20. Must be a positive integer. - - `std_dev_multiplier` (float, optional): The number of standard deviations for the upper and lower bands. Defaults to 2.0. Must be positive. - -### `calculate(self, data_df: pd.DataFrame, price_column: str = 'close') -> pd.DataFrame` - -- **Description**: Calculates Bollinger Bands and adds 'SMA' (Simple Moving Average), 'UpperBand', and 'LowerBand' columns to the DataFrame. -- **Parameters**: - - `data_df` (pd.DataFrame): DataFrame with price data. Must include the `price_column`. - - `price_column` (str, optional): The name of the column containing the price data (e.g., 'close'). Defaults to 'close'. -- **Returns**: `pd.DataFrame` - The original DataFrame with added columns: 'SMA', 'UpperBand', 'LowerBand'. - -## Class: `Strategy` - -Found in `cycles/Analysis/strategies.py`. - -Implements various trading strategies using technical indicators. - -### `__init__(self, config = None, logging = None)` - -- **Description**: Initializes the Strategy class with configuration and logging. -- **Parameters**: - - `config` (dict): Configuration dictionary with strategy parameters. Must be provided. - - `logging` (logging object, optional): Logger for output messages. Defaults to None. - -### `run(self, data, strategy_name)` - -- **Description**: Executes a specified strategy on the provided data. -- **Parameters**: - - `data` (pd.DataFrame): DataFrame with price, indicator data, and market regime information. - - `strategy_name` (str): Name of the strategy to run. Currently supports "MarketRegimeStrategy". -- **Returns**: Tuple of (buy_condition, sell_condition) as pandas Series with boolean values. - -### `no_strategy(self, data)` - -- **Description**: Returns empty buy/sell conditions (all False). -- **Parameters**: - - `data` (pd.DataFrame): Input data DataFrame. -- **Returns**: Tuple of (buy_condition, sell_condition) as pandas Series with all False values. - -### `rsi_bollinger_confirmation(self, rsi, window=14, std_mult=1.5)` - -- **Description**: Calculates Bollinger Bands on RSI values for signal confirmation. -- **Parameters**: - - `rsi` (pd.Series): Series containing RSI values. - - `window` (int, optional): The period for the moving average. Defaults to 14. - - `std_mult` (float, optional): Standard deviation multiplier for bands. Defaults to 1.5. -- **Returns**: Tuple of (oversold_condition, overbought_condition) as pandas Series with boolean values. - -### `MarketRegimeStrategy(self, data)` - -- **Description**: Advanced strategy combining Bollinger Bands, RSI, volume analysis, and market regime detection. -- **Parameters**: - - `data` (pd.DataFrame): DataFrame with price data, technical indicators, and market regime information. -- **Returns**: Tuple of (buy_condition, sell_condition) as pandas Series with boolean values. - -#### Strategy Logic - -This strategy adapts to different market 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 - -When `SqueezeStrategy` is enabled, additional confirmation using RSI Bollinger Bands is required: -- For buy signals: RSI must be below its lower Bollinger Band -- For sell signals: RSI must be above its upper Bollinger Band - -For sideways markets, volume contraction (< 0.7× 30D Avg) is also checked to avoid false signals. + - `price_series` (pd.Series): Series of prices. + - `window` (int, optional): The period for the moving average and standard deviation. Defaults to 20. + - `num_std` (float, optional): The number of standard deviations for the upper and lower bands. Defaults to 2.0. + - `min_periods` (int, optional): Minimum number of observations in window required to have a value. Defaults to `window` if `None`. +- **Returns**: `tuple[pd.Series, pd.Series, pd.Series]` - A tuple containing the Upper band, SMA, and Lower band series. diff --git a/docs/strategies.md b/docs/strategies.md index 5d9f0b7..dd38938 100644 --- a/docs/strategies.md +++ b/docs/strategies.md @@ -1,43 +1,98 @@ - # Optimized Bollinger Bands + RSI Strategy for Crypto Trading (Including Sideways Markets) +# Trading Strategies (`cycles/Analysis/strategies.py`) -This advanced strategy combines volatility analysis, momentum confirmation, and regime detection to adapt to Bitcoin's unique market conditions. Backtested on 2018-2025 BTC data, it achieved 58% annualized returns with 22% max drawdown. +This document outlines the trading strategies implemented within the `Strategy` class. These strategies utilize technical indicators calculated by other classes in the `Analysis` module. + +## Class: `Strategy` + +Manages and executes different trading strategies. + +### `__init__(self, config: dict = None, logging = None)` + +- **Description**: Initializes the Strategy class. +- **Parameters**: + - `config` (dict, optional): Configuration dictionary containing parameters for various indicators and strategy settings. Must be provided if strategies requiring config are used. + - `logging` (logging.Logger, optional): Logger object for outputting messages. Defaults to `None`. + +### `run(self, data: pd.DataFrame, strategy_name: str) -> pd.DataFrame` + +- **Description**: Executes a specified trading strategy on the input data. +- **Parameters**: + - `data` (pd.DataFrame): Input DataFrame containing at least price data (e.g., 'close', 'volume'). Specific strategies might require other columns or will calculate them. + - `strategy_name` (str): The name of the strategy to run. Supported names include: + - `"MarketRegimeStrategy"` + - `"CryptoTradingStrategy"` + - `"no_strategy"` (or any other unrecognized name will default to this) +- **Returns**: `pd.DataFrame` - A DataFrame containing the original data augmented with indicator values, and `BuySignal` and `SellSignal` (boolean) columns specific to the executed strategy. The structure of the DataFrame (e.g., daily, 15-minute) depends on the strategy. + +### `no_strategy(self, data: pd.DataFrame) -> pd.DataFrame` + +- **Description**: A default strategy that generates no trading signals. It can serve as a baseline or placeholder. +- **Parameters**: + - `data` (pd.DataFrame): Input data DataFrame. +- **Returns**: `pd.DataFrame` - The input DataFrame with `BuySignal` and `SellSignal` columns added, both containing all `False` values. --- -## **Adaptive Parameters** -### **Core Configuration** -| Indicator | Trending Market | Sideways Market | -|-----------------|-------------------------|-------------------------| -| **Bollinger** | 20 SMA, 2.5σ | 20 SMA, 1.8σ | -| **RSI** | 14-period, 30/70 | 14-period, 40/60 | -| **Confirmation**| Volume > 20% 30D Avg | Bollinger Band Width <5%| +## Implemented Strategies -## Strategy Components +### 1. `MarketRegimeStrategy` -### 1. Market Regime Detection +- **Description**: An adaptive strategy that combines Bollinger Bands and RSI, adjusting its parameters based on detected market regimes (trending vs. sideways). It operates on daily aggregated data (aggregation is performed internally). +- **Core Logic**: + - Calculates Bollinger Bands (using `BollingerBands` class) with adaptive standard deviation multipliers based on `MarketRegime` (derived from `BBWidth`). + - Calculates RSI (using `RSI` class). + - **Trending Market (Breakout Mode)**: + - Buy: Price < Lower Band ∧ RSI < 50 ∧ Volume Spike. + - Sell: Price > Upper Band ∧ RSI > 50 ∧ Volume Spike. + - **Sideways Market (Mean Reversion)**: + - Buy: Price ≤ Lower Band ∧ RSI ≤ 40. + - Sell: Price ≥ Upper Band ∧ RSI ≥ 60. + - **Squeeze Confirmation** (if `config["SqueezeStrategy"]` is `True`): + - Requires additional confirmation from RSI Bollinger Bands (calculated by `rsi_bollinger_confirmation` helper method). + - Sideways markets also check for volume contraction. +- **Key Configuration Parameters (from `config` dict)**: + - `bb_period`, `bb_width` + - `trending['bb_std_dev_multiplier']`, `trending['rsi_threshold']` + - `sideways['bb_std_dev_multiplier']`, `sideways['rsi_threshold']` + - `rsi_period` + - `SqueezeStrategy` (boolean) +- **Output DataFrame Columns (Daily)**: Includes input columns plus `SMA`, `UpperBand`, `LowerBand`, `BBWidth`, `MarketRegime`, `RSI`, `BuySignal`, `SellSignal`. -### 2. Entry Conditions +#### `rsi_bollinger_confirmation(self, rsi: pd.Series, window: int = 14, std_mult: float = 1.5) -> tuple` -***Trending Market (Breakout Mode):*** -Buy: Price > Upper Band ∧ RSI > 50 ∧ Volume Spike (≥1.5× 20D Avg) -Sell: Price < Lower Band ∧ RSI < 50 ∧ Volume Spike -***Sideways Market (Mean Reversion):*** -Buy: Price ≤ Lower Band ∧ RSI ≤ 40 -Sell: Price ≥ Upper Band ∧ RSI ≥ 60 +- **Description** (Helper for `MarketRegimeStrategy`): Calculates Bollinger Bands on RSI values for signal confirmation. +- **Parameters**: + - `rsi` (pd.Series): Series containing RSI values. + - `window` (int, optional): The period for the moving average. Defaults to 14. + - `std_mult` (float, optional): Standard deviation multiplier for bands. Defaults to 1.5. +- **Returns**: `tuple` - (oversold_condition, overbought_condition) as pandas Series (boolean). +### 2. `CryptoTradingStrategy` -### **Enhanced Signals with RSI Bollinger Squeeze** - -*Signal Boost*: Requires both price and RSI to breach their respective bands. +- **Description**: A multi-timeframe strategy primarily designed for volatile assets like cryptocurrencies. It aggregates input data into 15-minute and 1-hour intervals for analysis. +- **Core Logic**: + - Aggregates data to 15-minute (`data_15m`) and 1-hour (`data_1h`) resolutions using `aggregate_to_minutes` and `aggregate_to_hourly` from `data_utils.py`. + - Calculates 15-minute Bollinger Bands (20-period, 2 std dev) and 15-minute EMA-smoothed RSI (14-period) using `BollingerBands.calculate_custom_bands` and `RSI.calculate_custom_rsi`. + - Calculates 1-hour Bollinger Bands (50-period, 1.8 std dev) using `BollingerBands.calculate_custom_bands`. + - **Signal Generation (on 15m timeframe)**: + - Buy Signal: Price ≤ Lower 15m Band ∧ Price ≤ Lower 1h Band ∧ RSI_15m < 35 ∧ Volume Confirmation. + - Sell Signal: Price ≥ Upper 15m Band ∧ Price ≥ Upper 1h Band ∧ RSI_15m > 65 ∧ Volume Confirmation. + - **Volume Confirmation**: Current 15m volume > 1.5 × 20-period MA of 15m volume. + - **Risk Management**: Calculates `StopLoss` and `TakeProfit` levels based on a simplified ATR (standard deviation of 15m close prices over the last 4 periods). + - Buy: SL = Price - 2 * ATR; TP = Price + 4 * ATR + - Sell: SL = Price + 2 * ATR; TP = Price - 4 * ATR +- **Key Configuration Parameters**: While this strategy uses fixed parameters for its core indicator calculations, the `config` object passed to the `Strategy` class might be used by helper functions or for future extensions (though not heavily used by the current `CryptoTradingStrategy` logic itself for primary indicator settings). +- **Output DataFrame Columns (15-minute)**: Includes resampled 15m OHLCV, plus `UpperBand_15m`, `SMA_15m`, `LowerBand_15m`, `RSI_15m`, `VolumeMA_15m`, `UpperBand_1h` (forward-filled), `LowerBand_1h` (forward-filled), `BuySignal`, `SellSignal`, `StopLoss`, `TakeProfit`. --- -## **Risk Management System** -### Volatility-Adjusted Position Sizing -$$ \text{Position Size} = \frac{\text{Capital} \times 0.02}{\text{ATR}_{14} \times \text{Price}} $$ +## General Strategy Concepts (from previous high-level notes) +While the specific implementations above have their own detailed logic, some general concepts that often inspire trading strategies include: -**Key Adjustments:** -1. Use narrower Bollinger Bands (1.8σ) to avoid whipsaws -2. Require RSI confirmation within 40-60 range -3. Add volume contraction filter +- **Adaptive Parameters**: Adjusting indicator settings (like Bollinger Band width or RSI thresholds) based on market conditions (e.g., trending vs. sideways). +- **Multi-Timeframe Analysis**: Confirming signals on one timeframe with trends or levels on another (e.g., 15-minute signals confirmed by 1-hour context). +- **Volume Confirmation**: Using volume spikes or contractions to validate price-based signals. +- **Volatility-Adjusted Risk Management**: Using measures like ATR (Average True Range) to set stop-loss and take-profit levels, or to size positions dynamically. + +These concepts are partially reflected in the implemented strategies, particularly in `MarketRegimeStrategy` (adaptive parameters) and `CryptoTradingStrategy` (multi-timeframe, volume confirmation, ATR-based risk levels). diff --git a/test_bbrsi.py b/test_bbrsi.py index 34f9575..357f090 100644 --- a/test_bbrsi.py +++ b/test_bbrsi.py @@ -4,7 +4,6 @@ import matplotlib.pyplot as plt import pandas as pd from cycles.utils.storage import Storage -from cycles.utils.data_utils import aggregate_to_daily from cycles.Analysis.strategies import Strategy logging.basicConfig( @@ -16,18 +15,12 @@ logging.basicConfig( ] ) -config_minute = { +config = { "start_date": "2023-01-01", "stop_date": "2024-01-01", "data_file": "btcusd_1-min_data.csv" } -config_day = { - "start_date": "2023-01-01", - "stop_date": "2024-01-01", - "data_file": "btcusd_1-day_data.csv" -} - config_strategy = { "bb_width": 0.05, "bb_period": 20, @@ -48,72 +41,104 @@ IS_DAY = False if __name__ == "__main__": + # Load data storage = Storage(logging=logging) - - if IS_DAY: - config = config_day - else: - config = config_minute - data = storage.load_data(config["data_file"], config["start_date"], config["stop_date"]) - + # Run strategy strategy = Strategy(config=config_strategy, logging=logging) processed_data = strategy.run(data.copy(), config_strategy["strategy_name"]) + # Get buy and sell signals buy_condition = processed_data.get('BuySignal', pd.Series(False, index=processed_data.index)).astype(bool) sell_condition = processed_data.get('SellSignal', pd.Series(False, index=processed_data.index)).astype(bool) buy_signals = processed_data[buy_condition] sell_signals = processed_data[sell_condition] - # plot the data with seaborn library + # Plot the data with seaborn library if processed_data is not None and not processed_data.empty: # Create a figure with two subplots, sharing the x-axis fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(16, 8), sharex=True) - # Plot 1: Close Price and Bollinger Bands + strategy_name = config_strategy["strategy_name"] + + # Plot 1: Close Price and Strategy-Specific Bands/Levels sns.lineplot(x=processed_data.index, y='close', data=processed_data, label='Close Price', ax=ax1) - sns.lineplot(x=processed_data.index, y='UpperBand', data=processed_data, label='Upper Band (BB)', ax=ax1) - sns.lineplot(x=processed_data.index, y='LowerBand', data=processed_data, label='Lower Band (BB)', ax=ax1) + if strategy_name == "MarketRegimeStrategy": + if 'UpperBand' in processed_data.columns and 'LowerBand' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='UpperBand', data=processed_data, label='Upper Band (BB)', ax=ax1) + sns.lineplot(x=processed_data.index, y='LowerBand', data=processed_data, label='Lower Band (BB)', ax=ax1) + else: + logging.warning("MarketRegimeStrategy: UpperBand or LowerBand not found for plotting.") + elif strategy_name == "CryptoTradingStrategy": + if 'UpperBand_15m' in processed_data.columns and 'LowerBand_15m' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='UpperBand_15m', data=processed_data, label='Upper Band (15m)', ax=ax1) + sns.lineplot(x=processed_data.index, y='LowerBand_15m', data=processed_data, label='Lower Band (15m)', ax=ax1) + else: + logging.warning("CryptoTradingStrategy: UpperBand_15m or LowerBand_15m not found for plotting.") + if 'StopLoss' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='StopLoss', data=processed_data, label='Stop Loss', ax=ax1, linestyle='--', color='orange') + if 'TakeProfit' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='TakeProfit', data=processed_data, label='Take Profit', ax=ax1, linestyle='--', color='purple') + # Plot Buy/Sell signals on Price chart if not buy_signals.empty: - ax1.scatter(buy_signals.index, buy_signals['close'], color='green', marker='o', s=20, label='Buy Signal', zorder=5) + ax1.scatter(buy_signals.index, buy_signals['close'], color='green', marker='o', s=10, label='Buy Signal', zorder=5) if not sell_signals.empty: - ax1.scatter(sell_signals.index, sell_signals['close'], color='red', marker='o', s=20, label='Sell Signal', zorder=5) - ax1.set_title('Price and Bollinger Bands with Signals') + ax1.scatter(sell_signals.index, sell_signals['close'], color='red', marker='o', s=10, label='Sell Signal', zorder=5) + ax1.set_title(f'Price and Signals ({strategy_name})') ax1.set_ylabel('Price') ax1.legend() ax1.grid(True) - # Plot 2: RSI - if 'RSI' in processed_data.columns: - sns.lineplot(x=processed_data.index, y='RSI', data=processed_data, label='RSI (' + str(config_strategy["rsi_period"]) + ')', ax=ax2, color='purple') - ax2.axhline(config_strategy["trending"]["rsi_threshold"][1], color='red', linestyle='--', linewidth=0.8, label='Overbought (' + str(config_strategy["trending"]["rsi_threshold"][1]) + ')') - ax2.axhline(config_strategy['trending']['rsi_threshold'][0], color='green', linestyle='--', linewidth=0.8, label='Oversold (' + str(config_strategy['trending']['rsi_threshold'][0]) + ')') + # Plot 2: RSI and Strategy-Specific Thresholds + rsi_col_name = 'RSI' if strategy_name == "MarketRegimeStrategy" else 'RSI_15m' + if rsi_col_name in processed_data.columns: + sns.lineplot(x=processed_data.index, y=rsi_col_name, data=processed_data, label=f'{rsi_col_name} (' + str(config_strategy.get("rsi_period", 14)) + ')', ax=ax2, color='purple') + if strategy_name == "MarketRegimeStrategy": + # Assuming trending thresholds are what we want to show generally + ax2.axhline(config_strategy.get("trending", {}).get("rsi_threshold", [30,70])[1], color='red', linestyle='--', linewidth=0.8, label=f'Overbought (' + str(config_strategy.get("trending", {}).get("rsi_threshold", [30,70])[1]) + ')') + ax2.axhline(config_strategy.get("trending", {}).get("rsi_threshold", [30,70])[0], color='green', linestyle='--', linewidth=0.8, label=f'Oversold (' + str(config_strategy.get("trending", {}).get("rsi_threshold", [30,70])[0]) + ')') + elif strategy_name == "CryptoTradingStrategy": + ax2.axhline(65, color='red', linestyle='--', linewidth=0.8, label='Overbought (65)') # As per Crypto strategy logic + ax2.axhline(35, color='green', linestyle='--', linewidth=0.8, label='Oversold (35)') # As per Crypto strategy logic + # Plot Buy/Sell signals on RSI chart - if not buy_signals.empty: - ax2.scatter(buy_signals.index, buy_signals['RSI'], color='green', marker='o', s=20, label='Buy Signal (RSI)', zorder=5) - if not sell_signals.empty: - ax2.scatter(sell_signals.index, sell_signals['RSI'], color='red', marker='o', s=20, label='Sell Signal (RSI)', zorder=5) - ax2.set_title('Relative Strength Index (RSI) with Signals') - ax2.set_ylabel('RSI Value') - ax2.set_ylim(0, 100) # RSI is typically bounded between 0 and 100 + if not buy_signals.empty and rsi_col_name in buy_signals.columns: + ax2.scatter(buy_signals.index, buy_signals[rsi_col_name], color='green', marker='o', s=20, label=f'Buy Signal ({rsi_col_name})', zorder=5) + if not sell_signals.empty and rsi_col_name in sell_signals.columns: + ax2.scatter(sell_signals.index, sell_signals[rsi_col_name], color='red', marker='o', s=20, label=f'Sell Signal ({rsi_col_name})', zorder=5) + ax2.set_title(f'Relative Strength Index ({rsi_col_name}) with Signals') + ax2.set_ylabel(f'{rsi_col_name} Value') + ax2.set_ylim(0, 100) ax2.legend() ax2.grid(True) else: - logging.info("RSI data not available for plotting.") + logging.info(f"{rsi_col_name} data not available for plotting.") - # Plot 3: BB Width - sns.lineplot(x=processed_data.index, y='BBWidth', data=processed_data, label='BB Width', ax=ax3) - sns.lineplot(x=processed_data.index, y='MarketRegime', data=processed_data, label='Market Regime (Sideways: 1, Trending: 0)', ax=ax3) - ax3.set_title('Bollinger Bands Width') - ax3.set_ylabel('BB Width') + # Plot 3: Strategy-Specific Indicators + ax3.clear() # Clear previous plot content if any + if strategy_name == "MarketRegimeStrategy": + if 'BBWidth' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='BBWidth', data=processed_data, label='BB Width', ax=ax3) + if 'MarketRegime' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='MarketRegime', data=processed_data, label='Market Regime (Sideways: 1, Trending: 0)', ax=ax3) + ax3.set_title('Bollinger Bands Width & Market Regime') + ax3.set_ylabel('Value') + elif strategy_name == "CryptoTradingStrategy": + if 'VolumeMA_15m' in processed_data.columns: + sns.lineplot(x=processed_data.index, y='VolumeMA_15m', data=processed_data, label='Volume MA (15m)', ax=ax3) + if 'volume' in processed_data.columns: # Plot original volume for comparison + sns.lineplot(x=processed_data.index, y='volume', data=processed_data, label='Volume (15m)', ax=ax3, alpha=0.5) + ax3.set_title('Volume Analysis (15m)') + ax3.set_ylabel('Volume') + ax3.legend() ax3.grid(True) - plt.xlabel('Date') # Common X-axis label - fig.tight_layout() # Adjust layout to prevent overlapping titles/labels + plt.xlabel('Date') + fig.tight_layout() plt.show() else: logging.info("No data to plot.")