Systematic 5-Min Pin Bar Reversal (Hammer & Shooting Star)
Strategy Overview: The Systematic 5-Min Pin Bar Reversal strategy is a highly precise, candlestick-driven algorithmic framework designed to capture violent market rejections at key intraday extremes. Operating on a 5-minute timeframe, this system mathematically isolates ‘Hammer’ and ‘Shooting Star’ formations by analyzing the precise ratio of the candle’s body to its wicks. A bullish Hammer is detected when the lower wick is at least twice the size of the body during a short-term downtrend, signaling aggressive institutional buying and triggering a long entry. Conversely, a bearish Shooting Star is identified when a massive upper wick forms during an uptrend, revealing a sudden lack of demand and executing a short position. To filter out market noise, the algorithm mandates that the candle body exceeds a minimum point threshold, effectively ignoring flat dojis and low-volume chop. Local trend confirmation is strictly enforced by comparing the current closing price to the close from three periods prior, ensuring reversals are only traded against established momentum. Risk management is hard-coded into the execution logic, utilizing a strict 50-point stop loss to truncate drawdowns and a 100-point take-profit to secure alpha. This mechanical structure guarantees a highly asymmetric 1:2 risk-to-reward ratio on every single trade, allowing the system to remain profitable even with a sub-50% win rate. Additionally, the system features a built-in reversal protocol that flips the directional bias if an opposing pin bar formation immediately materializes. As a purely intraday quantitative model, it strictly enforces an automated 15:15 square-off sequence, completely neutralizing the portfolio against unpredictable overnight gaps and theta decay.
max(axis=1) and min(axis=1) functions, whilst verifying the local trend by comparing current prices to the shift(3) lookback period.
1. Imports and Configuration
Summary: This section initializes the required libraries and configures the core backtesting setup, including the historical data file path, a 15:15 intraday square-off time, a 5-minute timeframe, and fixed risk parameters of a 50-point stop loss and 100-point target.
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
# ==================================================2. Data Loading and Preparation
Summary: This section loads historical price data from the CSV, validates the presence of required columns, standardizes data types, cleans missing values, and resamples the dataset into structured 5-minute OHLC candles.
# ----------------- 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.time3. Candlestick Anatomy & Trend Calculation
Summary: Here, the algorithm dynamically calculates the anatomy of each candlestick. It calculates the size of the candle’s body, upper wick, and lower wick. It also determines the local trend by comparing the current closing price to the close from three periods prior.
# ----------------- INDICATORS --------------------
print("Detecting Hammer and Shooting Star (Pin Bar) Patterns...")
# 1. Calculate the components of the candlestick
df["max_oc"] = df[["open", "close"]].max(axis=1)
df["min_oc"] = df[["open", "close"]].min(axis=1)
df["body"] = df["max_oc"] - df["min_oc"]
df["upper_wick"] = df["high"] - df["max_oc"]
df["lower_wick"] = df["min_oc"] - df["low"]
# 2. Local Trend check (Comparing current price to 3 candles ago)
df["trend_down"] = df["close"] < df["close"].shift(3)
df["trend_up"] = df["close"] > df["close"].shift(3)4. Pattern Recognition Logic
Summary: This applies the mathematical rules to identify valid trade setups. A bullish Hammer requires a lower wick twice the size of the body during a downtrend. A bearish Shooting Star requires an upper wick twice the size of the body during an uptrend. Both filters mandate a minimum body size of 2 points to ignore “doji” market noise.
# 3. Hammer (Bullish Reversal) Logic:
# - Lower wick is at least twice the size of the body
# - Upper wick is very small (less than the body)
# - Body is larger than 2 points (to avoid flat dojis)
# - Occurs in a short-term downtrend
df["is_hammer"] = (
(df["lower_wick"] >= 2.0 * df["body"]) &
(df["upper_wick"] <= df["body"]) &
(df["body"] > 2.0) &
df["trend_down"]
)
# 4. Shooting Star (Bearish Reversal) Logic:
# - Upper wick is at least twice the size of the body
# - Lower wick is very small (less than the body)
# - Body is larger than 2 points
# - Occurs in a short-term uptrend
df["is_shooting_star"] = (
(df["upper_wick"] >= 2.0 * df["body"]) &
(df["lower_wick"] <= df["body"]) &
(df["body"] > 2.0) &
df["trend_up"]
)5. Utility Functions
Summary: These are helper functions. stats generates the performance summary (win rate, total PnL, etc.) from the list of executed trades. close_out calculates the raw points gained or lost on a specific trade based on its direction.
# -------------- 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)6. The Core Backtest Engine
Summary: This is the main loop that iterates through the prepared data row by row. It enforces the 15:15 intraday exit, monitors active trades for stop-loss or take-profit hits, initiates new long or short positions based on the candlestick signals, and includes logic to immediately reverse a position if an opposing pattern forms.
# -------------- STRATEGY: PIN BAR REVERSAL --------------
def backtest_pin_bar(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":"PIN_BAR", "dir":direction, "entry":entry,
"exit":r["close"], "pnl":pnl, "outcome":"TIME_EXIT"
})
in_trade = False
continue
if pd.isna(r["is_hammer"]):
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":"PIN_BAR", "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
go_long = r["is_hammer"]
go_short = r["is_shooting_star"]
# 3. ENTRY LOGIC
if not in_trade:
if go_long:
direction = "long"
entry = r["close"]
entry_time = r["dt"]
in_trade = True
elif go_short:
direction = "short"
entry = r["close"]
entry_time = r["dt"]
in_trade = True
# 4. REVERSAL LOGIC
else:
if direction=="long" and go_short:
exit_price = r["close"]
pnl = (exit_price - entry) - SLIPPAGE
trades.append({
"date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
"strategy":"PIN_BAR", "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 go_long:
exit_price = r["close"]
pnl = (entry - exit_price) - SLIPPAGE
trades.append({
"date": r["d"], "entry_time": entry_time, "exit_time": r["dt"],
"strategy":"PIN_BAR", "dir":"short", "entry":entry,
"exit":exit_price, "pnl":pnl, "outcome":"REVERSE"
})
# Reverse to long
direction = "long"
entry = r["close"]
entry_time = r["dt"]
return trades7. Execution and Reporting
Summary: This final section executes the backtest and outputs a concise performance summary to the console. It then organizes detailed trade logs into weekly and monthly breakdowns, computes drawdowns, and exports all results into a structured multi-sheet Excel file (pin_bar_strategy_results.xlsx).
# -------------- EXECUTION & REPORTING --------------
pin_trades = backtest_pin_bar(df)
summary = {"PIN_BAR": stats(pin_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 = "pin_bar_strategy_results.xlsx"
print(f"\nSaving results to {output_file}...")
with pd.ExcelWriter(output_file, engine="xlsxwriter") as writer:
if pin_trades:
trades_df = pd.DataFrame(pin_trades)
trades_df.to_excel(writer, sheet_name="PIN_BAR", 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)