backtest.py

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

class BacktestInformation:
    def __init__(self, df_pos, df_close, fee_rate=0.0005, strategy_type='full_equal',
                 allocation_fraction=1.0, target_active=None,
                 warn_threshold=True):
        """
        Enhanced backtest with uniform fee rate=0.05% for crypto futures.
        Fee applied to turnover, capturing changes including drift for realistic futures rebalancing.
        """
        # Align dataframes
        self.df_pos, self.df_close = df_pos.align(df_close, join='inner')
        self.symbols = self.df_pos.columns
        self.fee_rate = fee_rate
        self.strategy_type = strategy_type
        self.allocation_fraction = allocation_fraction
        self.target_active = target_active
        self.warn_threshold = warn_threshold

        # Ensure datetime index
        self.df_pos.index = pd.to_datetime(self.df_pos.index)
        self.df_close.index = pd.to_datetime(self.df_close.index)

        # Drop any NaN rows
        self.df_pos = self.df_pos.dropna(how='all')
        self.df_close = self.df_close.dropna(how='all')

        # Compute returns at original frequency
        self.df_ret = self.df_close.pct_change(fill_method=None).fillna(0)

        # Compute weights
        self._compute_weights()

        # Gross returns
        self.symbol_contrib = self.weight_df.shift(1).fillna(0) * self.df_ret
        self.gross_ret = self.symbol_contrib.sum(axis=1)

        # Turnover calculation
        pre_w = self.weight_df.shift(1).fillna(0)
        port_ret_t = (pre_w * self.df_ret).sum(axis=1)
        w_tilde = (pre_w * (1 + self.df_ret)).div(1 + port_ret_t, axis=0).fillna(0)

        self.turnover_by_symbol = (self.weight_df - w_tilde).abs()
        self.turnover = self.turnover_by_symbol.sum(axis=1)

        self.fee_by_symbol = self.fee_rate * self.turnover_by_symbol
        self.fee = self.fee_by_symbol.sum(axis=1)

        self.symbol_net_contrib = self.symbol_contrib - self.fee_by_symbol
        self.net_ret = self.symbol_net_contrib.sum(axis=1).dropna()

        # Fee: Uniform rate for all changes, suitable for crypto futures (taker fee per delta)
        self.fee = self.fee_rate * self.turnover

        # Net returns
        self.net_ret = self.gross_ret - self.fee
        self.net_ret = self.net_ret.dropna()

        # Resample to daily
        self._resample_to_daily()
        self._compute_trade_stats()

        # Cumulative PnL
        self.cum_pnl = self.net_ret_daily.cumsum()

        # Compute metrics
        self.compute_metrics()

    def _compute_weights(self):
        """Compute portfolio weights - Equal for active symbols"""
        if self.strategy_type not in ['full_equal', 'partial_equal']:
            raise ValueError("strategy_type must be 'full_equal' or 'partial_equal'")

        weight_df = pd.DataFrame(0.0, index=self.df_pos.index, columns=self.df_pos.columns)

        active_mask = self.df_pos != 0
        num_active = active_mask.sum(axis=1)

        if self.strategy_type == 'partial_equal' and self.warn_threshold and self.target_active is not None:
            mismatch = num_active != self.target_active
            if mismatch.any():
                for idx in mismatch[mismatch].index:
                    print(f"Warning at {idx}: Active symbols {num_active[idx]} != target {self.target_active}")

        if self.strategy_type == 'full_equal':
            n_symbols = len(self.symbols)
            equal_weight = self.allocation_fraction / n_symbols
            weight_df += active_mask * equal_weight * np.sign(self.df_pos)
        else:  # partial_equal
            equal_weight = self.allocation_fraction / num_active.replace(0, np.nan).fillna(1)
            weight_df += active_mask.multiply(equal_weight, axis=0) * np.sign(self.df_pos)

        self.weight_df = weight_df

    def _compute_trade_stats(self):
        trade_pnls = []
        for symbol in self.symbols:
            pos = self.df_pos[symbol]
            side = np.sign(pos).fillna(0)

            entry_mask = (side != side.shift()).fillna(False) & (side != 0)
            trade_ids = entry_mask.cumsum().where(side != 0)

            contrib = self.symbol_net_contrib[symbol]
            active_prev = side.shift(1) != 0
            trade_ids_active = trade_ids.shift(1).where(active_prev)
            valid = trade_ids_active.notna()

            if not valid.any():
                continue

            contrib_active = contrib[valid]
            trade_ids_active = trade_ids_active[valid].astype(int)
            per_trade = contrib_active.groupby(trade_ids_active).sum()
            trade_pnls.extend(per_trade.tolist())

        trade_pnls = np.array(trade_pnls, dtype=float)
        if trade_pnls.size == 0:
            self.avg_profit_per_win_trade_perc = 0
            self.avg_loss_per_loss_trade_perc = 0
            self.num_trades = 0
            return

        wins = trade_pnls[trade_pnls > 0]
        losses = trade_pnls[trade_pnls < 0]
        self.avg_profit_per_win_trade_perc = wins.mean() * 100 if wins.size else 0
        self.avg_loss_per_loss_trade_perc = losses.mean() * 100 if losses.size else 0
        self.num_trades = trade_pnls.size

    def _resample_to_daily(self):
        """Resample to daily frequency"""
        self.net_ret_daily = self.net_ret.resample('1D').sum().ffill().fillna(0)
        self.turnover_daily = self.turnover.resample('1D').sum().fillna(0)

    def compute_metrics(self):
        """Compute performance metrics"""
        if len(self.net_ret_daily) == 0:
            self.metrics = {
                'total_profit_perc': 0,
                'annual_return_perc': 0,
                'sharpe': 0,
                'daily_turnover_perc': 0,
                'fitness': 0,
                'max_drawdown_perc': 0,
                'calmar': 0,
                'hit_rate_perc': 0,
                'margin_bps': 0,
                'avg_long_exposure': 0,
                'avg_short_exposure': 0,
                'avg_net_exposure': 0,
                'avg_total_exposure': 0,
            }
            return

        # Exposure metrics
        long_exposure = self.weight_df.clip(lower=0).sum(axis=1)
        short_exposure = -self.weight_df.clip(upper=0).sum(axis=1)
        net_exposure = long_exposure - short_exposure
        total_exposure = long_exposure + short_exposure

        self.avg_long_exposure = long_exposure.mean()
        self.avg_short_exposure = short_exposure.mean()
        self.avg_net_exposure = net_exposure.mean()
        self.avg_total_exposure = total_exposure.mean()

        # Total profit %
        self.total_profit_perc = self.cum_pnl.iloc[-1] * 100

        # Annual return %
        total_days = (self.cum_pnl.index[-1] - self.cum_pnl.index[0]).days + 1
        total_years = total_days / 365.0
        self.annual_return_perc = (self.cum_pnl.iloc[-1] / total_years) * 100 if total_years > 0 else 0

        # Sharpe
        daily_mean = self.net_ret_daily.mean()
        std = self.net_ret_daily.std()
        self.sharpe = (daily_mean / std) * np.sqrt(365) if std != 0 else 0

        # Daily turnover %
        avg_daily_turnover = self.turnover_daily.mean()
        self.daily_turnover_perc = avg_daily_turnover * 100

        # Fitness
        abs_returns = abs(self.annual_return_perc / 100)
        turnover_for_fitness = max(avg_daily_turnover, 0.125)
        self.fitness = self.sharpe * np.sqrt(abs_returns / turnover_for_fitness) if turnover_for_fitness > 0 else 0

        # Max Drawdown %
        equity = 1 + self.cum_pnl
        peak = equity.cummax()
        drawdown = (equity / peak) - 1
        self.max_drawdown_perc = drawdown.min() * -100 if not drawdown.empty else 0

        # Calmar
        mdd_abs = abs(self.max_drawdown_perc / 100) if self.max_drawdown_perc != 0 else np.nan
        self.calmar = (self.annual_return_perc / 100) / mdd_abs if not np.isnan(mdd_abs) else 0

        # Hit Rate %
        self.hit_rate_perc = (self.net_ret_daily > 0).mean() * 100 if len(self.net_ret_daily) > 0 else 0

        # Margin (bps)
        self.margin_bps = (self.cum_pnl.iloc[-1] / self.avg_total_exposure) * 10000 if self.avg_total_exposure > 0 else 0

        self.metrics = {
            'total_profit_perc': self.total_profit_perc,
            'annual_return_perc': self.annual_return_perc,
            'sharpe': self.sharpe,
            'daily_turnover_perc': self.daily_turnover_perc,
            'fitness': self.fitness,
            'max_drawdown_perc': self.max_drawdown_perc,
            'calmar': self.calmar,
            'hit_rate_perc': self.hit_rate_perc,
            'margin_bps': self.margin_bps,
            'avg_profit_per_win_trade_perc': self.avg_profit_per_win_trade_perc,
            'avg_loss_per_loss_trade_perc': self.avg_loss_per_loss_trade_perc,
        }

    def print_metrics(self):
        """Print performance metrics"""
        print("=== Performance Metrics ===")
        for key, value in self.metrics.items():
            display_name = key.replace('_perc', ' (%)').replace('_bps', ' (bps)').replace('_', ' ').title()
            print(f"{display_name}: {value:.6f}")

    def print_allocation_summary(self):
        """Print allocation summary"""
        n_symbols = len(self.symbols)
        if self.strategy_type == 'full_equal':
            equal_weight = self.allocation_fraction * 100 / n_symbols
            print(f"\n=== Full Equal Allocation Summary ===")
            print(f"Universe size: {n_symbols} symbols")
            print(f"Equal weight per symbol: {equal_weight:.2f}%")
        elif self.strategy_type == 'partial_equal':
            print(f"\n=== Partial Equal Allocation Summary ===")
            print(f"Allocation fraction: {self.allocation_fraction*100:.2f}%")
            if self.target_active:
                print(f"Target active symbols: {self.target_active}")

    def plot_pnl(self, figsize=(12, 6)):
        """Plot cumulative PnL"""
        plt.figure(figsize=figsize)
        plt.plot(self.cum_pnl.index, self.cum_pnl * 100, label='Cumulative PnL after Fee (%)', color='blue')
        plt.xlabel('Date')
        plt.ylabel('Return (%)')
        plt.title('Portfolio PnL after Fees')
        plt.legend()
        plt.grid(True)
        plt.show()

    def plot_exposure(self, figsize=(12, 8)):
        """Plot portfolio exposure over time"""
        long_exposure = self.weight_df.clip(lower=0).sum(axis=1) * 100
        short_exposure = -self.weight_df.clip(upper=0).sum(axis=1) * 100
        net_exposure = long_exposure - short_exposure

        fig, (ax1, ax2) = plt.subplots(2, 1, figsize=figsize, sharex=True)

        ax1.plot(long_exposure.index, long_exposure, label='Long Exposure (%)', color='green', alpha=0.7)
        ax1.plot(short_exposure.index, short_exposure, label='Short Exposure (%)', color='red', alpha=0.7)
        ax1.fill_between(long_exposure.index, 0, long_exposure, color='green', alpha=0.3)
        ax1.fill_between(short_exposure.index, 0, short_exposure, color='red', alpha=0.3)
        ax1.set_ylabel('Exposure (%)')
        ax1.set_title('Portfolio Long/Short Exposure')
        ax1.legend()
        ax1.grid(True)

        ax2.plot(net_exposure.index, net_exposure, label='Net Exposure (%)', color='blue')
        ax2.axhline(y=0, color='black', linestyle='--', alpha=0.5)
        ax2.fill_between(net_exposure.index, 0, net_exposure, color='blue', alpha=0.3)
        ax2.set_xlabel('Date')
        ax2.set_ylabel('Net Exposure (%)')
        ax2.set_title('Portfolio Net Exposure')
        ax2.legend()
        ax2.grid(True)

        plt.tight_layout()
        plt.show()

    def plot_yearly_metrics_table(self, figsize=(12, 4)):
        """Plot yearly performance metrics table"""
        years = self.net_ret_daily.index.year.unique()
        data = []
        for year in years:
            net_ret_year = self.net_ret_daily[self.net_ret_daily.index.year == year]
            if len(net_ret_year) == 0:
                continue

            annual_ret_year = net_ret_year.sum() * 100

            mean_year = net_ret_year.mean()
            std_year = net_ret_year.std()
            sharpe_year = (mean_year / std_year) * np.sqrt(365) if std_year != 0 else 0

            turnover_year = self.turnover_daily[self.turnover_daily.index.year == year].mean() * 100

            cum_pnl_year = net_ret_year.cumsum()
            equity_year = 1 + cum_pnl_year
            peak_year = equity_year.cummax()
            dd_year = (equity_year / peak_year) - 1
            mdd_year = dd_year.min() * -100 if not dd_year.empty else 0

            abs_ret_year = abs(annual_ret_year / 100)
            to_year = max(turnover_year / 100, 0.125)
            fitness_year = sharpe_year * np.sqrt(abs_ret_year / to_year) if to_year > 0 else 0

            data.append([sharpe_year, mdd_year, turnover_year, annual_ret_year, fitness_year])

        if not data:
            print("No data for yearly metrics.")
            return

        df_yearly = pd.DataFrame(data, index=years,
                                 columns=['Sharpe', 'MDD (%)', 'Turnover (%)', 'Return (%)', 'Fitness'])
        df_yearly = df_yearly.round(2)

        fig, ax = plt.subplots(figsize=figsize)
        ax.axis('tight')
        ax.axis('off')
        table = ax.table(cellText=df_yearly.values, colLabels=df_yearly.columns,
                         rowLabels=df_yearly.index, loc='center', cellLoc='center')
        table.auto_set_font_size(False)
        table.set_fontsize(10)
        table.scale(1.2, 1.2)
        plt.title('Yearly Performance Metrics')
        plt.show()

    def plot_symbol_contributions(self, figsize=(12, 6), max_symbols=20):
        """Plot bar chart of cumulative contributions per symbol (% return) - Dynamic for large universes"""
        per_symbol_pnl = (self.weight_df.shift(1).fillna(0) * self.df_ret).cumsum().iloc[-1] * 100
        per_symbol_pnl = per_symbol_pnl.sort_values(ascending=False)

        if len(per_symbol_pnl) > max_symbols:
            top_pos = per_symbol_pnl.head(max_symbols // 2)
            bottom_neg = per_symbol_pnl.tail(max_symbols // 2)
            per_symbol_pnl = pd.concat([top_pos, bottom_neg])
            title = f'Top/Bottom Symbol Contributions (%) - Showing {max_symbols} out of {len(self.symbols)}'
        else:
            title = 'Symbol Contributions to Total PnL (%)'

        plt.figure(figsize=figsize)
        per_symbol_pnl.plot(kind='bar', color=['green' if x > 0 else 'red' for x in per_symbol_pnl])
        plt.title(title)
        plt.xlabel('Symbols')
        plt.ylabel('Cumulative Contribution (%)')
        plt.xticks(rotation=45, ha='right')
        plt.grid(True, axis='y')
        plt.tight_layout()
        plt.show()

    def plot_yearly_symbol_contributions(self, max_symbols=20, figsize=(15, 12)):
        """Plot bar charts of symbol contributions per year (% return) - Arranged in subplots with 2 per row"""
        # Compute daily contributions per symbol
        daily_contrib = self.weight_df.shift(1).fillna(0) * self.df_ret

        # Resample to yearly sum (total contribution per year per symbol)
        yearly_contrib = daily_contrib.resample('Y').sum()

        # Get unique years from index (end-of-year dates)
        years = yearly_contrib.index.year.unique()
        n_years = len(years)

        if n_years == 0:
            print("No data for yearly contributions.")
            return

        # Determine grid: 2 columns, rows = ceil(n_years / 2)
        nrows = (n_years + 1) // 2
        ncols = 2
        fig, axs = plt.subplots(nrows=nrows, ncols=ncols, figsize=figsize)
        axs = axs.flatten()  # Flatten for easy indexing

        for i, year in enumerate(years):
            # Get contribution for that year (single row, as resampled)
            per_year = yearly_contrib[yearly_contrib.index.year == year].iloc[0] * 100
            per_year = per_year.sort_values(ascending=False)

            # Limit to top 10 and bottom 10 if > max_symbols
            if len(per_year) > max_symbols:
                top = per_year.head(max_symbols // 2)
                bottom = per_year.tail(max_symbols // 2)
                per_year_limited = pd.concat([top, bottom])
                title = f'Top/Bottom {year} Contributions (%) - {max_symbols} out of {len(self.symbols)}'
            else:
                per_year_limited = per_year
                title = f'Symbol Contributions {year} (%)'

            # Plot on subplot
            per_year_limited.plot(kind='bar', ax=axs[i], color=['green' if x > 0 else 'red' for x in per_year_limited])
            axs[i].set_title(title)
            axs[i].set_xlabel('Symbols')
            axs[i].set_ylabel('Contribution (%)')
            axs[i].tick_params(axis='x', rotation=45)
            axs[i].grid(True, axis='y')

        # Hide unused subplots if any
        for j in range(i + 1, nrows * ncols):
            axs[j].axis('off')

        plt.suptitle('Yearly Symbol Contributions to PnL')
        plt.tight_layout()
        plt.show()

    def plot_symbol_turnover_table(self, freq='ME', top_n=None, figsize=(16, 16), annot=False, orientation='time'):
        turnover = self.turnover_by_symbol.resample(freq).sum()

        if top_n:
            keep = turnover.sum().nlargest(top_n).index
            turnover = turnover[keep]

        # format nhãn cho cột/row theo freq
        if freq.upper().startswith('M'):
            time_labels = turnover.index.strftime('%Y-%m')
        elif freq.upper().startswith('Q'):
            time_labels = turnover.index.to_period('Q').astype(str)
        elif freq.upper().startswith('Y'):
            time_labels = turnover.index.year.astype(str)
        else:
            time_labels = turnover.index.strftime('%Y-%m-%d')

        if orientation == 'symbols':
            pivot = turnover.T
            x_label, y_label = 'Datetime', 'Symbol'
            cols = time_labels
        else:
            pivot = turnover
            x_label, y_label = 'Symbol', 'Datetime'
            cols = turnover.columns
            pivot.index = time_labels

        plt.figure(figsize=figsize)
        sns.heatmap(pivot, cmap='Blues', annot=annot, fmt='.3f',
                    cbar_kws={'label': 'Turnover'})
        plt.title(f'Turnover in all symbol ({freq})')
        plt.xlabel(x_label)
        plt.ylabel(y_label)
        plt.tight_layout()
        plt.show()