import logging import seaborn as sns 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.boillinger_band import BollingerBands from cycles.Analysis.rsi import RSI logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("backtest.log"), logging.StreamHandler() ] ) config_minute = { "start_date": "2022-01-01", "stop_date": "2023-01-01", "data_file": "btcusd_1-min_data.csv" } config_day = { "start_date": "2022-01-01", "stop_date": "2023-01-01", "data_file": "btcusd_1-day_data.csv" } IS_DAY = True def no_strategy(data_bb, data_with_rsi): buy_condition = pd.Series([False] * len(data_bb), index=data_bb.index) sell_condition = pd.Series([False] * len(data_bb), index=data_bb.index) return buy_condition, sell_condition def strategy_1(data_bb, data_with_rsi): # Long trade: price move below lower Bollinger band and RSI go below 25 buy_condition = (data_bb['close'] < data_bb['LowerBand']) & (data_bb['RSI'] < 25) # Short only: price move above top Bollinger band and RSI goes over 75 sell_condition = (data_bb['close'] > data_bb['UpperBand']) & (data_bb['RSI'] > 75) return buy_condition, sell_condition if __name__ == "__main__": 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"]) if not IS_DAY: data_daily = aggregate_to_daily(data) storage.save_data(data, "btcusd_1-day_data.csv") df_to_plot = data_daily else: df_to_plot = data bb = BollingerBands(period=30, std_dev_multiplier=2.0) data_bb = bb.calculate(df_to_plot.copy()) rsi_calculator = RSI(period=13) data_with_rsi = rsi_calculator.calculate(df_to_plot.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 df_to_plot.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) logging.warning("RSI column not found or not calculated. Signals relying on RSI may not be generated.") strategy = 1 if strategy == 1: buy_condition, sell_condition = strategy_1(data_bb, data_with_rsi) else: buy_condition, sell_condition = no_strategy(data_bb, data_with_rsi) buy_signals = data_bb[buy_condition] sell_signals = data_bb[sell_condition] # plot the data with seaborn library if df_to_plot is not None and not df_to_plot.empty: # Create a figure with two subplots, sharing the x-axis fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(16, 8), sharex=True) # Plot 1: Close Price and Bollinger Bands sns.lineplot(x=data_bb.index, y='close', data=data_bb, label='Close Price', ax=ax1) sns.lineplot(x=data_bb.index, y='UpperBand', data=data_bb, label='Upper Band (BB)', ax=ax1) sns.lineplot(x=data_bb.index, y='LowerBand', data=data_bb, label='Lower Band (BB)', ax=ax1) # 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) 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.set_ylabel('Price') ax1.legend() ax1.grid(True) # Plot 2: RSI if 'RSI' in data_bb.columns: # Check data_bb now as it should contain RSI sns.lineplot(x=data_bb.index, y='RSI', data=data_bb, label='RSI (14)', ax=ax2, color='purple') ax2.axhline(75, color='red', linestyle='--', linewidth=0.8, label='Overbought (75)') ax2.axhline(25, color='green', linestyle='--', linewidth=0.8, label='Oversold (25)') # 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 ax2.legend() ax2.grid(True) else: logging.info("RSI data not available for plotting.") plt.xlabel('Date') # Common X-axis label fig.tight_layout() # Adjust layout to prevent overlapping titles/labels plt.show() else: logging.info("No data to plot.")