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

"""
SimpleTrendlineBTCUSD — 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,
    Trade,
    build_report,
    calc_profit,
    fill_price,
    load_bars,
    resolve_symbol,
)
from indicator_utils import calculate_adx, calculate_atr, calculate_dmi, calculate_ema, calculate_rsi  # noqa: E402

STRATEGY_ID = "SimpleTrendlineBTCUSD"


@dataclass
class SimState:
  side: str | None = None
  entry: float = 0.0
  entry_i: int = 0
  entry_time: object = None
  sl: float = 0.0
  tp: float = 0.0
  bars_against: int = 0
  rsi_against: bool = False


def run_single_position(
  df: pd.DataFrame,
  symbol: str,
  point: float,
  costs: CostModel,
  lot: float,
  tf_label: str,
  period_label: str,
  params: dict,
  initial_balance: float,
  on_bar,
) -> BacktestReport:
  trades: list[Trade] = []
  equity = [initial_balance]
  st = SimState()

  def close(i: int, mid: float, reason: str) -> None:
    nonlocal st
    if st.side is None:
      return
    exit_px = fill_price(mid, point, costs, st.side, entry=False)
    commission = costs.commission_per_lot * lot * 2.0
    profit = calc_profit(symbol, st.side, lot, st.entry, exit_px) - commission
    trades.append(
      Trade(
        side=st.side,
        open_time=st.entry_time,
        close_time=df.index[i],
        open_price=st.entry,
        close_price=exit_px,
        volume=lot,
        profit=profit,
        bars_held=i - st.entry_i,
        exit_reason=reason,
      )
    )
    equity.append(equity[-1] + profit)
    st = SimState()

  def open_pos(i: int, side: str, mid: float) -> None:
    nonlocal st
    st.side = side
    st.entry = fill_price(mid, point, costs, side, entry=True)
    st.entry_i = i
    st.entry_time = df.index[i]

  for i in range(1, len(df)):
    on_bar(i, st, open_pos, close)
    if len(equity) == len(trades) + 1:
      equity.append(equity[-1])

  if st.side is not None:
    close(len(df) - 1, float(df["close"].iloc[-1]), "eod")

  eq = pd.Series(equity[: len(df)], index=df.index[: len(equity)])
  return build_report(STRATEGY_ID, symbol, tf_label, period_label, trades, eq, initial_balance, params)


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)
  equity = bal0 + df["profit"].cumsum()

  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(df["close_time"], 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(df["close_time"], 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(df["close_time"], 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(df["close_time"], dd, 0, color="red", alpha=0.3)
  plt.plot(df["close_time"], 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:
  ma_period: int = 150
  touch_tolerance_pts: float = 170
  break_buffer_pts: float = 90
  lot_size: float = 0.10
  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
  hdf = df.resample("4h").agg({"open": "first", "high": "max", "low": "min", "close": "last"}).dropna()
  ma = calculate_ema(hdf["close"], params.ma_period).to_numpy()
  htimes = hdf.index.to_numpy()
  hcloses = hdf["close"].to_numpy()
  touch_tol = params.touch_tolerance_pts * point
  break_buf = params.break_buffer_pts * point
  p = params.to_dict()

  def on_bar(i, st, open_pos, close):
    if i < 5:
      return
    ts = df.index[i]
    hidx = int(np.searchsorted(htimes, ts, side="right")) - 1
    if hidx < params.ma_period + 5:
      return
    crosses_t, crosses_p = [], []
    for j in range(hidx, params.ma_period + 2, -1):
      if j >= len(ma) - 1:
        continue
      d0, d1 = hcloses[j] - ma[j], hcloses[j + 1] - ma[j + 1]
      if d0 == 0 or d1 == 0 or d0 * d1 < 0:
        crosses_t.append(htimes[j])
        crosses_p.append(hcloses[j])
        if len(crosses_t) >= 3:
          break
    if len(crosses_t) < 3:
      return
    t0 = crosses_t[2]

    def _secs(delta) -> float:
      if hasattr(delta, "total_seconds"):
        return float(delta.total_seconds())
      return float(delta.astype("timedelta64[s]").astype(float))

    xs = np.array([_secs(t - t0) for t in crosses_t[::-1]])
    ys = np.array(crosses_p[::-1])
    den = 3 * np.sum(xs ** 2) - np.sum(xs) ** 2
    if abs(den) < 1e-10:
      return
    a = (3 * np.sum(xs * ys) - np.sum(xs) * np.sum(ys)) / den
    b = (np.sum(ys) - a * np.sum(xs)) / 3
    t1, t2 = df.index[i - 1], df.index[i - 2]
    line1 = a * _secs(t1 - t0) + b
    line2 = a * _secs(t2 - t0) + b
    mid = float(df["open"].iloc[i])
    hi = float(df["high"].iloc[i - 1])
    lo = float(df["low"].iloc[i - 1])
    cl1 = float(df["close"].iloc[i - 1])
    op1 = float(df["open"].iloc[i - 1])
    cl2 = float(df["close"].iloc[i - 2])
    if st.side == "BUY" and cl1 < line1 - break_buf:
      close(i, mid, "break")
      return
    if st.side == "SELL" and cl1 > line1 + break_buf:
      close(i, mid, "break")
      return
    if st.side:
      return
    if a > 0 and lo <= line1 + touch_tol and cl1 > line1 and cl1 > op1 and cl2 >= line2 - touch_tol:
      open_pos(i, "BUY", mid)
    elif a < 0 and hi >= line1 - touch_tol and cl1 < line1 and cl1 < op1 and cl2 <= line2 + touch_tol:
      open_pos(i, "SELL", mid)

  return run_single_position(df, symbol, point, costs, params.lot_size, "H1", 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="BTCUSD")
  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_H1, 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()