Testing the 20/50 SMA Crossover Algo Strategy on Nifty 50

pandas>=2.0
  numpy>=1.24

"""
  Dailybulls backtest package initializer.
  dailybulls.in keeps this module intentionally lightweight.
  """
  __all__ = [
      "config",
      "data",
      "metrics",
      "strategy",
  ]

from __future__ import annotations

  from dataclasses import dataclass
  from pathlib import Path
  from typing import Optional


  @dataclass(slots=True)
  class StrategyConfig:
      """Configuration object shared across the Dailybulls SMA modules."""

      data_path: Path
      date_column: str = "date"
      open_column: str = "open"
      high_column: str = "high"
      low_column: str = "low"
      close_column: str = "close"
      volume_column: Optional[str] = None

      short_window: int = 20
      long_window: int = 50
      atr_window: int = 14

      risk_fraction: float = 0.01  # Used by fixed R:R variant
      atr_multiple: float = 1.5    # Used by ATR trailing variant

      def __post_init__(self) -> None:
          self.data_path = Path(self.data_path).expanduser().resolve()
          if self.short_window >= self.long_window:
              raise ValueError("short_window must be strictly less than long_window.")
          if self.risk_fraction <= 0:
              raise ValueError("risk_fraction must be positive.")
          if self.atr_multiple <= 0:
              raise ValueError("atr_multiple must be positive.")

from __future__ import annotations

  from typing import Final

  import pandas as pd

  from .config import StrategyConfig

  REQUIRED_COLUMNS: Final[tuple[str, ...]] = ("date", "open", "high", "low", "close")


  def load_price_data(config: StrategyConfig) -> pd.DataFrame:
      """Load and normalise OHLC data for the Dailybulls SMA engine."""

      df = pd.read_csv(config.data_path)

      missing = [
          c for c in (
              config.date_column,
              config.open_column,
              config.high_column,
              config.low_column,
              config.close_column,
          ) if c not in df.columns
      ]
      if missing:
          raise ValueError(f"Input file is missing required columns: {missing}")

      rename_map = {
          config.date_column: "date",
          config.open_column: "open",
          config.high_column: "high",
          config.low_column: "low",
          config.close_column: "close",
      }
      df = df.rename(columns=rename_map)

      columns = list(REQUIRED_COLUMNS)
      if config.volume_column and config.volume_column in df.columns:
          columns.append(config.volume_column)
      df = df[columns]

      df["date"] = pd.to_datetime(df["date"], utc=False)
      df = df.sort_values("date", ascending=True).dropna(subset=["open", "high", "low", "close"])

      for col in ("open", "high", "low", "close"):
          df[col] = pd.to_numeric(df[col], errors="coerce")

      df = df.dropna(subset=["open", "high", "low", "close"]).reset_index(drop=True)

      duplicates = df.duplicated(subset="date").sum()
      if duplicates:
          raise ValueError(
              f"Found {duplicates} duplicate timestamps in the input data. Please clean or aggregate the dataset."
          )

      return df

from __future__ import annotations

  from dataclasses import dataclass
  from typing import List, Optional

  import numpy as np
  import pandas as pd

  from .config import StrategyConfig


  @dataclass(slots=True)
  class Trade:
      signal_date: pd.Timestamp
      entry_date: pd.Timestamp
      entry_price: float
      exit_date: pd.Timestamp
      exit_price: float
      initial_stop_price: float
      stop_price: float
      target_price: Optional[float]
      exit_reason: str

      @property
      def pnl(self) -> float:
          return float(self.exit_price - self.entry_price)

      @property
      def return_pct(self) -> float:
          return float((self.exit_price / self.entry_price) - 1.0)

      @property
      def risk_amount(self) -> float:
          return float(self.entry_price - self.initial_stop_price)

      @property
      def r_multiple(self) -> float:
          risk = self.risk_amount
          if risk <= 0:
              return np.nan
          return self.pnl / risk

      @property
      def holding_days(self) -> int:
          return int((self.exit_date - self.entry_date).days)


  def _prepare_indicators(frame: pd.DataFrame, config: StrategyConfig) -> pd.DataFrame:
      data = frame.copy()
      data["sma_short"] = data["close"].rolling(config.short_window, min_periods=config.short_window).mean()
      data["sma_long"] = data["close"].rolling(config.long_window, min_periods=config.long_window).mean()

      prev_close = data["close"].shift(1)
      tr = pd.concat(
          [
              data["high"] - data["low"],
              (data["high"] - prev_close).abs(),
              (data["low"] - prev_close).abs(),
          ],
          axis=1,
      )
      data["true_range"] = tr.max(axis=1)
      data["atr"] = data["true_range"].rolling(config.atr_window, min_periods=config.atr_window).mean()
      return data.reset_index(drop=True)


  def run_fixed_sma_strategy(data: pd.DataFrame, config: StrategyConfig) -> List[Trade]:
      priced = _prepare_indicators(data, config)
      trades: List[Trade] = []
      in_position = False
      entry_price = initial_stop = target_price = 0.0
      signal_index: Optional[int] = None
      entry_index: Optional[int] = None

      for idx in range(1, len(priced) - 1):
          today = priced.iloc[idx]
          yesterday = priced.iloc[idx - 1]

          if not in_position:
              if np.isnan(today["sma_short"]) or np.isnan(today["sma_long"]):
                  continue

              crossed = yesterday["sma_short"] <= yesterday["sma_long"] and today["sma_short"] > today["sma_long"]
              if crossed:
                  next_bar = priced.iloc[idx + 1]
                  entry_price = float(next_bar["open"])
                  risk_distance = entry_price * config.risk_fraction
                  initial_stop = entry_price - risk_distance
                  if initial_stop <= 0:
                      continue
                  target_price = entry_price + risk_distance * 2.0

                  signal_index = idx
                  entry_index = idx + 1
                  in_position = True
          else:
              assert entry_index is not None and signal_index is not None
              next_bar = priced.iloc[idx + 1]

              exit_reason = None
              open_price = float(next_bar["open"])
              high_price = float(next_bar["high"])
              low_price = float(next_bar["low"])
              close_price = float(next_bar["close"])

              if open_price <= initial_stop:
                  exit_price = open_price
                  exit_reason = "stop_gap"
              elif open_price >= target_price:
                  exit_price = open_price
                  exit_reason = "target_gap"
              elif low_price <= initial_stop and high_price >= target_price:
                  exit_price = initial_stop
                  exit_reason = "stop_hit"
              elif low_price <= initial_stop:
                  exit_price = initial_stop
                  exit_reason = "stop_hit"
              elif high_price >= target_price:
                  exit_price = target_price
                  exit_reason = "target_hit"
              elif priced.iloc[idx]["sma_short"] <= priced.iloc[idx]["sma_long"]:
                  exit_price = close_price
                  exit_reason = "crossover_exit"
              else:
                  exit_price = None

              if exit_reason:
                  trades.append(
                      Trade(
                          signal_date=priced.iloc[signal_index]["date"],
                          entry_date=priced.iloc[entry_index]["date"],
                          entry_price=entry_price,
                          exit_date=next_bar["date"],
                          exit_price=float(exit_price),
                          initial_stop_price=initial_stop,
                          stop_price=initial_stop,
                          target_price=target_price,
                          exit_reason=exit_reason,
                      )
                  )
                  in_position = False
                  entry_index = None
                  signal_index = None

      if in_position and entry_index is not None and signal_index is not None:
          last_bar = priced.iloc[-1]
          trades.append(
              Trade(
                  signal_date=priced.iloc[signal_index]["date"],
                  entry_date=priced.iloc[entry_index]["date"],
                  entry_price=entry_price,
                  exit_date=last_bar["date"],
                  exit_price=float(last_bar["close"]),
                  initial_stop_price=initial_stop,
                  stop_price=initial_stop,
                  target_price=target_price,
                  exit_reason="end_of_data",
              )
          )
      return trades


  def run_atr_trailing_strategy(data: pd.DataFrame, config: StrategyConfig) -> List[Trade]:
      priced = _prepare_indicators(data, config)
      trades: List[Trade] = []
      in_position = False
      entry_price = initial_stop = trailing_stop = 0.0
      signal_index: Optional[int] = None
      entry_index: Optional[int] = None

      for idx in range(1, len(priced) - 1):
          today = priced.iloc[idx]
          yesterday = priced.iloc[idx - 1]

          if not in_position:
              if np.isnan(today["sma_short"]) or np.isnan(today["sma_long"]):
                  continue

              crossed = yesterday["sma_short"] <= yesterday["sma_long"] and today["sma_short"] > today["sma_long"]
              if crossed:
                  next_bar = priced.iloc[idx + 1]
                  atr_value = float(next_bar["atr"])
                  if np.isnan(atr_value) or atr_value <= 0:
                      continue
                  entry_price = float(next_bar["open"])
                  initial_stop = entry_price - config.atr_multiple * atr_value
                  if initial_stop <= 0:
                      continue
                  trailing_stop = initial_stop
                  signal_index = idx
                  entry_index = idx + 1
                  in_position = True
          else:
              assert entry_index is not None and signal_index is not None
              next_bar = priced.iloc[idx + 1]

              atr_value = float(next_bar["atr"])
              if not np.isnan(atr_value) and atr_value > 0:
                  candidate_stop = float(next_bar["close"]) - config.atr_multiple * atr_value
                  trailing_stop = max(trailing_stop, candidate_stop)

              exit_price = None
              exit_reason = None
              open_price = float(next_bar["open"])
              low_price = float(next_bar["low"])
              close_price = float(next_bar["close"])

              if open_price <= trailing_stop:
                  exit_price = open_price
                  exit_reason = "stop_gap"
              elif low_price <= trailing_stop:
                  exit_price = trailing_stop
                  exit_reason = "stop_hit"
              elif priced.iloc[idx]["sma_short"] <= priced.iloc[idx]["sma_long"]:
                  exit_price = close_price
                  exit_reason = "crossover_exit"

              if exit_reason:
                  trades.append(
                      Trade(
                          signal_date=priced.iloc[signal_index]["date"],
                          entry_date=priced.iloc[entry_index]["date"],
                          entry_price=entry_price,
                          exit_date=next_bar["date"],
                          exit_price=float(exit_price),
                          initial_stop_price=initial_stop,
                          stop_price=trailing_stop,
                          target_price=None,
                          exit_reason=exit_reason,
                      )
                  )
                  in_position = False
                  entry_index = None
                  signal_index = None

      if in_position and entry_index is not None and signal_index is not None:
          last_bar = priced.iloc[-1]
          trades.append(
              Trade(
                  signal_date=priced.iloc[signal_index]["date"],
                  entry_date=priced.iloc[entry_index]["date"],
                  entry_price=entry_price,
                  exit_date=last_bar["date"],
                  exit_price=float(last_bar["close"]),
                  initial_stop_price=initial_stop,
                  stop_price=trailing_stop,
                  target_price=None,
                  exit_reason="end_of_data",
              )
          )
      return trades
  

from __future__ import annotations

  from dataclasses import dataclass
  from typing import Iterable

  import numpy as np
  import pandas as pd

  from .strategy import Trade


  @dataclass(slots=True)
  class Summary:
      trade_count: int
      wins: int
      losses: int
      win_rate: float
      expectancy_r: float
      total_return_pct: float
      compounded_return_pct: float
      max_drawdown_pct: float


  def trades_to_frame(trades: Iterable[Trade]) -> pd.DataFrame:
      rows = []
      for trade in trades:
          rows.append(
              {
                  "signal_date": trade.signal_date,
                  "entry_date": trade.entry_date,
                  "entry_price": trade.entry_price,
                  "exit_date": trade.exit_date,
                  "exit_price": trade.exit_price,
                  "initial_stop_price": trade.initial_stop_price,
                  "stop_price": trade.stop_price,
                  "target_price": trade.target_price,
                  "exit_reason": trade.exit_reason,
                  "pnl": trade.pnl,
                  "return_pct": trade.return_pct,
                  "r_multiple": trade.r_multiple,
                  "holding_days": trade.holding_days,
              }
          )
      frame = pd.DataFrame(rows)
      if not frame.empty:
          frame = frame.sort_values("exit_date").reset_index(drop=True)
      return frame


  def summarise_trades(trades: Iterable[Trade]) -> Summary:
      frame = trades_to_frame(trades)
      if frame.empty:
          return Summary(0, 0, 0, 0.0, 0.0, 0.0, 0.0, 0.0)

      wins = int((frame["pnl"] > 0).sum())
      losses = int((frame["pnl"] <= 0).sum())
      win_rate = wins / len(frame)

      r_values = frame["r_multiple"].replace([np.inf, -np.inf], np.nan).dropna()
      expectancy = float(r_values.mean()) if not r_values.empty else 0.0

      total_return_pct = float(frame["return_pct"].sum())
      compounded_return_pct = float(np.prod(1 + frame["return_pct"]) - 1)

      equity = (1 + frame["return_pct"]).cumprod()
      peak = equity.cummax()
      drawdown = (equity - peak) / peak
      max_dd = float(drawdown.min())

      return Summary(
          trade_count=len(frame),
          wins=wins,
          losses=losses,
          win_rate=win_rate,
          expectancy_r=expectancy,
          total_return_pct=total_return_pct,
          compounded_return_pct=compounded_return_pct,
          max_drawdown_pct=max_dd,
      )


  def equity_curve(trades: Iterable[Trade]) -> pd.DataFrame:
      frame = trades_to_frame(trades)
      if frame.empty:
          return pd.DataFrame(columns=["timestamp", "equity", "trade_index"])

      equity = (1 + frame["return_pct"]).cumprod()
      return pd.DataFrame(
          {
              "timestamp": frame["exit_date"].values,
              "equity": equity.values,
              "trade_index": np.arange(1, len(frame) + 1),
          }
      )

from __future__ import annotations

  import argparse
  from pathlib import Path

  import pandas as pd

  from backtest.config import StrategyConfig
  from backtest.data import load_price_data
  from backtest.metrics import equity_curve, summarise_trades, trades_to_frame
  from backtest.strategy import run_atr_trailing_strategy, run_fixed_sma_strategy


  def parse_args() -> argparse.Namespace:
      parser = argparse.ArgumentParser(
          description="Dailybulls 20/50 SMA strategy runner (dailybulls.in)."
      )
      parser.add_argument(
          "--data-path",
          type=Path,
          required=True,
          help="Path to CSV containing OHLC data (date, open, high, low, close).",
      )
      parser.add_argument(
          "--mode",
          choices=("fixed", "atr"),
          default="fixed",
          help="Strategy variant: fixed 1:2 risk/reward or ATR trailing.",
      )
      parser.add_argument(
          "--short-window",
          type=int,
          default=20,
          help="Lookback for the short SMA (default: 20).",
      )
      parser.add_argument(
          "--long-window",
          type=int,
          default=50,
          help="Lookback for the long SMA (default: 50).",
      )
      parser.add_argument(
          "--atr-window",
          type=int,
          default=14,
          help="ATR lookback for the trailing stop variant (default: 14).",
      )
      parser.add_argument(
          "--risk-fraction",
          type=float,
          default=0.01,
          help="Risk fraction for the fixed strategy (default: 1%).",
      )
      parser.add_argument(
          "--atr-multiple",
          type=float,
          default=1.5,
          help="ATR multiple for the trailing stop (default: 1.5).",
      )
      parser.add_argument(
          "--export-trades",
          type=Path,
          help="Optional path to export the trade ledger CSV.",
      )
      parser.add_argument(
          "--export-equity",
          type=Path,
          help="Optional path to export the equity curve CSV.",
      )
      return parser.parse_args()


  def main() -> None:
      args = parse_args()
      config = StrategyConfig(
          data_path=args.data_path,
          short_window=args.short_window,
          long_window=args.long_window,
          atr_window=args.atr_window,
          risk_fraction=args.risk_fraction,
          atr_multiple=args.atr_multiple,
      )

      price_data = load_price_data(config)
      if price_data.empty:
          raise RuntimeError("No rows found after loading the CSV. Please verify the dataset.")

      if args.mode == "atr":
          trades = run_atr_trailing_strategy(price_data, config)
      else:
          trades = run_fixed_sma_strategy(price_data, config)

      if not trades:
          print("No trades were generated. Consider adjusting the lookback windows or checking the dataset.")
          return

      summary = summarise_trades(trades)
      ledger = trades_to_frame(trades)
      equity = equity_curve(trades)

      print("\n=== Dailybulls 20/50 SMA Strategy Summary ===")
      print(f"Total trades      : {summary.trade_count}")
      print(f"Wins / Losses     : {summary.wins} / {summary.losses}")
      print(f"Win rate          : {summary.win_rate:.1%}")
      print(f"Expectancy (R)    : {summary.expectancy_r:.2f}")
      print(f"Total return      : {summary.total_return_pct:.2%}")
      print(f"Compounded return : {summary.compounded_return_pct:.2%}")
      print(f"Max drawdown      : {summary.max_drawdown_pct:.2%}")
      print(" =============================================\n")

      print("Last five trades:")
      print(ledger.tail(5).to_string(index=False))

      if args.export_trades:
          ledger.to_csv(args.export_trades, index=False)
          print(f"\nTrade ledger exported to: {args.export_trades.resolve()}")

      if args.export_equity:
          equity.to_csv(args.export_equity, index=False)
          print(f"Equity curve exported to: {args.export_equity.resolve()}")


  if __name__ == "__main__":
      main()