thank you for your reply.. i have implemented a strategy after doing tested of 6 month data in live market after watching your last algo series but now started showing negative result. yesterday i did back testing of the same on last 5 years data and that is also not good after deduction of fees. so doesn’t make any sense thank you so much in helping to download bank nifty data from 2021. and providing valuable knowledge which is important. please help in reviewing this code or anybody else have suggestion to improve this strategy to reduce drawdown and good equity curve. import pandas as pd
=========================
CONFIG
=========================
CSV_FILE = “Historical Data/BANKNIFTY_1M_clean.csv”
QTY_PER_LEG = 30
TRADE_START_TIME = “09:45”
NO_TRADE_AFTER = “14:00”
FORCE_EXIT_TIME = “15:10”
EXIT_TIME = “15:15”
RISK_REWARD = 3
=========================
LOAD DATA
=========================
df = pd.read_csv(CSV_FILE)
for c in [“open”, “high”, “low”, “close”, “volume”]:
df[c] = df[c].astype(str).str.replace(“,”, “”).astype(float)
df[“timestamp”] = pd.to_datetime(df[“timestamp”])
df.set_index(“timestamp”, inplace=True)
df[“date”] = df.index.date
activity_log =
orderbook =
=========================
LOGGER
=========================
def log_activity(ts, row, day, activity, trade_side,break_side, ref_high, ref_low, step, active_legs, unrealized):
position_qty = sum(l["qty"] for l in active_legs if l["active"])
activity_log.append({
"timestamp": ts,
"date": day,
"open": row["open"],
"high": row["high"],
"low": row["low"],
"close": row["close"],
"activity": activity,
"trade_side": trade_side,
"break_side" :break_side,
"ref_high": ref_high,
"ref_low": ref_low,
"step": step,
"active_legs": ",".join(f"L{l['leg']}" for l in active_legs if l["active"]),
"position_qty": position_qty,
"unrealized_pnl": unrealized
})
def all_legs_closed(legs):
return all(not l[“active”] for l in legs) if legs else False
=========================
BACKTEST
=========================
for day, day_df in df.groupby(“date”):
active_trade = None
active_legs = []
next_leg = 0
trade_done = {"BUY": False, "SELL": False}
# ---- Reference candle (09:30–09:44) ----
ref = day_df.between_time("09:30", "09:44")
if ref.empty:
continue
ref_high = ref["high"].max()
ref_low = ref["low"].min()
adj_range = round(min(max(ref_high - ref_low, 50), 150),2)
step = adj_range / 4
for ts, row in day_df.iterrows():
sl_hits = []
high, low, close = row["high"], row["low"], row["close"]
activity = "WAIT"
break_side = None
# ---- PRE MARKET ----
if ts.strftime("%H:%M") < TRADE_START_TIME:
log_activity(ts, row, day, "PRE_MARKET_WAIT", None, break_side, ref_high, ref_low, step, [], 0)
continue
# ---- ENTRY DIRECTION ----
if active_trade is None and ts.strftime("%H:%M") < NO_TRADE_AFTER:
if not trade_done["BUY"] and high > ref_high:
active_trade = "BUY"
trade_done["BUY"] = True
next_leg = 0
activity = "BUY_BREAK"
break_side = "BUY_BREAK"
elif not trade_done["SELL"] and low < ref_low:
active_trade = "SELL"
trade_done["SELL"] = True
next_leg = 0
activity = "SELL_BREAK"
break_side = "SELL_BREAK"
# ---- PYRAMID ADD ----
if active_trade == "BUY" and next_leg < 4:
entry = ref_high + next_leg * step
if high >= entry:
#sl = ref_low if next_leg == 0 else active_legs[-1]["entry"]
sl = entry - (adj_range * 0.7)
risk = entry - sl
tp = entry + (risk * RISK_REWARD)
active_legs.append({
"side": "BUY", "leg": next_leg + 1,
"entry": entry, "sl": sl, "tp": tp,
"rr": RISK_REWARD, "qty": QTY_PER_LEG, "active": True
})
activity = f"BUY_L{next_leg+1}"
next_leg += 1
if active_trade == "SELL" and next_leg < 4:
entry = ref_low - next_leg * step
if low <= entry:
#sl = ref_high if next_leg == 0 else active_legs[-1]["entry"]
sl = entry + (adj_range * 0.7)
risk = sl - entry
tp = entry - (risk * RISK_REWARD)
active_legs.append({
"side": "SELL", "leg": next_leg + 1,
"entry": entry, "sl": sl, "tp": tp,
"rr": RISK_REWARD, "qty": QTY_PER_LEG, "active": True
})
activity = f"SELL_L{next_leg+1}"
next_leg += 1
# ---- SL / TP CHECK (SL FIRST) ----
for leg in active_legs:
if not leg["active"]:
continue
# SL
if leg["side"] == "BUY" and low <= leg["sl"]:
leg["active"] = False
pnl = (leg["sl"] - leg["entry"]) * leg["qty"]
orderbook.append({
"date": day, "side": "BUY", "leg": leg["leg"],
"entry": leg["entry"], "sl": leg["sl"],
"tp": leg["tp"], "exit": leg["sl"],
"rr": leg["rr"], "exit_reason": "SL", "pnl": pnl
})
sl_hits.append(f"L{leg['leg']}_SL_HIT")
continue
if leg["side"] == "SELL" and high >= leg["sl"]:
leg["active"] = False
pnl = (leg["entry"] - leg["sl"]) * leg["qty"]
orderbook.append({
"date": day, "side": "SELL", "leg": leg["leg"],
"entry": leg["entry"], "sl": leg["sl"],
"tp": leg["tp"], "exit": leg["sl"],
"rr": leg["rr"], "exit_reason": "SL", "pnl": pnl
})
sl_hits.append(f"L{leg['leg']}_SL_HIT")
continue
# TP
if leg["side"] == "BUY" and high >= leg["tp"]:
leg["active"] = False
pnl = (leg["tp"] - leg["entry"]) * leg["qty"]
orderbook.append({
"date": day, "side": "BUY", "leg": leg["leg"],
"entry": leg["entry"], "sl": leg["sl"],
"tp": leg["tp"], "exit": leg["tp"],
"rr": leg["rr"], "exit_reason": "TP", "pnl": pnl
})
sl_hits.append(f"L{leg['leg']}_TP_HIT")
if leg["side"] == "SELL" and low <= leg["tp"]:
leg["active"] = False
pnl = (leg["entry"] - leg["tp"]) * leg["qty"]
orderbook.append({
"date": day, "side": "SELL", "leg": leg["leg"],
"entry": leg["entry"], "sl": leg["sl"],
"tp": leg["tp"], "exit": leg["tp"],
"rr": leg["rr"], "exit_reason": "TP", "pnl": pnl
})
sl_hits.append(f"L{leg['leg']}_TP_HIT")
# ---- FORCE EXIT AT 15:10 ----
#if ts.strftime("%H:%M") >= FORCE_EXIT_TIME:
# ---- FORCE EXIT AT 15:10 (ONLY IF STILL ACTIVE) ----
if ts.strftime("%H:%M") >= FORCE_EXIT_TIME:
for leg in active_legs:
if not leg["active"]:
continue
leg["active"] = False
if leg["side"] == "BUY":
pnl = (close - leg["entry"]) * leg["qty"]
else:
pnl = (leg["entry"] - close) * leg["qty"]
orderbook.append({
"date": day,
"side": leg["side"],
"leg": leg["leg"],
"entry": leg["entry"],
"sl": leg["sl"],
"tp": leg["tp"],
"exit": close,
"rr": leg["rr"],
"exit_reason": "TIME_EXIT_1510",
"pnl": pnl
})
sl_hits.append(f"L{leg['leg']}_TIME_EXIT")
if sl_hits:
activity = " | ".join(sl_hits)
# ---- CHECK IF TRADE COMPLETED ----
if active_trade is not None and all_legs_closed(active_legs):
active_trade = None
active_legs = []
next_leg = 0
# ---- UNREALIZED ----
unrealized = 0
for leg in active_legs:
if leg["active"]:
unrealized += (
(close - leg["entry"]) * leg["qty"]
if leg["side"] == "BUY"
else (leg["entry"] - close) * leg["qty"]
)
log_activity(ts, row, day, activity, active_trade,break_side, ref_high, ref_low, step, active_legs, unrealized)
=========================
EXPORT
=========================
pd.DataFrame(activity_log).to_csv(“BANKNIFTY_activity_log.csv”, index=False)
pd.DataFrame(orderbook).to_csv(“BANKNIFTY_orderbook.csv”, index=False)
print(“
Backtest completed successfully”)
