bt_utils.py

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