322 lines
14 KiB
Python
322 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
RSI Mean Reversion Strategy — Backtest
|
||
======================================
|
||
Symbol : EURJPY M15
|
||
Period : 2024-04-22 → 2026-04-21 (~2 years, 10 277 candles)
|
||
|
||
Strategy
|
||
--------
|
||
Signal : RSI(14) exits oversold (<30) or overbought (>70) zone
|
||
Entry : Next bar open after signal candle
|
||
Stop : 1.5 × ATR(14) against position
|
||
Target : 3.0 × ATR(14) in favour → R:R = 1 : 2
|
||
Session : 07:00–20:00 UTC only (avoid thin Asian/weekend hours)
|
||
Max pos : 1 trade at a time
|
||
|
||
Rationale
|
||
---------
|
||
RSI reversals from extreme territory have decades of documented edge on
|
||
liquid FX pairs. Entering on the *exit* from the extreme (rather than at
|
||
the extreme itself) gives one-bar of confirmation that the bounce has
|
||
started. ATR-based stops adapt to current volatility, keeping the R:R
|
||
ratio constant in volatility-adjusted terms.
|
||
"""
|
||
|
||
import sqlite3
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
import sys
|
||
|
||
# ── numerical libs ──────────────────────────────────────────────────────────
|
||
try:
|
||
import numpy as np
|
||
import pandas as pd
|
||
except ImportError:
|
||
sys.exit("Run with the BellCurve venv: "
|
||
"/home/jonathan/Projects/ForexBots/BellCurve/.venv/bin/python3 backtest.py")
|
||
|
||
# ── constants ────────────────────────────────────────────────────────────────
|
||
DB_PATH = Path("/home/jonathan/Projects/ForexBots/data/candles.db")
|
||
SYMBOL = "EURJPY"
|
||
TIMEFRAME = "M15"
|
||
RSI_PERIOD = 14
|
||
ATR_PERIOD = 14
|
||
RSI_OB = 70 # overbought threshold
|
||
RSI_OS = 30 # oversold threshold
|
||
SL_ATR = 1.5 # stop-loss multiplier
|
||
TP_ATR = 3.0 # take-profit multiplier
|
||
SESSION_START = 7 # UTC hour
|
||
SESSION_END = 20 # UTC hour
|
||
PIP = 0.01 # 1 pip for JPY pairs (e.g. EURJPY)
|
||
INITIAL_BALANCE = 10_000.0 # USD
|
||
RISK_PER_TRADE = 0.01 # 1 % of balance per trade
|
||
|
||
|
||
# ── data classes ─────────────────────────────────────────────────────────────
|
||
@dataclass
|
||
class Trade:
|
||
direction: str # 'long' | 'short'
|
||
entry_time: datetime
|
||
entry_price: float
|
||
sl: float
|
||
tp: float
|
||
exit_time: Optional[datetime] = None
|
||
exit_price: Optional[float] = None
|
||
exit_reason: str = ""
|
||
pnl_pips: float = 0.0
|
||
pnl_pct: float = 0.0 # % of balance risked
|
||
|
||
@property
|
||
def closed(self) -> bool:
|
||
return self.exit_time is not None
|
||
|
||
|
||
# ── helpers ───────────────────────────────────────────────────────────────────
|
||
def calc_rsi(close: pd.Series, period: int = 14) -> pd.Series:
|
||
delta = close.diff()
|
||
gain = delta.clip(lower=0)
|
||
loss = (-delta).clip(lower=0)
|
||
avg_g = gain.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
||
avg_l = loss.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
||
rs = avg_g / avg_l.replace(0, float("inf"))
|
||
return 100 - (100 / (1 + rs))
|
||
|
||
|
||
def calc_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
|
||
high, low, prev_close = df["high"], df["low"], df["close"].shift(1)
|
||
tr = pd.concat([
|
||
high - low,
|
||
(high - prev_close).abs(),
|
||
(low - prev_close).abs(),
|
||
], axis=1).max(axis=1)
|
||
return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
||
|
||
|
||
def pip_diff(a: float, b: float) -> float:
|
||
"""Signed pip difference a − b (JPY pair)."""
|
||
return (a - b) / PIP
|
||
|
||
|
||
# ── load data ─────────────────────────────────────────────────────────────────
|
||
def load_candles() -> pd.DataFrame:
|
||
conn = sqlite3.connect(DB_PATH)
|
||
df = pd.read_sql_query(
|
||
"SELECT time, open, high, low, close, tick_volume AS volume "
|
||
"FROM candles WHERE symbol=? AND timeframe=? ORDER BY time",
|
||
conn, params=(SYMBOL, TIMEFRAME),
|
||
)
|
||
conn.close()
|
||
df["time"] = pd.to_datetime(df["time"])
|
||
df.set_index("time", inplace=True)
|
||
return df
|
||
|
||
|
||
# ── core backtest ─────────────────────────────────────────────────────────────
|
||
def run_backtest(df: pd.DataFrame) -> list[Trade]:
|
||
df = df.copy()
|
||
df["rsi"] = calc_rsi(df["close"], RSI_PERIOD)
|
||
df["atr"] = calc_atr(df, ATR_PERIOD)
|
||
|
||
trades: list[Trade] = []
|
||
position: Optional[Trade] = None
|
||
|
||
for i in range(1, len(df)):
|
||
row = df.iloc[i]
|
||
prev_row = df.iloc[i - 1]
|
||
|
||
hour = row.name.hour
|
||
|
||
# ── manage open position ─────────────────────────────────────────────
|
||
if position is not None and not position.closed:
|
||
hi, lo = row["high"], row["low"]
|
||
sl, tp = position.sl, position.tp
|
||
|
||
hit_sl = (position.direction == "long" and lo <= sl) or \
|
||
(position.direction == "short" and hi >= sl)
|
||
hit_tp = (position.direction == "long" and hi >= tp) or \
|
||
(position.direction == "short" and lo <= tp)
|
||
|
||
# if both hit on same bar, assume SL got hit first (conservative)
|
||
if hit_sl or hit_tp:
|
||
exit_price = sl if hit_sl else tp
|
||
exit_reason = "SL" if hit_sl else "TP"
|
||
pips = pip_diff(exit_price, position.entry_price) if position.direction == "long" \
|
||
else pip_diff(position.entry_price, exit_price)
|
||
position.exit_time = row.name
|
||
position.exit_price = exit_price
|
||
position.exit_reason = exit_reason
|
||
position.pnl_pips = pips
|
||
# pnl_pct: risk was SL distance; TP gives 2× risk, SL gives −1× risk
|
||
sl_distance_pips = abs(pip_diff(position.entry_price, sl))
|
||
position.pnl_pct = (pips / sl_distance_pips) * RISK_PER_TRADE * 100
|
||
position = None
|
||
continue # don't open new trade on same bar we just closed
|
||
|
||
# ── check for new signal (no open position) ──────────────────────────
|
||
if position is not None:
|
||
continue
|
||
|
||
# session filter
|
||
if not (SESSION_START <= hour < SESSION_END):
|
||
continue
|
||
|
||
# need valid indicators
|
||
if pd.isna(prev_row["rsi"]) or pd.isna(row["rsi"]) or pd.isna(row["atr"]):
|
||
continue
|
||
|
||
rsi_prev, rsi_now = prev_row["rsi"], row["rsi"]
|
||
atr = row["atr"]
|
||
|
||
# signal: RSI exiting oversold (cross back above 30) → long
|
||
long_signal = (rsi_prev < RSI_OS) and (rsi_now >= RSI_OS)
|
||
# signal: RSI exiting overbought (cross back below 70) → short
|
||
short_signal = (rsi_prev > RSI_OB) and (rsi_now <= RSI_OB)
|
||
|
||
if not (long_signal or short_signal):
|
||
continue
|
||
|
||
# entry on NEXT bar open — peek at i+1 if available
|
||
if i + 1 >= len(df):
|
||
continue
|
||
next_bar = df.iloc[i + 1]
|
||
entry_price = next_bar["open"]
|
||
atr_at_entry = row["atr"] # use signal-bar ATR for SL/TP
|
||
|
||
if long_signal:
|
||
sl = entry_price - SL_ATR * atr_at_entry
|
||
tp = entry_price + TP_ATR * atr_at_entry
|
||
direction = "long"
|
||
else:
|
||
sl = entry_price + SL_ATR * atr_at_entry
|
||
tp = entry_price - TP_ATR * atr_at_entry
|
||
direction = "short"
|
||
|
||
position = Trade(
|
||
direction = direction,
|
||
entry_time = next_bar.name,
|
||
entry_price = entry_price,
|
||
sl = sl,
|
||
tp = tp,
|
||
)
|
||
trades.append(position)
|
||
|
||
return trades
|
||
|
||
|
||
# ── metrics ───────────────────────────────────────────────────────────────────
|
||
def print_metrics(trades: list[Trade], df: pd.DataFrame) -> None:
|
||
closed = [t for t in trades if t.closed]
|
||
if not closed:
|
||
print("No closed trades.")
|
||
return
|
||
|
||
pips = [t.pnl_pips for t in closed]
|
||
wins = [p for p in pips if p > 0]
|
||
losses = [p for p in pips if p <= 0]
|
||
|
||
gross_profit = sum(wins) if wins else 0
|
||
gross_loss = abs(sum(losses)) if losses else 0
|
||
profit_factor = gross_profit / gross_loss if gross_loss else float("inf")
|
||
|
||
# equity curve (pnl_pct accumulates on balance)
|
||
balance = INITIAL_BALANCE
|
||
equity = [balance]
|
||
for t in closed:
|
||
balance += balance * (t.pnl_pct / 100)
|
||
equity.append(balance)
|
||
equity = np.array(equity)
|
||
peak = np.maximum.accumulate(equity)
|
||
dd = (equity - peak) / peak * 100
|
||
max_dd = dd.min()
|
||
|
||
# annualised Sharpe (daily pnl_pct, then annualise √252)
|
||
# group trades by calendar day, sum pnl_pct per day
|
||
by_day: dict[str, float] = {}
|
||
for t in closed:
|
||
day = t.exit_time.date().isoformat() # type: ignore[union-attr]
|
||
by_day[day] = by_day.get(day, 0) + t.pnl_pct
|
||
daily = np.array(list(by_day.values()))
|
||
sharpe = (daily.mean() / daily.std() * np.sqrt(252)) if daily.std() > 0 else 0
|
||
|
||
open_trades = [t for t in trades if not t.closed]
|
||
|
||
total_pips = sum(pips)
|
||
total_return = (equity[-1] / INITIAL_BALANCE - 1) * 100
|
||
avg_win = np.mean(wins) if wins else 0
|
||
avg_loss = np.mean(losses) if losses else 0
|
||
win_rate = len(wins) / len(closed) * 100
|
||
|
||
# trade duration
|
||
durations = [(t.exit_time - t.entry_time).total_seconds() / 3600
|
||
for t in closed]
|
||
|
||
longs = [t for t in closed if t.direction == "long"]
|
||
shorts = [t for t in closed if t.direction == "short"]
|
||
|
||
print("=" * 60)
|
||
print(" RSI Mean Reversion — EURJPY M15 Backtest Results")
|
||
print("=" * 60)
|
||
data_start = df.index[0].strftime("%Y-%m-%d")
|
||
data_end = df.index[-1].strftime("%Y-%m-%d")
|
||
print(f" Period : {data_start} → {data_end}")
|
||
print(f" Candles : {len(df):,}")
|
||
print(f" Initial balance : ${INITIAL_BALANCE:,.0f}")
|
||
print()
|
||
print("── Trade Summary ─────────────────────────────────────────")
|
||
print(f" Total trades : {len(closed)}")
|
||
print(f" Open (unresolved): {len(open_trades)}")
|
||
print(f" Longs : {len(longs)}")
|
||
print(f" Shorts : {len(shorts)}")
|
||
print(f" Wins : {len(wins)}")
|
||
print(f" Losses : {len(losses)}")
|
||
print(f" Win rate : {win_rate:.1f}%")
|
||
print()
|
||
print("── P&L ───────────────────────────────────────────────────")
|
||
print(f" Total pips : {total_pips:+.1f}")
|
||
print(f" Total return : {total_return:+.2f}%")
|
||
print(f" Final balance : ${equity[-1]:,.2f}")
|
||
print(f" Avg win (pips) : {avg_win:+.1f}")
|
||
print(f" Avg loss (pips) : {avg_loss:+.1f}")
|
||
print(f" Reward/Risk : {abs(avg_win/avg_loss):.2f}" if avg_loss else " Reward/Risk : ∞")
|
||
print(f" Profit factor : {profit_factor:.2f}")
|
||
print()
|
||
print("── Risk ──────────────────────────────────────────────────")
|
||
print(f" Max drawdown : {max_dd:.2f}%")
|
||
print(f" Sharpe ratio : {sharpe:.2f}")
|
||
print(f" Risk/trade : {RISK_PER_TRADE*100:.1f}% of balance")
|
||
print()
|
||
print("── Timing ────────────────────────────────────────────────")
|
||
print(f" Avg duration : {np.mean(durations):.1f} hours")
|
||
print(f" Median duration : {np.median(durations):.1f} hours")
|
||
print()
|
||
print("── SL/TP Breakdown ───────────────────────────────────────")
|
||
by_exit: dict[str, int] = {}
|
||
for t in closed:
|
||
by_exit[t.exit_reason] = by_exit.get(t.exit_reason, 0) + 1
|
||
for reason, count in sorted(by_exit.items()):
|
||
print(f" {reason:5s} : {count} trades ({count/len(closed)*100:.1f}%)")
|
||
print("=" * 60)
|
||
|
||
# sample trades
|
||
print("\n── Last 10 closed trades ─────────────────────────────────")
|
||
print(f" {'Entry':19s} {'Exit':19s} {'Dir':5s} {'Pips':>8s} {'Reason':6s}")
|
||
for t in closed[-10:]:
|
||
print(f" {str(t.entry_time):19s} {str(t.exit_time):19s} "
|
||
f"{t.direction:5s} {t.pnl_pips:+8.1f} {t.exit_reason}")
|
||
|
||
|
||
# ── main ──────────────────────────────────────────────────────────────────────
|
||
if __name__ == "__main__":
|
||
print(f"Loading {SYMBOL} {TIMEFRAME} candles...")
|
||
df = load_candles()
|
||
print(f" {len(df):,} candles "
|
||
f"({df.index[0].date()} → {df.index[-1].date()})")
|
||
|
||
print("Running backtest...")
|
||
trades = run_backtest(df)
|
||
|
||
print_metrics(trades, df)
|