Important This algorithm utilizes Pandas groupby("d").agg() to calculate the previous day’s High, Low, and Close, subsequently mapping the Floor Pivot, R1, and S1 levels onto the intraday 5-minute data.

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"   

# 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("Calculating Daily Pivot Points (R1 & S1)...")

# 1. Group by date to find daily High, Low, and Close
daily_df = df.groupby("d").agg(
    day_high=("high", "max"),
    day_low=("low", "min"),
    day_close=("close", "last")
).reset_index()

# 2. Shift to get the PREVIOUS day's values
daily_df["prev_high"] = daily_df["day_high"].shift(1)
daily_df["prev_low"] = daily_df["day_low"].shift(1)
daily_df["prev_close"] = daily_df["day_close"].shift(1)

# 3. Calculate Floor Pivot Points
daily_df["pivot"] = (daily_df["prev_high"] + daily_df["prev_low"] + daily_df["prev_close"]) / 3
daily_df["R1"] = (2 * daily_df["pivot"]) - daily_df["prev_low"]
daily_df["S1"] = (2 * daily_df["pivot"]) - daily_df["prev_high"]

# 4. Merge the specific daily pivot levels back into the intraday timeframe dataframe
pivot_levels = daily_df[["d", "R1", "S1", "pivot"]]
df = df.merge(pivot_levels, on="d", how="left")

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

        
        if pd.isna(r["R1"]) or pd.isna(r["S1"]):
            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":"PIVOT_BREAKOUT", "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 - Breakout beyond the Resistance 1 or Support 1
        breakout_up = r["close"] > r["R1"]
        breakout_down = r["close"] < r["S1"]

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

        # 4. REVERSAL LOGIC
        else:
            if direction=="long" and breakout_down:
                exit_price = r["close"]
                pnl = (exit_price - entry) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"PIVOT_BREAKOUT", "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 breakout_up:
                exit_price = r["close"]
                pnl = (entry - exit_price) - SLIPPAGE
                trades.append({
                    "date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
                    "strategy":"PIVOT_BREAKOUT", "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 --------------
pivot_trades = backtest_pivots(df)
summary = {"PIVOT_BREAKOUT": stats(pivot_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 = "pivot_breakout_results.xlsx"
print(f"\nSaving results to {output_file}...")

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