import os os.environ.setdefault("MPLBACKEND", "Agg") from datetime import datetime, timedelta from io import BytesIO import backtrader as bt import matplotlib import matplotlib.pyplot as plt import pandas as pd matplotlib.use("Agg") # Set the backend to non-interactive fmp_intervals_to_int = {"1min":1, "5min":5, "15min":15, "1hour":60, "4hour":240} class TransactionRecorder(bt.Analyzer): """ Records all buy/sell transactions with details.""" def __init__(self): self.records = [] def notify_order(self, order): if order.status != order.Completed: return self.records.append({ 'datetime': self.strategy.data.datetime.datetime(), 'type': 'BUY' if order.isbuy() else 'SELL', 'price': order.executed.price, 'size': order.executed.size, 'value': order.executed.value, 'commission': order.executed.comm, }) def get_analysis(self): return pd.DataFrame(self.records) class TradeRecorder(bt.Analyzer): """ Records detailed trade information including entry/exit prices and PnL.""" def __init__(self): self.records = [] self.trade_id = 0 def notify_trade(self, trade): # RECORD ENTRY if trade.isopen: self.current_entry_price = trade.price self.current_size = trade.size return # RECORD EXIT if trade.isclosed: self.trade_id += 1 # Backtrader clears trade.size to 0 on close, so restore saved size size = self.current_size # Compute exit price from PnL formula: # pnl = (exit - entry) * size if size is not None and size != 0: exit_price = self.current_entry_price + (trade.pnl / size) else: exit_price = None self.records.append({ 'trade_id': self.trade_id, 'size': size, 'entry_price': self.current_entry_price, 'exit_price': exit_price, 'bruto_profitloss': trade.pnl, 'neto_profitloss': trade.pnlcomm, 'date_open': self._format_dt(trade.dtopen), 'date_close': self._format_dt(trade.dtclose), }) # Reset after close self.current_entry_price = None self.current_size = None def get_analysis(self): return pd.DataFrame(self.records) def _format_dt(self, val): """Convert a backtrader/matplotlib numeric datetime to a readable string.""" if val is None: return None try: ord_day = int(val) frac = val - ord_day dt = datetime.fromordinal(ord_day) + timedelta(days=frac) return dt.strftime("%Y-%m-%d %H:%M:%S") except Exception as e: return "Error formatting dt: " + str(e) def _fig_to_numpy(fig, dpi=150): buf = BytesIO() fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight") buf.seek(0) img = plt.imread(buf, format="png") buf.close() return img #TODO: Dates plotting not working def plot_bt(figs, symbol, market_name, save_img=True): """ Plot backtrader results with market name in title and readable date ticks on all x-axes. """ images = [] plt.ioff() # Turn off interactive mode for i, fig_list in enumerate(figs): for j, fig in enumerate(fig_list): # Decorate titles if fig.axes: main_ax = fig.axes[0] current_title = main_ax.get_title() new_title = f"{market_name} ({symbol.upper()}) - {current_title or 'Price Chart'}" main_ax.set_title(new_title, fontsize=8, fontweight="bold") fig.suptitle( f"Trading Strategy Analysis: {market_name}", fontsize=10, fontweight="bold", y=0.98, ) fig.set_size_inches(12, 6) fig.autofmt_xdate() fig.tight_layout(rect=[0, 0, 1, 0.95]) # Convert to numpy img_data = _fig_to_numpy(fig) images.append(img_data) if save_img: filename = f"plot_{symbol}_{i}_{j}.png" fig.savefig(filename, dpi=300, bbox_inches="tight") print(f"Chart saved as: {filename}") plt.close(fig) # Close the figure to free memory return images def run_bt(cerebro, tckr_symbl="SPY", replay=False, replay_compression=15, save_img=False, interval=None, market_name = "Complete Market Name here", initial_capital=10000.0, commission=0.001, slippage_percent=0.01, df=None): """ Run backtrader strategy with enhanced plotting Args: strategy: Backtrader strategy class already init tckr_symbl: Stock ticker symbol save_img: Whether to save plot images initial_capital: Starting capital for the broker commission: Commission per share slippage_percent: Percent (e.g., 0.01 for 0.01%) applied as slippage df: pandas DataFrame with datetime index. """ print(f"Running strategy on: {market_name} ({tckr_symbl.upper()})") print("-" * 50) print(f"Initial capital: {initial_capital}, Commission: {commission}, Slippage%: {slippage_percent}") if df is None: raise ValueError("df is required. Load data with get_data before calling run_bt.") if replay: print(f"Replaying data on: {market_name} ({tckr_symbl.upper()}) at interval {interval} with compression {replay_compression}") data = bt.feeds.PandasData(dataname=df, timeframe=bt.TimeFrame.Minutes, compression=fmp_intervals_to_int[interval]) cerebro.replaydata(data, timeframe=bt.TimeFrame.Minutes, compression=replay_compression) else: data = bt.feeds.PandasData(dataname=df) cerebro.adddata(data) # Set initial cash and commission initial_cash = float(initial_capital) cerebro.broker.setcash(initial_cash) cerebro.broker.setcommission(commission=float(commission)) slippage_decimal = float(slippage_percent) / 100.0 cerebro.broker.set_slippage_perc(slippage_decimal) # Add analyzers for better performance metrics cerebro.addanalyzer(bt.analyzers.Returns, _name='returns') cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') cerebro.addanalyzer(TransactionRecorder, _name='transactions') cerebro.addanalyzer(TradeRecorder, _name='trades') # Print starting conditions print(f'Starting Portfolio Value: ${cerebro.broker.getvalue():,.2f}') # Run strategy results = cerebro.run() # Calculate and display results final_value = cerebro.broker.getvalue() total_return = (final_value - initial_cash) / initial_cash * 100 print(f'Final Portfolio Value: ${final_value:,.2f}') print(f'Total Return: {total_return:.2f}%') # Print analyzer results strat = results[0] try: sharpe = strat.analyzers.sharpe.get_analysis().get('sharperatio', 'N/A') if sharpe != 'N/A': print(f'Sharpe Ratio: {sharpe:.3f}') else: print('Sharpe Ratio: N/A') except: print('Sharpe Ratio: N/A') try: max_dd = strat.analyzers.drawdown.get_analysis()['max']['drawdown'] print(f'Max Drawdown: {max_dd:.2f}%') except: print('Max Drawdown: N/A') # Trades and transaction tables df_transactions = strat.analyzers.transactions.get_analysis() df_trades = strat.analyzers.trades.get_analysis() print(f"Number of Trades: {len(df_trades)}") print("-" * 50) if not df_transactions.empty: print("Transactions logs generated") else: print("No transactions recorded.") if not df_trades.empty: print("Trades logs generated") else: print("No trades recorded.") # Generate plot with market name (disable interactive plotting to avoid GUI in threads) fig = [] try: figs = cerebro.plot( style='candlestick', barstyle='candlestick', subplot=True, plotabove=False, downsample=False, plotstyle='multiple', iplot=False, show=False, dpi=120, ) print("How many charts where created: ", len(figs)) fig = plot_bt(figs, market_name, tckr_symbl, save_img) except Exception as e: print(f"Plot skipped due to error: {e}") return final_value, total_return, fig, df_trades, df_transactions