profitable-expert-advisor/frontline/units/RSIScalpingAPPL/run_backtest.py

"""
RSIScalpingAPPL — bar backtest mirroring main.mq5 inputs.

Outputs in this folder:
  backtest_report.json, trades.csv, report.png,
  equity_curve.png, drawdown.png, monthly_returns.png,
  pnl_distribution.png, exit_reasons.png

Usage:
  python run_backtest.py
  python run_backtest.py --start 2021-01-01 --end 2026-01-01
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import MetaTrader5 as mt5
import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(ROOT / "backtesting" / "MT5"))

from cluster_audit.backtest_core import (  # noqa: E402
    BacktestReport,
    CostModel,
    load_bars,
    resolve_symbol,
    run_single_position,
)
from indicator_utils import calculate_adx, calculate_atr, calculate_dmi, calculate_ema, calculate_rsi  # noqa: E402

STRATEGY_ID = "RSIScalpingAPPL"


def save_reports(report: BacktestReport, out_dir: Path) -> None:
  rows = [
    {
      "side": t.side,
      "open_time": t.open_time,
      "close_time": t.close_time,
      "open_price": t.open_price,
      "close_price": t.close_price,
      "volume": t.volume,
      "profit": t.profit,
      "bars_held": t.bars_held,
      "exit_reason": t.exit_reason,
    }
    for t in report.trades_list
  ]
  pd.DataFrame(rows).to_csv(out_dir / "trades.csv", index=False)
  with open(out_dir / "backtest_report.json", "w", encoding="utf-8") as f:
    json.dump(report.to_dict(), f, indent=2, ensure_ascii=False)

  trades = report.trades_list
  if not trades:
    fig, ax = plt.subplots(figsize=(10, 4))
    ax.text(0.5, 0.5, "No trades in backtest window", ha="center", va="center", fontsize=14)
    ax.axis("off")
    fig.savefig(out_dir / "report.png", dpi=200, bbox_inches="tight")
    plt.close(fig)
    return

  df = pd.DataFrame(rows)
  df["close_time"] = pd.to_datetime(df["close_time"])
  df = df.sort_values("close_time")
  bal0 = report.params.get("initial_balance", 10_000.0)
  if report.equity_curve is not None and len(report.equity_curve) > 1:
    eq = report.equity_curve
  else:
    eq = pd.Series(bal0 + df["profit"].cumsum().values, index=df["close_time"])
  equity_times = eq.index
  equity = eq

  fig = plt.figure(figsize=(14, 10))
  gs = fig.add_gridspec(3, 2, height_ratios=[2, 1.2, 1.2])
  ax1 = fig.add_subplot(gs[0, :])
  ax1.plot(equity_times, equity, lw=1.8)
  ax1.axhline(bal0, color="gray", ls="--")
  ax1.set_title("Equity Curve")
  ax1.grid(alpha=0.3)
  ax2 = fig.add_subplot(gs[1, 0])
  dd = (equity - equity.cummax()) / equity.cummax() * 100
  ax2.fill_between(equity_times, dd, 0, color="#d62728", alpha=0.35)
  ax2.set_title("Drawdown %")
  ax2.grid(alpha=0.3)
  ax3 = fig.add_subplot(gs[1, 1])
  df["month"] = df["close_time"].dt.to_period("M")
  monthly = df.groupby("month")["profit"].sum()
  ax3.bar(range(len(monthly)), monthly.values, color=["#2ca02c" if v >= 0 else "#d62728" for v in monthly])
  ax3.set_title("Monthly PnL")
  ax3.axhline(0, color="black", lw=0.6)
  ax4 = fig.add_subplot(gs[2, 0])
  ax4.hist(df["profit"], bins=30, color="#9467bd", alpha=0.85)
  ax4.axvline(0, color="black")
  ax4.set_title("Trade PnL Distribution")
  ax5 = fig.add_subplot(gs[2, 1])
  rc = df["exit_reason"].value_counts()
  ax5.bar(rc.index.astype(str), rc.values, color="#ff7f0e")
  ax5.set_title("Exit Reasons")
  fig.suptitle(
    f"{STRATEGY_ID} — Net ${report.net_profit:,.2f} | Trades {report.total_trades} | "
    f"WR {report.win_rate:.1f}% | PF {report.profit_factor:.2f} | MaxDD {report.max_drawdown_pct:.2f}%",
    fontsize=11,
  )
  fig.tight_layout(rect=[0, 0, 1, 0.96])
  fig.savefig(out_dir / "report.png", dpi=200, bbox_inches="tight")
  plt.close(fig)

  plt.figure(figsize=(12, 5))
  plt.plot(equity_times, equity, lw=2)
  plt.title("Equity Curve")
  plt.grid(alpha=0.3)
  plt.tight_layout()
  plt.savefig(out_dir / "equity_curve.png", dpi=200, bbox_inches="tight")
  plt.close()

  plt.figure(figsize=(12, 5))
  plt.fill_between(equity_times, dd, 0, color="red", alpha=0.3)
  plt.plot(equity_times, dd, color="darkred")
  plt.title("Drawdown %")
  plt.grid(alpha=0.3)
  plt.tight_layout()
  plt.savefig(out_dir / "drawdown.png", dpi=200, bbox_inches="tight")
  plt.close()

  plt.figure(figsize=(12, 5))
  plt.bar(range(len(monthly)), monthly.values, color=["green" if v >= 0 else "red" for v in monthly], alpha=0.75)
  plt.title("Monthly PnL")
  plt.axhline(0, color="black")
  plt.grid(alpha=0.3, axis="y")
  plt.tight_layout()
  plt.savefig(out_dir / "monthly_returns.png", dpi=200, bbox_inches="tight")
  plt.close()

  plt.figure(figsize=(10, 5))
  plt.hist(df["profit"], bins=40, color="#6a5acd", alpha=0.85)
  plt.axvline(0, color="black")
  plt.title("Per-Trade PnL Distribution")
  plt.tight_layout()
  plt.savefig(out_dir / "pnl_distribution.png", dpi=200, bbox_inches="tight")
  plt.close()

  if report.exit_reason_breakdown:
    labels = list(report.exit_reason_breakdown.keys())
    counts = [report.exit_reason_breakdown[k]["count"] for k in labels]
    plt.figure(figsize=(8, 5))
    plt.bar(labels, counts, color="#e377c2")
    plt.title("Exit Reason Counts")
    plt.tight_layout()
    plt.savefig(out_dir / "exit_reasons.png", dpi=200, bbox_inches="tight")
    plt.close()


@dataclass
class StrategyParams:
  rsi_period: int = 14
  rsi_overbought: float = 80
  rsi_oversold: float = 78
  rsi_target_buy: float = 94
  rsi_target_sell: float = 44
  bars_to_wait: int = 7
  lot_size: float = 25
  use_reversal_escape: bool = False
  reversal_atr_period: int = 14
  reversal_adverse_atr_mult: float = 1.5
  reversal_signs_required: int = 2
  reversal_rsi_velocity: float = 8.0
  initial_balance: float = 10_000.0

  def to_dict(self) -> dict:
    return asdict(self)


def make_params(balance: float) -> StrategyParams:
  return StrategyParams(initial_balance=balance)



def run_backtest(df, symbol, params: StrategyParams, costs, period_label):
  info = mt5.symbol_info(symbol)
  point = float(info.point) if info else 0.01
  rsi = calculate_rsi(df["close"], params.rsi_period).to_numpy()
  atr = calculate_atr(df, params.reversal_atr_period).to_numpy()
  p = params.to_dict()

  def on_bar(i, st, open_pos, close):
    if i < 3 or np.isnan(rsi[i - 1]):
      return
    sig, prev, two = rsi[i - 1], rsi[i - 2], rsi[i - 3]
    mid = float(df["open"].iloc[i])
    hi, lo = float(df["high"].iloc[i]), float(df["low"].iloc[i])

    if st.side and params.use_reversal_escape:
      a = float(atr[i - 1]) if not np.isnan(atr[i - 1]) else 0.0
      if a > 0:
        signs = 0
        if st.side == "BUY":
          if st.entry - lo >= params.reversal_adverse_atr_mult * a:
            signs += 1
          if sig - prev >= params.reversal_rsi_velocity:
            signs += 1
        else:
          if hi - st.entry >= params.reversal_adverse_atr_mult * a:
            signs += 1
          if prev - sig >= params.reversal_rsi_velocity:
            signs += 1
        if signs >= params.reversal_signs_required:
          close(i, mid, "reversal_escape")
          return

    if st.side == "BUY":
      if sig < params.rsi_oversold:
        st.bars_against = st.bars_against + 1 if st.rsi_against else 1
        st.rsi_against = True
        if st.bars_against >= params.bars_to_wait:
          close(i, mid, "rsi_against")
      else:
        st.rsi_against = False
        st.bars_against = 0
        if sig >= params.rsi_target_buy:
          close(i, mid, "target")
    elif st.side == "SELL":
      if sig > params.rsi_overbought:
        st.bars_against = st.bars_against + 1 if st.rsi_against else 1
        st.rsi_against = True
        if st.bars_against >= params.bars_to_wait:
          close(i, mid, "rsi_against")
      else:
        st.rsi_against = False
        st.bars_against = 0
        if sig <= params.rsi_target_sell:
          close(i, mid, "target")
    else:
      if two <= params.rsi_oversold and prev > params.rsi_oversold:
        open_pos(i, "BUY", mid)
      elif two >= params.rsi_overbought and prev < params.rsi_overbought:
        open_pos(i, "SELL", mid)

  return run_single_position(
    df, symbol, point, costs, params.lot_size, STRATEGY_ID, "M10", period_label, p, params.initial_balance, on_bar
  )



def parse_args() -> argparse.Namespace:
  p = argparse.ArgumentParser(description=f"{STRATEGY_ID} Python backtest")
  p.add_argument("--symbol", default="AAPL")
  p.add_argument("--start", default="2021-01-01")
  p.add_argument("--end", default="2026-01-01")
  p.add_argument("--balance", type=float, default=10_000.0)
  return p.parse_args()


def main() -> None:
  args = parse_args()
  out_dir = Path(__file__).resolve().parent
  params = make_params(args.balance)
  if not mt5.initialize():
    raise SystemExit("MetaTrader5 initialize() failed")
  try:
    symbol = resolve_symbol(args.symbol)
    start = datetime.fromisoformat(args.start)
    end = datetime.fromisoformat(args.end)
    period_label = f"{args.start}_{args.end}"
    print(f"Loading {symbol} bars ...")
    df = load_bars(symbol, mt5.TIMEFRAME_M10, start, end)
    costs = CostModel.for_symbol(symbol)
    report = run_backtest(df, symbol, params, costs, period_label)
    save_reports(report, out_dir)
    print(f"Net: ${report.net_profit:,.2f} | Trades: {report.total_trades} | WR: {report.win_rate:.1f}% | PF: {report.profit_factor:.2f}")
    print(f"Saved to {out_dir}")
  finally:
    mt5.shutdown()


if __name__ == "__main__":
  main()