Equity 5-Min RSI Momentum & EMA Trend
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.
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.
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.
# ----------------- 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.
# ----------------- 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.
# -------------- 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.
# -------------- 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.
# -------------- 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.")
Browse the Full Quantitative Repository:
- â Intraday 10/30 Moving Average Crossover
- â Index 5-Min Bollinger Bands Mean Reversion
- â Algorithmic 5-Min EMA & MACD Confluence
- â Equity 5-Min RSI Momentum & EMA Trend
- â Volatility 5-Min Donchian Channel Breakout
- â Global Macro 30-Min Inside Bar Breakout
- â Institutional 15-Min Opening Range Breakout
- â Quantitative 5-Min Daily Floor Pivots Breakout
- â Price Action 30-Min Candlestick Engulfing Pattern
- â Systematic 5-Min Pin Bar Reversal (Hammer & Shooting Star)
