Important This algorithm leverages Pandas ewm (Exponential Weighted Functions) to calculate the MACD and EMA dynamically. It automatically aggregates your base 1-minute data into a 5-minute timeframe for smoother signal generation.

Disclaimer These results are generated based on automated backtesting performed using Python code and algorithms. Actual results may vary, and manual backtesting outcomes could differ due to varying assumptions, data interpretation, and market conditions. This information is provided for educational and informational purposes only and should not be considered as financial advice. Before making any trading or investment decisions, please consult with a qualified financial advisor or professional.

import pandas as pd
import numpy as np
import datetime as dt

# ===================== CONFIG =====================
FILE_PATH = "Index_1_minute.csv"  
DAYFIRST = True                    
EXIT_TIME = dt.time(15, 15)        
SLIPPAGE = 0.5                     

# Strategy params
TIMEFRAME = "5min"   
FAST_EMA = 9
SLOW_EMA = 21
MACD_FAST = 12
MACD_SLOW = 26
MACD_SIGNAL = 9
# ==================================================

# ----------------- LOAD & PREP --------------------
print("Loading and preparing data...")
df = pd.read_csv(FILE_PATH)
df.columns = [c.strip().lower() for c in df.columns]
required = {"date","open","high","low","close"}
missing = required - set(df.columns)
if missing:
    raise ValueError(f"CSV missing columns: {missing}. Need exactly {sorted(required)}")

df["dt"] = pd.to_datetime(df["date"], dayfirst=DAYFIRST)
df = df.sort_values("dt").reset_index(drop=True)

for c in ["open","high","low","close"]:
    df[c] = pd.to_numeric(df[c], errors="coerce")
df = df.dropna(subset=["open","high","low","close","dt"]).copy()


df = df.set_index("dt").resample(TIMEFRAME).agg({
    "open": "first",
    "high": "max",
    "low": "min",
    "close": "last"
}).dropna().reset_index()

df["d"] = df["dt"].dt.date
df["t"] = df["dt"].dt.time

print("Calculating EMA and MACD indicators...")

df["ema_fast"] = df["close"].ewm(span=FAST_EMA, adjust=False).mean()
df["ema_slow"] = df["close"].ewm(span=SLOW_EMA, adjust=False).mean()


macd_fast_line = df["close"].ewm(span=MACD_FAST, adjust=False).mean()
macd_slow_line = df["close"].ewm(span=MACD_SLOW, adjust=False).mean()
df["macd_line"] = macd_fast_line - macd_slow_line
df["macd_signal"] = df["macd_line"].ewm(span=MACD_SIGNAL, adjust=False).mean()
df["macd_hist"] = df["macd_line"] - df["macd_signal"]

# -------------- UTILITIES ----------------
def stats(trades):
    if not trades:
        return {"trades":0,"wins":0,"win_rate":0.0,"avg_pnl":0.0,"total_pnl":0.0}
    pnl = np.array([x["pnl"] for x in trades])
    wins = (pnl > 0).sum()
    return {
        "trades": len(trades),
        "wins": int(wins),
        "win_rate": round(100*wins/len(trades), 2),
        "avg_pnl": round(pnl.mean(), 2),
        "total_pnl": round(pnl.sum(), 2),
    }

def close_out(direction, entry, row_close):
    return (row_close - entry) if direction=="long" else (entry - row_close)

# -------------- STRATEGY: EMA + MACD --------------
def backtest_ema_macd(df):
    print("Running backtest loop...")
    trades = []
    in_trade = False
    direction = None
    entry = None
    entry_time = None

    for _, r in df.iterrows():
        # 1. INTRADAY TIME EXIT
        if r["t"] >= EXIT_TIME:
            if in_trade:
                pnl = close_out(direction, entry, r["close"]) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"EMA_MACD", "dir":direction, "entry":entry,
                    "exit":r["close"], "pnl":pnl, "outcome":"TIME_EXIT"
                })
                in_trade = False
            continue

        if pd.isna(r["ema_fast"]) or pd.isna(r["ema_slow"]) or pd.isna(r["macd_hist"]):
            continue

        # Signal Logic
        uptrend = (r["ema_fast"] > r["ema_slow"]) and (r["macd_hist"] > 0)
        downtrend = (r["ema_fast"] < r["ema_slow"]) and (r["macd_hist"] < 0)

        # 2. ENTRY LOGIC
        if not in_trade:
            if uptrend:
                direction = "long"
                entry = r["close"]
                entry_time = r["dt"]
                in_trade = True
            elif downtrend:
                direction = "short"
                entry = r["close"]
                entry_time = r["dt"]
                in_trade = True

        # 3. REVERSAL LOGIC
        else:
            if direction=="long" and downtrend:
                exit_price = r["close"]
                pnl = (exit_price - entry) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"EMA_MACD", "dir":"long", "entry":entry,
                    "exit":exit_price, "pnl":pnl, "outcome":"REVERSE"
                })
                # Reverse to short
                direction = "short"
                entry = r["close"]
                entry_time = r["dt"]

            elif direction=="short" and uptrend:
                exit_price = r["close"]
                pnl = (entry - exit_price) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"EMA_MACD", "dir":"short", "entry":entry,
                    "exit":exit_price, "pnl":pnl, "outcome":"REVERSE"
                })
                # Reverse to long
                direction = "long"
                entry = r["close"]
                entry_time = r["dt"]

    return trades

# -------------- EXECUTION & REPORTING --------------
macd_trades = backtest_ema_macd(df)
summary = {"EMA_MACD": stats(macd_trades)}

print(f"✅ Backtest complete ({TIMEFRAME} candles)")
for name, s in summary.items():
    print(f"\n{name} -> Trades: {s['trades']}, Wins: {s['wins']}, WinRate: {s['win_rate']}%, "
          f"AvgPnL: {s['avg_pnl']}, TotalPnL: {s['total_pnl']}")


output_file = "ema_macd_results.xlsx"
print(f"\nSaving results to {output_file}...")

with pd.ExcelWriter(output_file, engine="xlsxwriter") as writer:
    if macd_trades:
        trades_df = pd.DataFrame(macd_trades)
        trades_df.to_excel(writer, sheet_name="EMA_MACD", index=False)

        trades_df["date_dt"] = pd.to_datetime(trades_df["date"])
        trades_df["week"] = trades_df["date_dt"].dt.to_period("W-FRI")
        weekly = trades_df.groupby("week", as_index=False)["pnl"].sum()
        weekly.rename(columns={"pnl": "total_points"}, inplace=True)
        weekly["week"] = weekly["week"].astype(str)
        weekly.to_excel(writer, sheet_name="Weekly_Report", index=False)

        trades_df["month"] = trades_df["date_dt"].dt.to_period("M")
        monthly = trades_df.groupby("month", as_index=False)["pnl"].sum()
        monthly.rename(columns={"pnl": "total_points"}, inplace=True)
        monthly["month"] = monthly["month"].astype(str)
        monthly.to_excel(writer, sheet_name="Monthly_Report", index=False)

        dd_rows = []
        for m, mdf in trades_df.groupby("month"):
            mdf = mdf.sort_values("exit_time")
            cum = mdf["pnl"].cumsum()
            max_dd = cum.min()
            dd_rows.append({"month": str(m), "max_drawdown": max_dd})
        dd_df = pd.DataFrame(dd_rows)
        dd_df.to_excel(writer, sheet_name="Drawdown_Report", index=False)

    pd.DataFrame([{"strategy": k, **v} for k, v in summary.items()]).to_excel(
        writer, sheet_name="Summary", index=False
    )

print("Process finished successfully.")