BTST Trades: Why It Looks Great on Paper, But Rarely Works in Reality

Baseline (EOD close entry and next day open entry) - 4lac max pos size limit
===== Performance Summary =====
Starting Balance: 100000.00
Ending Balance:   3094845.21
Total Net PnL:    2994845.21
Total Net PnL %:  2994.85%
CAGR:             98.76%
Max Drawdown (%): 9.99%
Max Drawdown ($): 3341091.74
Number of Trades: 1242
Win Rate (%):     53.54%
Average Win (%):  2.2719%
Average Loss (%): -0.8616%
Average Win Amount:  7174.9250
Average Loss Amount: -3078.8213
Risk/Reward Ratio: 2.330412932197329
Sharpe Ratio:      3.8911
Sortino Ratio:     7.4598
===============================

# ------------------------------------------------------------------------------------
#                          FabTrader Algorithmic Trading Platform
# ------------------------------------------------------------------------------------
# Copyright (c) 2022 FabTrader (Unit of Rough Sketch Company)
#
# LICENSE: PROPRIETARY SOFTWARE
# - This software is the exclusive property of FabTrader.
# - Unauthorized copying, modification, distribution, or use is strictly prohibited.
# - Written permission from the author is required for any use beyond personal,
#   non-commercial purposes.
#
# CONTACT:
# - Website: https://fabtrader.in
# - Email: [email protected]
#
# Usage: Internal use only. Not for commercial redistribution.
# Permissions and licensing inquiries should be directed to the contact email.
"""
GapBees BTST Backtester (Long-only)
-----------------------------------
Strategy:
 - Name: GapBees
 - Market: NSE (instrument: NIFTYBEES)
 - Type: Buy Today Sell Tomorrow (enter at close, exit next-day open)
 - Direction: Long-only

Features:
 - Costs: fixed brokerage per trade (13) and slippage (0.3% of execution price)
 - Input data function placeholder: get_historical_data(...)
 - Handles missing data and final-day-without-next-open cases
 - Produces trade log DataFrame and prints performance metrics requested

Requirements:
 - Python 3.11+
 - pandas, numpy
"""

from __future__ import annotations
from dataclasses import dataclass, asdict
from datetime import date, datetime
from typing import Optional, List, Tuple
import math
import os

import numpy as np
import pandas as pd

# ---------------------------
# Placeholder for data source
# ---------------------------
def get_historical_data(symbol: str, start_date: date, end_date: date, interval: str = "1d") -> pd.DataFrame:

    df = Instruments.get_historical_data(symbol, start_date, end_date)
    return df

def get_minute_data(symbol: str, start_date: date, end_date: date) -> pd.DataFrame:

    df = Instruments.get_historical_data(symbol, start_date, end_date, 'minute')
    if df.empty:
        return pd.DataFrame()
    return df

# ---------------------------
# Trade dataclass
# ---------------------------
@dataclass
class Trade:
    tradingSymbol: str
    tradeState: str  # 'completed' or 'open'
    entry_date: pd.Timestamp
    direction: str  # 'Long'/'Short' (here always 'Long')
    qty: int
    entry: float
    Target: Optional[float] = math.nan
    StopLoss: Optional[float] = math.nan
    exit_date: Optional[pd.Timestamp] = pd.NaT
    exit: Optional[float] = math.nan
    pnl: Optional[float] = 0.0
    pnl_pct: Optional[float] = 0.0
    brokerage: Optional[float] = 0.0
    netPnl: Optional[float] = 0.0
    netPnl_pct: Optional[float] = 0.0

    def to_dict(self):
        d = asdict(self)
        # rename pnl_pct -> 'pnl%' and netPnl_pct -> 'netPnl%'
        d["pnl%"] = d.pop("pnl_pct")
        d["netPnl%"] = d.pop("netPnl_pct")
        # ensure NaN / NaT values are pandas-friendly
        return d


# ---------------------------
# Backtest engine
# ---------------------------
class BacktestEngine:
    """
    BacktestEngine for the GapBees BTST strategy.

    Usage:
        engine = BacktestEngine(
            symbol="NIFTYBEES",
            start_date=date(2020,1,1),
            end_date=date(2021,12,31),
            initial_capital=100000,
            allocation=0.5,
            brokerage_per_trade=13,
            slippage_pct=0.003
        )
        engine.run_backtest(data_loader=get_historical_data)
        trades_df = engine.trades_df
        engine.print_performance()
    """

    def __init__(
        self,
        symbol: str,
        start_date: date,
        end_date: date,
        initial_capital: float = 100000.0,
        allocation: float = 0.5,  # fraction of portfolio to allocate each trade
        brokerage_per_trade: float = 13.0,
        slippage_pct: float = 0.003,  # 0.3%
        timeframe: str = "1d",
        trading_days_per_year: int = 252,
    ):
        self.symbol = symbol
        self.start_date = start_date
        self.end_date = end_date
        self.initial_capital = float(initial_capital)
        self.allocation = float(allocation)
        self.brokerage = float(brokerage_per_trade)
        self.slippage_pct = float(slippage_pct)
        self.timeframe = timeframe
        self.trading_days_per_year = trading_days_per_year

        # runtime containers
        self.equity_curve = pd.Series(dtype=float)
        self.trades: List[Trade] = []
        self.trades_df: Optional[pd.DataFrame] = None
        self.daily_portfolio_values = pd.Series(dtype=float)

    # ---------------------------
    # Utility functions
    # ---------------------------
    @staticmethod
    def _annualize_returns(cum_return: float, days: int) -> float:
        if days <= 0:
            return 0.0
        years = days / 365.25
        if years <= 0:
            return 0.0
        return (1.0 + cum_return) ** (1.0 / years) - 1.0

    @staticmethod
    def _max_drawdown(equity_series: pd.Series) -> Tuple[float, float]:
        """
        Returns (max_drawdown_pct, max_drawdown_amount)
        equity_series: cumulative equity values indexed by date
        """
        if equity_series.empty:
            return 0.0, 0.0
        running_max = equity_series.cummax()
        drawdown = (equity_series - running_max) / running_max
        max_dd_pct = drawdown.min()  # negative
        # find drawdown amount (peak - trough)
        peak_idx = (running_max - equity_series).idxmin()  # index where distance from running_max minimal? Careful...
        # Simpler: compute max peak-trough difference
        peak = running_max.max()
        trough = equity_series.min()
        max_dd_amt = peak - trough
        return float(abs(max_dd_pct) * 100.0), float(max_dd_amt)

    @staticmethod
    def _sharpe_ratio(daily_returns: pd.Series, trading_days_per_year: int = 252, risk_free_rate: float = 0.0) -> float:
        """
        Annualized Sharpe ratio (using daily returns).
        risk_free_rate is annual; convert to daily.
        """
        if daily_returns.empty:
            return float("nan")
        excess_daily = daily_returns - (risk_free_rate / trading_days_per_year)
        mean_excess = excess_daily.mean()
        std = excess_daily.std(ddof=1)
        if std == 0 or np.isnan(std):
            return float("nan")
        annualized = (mean_excess * trading_days_per_year) / (std * math.sqrt(trading_days_per_year))
        return float(annualized)

    @staticmethod
    def _sortino_ratio(daily_returns: pd.Series, trading_days_per_year: int = 252, risk_free_rate: float = 0.0) -> float:
        """
        Annualized Sortino ratio using daily returns; downside deviation uses negative returns only.
        """
        if daily_returns.empty:
            return float("nan")
        target = risk_free_rate / trading_days_per_year
        negative_diff = daily_returns - target
        downside = negative_diff[negative_diff < 0]
        if downside.size == 0:
            return float("nan")
        downside_std = downside.std(ddof=1)
        if downside_std == 0 or np.isnan(downside_std):
            return float("nan")
        mean_excess = (daily_returns.mean() - target)
        annualized = (mean_excess * trading_days_per_year) / (downside_std * math.sqrt(trading_days_per_year))
        return float(annualized)

    # ---------------------------
    # Core backtest logic
    # ---------------------------
    def run_backtest(self, data_loader=get_historical_data, verbose: bool = True):
        """
        Run the BTST backtest.

        data_loader: function with signature (symbol, start_date, end_date, interval) -> pd.DataFrame
        (index must be datetime-like and named 'Date' or have DatetimeIndex)
        """
        # load data
        try:
            df = data_loader(self.symbol, self.start_date, self.end_date, self.timeframe)
        except Exception as e:
            raise RuntimeError("Data loader failed. Ensure get_historical_data returns a DataFrame.") from e

        if not isinstance(df, pd.DataFrame):
            raise TypeError("Data loader must return a pandas DataFrame.")

        # Normalize index and columns
        if df.index.name != "Date":
            # ensure timezone-naive pandas Timestamp index and rename index
            df.index = pd.to_datetime(df.index)
            df.index.name = "Date"

        required_cols = {"Open", "High", "Low", "Close", "Volume"}
        if not required_cols.issubset(set(df.columns)):
            raise ValueError(f"DataFrame missing required columns. Found columns: {df.columns}")

        # Sort by date
        df = df.sort_index()

        # Drop rows with no Close or no Open (can't enter / exit)
        # But be careful: we only need next-day Open to exit; we will handle last day later.
        df_clean = df.copy()

        # Initialize bookkeeping
        portfolio_value = self.initial_capital
        equity_list = []
        daily_values = []

        # We will iterate through rows (by date). For each date t we:
        #  - generate entry signal at end-of-day t (always enter long)
        #  - execute entry at Close_t (apply slippage and brokerage)
        #  - exit at next trading day's Open_{t+1}
        # To compute pnl, we need next day open. Skip if next day is not available or missing.

        dates = df_clean.index.to_list()
        n = len(dates)

        for i, dt in enumerate(dates):
            # record starting-of-day portfolio value for dt
            daily_values.append((dt, portfolio_value))

            # cannot open a BTST trade on the last day because we don't have next day's open
            if i == n - 1:
                # no next day to exit
                continue

            row = df_clean.loc[dt]
            # print("Curr date ", dt.date())
            df = get_minute_data(self.symbol, dt.date(), dt.date())
            if df.empty:
                print("No data")
                continue
            close = df['Open'].iloc[-1]
            # print("Close ", close)
            # basic data quality checks
            # close = row.get("Close", np.nan)
            if pd.isna(close):
                # skip if missing close
                if verbose:
                    print(f"Skipping {dt.date()}: Close missing")
                continue

            next_dt = dates[i + 1]
            # print("Next dt ", next_dt.date())
            df = get_minute_data(self.symbol, next_dt.date(), next_dt.date())
            if df.empty:
                print("No data")
                continue
            next_open = df['Open'].iloc[1]
            # print("next_open ", next_open)

            # next_row = df_clean.loc[next_dt]
            # next_open = next_row.get("Open", np.nan)
            if pd.isna(next_open):
                # skip trade if next open missing
                if verbose:
                    print(f"Skipping entry on {dt.date()}: next day's Open ({next_dt.date()}) missing")
                continue

            # Position sizing: allocate fraction of current portfolio value
            # allocation_amount = min(400000, portfolio_value * self.allocation)
            allocation_amount = min(400000, portfolio_value * self.allocation)
            # compute entry executed price including slippage (we model slippage as adverse move from executed price)
            executed_entry_price = close * (1.0 + self.slippage_pct)  # buying, so slippage increases price
            # compute integer quantity
            qty = int(allocation_amount // executed_entry_price)
            if qty <= 0:
                if verbose:
                    print(f"Skipping {dt.date()}: insufficient funds to buy 1 unit (alloc {allocation_amount:.2f}, price {executed_entry_price:.2f})")
                continue

            # compute entry cost
            entry_cost = qty * executed_entry_price
            # brokerage on entry
            brokerage_entry = self.brokerage

            # Exit at next day's open with slippage (sell, so slippage reduces exit price)
            executed_exit_price = next_open * (1.0 - self.slippage_pct)
            exit_proceeds = qty * executed_exit_price
            brokerage_exit = self.brokerage

            # raw pnl
            pnl = exit_proceeds - entry_cost

            # commission cost
            total_brokerage = brokerage_entry + brokerage_exit

            # net pnl after brokerage
            net_pnl = pnl - total_brokerage

            # Update portfolio value
            portfolio_value += net_pnl  # portfolio includes realized PnL

            # Prepare trade record
            trade = Trade(
                tradingSymbol=self.symbol,
                tradeState="completed",
                entry_date=pd.Timestamp(dt),
                direction="Long",
                qty=qty,
                entry=float(round(executed_entry_price, 8)),
                Target=math.nan,
                StopLoss=math.nan,
                exit_date=pd.Timestamp(next_dt),
                exit=float(round(executed_exit_price, 8)),
                pnl=float(round(pnl, 8)),
                pnl_pct=float((pnl / entry_cost) * 100.0) if entry_cost != 0 else 0.0,
                brokerage=float(round(total_brokerage, 8)),
                netPnl=float(round(net_pnl, 8)),
                netPnl_pct=float((net_pnl / entry_cost) * 100.0) if entry_cost != 0 else 0.0,
            )

            self.trades.append(trade)

        # After iterating, build equity curve and trades df
        # Append final portfolio value as of last date
        if dates:
            final_date = dates[-1]
            daily_values.append((final_date, portfolio_value))

        self.daily_portfolio_values = pd.Series(
            [v for (_, v) in daily_values],
            index=pd.DatetimeIndex([d for (d, _) in daily_values]),
            name="portfolio_value",
        ).sort_index()

        # equity curve: use daily portfolio values (fwd-filled)
        self.equity_curve = self.daily_portfolio_values

        # Build trades_df
        trades_records = [t.to_dict() for t in self.trades]
        trades_df = pd.DataFrame(trades_records)
        # ensure columns order and names as requested
        expected_cols = [
            "tradingSymbol",
            "tradeState",
            "entry_date",
            "direction",
            "qty",
            "entry",
            "Target",
            "StopLoss",
            "exit_date",
            "exit",
            "pnl",
            "pnl%",
            "brokerage",
            "netPnl",
            "netPnl%",
        ]
        # If trades_df is empty, create empty with expected cols
        if trades_df.empty:
            trades_df = pd.DataFrame(columns=expected_cols)
        else:
            # Ensure column types and ordering
            for c in expected_cols:
                if c not in trades_df.columns:
                    trades_df[c] = np.nan
            trades_df = trades_df[expected_cols]

        # Convert date columns to datetime for safety
        if "entry_date" in trades_df.columns:
            trades_df["entry_date"] = pd.to_datetime(trades_df["entry_date"])
        if "exit_date" in trades_df.columns:
            trades_df["exit_date"] = pd.to_datetime(trades_df["exit_date"])

        # Chronological order by entry_date
        trades_df = trades_df.sort_values(by="entry_date").reset_index(drop=True)

        self.trades_df = trades_df
        # Save closed trades only
        save_path = os.path.join("Results", "SimpleBTST.csv")
        trades_df.to_csv(save_path, index=False)
        if verbose:
            print(f"Backtest complete. Trades executed: {len(self.trades_df)}")
        return self.trades_df

    # ---------------------------
    # Performance metrics
    # ---------------------------
    def compute_performance(self, risk_free_rate: float = 0.0) -> dict:
        """
        Compute requested metrics:
         - Starting Balance, Ending Balance, Total Net PnL, Total Net PnL %, CAGR,
           Max Drawdown (%), Max Drawdown (Amount), Number of Trades, Win Rate (%),
           Average Win (%), Average Loss (%), Average Win Amount, Average Loss Amount,
           Risk/Reward Ratio, Sharpe, Sortino.

        Returns a dict of metrics.
        """
        if self.trades_df is None:
            raise RuntimeError("No trades run. Call run_backtest(...) first.")

        starting_balance = float(self.initial_capital)
        ending_balance = float(self.equity_curve.iloc[-1]) if not self.equity_curve.empty else starting_balance
        total_net_pnl = ending_balance - starting_balance
        total_net_pnl_pct = (total_net_pnl / starting_balance) * 100.0 if starting_balance != 0 else 0.0

        # compute CAGR using overall return across calendar time between first and last equity indices
        if not self.equity_curve.empty:
            days = (self.equity_curve.index[-1] - self.equity_curve.index[0]).days or 1
            cum_return = (ending_balance / starting_balance) - 1.0
            cagr = self._annualize_returns(cum_return, days)
        else:
            cagr = 0.0
            days = 0

        # Max drawdown:
        max_dd_pct, max_dd_amt = self._max_drawdown(self.equity_curve)

        # Trades stats
        trades = self.trades_df.copy()
        num_trades = len(trades)
        wins = trades[trades["netPnl"] > 0]
        losses = trades[trades["netPnl"] <= 0]
        win_rate = (len(wins) / num_trades) * 100.0 if num_trades > 0 else 0.0

        avg_win_pct = wins["pnl%"].mean() if not wins.empty else 0.0
        avg_loss_pct = losses["pnl%"].mean() if not losses.empty else 0.0
        avg_win_amt = wins["netPnl"].mean() if not wins.empty else 0.0
        avg_loss_amt = losses["netPnl"].mean() if not losses.empty else 0.0

        # Risk/Reward ratio (avg win amount / abs(avg loss amount))
        risk_reward_ratio = float("nan")
        if avg_loss_amt != 0.0:
            # note avg_loss_amt is likely negative or zero
            risk_reward_ratio = (avg_win_amt / abs(avg_loss_amt)) if avg_loss_amt != 0 else float("nan")

        # Build daily returns series from equity curve
        if not self.equity_curve.empty and len(self.equity_curve) > 1:
            daily_returns = self.equity_curve.pct_change().dropna()
        else:
            daily_returns = pd.Series(dtype=float)

        sharpe = self._sharpe_ratio(daily_returns, trading_days_per_year=self.trading_days_per_year, risk_free_rate=risk_free_rate)
        sortino = self._sortino_ratio(daily_returns, trading_days_per_year=self.trading_days_per_year, risk_free_rate=risk_free_rate)

        metrics = {
            "Starting Balance": starting_balance,
            "Ending Balance": ending_balance,
            "Total Net PnL": total_net_pnl,
            "Total Net PnL %": total_net_pnl_pct,
            "CAGR": cagr * 100.0,  # present in percent
            "Max Drawdown (%)": max_dd_pct,
            "Max Drawdown (Amount)": max_dd_amt,
            "Number of Trades": num_trades,
            "Win Rate (%)": win_rate,
            "Average Win (%)": avg_win_pct,
            "Average Loss (%)": avg_loss_pct,
            "Average Win Amount": avg_win_amt,
            "Average Loss Amount": avg_loss_amt,
            "Risk/Reward Ratio": risk_reward_ratio,
            "Sharpe Ratio": sharpe,
            "Sortino Ratio": sortino,
        }

        return metrics

    def print_performance(self, risk_free_rate: float = 0.0):
        """
        Print metrics clearly labelled as requested in the output requirements.
        """
        metrics = self.compute_performance(risk_free_rate=risk_free_rate)
        print("\n===== Performance Summary =====")
        print(f"Starting Balance: {metrics['Starting Balance']:.2f}")
        print(f"Ending Balance:   {metrics['Ending Balance']:.2f}")
        print(f"Total Net PnL:    {metrics['Total Net PnL']:.2f}")
        print(f"Total Net PnL %:  {metrics['Total Net PnL %']:.2f}%")
        print(f"CAGR:             {metrics['CAGR']:.2f}%")
        print(f"Max Drawdown (%): {metrics['Max Drawdown (%)']:.2f}%")
        print(f"Max Drawdown ($): {metrics['Max Drawdown (Amount)']:.2f}")
        print(f"Number of Trades: {metrics['Number of Trades']}")
        print(f"Win Rate (%):     {metrics['Win Rate (%)']:.2f}%")
        print(f"Average Win (%):  {metrics['Average Win (%)']:.4f}%")
        print(f"Average Loss (%): {metrics['Average Loss (%)']:.4f}%")
        print(f"Average Win Amount:  {metrics['Average Win Amount']:.4f}")
        print(f"Average Loss Amount: {metrics['Average Loss Amount']:.4f}")
        print(f"Risk/Reward Ratio: {metrics['Risk/Reward Ratio']}")
        print(f"Sharpe Ratio:      {metrics['Sharpe Ratio']:.4f}")
        print(f"Sortino Ratio:     {metrics['Sortino Ratio']:.4f}")
        print("===============================\n")

    def get_equity_curve(self) -> pd.Series:
        return self.equity_curve.copy()


# ---------------------------
# Example usage (replace data loader)
# ---------------------------
if __name__ == "__main__":

    from SymbolMaster import Instruments
    import BrokerConnector
    pd.set_option("display.max_rows", None, "display.max_columns", None)
    # Login into broker
    BrokerConnector.login()
    # Download Symbol Master from Broker (Master Contracts)
    Instruments.fetch_instruments()

    # start_date = date(2024, 1, 15)
    # end_date = date(2024, 10, 14)

    start_date = date(2020, 7, 1)
    end_date = date(2025, 6, 30)

    engine = BacktestEngine("NIFTYBEES", start_date, end_date)
    engine.run_backtest(get_historical_data)
    engine.print_performance()