Important This algorithm strictly utilizes Candlestick Pattern recognition by comparing prev_open, prev_close, open, and close data points to identify structural engulfing formations dynamically on the 30-minute timeframe.

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 = "30min"   

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

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

# ----------------- INDICATORS --------------------
print("Detecting Engulfing Candlestick Patterns...")

# Shift data to compare with the previous candle's open and close
df["prev_open"] = df["open"].shift(1)
df["prev_close"] = df["close"].shift(1)

# Bullish Engulfing: 
# 1. Previous candle is red
# 2. Current candle is green
# 3. Current body completely covers previous body
df["bullish_engulfing"] = (
    (df["prev_close"] < df["prev_open"]) & 
    (df["close"] > df["open"]) & 
    (df["close"] >= df["prev_open"]) &
    (df["open"] <= df["prev_close"])
)

# Bearish Engulfing:
# 1. Previous candle is green
# 2. Current candle is red
# 3. Current body completely covers previous body
df["bearish_engulfing"] = (
    (df["prev_close"] > df["prev_open"]) & 
    (df["close"] < df["open"]) & 
    (df["close"] <= df["prev_open"]) & 
    (df["open"] >= df["prev_close"])
)

# -------------- 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: ENGULFING PATTERN --------------
def backtest_engulfing(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":"ENGULFING", "dir":direction, "entry":entry,
                    "exit":r["close"], "pnl":pnl, "outcome":"TIME_EXIT"
                })
                in_trade = False
            continue

        if pd.isna(r["prev_open"]):
            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":"ENGULFING", "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 - Pure Candlestick Pattern Match
        signal_long = r["bullish_engulfing"]
        signal_short = r["bearish_engulfing"]

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

        # 4. REVERSAL LOGIC
        else:
            if direction=="long" and signal_short:
                exit_price = r["close"]
                pnl = (exit_price - entry) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"ENGULFING", "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 signal_long:
                exit_price = r["close"]
                pnl = (entry - exit_price) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"ENGULFING", "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 --------------
engulfing_trades = backtest_engulfing(df)
summary = {"ENGULFING": stats(engulfing_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 = "engulfing_pattern_results.xlsx"
print(f"\nSaving results to {output_file}...")

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