Equity 5-Min RSI Momentum & EMA Trend

đŸĻŠ Premium Script

Strategy Overview: A comprehensive momentum strategy engineered to execute precisely when directional velocity aligns with the prevailing market trend on the 5-minute timeframe. This dual-indicator algorithm evaluates trend context using a 50-period Exponential Moving Average (EMA) while dynamically assessing momentum using the 14-period Relative Strength Index (RSI). It executes long setups exclusively when the closing price is positioned above the 50 EMA and the RSI breaches the 60 threshold, indicating strong bullish conviction. Conversely, short setups are initiated only when the price rests below the 50 EMA and the RSI plunges beneath 40, signaling bearish dominance. The script incorporates robust risk management architecture, actively managing open positions with strict predefined stop-loss and take-profit parameters measured in absolute points to protect capital. Furthermore, it embeds a continuous reversal mechanism to flip bias when market regimes shift abruptly. Ensuring disciplined execution, the algorithm enforces a rigid 15:15 intraday time exit, neutralizing all exposure and immunizing the portfolio against overnight gap volatility.

📈 EXPECTED WIN RATE
41.21%
📉 MAX DRAWDOWN
907 Pts
💰 TOTAL PNL
+12,140 Pts
🔄 TOTAL TRADES
4254
🏆 TOTAL WINS
1753
!
Important This algorithm incorporates Wilder’s Smoothing method for calculating the Relative Strength Index (RSI) dynamically using pandas exponential functions. It integrates strict absolute point-based stop-loss and target-profit mechanisms.
â„šī¸
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.

1. Imports and Configuration

Summary: This section establishes the base libraries and configures the core operational logic. It sets the 5-minute timeframe, defines the 50-period EMA, configures the 14-period RSI (with buy triggers above 60 and sell triggers below 40), and locks in the rigid risk parameters (50-point SL and 100-point TP) alongside the 15:15 end-of-day exit.

python script Copy Code
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"   
EMA_PERIOD = 50
RSI_PERIOD = 14
RSI_BUY_LEVEL = 60
RSI_SELL_LEVEL = 40

# Risk Management
STOP_LOSS_POINTS = 50.0            
TARGET_POINTS = 100.0              
# ==================================================

2. Data Loading and Preparation

Summary: This code segment handles loading the historical market data from the CSV file. It ensures all required OHLC columns exist, converts date strings to proper datetime formats, handles missing values, and resamples the minute-level base data into robust 5-minute intervals.

python script Copy Code
# ----------------- 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()

# Resampling
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

3. EMA & RSI Indicator Calculations

Summary: This calculates the trend filter (50-period Exponential Moving Average) and the momentum oscillator (14-period RSI) directly on the pandas dataframe. Crucially, the RSI is calculated utilizing Wilder’s Smoothing method via the exponentially weighted moving average (`ewm`) function to accurately reflect classical RSI formulas.

python script Copy Code
# ----------------- INDICATORS --------------------
print("Calculating EMA and RSI...")

# Calculate 50-period EMA
df["ema"] = df["close"].ewm(span=EMA_PERIOD, adjust=False).mean()

# Calculate 14-period RSI
delta = df["close"].diff()
gain = delta.where(delta > 0, 0.0)
loss = -delta.where(delta < 0, 0.0)

# Wilder's Smoothing for RSI (using EMA)
avg_gain = gain.ewm(alpha=1/RSI_PERIOD, adjust=False).mean()
avg_loss = loss.ewm(alpha=1/RSI_PERIOD, adjust=False).mean()

rs = avg_gain / avg_loss
df["rsi"] = 100 - (100 / (1 + rs))

4. Utility Functions

Summary: Helper functions essential for backtesting mechanics. The stats function aggregates the completed trades into performance metrics like win rate and total PnL. The close_out function uniformly calculates the absolute point differences for exited long and short positions.

python script Copy Code
# -------------- 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)

5. The Core Backtest Engine

Summary: The primary backtesting iteration loop. It constantly monitors for an automated 15:15 market exit, assesses Stop-Loss and Take-Profit triggers dynamically based on intrabar Highs and Lows, initiates fresh positions when the RSI breaks the momentum thresholds while confirming trend direction against the EMA, and manages immediate position reversals.

python script Copy Code
# -------------- STRATEGY: RSI MOMENTUM --------------
def backtest_rsi_momentum(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":"RSI_MOMENTUM", "dir":direction, "entry":entry,
                    "exit":r["close"], "pnl":pnl, "outcome":"TIME_EXIT"
                })
                in_trade = False
            continue

        if pd.isna(r["ema"]) or pd.isna(r["rsi"]):
            continue

        # 2. SL / TP LOGIC
        if in_trade:
            sl_hit = False
            tp_hit = False
            exit_price = None
            outcome = None

            # Check conservative High/Low hits within the current candle
            if direction == "long":
                if r["low"] <= entry - STOP_LOSS_POINTS:
                    sl_hit = True
                    exit_price = entry - STOP_LOSS_POINTS
                    outcome = "STOP_LOSS"
                elif r["high"] >= entry + TARGET_POINTS:
                    tp_hit = True
                    exit_price = entry + TARGET_POINTS
                    outcome = "TARGET"
            elif direction == "short":
                if r["high"] >= entry + STOP_LOSS_POINTS:
                    sl_hit = True
                    exit_price = entry + STOP_LOSS_POINTS
                    outcome = "STOP_LOSS"
                elif r["low"] <= entry - TARGET_POINTS:
                    tp_hit = True
                    exit_price = entry - TARGET_POINTS
                    outcome = "TARGET"

            # If stopped out or target reached, close the trade
            if sl_hit or tp_hit:
                pnl = close_out(direction, entry, exit_price) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"RSI_MOMENTUM", "dir":direction, "entry":entry,
                    "exit":exit_price, "pnl":pnl, "outcome":outcome
                })
                in_trade = False
                direction = None
                entry = None
                entry_time = None
                # Skip entry logic on the same candle we hit SL/TP
                continue

        # Signal Logic
        bullish = (r["rsi"] > RSI_BUY_LEVEL) and (r["close"] > r["ema"])
        bearish = (r["rsi"] < RSI_SELL_LEVEL) and (r["close"] < r["ema"])

        # 3. ENTRY LOGIC
        if not in_trade:
            if bullish:
                direction = "long"
                entry = r["close"]
                entry_time = r["dt"]
                in_trade = True
            elif bearish:
                direction = "short"
                entry = r["close"]
                entry_time = r["dt"]
                in_trade = True

        # 4. REVERSAL LOGIC
        else:
            if direction=="long" and bearish:
                exit_price = r["close"]
                pnl = (exit_price - entry) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"RSI_MOMENTUM", "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 bullish:
                exit_price = r["close"]
                pnl = (entry - exit_price) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"RSI_MOMENTUM", "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

6. Execution and Reporting

Summary: Triggers the backtesting engine and reports the results. It begins by echoing a high-level summary to the console, and then it systematically processes the array of completed trades into an exhaustive, multi-tab Excel document (rsi_momentum_results.xlsx) including weekly summaries, monthly aggregates, and detailed drawdowns.

python script Copy Code
# -------------- EXECUTION & REPORTING --------------
rsi_trades = backtest_rsi_momentum(df)
summary = {"RSI_MOMENTUM": stats(rsi_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']}")

# Write results to Excel
output_file = "rsi_momentum_results.xlsx"
print(f"\nSaving results to {output_file}...")

with pd.ExcelWriter(output_file, engine="xlsxwriter") as writer:
    if rsi_trades:
        trades_df = pd.DataFrame(rsi_trades)
        trades_df.to_excel(writer, sheet_name="RSI_MOMENTUM", 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.")