import os import matplotlib.pyplot as plt import seaborn as sns import pandas as pd import numpy as np class BacktestCharts: @staticmethod def plot(df, meta_trend): """ Plot close price line chart with a bar at the bottom: green when trend is 1, red when trend is 0. The bar stays at the bottom even when zooming/panning. - df: DataFrame with columns ['close', ...] and a datetime index or 'timestamp' column. - meta_trend: array-like, same length as df, values 1 (green) or 0 (red). """ fig, (ax_price, ax_bar) = plt.subplots( nrows=2, ncols=1, figsize=(16, 8), sharex=True, gridspec_kw={'height_ratios': [12, 1]} ) sns.lineplot(x=df.index, y=df['close'], label='Close Price', color='blue', ax=ax_price) ax_price.set_title('Close Price with Trend Bar (Green=1, Red=0)') ax_price.set_ylabel('Price') ax_price.grid(True, alpha=0.3) ax_price.legend() # Clean meta_trend: ensure only 0/1, handle NaNs by forward-fill then fill remaining with 0 meta_trend_arr = np.asarray(meta_trend) if not np.issubdtype(meta_trend_arr.dtype, np.number): meta_trend_arr = pd.Series(meta_trend_arr).astype(float).to_numpy() if np.isnan(meta_trend_arr).any(): meta_trend_arr = pd.Series(meta_trend_arr).fillna(method='ffill').fillna(0).astype(int).to_numpy() else: meta_trend_arr = meta_trend_arr.astype(int) meta_trend_arr = np.where(meta_trend_arr != 1, 0, 1) # force only 0 or 1 if hasattr(df.index, 'to_numpy'): x_vals = df.index.to_numpy() else: x_vals = np.array(df.index) # Find contiguous regions regions = [] start = 0 for i in range(1, len(meta_trend_arr)): if meta_trend_arr[i] != meta_trend_arr[i-1]: regions.append((start, i-1, meta_trend_arr[i-1])) start = i regions.append((start, len(meta_trend_arr)-1, meta_trend_arr[-1])) # Draw red vertical lines at the start of each new region (except the first) for region_idx in range(1, len(regions)): region_start = regions[region_idx][0] ax_price.axvline(x=x_vals[region_start], color='black', linestyle='--', alpha=0.7, linewidth=1) for start, end, trend in regions: color = '#089981' if trend == 1 else '#F23645' # Offset by 1 on x: span from x_vals[start] to x_vals[end+1] if possible x_start = x_vals[start] x_end = x_vals[end+1] if end+1 < len(x_vals) else x_vals[end] ax_bar.axvspan(x_start, x_end, color=color, alpha=1, ymin=0, ymax=1) ax_bar.set_ylim(0, 1) ax_bar.set_yticks([]) ax_bar.set_ylabel('Trend') ax_bar.set_xlabel('Time') ax_bar.grid(False) ax_bar.set_title('Meta Trend') plt.tight_layout(h_pad=0.1) plt.show() @staticmethod def format_strategy_data_with_trades(strategy_data, backtest_results): """ Format strategy data for universal plotting with actual executed trades. Converts strategy output into the expected column format: "x_type_name" Args: strategy_data (DataFrame): Output from strategy with columns like 'close', 'UpperBand', 'LowerBand', 'RSI' backtest_results (dict): Results from backtest.run() containing actual executed trades Returns: DataFrame: Formatted data ready for plot_data function """ formatted_df = pd.DataFrame(index=strategy_data.index) # Plot 1: Price data with Bollinger Bands and actual trade signals if 'close' in strategy_data.columns: formatted_df['1_line_close'] = strategy_data['close'] # Bollinger Bands area (prefer standard names, fallback to timeframe-specific) upper_band_col = None lower_band_col = None sma_col = None # Check for standard BB columns first if 'UpperBand' in strategy_data.columns and 'LowerBand' in strategy_data.columns: upper_band_col = 'UpperBand' lower_band_col = 'LowerBand' # Check for 15m BB columns elif 'UpperBand_15m' in strategy_data.columns and 'LowerBand_15m' in strategy_data.columns: upper_band_col = 'UpperBand_15m' lower_band_col = 'LowerBand_15m' if upper_band_col and lower_band_col: formatted_df['1_area_bb_upper'] = strategy_data[upper_band_col] formatted_df['1_area_bb_lower'] = strategy_data[lower_band_col] # SMA/Moving Average line if 'SMA' in strategy_data.columns: sma_col = 'SMA' elif 'SMA_15m' in strategy_data.columns: sma_col = 'SMA_15m' if sma_col: formatted_df['1_line_sma'] = strategy_data[sma_col] # Strategy buy/sell signals (all signals from strategy) as smaller scatter points if 'BuySignal' in strategy_data.columns and 'close' in strategy_data.columns: strategy_buy_points = strategy_data['close'].where(strategy_data['BuySignal'], np.nan) formatted_df['1_scatter_strategy_buy'] = strategy_buy_points if 'SellSignal' in strategy_data.columns and 'close' in strategy_data.columns: strategy_sell_points = strategy_data['close'].where(strategy_data['SellSignal'], np.nan) formatted_df['1_scatter_strategy_sell'] = strategy_sell_points # Actual executed trades from backtest results (larger, more prominent) if 'trades' in backtest_results and backtest_results['trades']: # Create series for buy and sell points buy_points = pd.Series(np.nan, index=strategy_data.index) sell_points = pd.Series(np.nan, index=strategy_data.index) for trade in backtest_results['trades']: entry_time = trade.get('entry_time') exit_time = trade.get('exit_time') entry_price = trade.get('entry') exit_price = trade.get('exit') # Find closest index for entry time if entry_time is not None and entry_price is not None: try: if isinstance(entry_time, str): entry_time = pd.to_datetime(entry_time) # Find the closest index to entry_time closest_entry_idx = strategy_data.index.get_indexer([entry_time], method='nearest')[0] if closest_entry_idx >= 0: buy_points.iloc[closest_entry_idx] = entry_price except (ValueError, IndexError, TypeError): pass # Skip if can't find matching time # Find closest index for exit time if exit_time is not None and exit_price is not None: try: if isinstance(exit_time, str): exit_time = pd.to_datetime(exit_time) # Find the closest index to exit_time closest_exit_idx = strategy_data.index.get_indexer([exit_time], method='nearest')[0] if closest_exit_idx >= 0: sell_points.iloc[closest_exit_idx] = exit_price except (ValueError, IndexError, TypeError): pass # Skip if can't find matching time formatted_df['1_scatter_actual_buy'] = buy_points formatted_df['1_scatter_actual_sell'] = sell_points # Stop Loss and Take Profit levels if 'StopLoss' in strategy_data.columns: formatted_df['1_line_stop_loss'] = strategy_data['StopLoss'] if 'TakeProfit' in strategy_data.columns: formatted_df['1_line_take_profit'] = strategy_data['TakeProfit'] # Plot 2: RSI rsi_col = None if 'RSI' in strategy_data.columns: rsi_col = 'RSI' elif 'RSI_15m' in strategy_data.columns: rsi_col = 'RSI_15m' if rsi_col: formatted_df['2_line_rsi'] = strategy_data[rsi_col] # Add RSI overbought/oversold levels formatted_df['2_line_rsi_overbought'] = 70 formatted_df['2_line_rsi_oversold'] = 30 # Plot 3: Volume (if available) if 'volume' in strategy_data.columns: formatted_df['3_bar_volume'] = strategy_data['volume'] # Add volume moving average if available if 'VolumeMA_15m' in strategy_data.columns: formatted_df['3_line_volume_ma'] = strategy_data['VolumeMA_15m'] return formatted_df @staticmethod def format_strategy_data(strategy_data): """ Format strategy data for universal plotting (without trade signals). Converts strategy output into the expected column format: "x_type_name" Args: strategy_data (DataFrame): Output from strategy with columns like 'close', 'UpperBand', 'LowerBand', 'RSI' Returns: DataFrame: Formatted data ready for plot_data function """ formatted_df = pd.DataFrame(index=strategy_data.index) # Plot 1: Price data with Bollinger Bands if 'close' in strategy_data.columns: formatted_df['1_line_close'] = strategy_data['close'] # Bollinger Bands area (prefer standard names, fallback to timeframe-specific) upper_band_col = None lower_band_col = None sma_col = None # Check for standard BB columns first if 'UpperBand' in strategy_data.columns and 'LowerBand' in strategy_data.columns: upper_band_col = 'UpperBand' lower_band_col = 'LowerBand' # Check for 15m BB columns elif 'UpperBand_15m' in strategy_data.columns and 'LowerBand_15m' in strategy_data.columns: upper_band_col = 'UpperBand_15m' lower_band_col = 'LowerBand_15m' if upper_band_col and lower_band_col: formatted_df['1_area_bb_upper'] = strategy_data[upper_band_col] formatted_df['1_area_bb_lower'] = strategy_data[lower_band_col] # SMA/Moving Average line if 'SMA' in strategy_data.columns: sma_col = 'SMA' elif 'SMA_15m' in strategy_data.columns: sma_col = 'SMA_15m' if sma_col: formatted_df['1_line_sma'] = strategy_data[sma_col] # Stop Loss and Take Profit levels if 'StopLoss' in strategy_data.columns: formatted_df['1_line_stop_loss'] = strategy_data['StopLoss'] if 'TakeProfit' in strategy_data.columns: formatted_df['1_line_take_profit'] = strategy_data['TakeProfit'] # Plot 2: RSI rsi_col = None if 'RSI' in strategy_data.columns: rsi_col = 'RSI' elif 'RSI_15m' in strategy_data.columns: rsi_col = 'RSI_15m' if rsi_col: formatted_df['2_line_rsi'] = strategy_data[rsi_col] # Add RSI overbought/oversold levels formatted_df['2_line_rsi_overbought'] = 70 formatted_df['2_line_rsi_oversold'] = 30 # Plot 3: Volume (if available) if 'volume' in strategy_data.columns: formatted_df['3_bar_volume'] = strategy_data['volume'] # Add volume moving average if available if 'VolumeMA_15m' in strategy_data.columns: formatted_df['3_line_volume_ma'] = strategy_data['VolumeMA_15m'] return formatted_df @staticmethod def plot_data(df): """ Universal plot function for any formatted data. - df: DataFrame with column names in format "x_type_name" where: x = plot number (subplot) type = plot type (line, area, scatter, bar, etc.) name = descriptive name for the data series """ if df.empty: print("No data to plot") return # Parse all columns plot_info = [] for column in df.columns: parts = column.split('_', 2) # Split into max 3 parts if len(parts) < 3: print(f"Warning: Skipping column '{column}' - invalid format. Expected 'x_type_name'") continue try: plot_number = int(parts[0]) plot_type = parts[1].lower() plot_name = parts[2] plot_info.append((plot_number, plot_type, plot_name, column)) except ValueError: print(f"Warning: Skipping column '{column}' - invalid plot number") continue if not plot_info: print("No valid columns found for plotting") return # Group by plot number plots = {} for plot_num, plot_type, plot_name, column in plot_info: if plot_num not in plots: plots[plot_num] = [] plots[plot_num].append((plot_type, plot_name, column)) # Sort plot numbers plot_numbers = sorted(plots.keys()) n_plots = len(plot_numbers) # Create subplots fig, axs = plt.subplots(n_plots, 1, figsize=(16, 6 * n_plots), sharex=True) if n_plots == 1: axs = [axs] # Ensure axs is always a list # Plot each subplot for i, plot_num in enumerate(plot_numbers): ax = axs[i] plot_items = plots[plot_num] # Handle Bollinger Bands area first (needs special handling) bb_upper = None bb_lower = None for plot_type, plot_name, column in plot_items: if plot_type == 'area' and 'bb_upper' in plot_name: bb_upper = df[column] elif plot_type == 'area' and 'bb_lower' in plot_name: bb_lower = df[column] # Plot Bollinger Bands area if both bounds exist if bb_upper is not None and bb_lower is not None: ax.fill_between(df.index, bb_upper, bb_lower, alpha=0.2, color='gray', label='Bollinger Bands') # Plot other items for plot_type, plot_name, column in plot_items: if plot_type == 'area' and ('bb_upper' in plot_name or 'bb_lower' in plot_name): continue # Already handled above data = df[column].dropna() # Remove NaN values for cleaner plots if plot_type == 'line': color = None linestyle = '-' alpha = 1.0 # Special styling for different line types if 'overbought' in plot_name: color = 'red' linestyle = '--' alpha = 0.7 elif 'oversold' in plot_name: color = 'green' linestyle = '--' alpha = 0.7 elif 'stop_loss' in plot_name: color = 'red' linestyle = ':' alpha = 0.8 elif 'take_profit' in plot_name: color = 'green' linestyle = ':' alpha = 0.8 elif 'sma' in plot_name: color = 'orange' alpha = 0.8 elif 'volume_ma' in plot_name: color = 'purple' alpha = 0.7 ax.plot(data.index, data, label=plot_name.replace('_', ' ').title(), color=color, linestyle=linestyle, alpha=alpha) elif plot_type == 'scatter': color = 'green' if 'buy' in plot_name else 'red' if 'sell' in plot_name else 'blue' marker = '^' if 'buy' in plot_name else 'v' if 'sell' in plot_name else 'o' size = 100 if 'buy' in plot_name or 'sell' in plot_name else 50 alpha = 0.8 zorder = 5 label_name = plot_name.replace('_', ' ').title() # Special styling for different signal types if 'actual_buy' in plot_name: color = 'darkgreen' marker = '^' size = 120 alpha = 1.0 zorder = 10 # Higher z-order to appear on top label_name = 'Actual Buy Trades' elif 'actual_sell' in plot_name: color = 'darkred' marker = 'v' size = 120 alpha = 1.0 zorder = 10 # Higher z-order to appear on top label_name = 'Actual Sell Trades' elif 'strategy_buy' in plot_name: color = 'lightgreen' marker = '^' size = 60 alpha = 0.6 zorder = 3 # Lower z-order to appear behind actual trades label_name = 'Strategy Buy Signals' elif 'strategy_sell' in plot_name: color = 'lightcoral' marker = 'v' size = 60 alpha = 0.6 zorder = 3 # Lower z-order to appear behind actual trades label_name = 'Strategy Sell Signals' ax.scatter(data.index, data, label=label_name, color=color, marker=marker, s=size, alpha=alpha, zorder=zorder) elif plot_type == 'area': ax.fill_between(data.index, data, alpha=0.5, label=plot_name.replace('_', ' ').title()) elif plot_type == 'bar': ax.bar(data.index, data, alpha=0.7, label=plot_name.replace('_', ' ').title()) else: print(f"Warning: Plot type '{plot_type}' not supported for column '{column}'") # Customize subplot ax.grid(True, alpha=0.3) ax.legend() # Set titles and labels if plot_num == 1: ax.set_title('Price Chart with Bollinger Bands and Signals') ax.set_ylabel('Price') elif plot_num == 2: ax.set_title('RSI Indicator') ax.set_ylabel('RSI') ax.set_ylim(0, 100) elif plot_num == 3: ax.set_title('Volume') ax.set_ylabel('Volume') else: ax.set_title(f'Plot {plot_num}') # Set x-axis label only on the bottom subplot axs[-1].set_xlabel('Time') plt.tight_layout() plt.show()