225 lines
8.8 KiB
Python
225 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Main backtest runner.
|
|
|
|
Steps
|
|
-----
|
|
1. Load EURJPY M15 data (2 years).
|
|
2. Run baseline RSI Mean Reversion (original strategy, no trend filter).
|
|
3. Run improved RSI Trend strategy with default params.
|
|
4. Grid-search optimise on in-sample (first 70%).
|
|
5. Walk-forward: run best params on out-of-sample (last 30%).
|
|
6. Print full reports and save equity-curve charts.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
import numpy as np
|
|
|
|
from engine.backtest import BacktestEngine
|
|
from engine.data import load_candles
|
|
from engine.metrics import compute, Metrics
|
|
from engine.optimizer import grid_search
|
|
from strategies.rsi_trend import RSITrendStrategy
|
|
|
|
try:
|
|
from reports.chart import save_equity_chart
|
|
CHARTS = True
|
|
except ImportError:
|
|
CHARTS = False
|
|
|
|
# ── config ────────────────────────────────────────────────────────────────────
|
|
|
|
SYMBOL = "EURJPY"
|
|
TIMEFRAME = "M15"
|
|
INITIAL_BALANCE = 10_000.0
|
|
RISK_PER_TRADE = 0.01
|
|
SPREAD_PIPS = 0.5
|
|
SPLIT_RATIO = 0.70 # in-sample fraction
|
|
|
|
# ── parameter grid for optimisation ──────────────────────────────────────────
|
|
|
|
PARAM_GRID = {
|
|
"os_level": [25, 30, 35],
|
|
"ob_level": [65, 70, 75],
|
|
"sl_atr": [1.0, 1.5, 2.0],
|
|
"tp_atr": [2.0, 3.0, 4.0],
|
|
"trend_fast": [50],
|
|
"trend_slow": [200],
|
|
"session_start": [7],
|
|
"session_end": [20],
|
|
}
|
|
|
|
|
|
# ── helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def divider(title: str = "") -> None:
|
|
if title:
|
|
pad = (60 - len(title) - 2) // 2
|
|
print(f"\n{'─'*pad} {title} {'─'*pad}")
|
|
else:
|
|
print("─" * 62)
|
|
|
|
|
|
def print_report(label: str, metrics: Metrics, period: str = "") -> None:
|
|
divider(label)
|
|
if period:
|
|
print(f" Period : {period}")
|
|
print(metrics)
|
|
|
|
|
|
# ── main ──────────────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None:
|
|
print("=" * 62)
|
|
print(" NewBot — RSI Trend Pullback | EURJPY M15")
|
|
print("=" * 62)
|
|
|
|
# 1. Load data
|
|
print(f"\nLoading {SYMBOL} {TIMEFRAME}…")
|
|
df = load_candles(SYMBOL, TIMEFRAME)
|
|
print(f" {len(df):,} candles ({df.index[0].date()} → {df.index[-1].date()})")
|
|
|
|
split_idx = int(len(df) * SPLIT_RATIO)
|
|
df_in = df.iloc[:split_idx]
|
|
df_out = df.iloc[split_idx:]
|
|
print(f" In-sample : {df_in.index[0].date()} → {df_in.index[-1].date()} ({len(df_in):,} bars)")
|
|
print(f" Out-sample : {df_out.index[0].date()} → {df_out.index[-1].date()} ({len(df_out):,} bars)")
|
|
|
|
engine = BacktestEngine(
|
|
SYMBOL,
|
|
initial_balance = INITIAL_BALANCE,
|
|
risk_per_trade = RISK_PER_TRADE,
|
|
spread_pips = SPREAD_PIPS,
|
|
)
|
|
|
|
# 2. Baseline: original RSI mean-reversion (no trend filter)
|
|
# Simulated by using trend_fast=1, trend_slow=1 so trend is always 0 —
|
|
# that short-circuits the trend guard. Instead we reuse the same class
|
|
# but force trend filter OFF by setting trend EMAs equal (trend ≈ 0 always).
|
|
# Simpler: use a tiny fast period so the two EMAs are always close → trend~0
|
|
# Actually the cleanest baseline: subclass and override signal to ignore trend.
|
|
|
|
class BaselineRSI(RSITrendStrategy):
|
|
"""RSI mean reversion without any trend filter — take both longs & shorts."""
|
|
def signal(self, df, i):
|
|
if i < 1:
|
|
return None
|
|
row = df.iloc[i]
|
|
prev = df.iloc[i - 1]
|
|
if any(df.iloc[i][c] != df.iloc[i][c] for c in ("rsi", "atr")):
|
|
return None
|
|
hour = row.name.hour
|
|
if not (self.session_start <= hour < self.session_end):
|
|
return None
|
|
rsi_now, rsi_prev = row["rsi"], prev["rsi"]
|
|
if __import__("pandas").isna(rsi_prev):
|
|
return None
|
|
atr = row["atr"]
|
|
ref = row["close"]
|
|
if rsi_prev < self.os_level and rsi_now >= self.os_level:
|
|
return "long", ref - self.sl_atr * atr, ref + self.tp_atr * atr
|
|
if rsi_prev > self.ob_level and rsi_now <= self.ob_level:
|
|
return "short", ref + self.sl_atr * atr, ref - self.tp_atr * atr
|
|
return None
|
|
|
|
baseline = BaselineRSI(os_level=30, ob_level=70, sl_atr=1.5, tp_atr=3.0)
|
|
bl_trades, bl_equity = engine.run(df, baseline)
|
|
bl_metrics = compute(bl_trades, bl_equity, INITIAL_BALANCE)
|
|
print_report("Baseline RSI (no trend filter) — full dataset",
|
|
bl_metrics,
|
|
f"{df.index[0].date()} → {df.index[-1].date()}")
|
|
|
|
# 3. Improved strategy — default params, full dataset
|
|
improved = RSITrendStrategy(os_level=30, ob_level=70, sl_atr=1.5, tp_atr=3.0)
|
|
im_trades, im_equity = engine.run(df, improved)
|
|
im_metrics = compute(im_trades, im_equity, INITIAL_BALANCE)
|
|
print_report("Improved RSI Trend (default params) — full dataset",
|
|
im_metrics,
|
|
f"{df.index[0].date()} → {df.index[-1].date()}")
|
|
|
|
# 4. Optimise on in-sample
|
|
divider("Grid Search (in-sample optimisation)")
|
|
n_combos = 1
|
|
for v in PARAM_GRID.values():
|
|
n_combos *= len(v)
|
|
print(f" Testing {n_combos} parameter combinations on in-sample data…")
|
|
|
|
results = grid_search(
|
|
RSITrendStrategy, PARAM_GRID, df_in, engine, verbose=True
|
|
)
|
|
best = results[0]
|
|
print(f"\n Best score : {best.score:.4f}")
|
|
print(f" Params : {best.params}")
|
|
print(f"\n In-sample metrics:")
|
|
print(best.metrics)
|
|
|
|
print("\n Top-5 combinations:")
|
|
print(f" {'Score':>8} {'WR%':>5} {'PF':>5} {'Sharpe':>6} {'Trades':>6} Params")
|
|
for r in results[:5]:
|
|
print(f" {r.score:8.4f} {r.metrics.win_rate*100:5.1f} "
|
|
f"{r.metrics.profit_factor:5.2f} {r.metrics.sharpe:6.2f} "
|
|
f"{r.metrics.n_trades:6d} {r.params}")
|
|
|
|
# 5. Walk-forward: run best params on out-of-sample
|
|
opt_strategy = RSITrendStrategy(**best.params)
|
|
oos_trades, oos_equity = engine.run(df_out, opt_strategy)
|
|
oos_metrics = compute(oos_trades, oos_equity, INITIAL_BALANCE)
|
|
print_report("Walk-Forward — out-of-sample (best params)",
|
|
oos_metrics,
|
|
f"{df_out.index[0].date()} → {df_out.index[-1].date()}")
|
|
|
|
# 6. Also run best params on full dataset for the final picture
|
|
full_strategy = RSITrendStrategy(**best.params)
|
|
full_trades, full_equity = engine.run(df, full_strategy)
|
|
full_metrics = compute(full_trades, full_equity, INITIAL_BALANCE)
|
|
print_report("Optimised Strategy — full dataset",
|
|
full_metrics,
|
|
f"{df.index[0].date()} → {df.index[-1].date()}")
|
|
|
|
# 7. Summary comparison table
|
|
divider("Strategy Comparison Summary")
|
|
print(f" {'Strategy':<35} {'Trades':>6} {'WR%':>5} {'PF':>5} {'Sharpe':>6} {'Return%':>7} {'MDD%':>6}")
|
|
rows = [
|
|
("Baseline RSI (no filter)", bl_metrics),
|
|
("RSI Trend (default params)", im_metrics),
|
|
(f"RSI Trend (optimised, in-sample)", best.metrics),
|
|
(f"RSI Trend (optimised, out-sample)", oos_metrics),
|
|
(f"RSI Trend (optimised, full data)", full_metrics),
|
|
]
|
|
for name, m in rows:
|
|
print(f" {name:<35} {m.n_trades:>6} {m.win_rate*100:>5.1f} "
|
|
f"{m.profit_factor:>5.2f} {m.sharpe:>6.2f} "
|
|
f"{m.total_return*100:>+7.2f} {m.max_drawdown*100:>6.2f}")
|
|
|
|
# 8. Charts
|
|
if CHARTS:
|
|
divider("Saving charts")
|
|
charts_dir = Path(__file__).parent / "reports"
|
|
charts_dir.mkdir(exist_ok=True)
|
|
|
|
paths = []
|
|
pairs = [
|
|
(bl_equity, bl_trades, bl_metrics, "Baseline RSI (no filter)", "baseline.png"),
|
|
(im_equity, im_trades, im_metrics, "RSI Trend — default params", "improved_default.png"),
|
|
(full_equity, full_trades, full_metrics, "RSI Trend — optimised (full data)","optimised_full.png"),
|
|
(oos_equity, oos_trades, oos_metrics, "RSI Trend — out-of-sample (WF)", "walkforward_oos.png"),
|
|
]
|
|
for eq, tr, me, lbl, fname in pairs:
|
|
p = save_equity_chart(eq, tr, me, lbl, fname)
|
|
paths.append(p)
|
|
print(f" Saved: {p}")
|
|
else:
|
|
print("\n (matplotlib not available — skipping charts)")
|
|
|
|
print("\n" + "=" * 62)
|
|
print(" Done.")
|
|
print("=" * 62)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|