Harwinder
6524238ce9
write_text() defaults to the platform encoding (cp1252 on Windows), which can't encode the arrow character used in the summary header. Force UTF-8 explicitly so backtests complete on Windows hosts. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
713 lines
26 KiB
Python
713 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Walk-forward backtest harness for the Forex signal engine.
|
||
|
||
Replays historical OHLCV (loaded from history/<PAIR>_<TF>.csv) through
|
||
the existing SignalAggregator with strict no-look-ahead semantics: at
|
||
each primary-TF bar close, the engine sees only bars that have *fully
|
||
closed by that moment* — never the bar still in formation.
|
||
|
||
Trade simulation: first-touch (SL vs TP1) on the primary timeframe,
|
||
worst-case if both touched in the same bar (assume SL first), with
|
||
per-pip spread cost deducted from the final R-multiple.
|
||
|
||
Outputs:
|
||
backtest_results/trades.csv — every simulated trade
|
||
backtest_results/summary.txt — per-grade win rate / avg R / PF
|
||
backtest_results/equity.png — cumulative R curve (if --plot)
|
||
backtest_results/by_grade.png — win rate by grade (if --plot)
|
||
|
||
Usage:
|
||
python backtest.py --primary M15 --pairs EURUSD
|
||
python backtest.py --primary H1 --pairs EURUSD GBPUSD USDJPY --plot
|
||
python backtest.py --primary M15 --grades A+ A
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import logging
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
import pandas as pd
|
||
|
||
from config import PlatformConfig, SignalStrength, Timeframe, MAJOR_PAIRS
|
||
from data_provider import DataUnavailableError, ForexDataProvider
|
||
from signal_aggregator import SignalAggregator, SignalReport
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
HISTORY_DIR = Path("history")
|
||
RESULTS_DIR = Path("backtest_results")
|
||
|
||
_TF_MAP = {
|
||
"M1": Timeframe.M1, "M5": Timeframe.M5,
|
||
"M15": Timeframe.M15, "M30": Timeframe.M30,
|
||
"H1": Timeframe.H1, "H4": Timeframe.H4,
|
||
"D1": Timeframe.D1, "W1": Timeframe.W1,
|
||
}
|
||
|
||
_TF_INTERVAL = {
|
||
Timeframe.M1: pd.Timedelta(minutes=1),
|
||
Timeframe.M5: pd.Timedelta(minutes=5),
|
||
Timeframe.M15: pd.Timedelta(minutes=15),
|
||
Timeframe.M30: pd.Timedelta(minutes=30),
|
||
Timeframe.H1: pd.Timedelta(hours=1),
|
||
Timeframe.H4: pd.Timedelta(hours=4),
|
||
Timeframe.D1: pd.Timedelta(days=1),
|
||
Timeframe.W1: pd.Timedelta(weeks=1),
|
||
}
|
||
|
||
# Confirmation TFs per primary — same shape Telegram bot uses
|
||
_TF_CONFIRMATIONS: Dict[Timeframe, List[Timeframe]] = {
|
||
Timeframe.M15: [Timeframe.H1, Timeframe.D1],
|
||
Timeframe.M30: [Timeframe.H1, Timeframe.D1],
|
||
Timeframe.H1: [Timeframe.M15, Timeframe.D1],
|
||
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
|
||
Timeframe.D1: [Timeframe.H4],
|
||
}
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# Stub news engine (no historical news data — returns neutral)
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
class _StubFundamentalContext:
|
||
"""Placeholder — backtest doesn't replay news, so context is empty."""
|
||
pass
|
||
|
||
|
||
class StubNewsEngine:
|
||
"""
|
||
Drop-in for `NewsEngine` in backtest mode. Returns neutral sentiment
|
||
and 'clear' event risk for every call. The signal aggregator's news
|
||
adjustments thus become no-ops, isolating the test to pure technicals.
|
||
"""
|
||
|
||
def get_fundamental_context(self):
|
||
return _StubFundamentalContext()
|
||
|
||
def get_event_risk_level(self, symbol, context):
|
||
return ("clear", "")
|
||
|
||
def get_pair_sentiment(self, symbol, context):
|
||
return (0.0, "")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# Backtest data provider (slices history at "as of" timestamp)
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
class BacktestProvider(ForexDataProvider):
|
||
"""
|
||
Replaces network fetches with in-memory slices of preloaded OHLCV.
|
||
|
||
`set_time(ts)` sets the cut-off; `fetch_ohlcv` returns only bars whose
|
||
*close* time is ≤ ts. Bars that are still in formation (open ≤ ts but
|
||
close > ts) are excluded — this is the no-look-ahead guarantee.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
config: PlatformConfig,
|
||
history: Dict[Tuple[str, Timeframe], pd.DataFrame],
|
||
):
|
||
super().__init__(config)
|
||
self._history = history
|
||
self._current_ts: Optional[pd.Timestamp] = None
|
||
|
||
def set_time(self, ts: pd.Timestamp) -> None:
|
||
if ts.tzinfo is None:
|
||
ts = ts.tz_localize("UTC")
|
||
self._current_ts = ts
|
||
|
||
def fetch_ohlcv(
|
||
self,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
lookback_days: Optional[int] = None,
|
||
) -> pd.DataFrame:
|
||
df = self._history.get((symbol, timeframe))
|
||
if df is None:
|
||
raise DataUnavailableError(
|
||
f"No history loaded for {symbol} {timeframe.value}. "
|
||
f"Run seed_history.py with --tfs including this TF."
|
||
)
|
||
if self._current_ts is not None:
|
||
interval = _TF_INTERVAL[timeframe]
|
||
df = df[df.index + interval <= self._current_ts]
|
||
if df.empty:
|
||
raise DataUnavailableError(
|
||
f"No bars before {self._current_ts} for {symbol} {timeframe.value}"
|
||
)
|
||
return df
|
||
|
||
def prefetch_batch(self, symbols, timeframes, lookback_days=None):
|
||
return # no-op: data is in memory
|
||
|
||
def get_spread_estimate(self, symbol: str) -> float:
|
||
# Backtest doesn't need a live spread estimate; return a fixed value
|
||
# so the aggregator's spread field is filled but does not affect SL/TP.
|
||
return 0.0
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# History loader
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
def load_history(
|
||
pairs: List[str],
|
||
timeframes: List[Timeframe],
|
||
history_dir: Path = HISTORY_DIR,
|
||
) -> Dict[Tuple[str, Timeframe], pd.DataFrame]:
|
||
"""
|
||
Load CSVs produced by seed_history.py into a dict keyed by (symbol, tf).
|
||
|
||
Drops CSVs that are missing — caller decides whether to abort or continue
|
||
with reduced TF coverage.
|
||
"""
|
||
history: Dict[Tuple[str, Timeframe], pd.DataFrame] = {}
|
||
missing: List[str] = []
|
||
|
||
for symbol in pairs:
|
||
for tf in timeframes:
|
||
csv_p = history_dir / f"{symbol.replace('=X', '')}_{tf.value}.csv"
|
||
if not csv_p.exists():
|
||
missing.append(str(csv_p))
|
||
continue
|
||
df = pd.read_csv(csv_p, index_col="Datetime", parse_dates=["Datetime"])
|
||
if df.index.tz is None:
|
||
df.index = df.index.tz_localize("UTC")
|
||
history[(symbol, tf)] = df.sort_index()
|
||
logger.info(
|
||
f"loaded {symbol} {tf.value}: {len(df):,} bars "
|
||
f"({df.index.min().date()} → {df.index.max().date()})"
|
||
)
|
||
|
||
if missing:
|
||
logger.warning(f"missing CSVs ({len(missing)}): {missing[:5]}…")
|
||
|
||
return history
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# Trade record + first-touch simulator
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class Trade:
|
||
symbol: str
|
||
pair_display: str
|
||
primary_tf: str
|
||
grade: str
|
||
direction: str
|
||
entry_time: pd.Timestamp
|
||
entry_price: float
|
||
sl: float
|
||
tp1: float
|
||
tp2: float
|
||
risk_pips: float
|
||
reward_pips: float
|
||
score: float
|
||
confidence: float
|
||
regime: str
|
||
mtf_aligned: bool
|
||
|
||
# Filled by simulation
|
||
exit_time: Optional[pd.Timestamp] = None
|
||
exit_price: float = 0.0
|
||
outcome: str = "OPEN" # WIN_TP1, LOSS_SL, TIMEOUT, INVALID
|
||
bars_held: int = 0
|
||
r_multiple: float = 0.0 # net R after spread cost
|
||
|
||
|
||
def simulate_trade(
|
||
trade: Trade,
|
||
forward_bars: pd.DataFrame,
|
||
max_bars: int,
|
||
spread_pips: float,
|
||
) -> Trade:
|
||
"""
|
||
First-touch simulation on the primary timeframe.
|
||
|
||
Conservatives:
|
||
- If both SL and TP1 touch within the same bar, assume SL first.
|
||
- Spread cost (entry + exit round-trip) is deducted from final R.
|
||
- TP2 is recorded but not used in this baseline simulation — we cap at
|
||
TP1 to keep the R-multiple math clean. Extend with breakeven-trail
|
||
management later if needed.
|
||
"""
|
||
if trade.risk_pips <= 0:
|
||
trade.outcome = "INVALID"
|
||
return trade
|
||
|
||
spread_r = spread_pips / trade.risk_pips
|
||
bars_examined = 0
|
||
|
||
for ts, bar in forward_bars.iterrows():
|
||
bars_examined += 1
|
||
if bars_examined > max_bars:
|
||
break
|
||
|
||
bar_high = bar["High"]
|
||
bar_low = bar["Low"]
|
||
|
||
if trade.direction == "BUY":
|
||
sl_hit = bar_low <= trade.sl
|
||
tp_hit = bar_high >= trade.tp1
|
||
else:
|
||
sl_hit = bar_high >= trade.sl
|
||
tp_hit = bar_low <= trade.tp1
|
||
|
||
if sl_hit:
|
||
trade.outcome = "LOSS_SL"
|
||
trade.exit_time = ts
|
||
trade.exit_price = trade.sl
|
||
trade.bars_held = bars_examined
|
||
trade.r_multiple = -1.0 - spread_r
|
||
return trade
|
||
|
||
if tp_hit:
|
||
trade.outcome = "WIN_TP1"
|
||
trade.exit_time = ts
|
||
trade.exit_price = trade.tp1
|
||
trade.bars_held = bars_examined
|
||
# TP1 sits at risk × RR away from entry → R = reward_pips / risk_pips
|
||
trade.r_multiple = (trade.reward_pips / trade.risk_pips) - spread_r
|
||
return trade
|
||
|
||
# Timeout — exit at last seen close
|
||
if bars_examined == 0:
|
||
trade.outcome = "INVALID"
|
||
return trade
|
||
|
||
last_idx = bars_examined - 1
|
||
last = forward_bars.iloc[last_idx]
|
||
last_ts = forward_bars.index[last_idx]
|
||
|
||
if trade.direction == "BUY":
|
||
pnl_price = last["Close"] - trade.entry_price
|
||
else:
|
||
pnl_price = trade.entry_price - last["Close"]
|
||
|
||
sl_distance_price = abs(trade.entry_price - trade.sl)
|
||
pnl_r = pnl_price / sl_distance_price if sl_distance_price > 0 else 0.0
|
||
|
||
trade.outcome = "TIMEOUT"
|
||
trade.exit_time = last_ts
|
||
trade.exit_price = float(last["Close"])
|
||
trade.bars_held = bars_examined
|
||
trade.r_multiple = pnl_r - spread_r
|
||
return trade
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# Backtester (the replay loop)
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
class Backtester:
|
||
def __init__(
|
||
self,
|
||
primary_tf: Timeframe,
|
||
pairs: List[str],
|
||
history: Dict[Tuple[str, Timeframe], pd.DataFrame],
|
||
target_grades: List[str],
|
||
max_hold_bars: int = 100,
|
||
spread_pips: float = 1.0,
|
||
warmup_bars: int = 250,
|
||
):
|
||
self.primary_tf = primary_tf
|
||
self.pairs = pairs
|
||
self.history = history
|
||
self.target_grades = set(target_grades)
|
||
self.max_hold_bars = max_hold_bars
|
||
self.spread_pips = spread_pips
|
||
self.warmup_bars = warmup_bars
|
||
self.confirmation_tfs = _TF_CONFIRMATIONS.get(primary_tf, [])
|
||
|
||
def _build_config(self, symbol: str) -> PlatformConfig:
|
||
cfg = PlatformConfig()
|
||
cfg.primary_timeframe = self.primary_tf
|
||
cfg.confirmation_timeframes = list(self.confirmation_tfs)
|
||
cfg.watch_pairs = [symbol]
|
||
return cfg
|
||
|
||
def _make_aggregator(self, config: PlatformConfig, provider: BacktestProvider) -> SignalAggregator:
|
||
agg = SignalAggregator(config, data_provider=provider)
|
||
# Replace live news engine with neutral stub (no historical news data)
|
||
agg.news_engine = StubNewsEngine()
|
||
return agg
|
||
|
||
def run(self) -> List[Trade]:
|
||
trades: List[Trade] = []
|
||
provider = BacktestProvider(PlatformConfig(), self.history)
|
||
|
||
for symbol in self.pairs:
|
||
primary_df = self.history.get((symbol, self.primary_tf))
|
||
if primary_df is None or primary_df.empty:
|
||
logger.warning(f"{symbol}: no primary {self.primary_tf.value} data, skipping")
|
||
continue
|
||
|
||
config = self._build_config(symbol)
|
||
aggregator = self._make_aggregator(config, provider)
|
||
|
||
interval = _TF_INTERVAL[self.primary_tf]
|
||
timestamps = primary_df.index
|
||
n_total = len(timestamps)
|
||
|
||
print(f"\n📊 {symbol} {self.primary_tf.value}: "
|
||
f"{n_total:,} bars, warmup={self.warmup_bars}")
|
||
|
||
open_until: Optional[pd.Timestamp] = None # block overlapping entries
|
||
n_signals = 0
|
||
|
||
for i, bar_open_ts in enumerate(timestamps):
|
||
if i < self.warmup_bars:
|
||
continue
|
||
if open_until is not None and bar_open_ts < open_until:
|
||
continue # previous trade still open — skip
|
||
|
||
# Run engine "as if it's now this bar's close"
|
||
bar_close_ts = bar_open_ts + interval
|
||
provider.set_time(bar_close_ts)
|
||
|
||
try:
|
||
report = aggregator.analyze_pair(symbol)
|
||
except DataUnavailableError:
|
||
continue
|
||
except Exception as e:
|
||
logger.debug(f"engine error at {bar_open_ts}: {e}")
|
||
continue
|
||
|
||
if report.signal_quality not in self.target_grades:
|
||
continue
|
||
if report.trade_setup is None:
|
||
continue
|
||
if report.final_signal == SignalStrength.NEUTRAL:
|
||
continue
|
||
|
||
ts = report.trade_setup
|
||
trade = Trade(
|
||
symbol = symbol,
|
||
pair_display = report.display_name,
|
||
primary_tf = self.primary_tf.value,
|
||
grade = report.signal_quality,
|
||
direction = ts.direction,
|
||
entry_time = bar_open_ts,
|
||
entry_price = ts.entry_price,
|
||
sl = ts.stop_loss,
|
||
tp1 = ts.take_profit_1,
|
||
tp2 = ts.take_profit_2,
|
||
risk_pips = ts.risk_pips,
|
||
reward_pips = ts.reward_pips,
|
||
score = report.final_score,
|
||
confidence = report.final_confidence,
|
||
regime = report.market_regime,
|
||
mtf_aligned = report.mtf_agreement,
|
||
)
|
||
|
||
# Forward window for simulation
|
||
forward = primary_df.iloc[i + 1 : i + 1 + self.max_hold_bars]
|
||
if forward.empty:
|
||
continue
|
||
|
||
trade = simulate_trade(
|
||
trade,
|
||
forward,
|
||
max_bars=self.max_hold_bars,
|
||
spread_pips=self.spread_pips,
|
||
)
|
||
trades.append(trade)
|
||
n_signals += 1
|
||
|
||
if trade.exit_time is not None:
|
||
open_until = trade.exit_time
|
||
|
||
if n_signals % 25 == 0:
|
||
print(f" [{i:>6}/{n_total}] {n_signals} trades so far "
|
||
f"(latest: {trade.grade} {trade.direction} {trade.outcome} "
|
||
f"R={trade.r_multiple:+.2f})")
|
||
|
||
print(f" ✅ {symbol}: {n_signals} trades")
|
||
|
||
return trades
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# Reporting
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame:
|
||
rows = [
|
||
{
|
||
"entry_time": t.entry_time,
|
||
"exit_time": t.exit_time,
|
||
"symbol": t.symbol,
|
||
"pair": t.pair_display,
|
||
"tf": t.primary_tf,
|
||
"grade": t.grade,
|
||
"direction": t.direction,
|
||
"entry": t.entry_price,
|
||
"sl": t.sl,
|
||
"tp1": t.tp1,
|
||
"exit_price": t.exit_price,
|
||
"outcome": t.outcome,
|
||
"bars_held": t.bars_held,
|
||
"r_multiple": t.r_multiple,
|
||
"score": t.score,
|
||
"confidence": t.confidence,
|
||
"regime": t.regime,
|
||
"mtf_aligned": t.mtf_aligned,
|
||
"risk_pips": t.risk_pips,
|
||
"reward_pips": t.reward_pips,
|
||
}
|
||
for t in trades
|
||
]
|
||
return pd.DataFrame(rows)
|
||
|
||
|
||
def summarize(df: pd.DataFrame) -> str:
|
||
"""Per-grade win rate / average R / profit factor summary."""
|
||
if df.empty:
|
||
return "No trades recorded."
|
||
|
||
lines: List[str] = []
|
||
lines.append("=" * 78)
|
||
lines.append(f"BACKTEST SUMMARY ({len(df)} trades)")
|
||
lines.append(f" Date range: {df['entry_time'].min()} → {df['entry_time'].max()}")
|
||
lines.append("=" * 78)
|
||
lines.append("")
|
||
|
||
overall = _grade_block("ALL", df)
|
||
lines.append(overall)
|
||
lines.append("")
|
||
|
||
for grade in ["A+", "A", "B", "C", "D"]:
|
||
sub = df[df["grade"] == grade]
|
||
if sub.empty:
|
||
continue
|
||
lines.append(_grade_block(grade, sub))
|
||
lines.append("")
|
||
|
||
# Per-pair breakdown
|
||
lines.append("─" * 78)
|
||
lines.append("PER-PAIR SUMMARY")
|
||
lines.append("─" * 78)
|
||
by_pair = df.groupby("pair").agg(
|
||
trades=("r_multiple", "count"),
|
||
win_rate=("r_multiple", lambda s: (s > 0).mean() * 100),
|
||
avg_r=("r_multiple", "mean"),
|
||
total_r=("r_multiple", "sum"),
|
||
).round(2)
|
||
lines.append(by_pair.to_string())
|
||
lines.append("")
|
||
|
||
# Per-direction
|
||
lines.append("─" * 78)
|
||
lines.append("BY DIRECTION")
|
||
lines.append("─" * 78)
|
||
by_dir = df.groupby("direction").agg(
|
||
trades=("r_multiple", "count"),
|
||
win_rate=("r_multiple", lambda s: (s > 0).mean() * 100),
|
||
avg_r=("r_multiple", "mean"),
|
||
).round(2)
|
||
lines.append(by_dir.to_string())
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _grade_block(label: str, sub: pd.DataFrame) -> str:
|
||
n = len(sub)
|
||
wins = (sub["r_multiple"] > 0).sum()
|
||
losses = (sub["r_multiple"] <= 0).sum()
|
||
win_rate = wins / n * 100 if n else 0.0
|
||
avg_r = sub["r_multiple"].mean()
|
||
sum_r = sub["r_multiple"].sum()
|
||
|
||
gross_win = sub.loc[sub["r_multiple"] > 0, "r_multiple"].sum()
|
||
gross_loss = -sub.loc[sub["r_multiple"] <= 0, "r_multiple"].sum()
|
||
profit_factor = (gross_win / gross_loss) if gross_loss > 0 else float("inf")
|
||
|
||
# Equity-curve drawdown (in R)
|
||
cum = sub["r_multiple"].cumsum()
|
||
running_peak = cum.cummax()
|
||
drawdown = (cum - running_peak).min()
|
||
|
||
return (
|
||
f"{label:>4}: {n:>4} trades | "
|
||
f"win {win_rate:5.1f}% ({wins}W/{losses}L) | "
|
||
f"avg R {avg_r:+.3f} | total R {sum_r:+.2f} | "
|
||
f"PF {profit_factor:.2f} | maxDD {drawdown:.2f}R"
|
||
)
|
||
|
||
|
||
def plot_results(df: pd.DataFrame, out_dir: Path) -> None:
|
||
"""Generate equity curve and per-grade bar chart. Requires matplotlib."""
|
||
try:
|
||
import matplotlib
|
||
matplotlib.use("Agg") # headless
|
||
import matplotlib.pyplot as plt
|
||
except ImportError:
|
||
print("⚠️ matplotlib not installed — skipping plots. "
|
||
"pip install matplotlib")
|
||
return
|
||
|
||
if df.empty:
|
||
return
|
||
|
||
df = df.sort_values("entry_time").reset_index(drop=True)
|
||
df["cum_r"] = df["r_multiple"].cumsum()
|
||
|
||
# Equity curve
|
||
fig, ax = plt.subplots(figsize=(11, 5))
|
||
ax.plot(df["entry_time"], df["cum_r"], linewidth=1.5)
|
||
ax.axhline(0, color="gray", linewidth=0.5)
|
||
ax.fill_between(df["entry_time"], df["cum_r"], 0,
|
||
where=df["cum_r"] >= 0, alpha=0.2, color="green")
|
||
ax.fill_between(df["entry_time"], df["cum_r"], 0,
|
||
where=df["cum_r"] < 0, alpha=0.2, color="red")
|
||
ax.set_title(f"Cumulative R ({len(df)} trades, total {df['cum_r'].iloc[-1]:+.2f}R)")
|
||
ax.set_xlabel("Trade entry time")
|
||
ax.set_ylabel("Cumulative R (1R = risked amount per trade)")
|
||
ax.grid(True, alpha=0.3)
|
||
fig.tight_layout()
|
||
fig.savefig(out_dir / "equity.png", dpi=120)
|
||
plt.close(fig)
|
||
|
||
# Win rate by grade
|
||
grades = ["A+", "A", "B", "C", "D"]
|
||
grade_stats = []
|
||
for g in grades:
|
||
sub = df[df["grade"] == g]
|
||
if sub.empty:
|
||
continue
|
||
wr = (sub["r_multiple"] > 0).mean() * 100
|
||
avg_r = sub["r_multiple"].mean()
|
||
grade_stats.append((g, len(sub), wr, avg_r))
|
||
|
||
if grade_stats:
|
||
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 4.5))
|
||
labels = [f"{g}\n(n={n})" for g, n, _, _ in grade_stats]
|
||
wrs = [w for _, _, w, _ in grade_stats]
|
||
avgs = [r for _, _, _, r in grade_stats]
|
||
|
||
ax1.bar(labels, wrs, color=["#16a34a", "#22c55e", "#84cc16", "#eab308", "#f97316"][:len(labels)])
|
||
ax1.axhline(50, linestyle="--", color="gray", linewidth=0.8)
|
||
ax1.set_ylabel("Win rate (%)")
|
||
ax1.set_title("Win rate by grade")
|
||
for i, w in enumerate(wrs):
|
||
ax1.text(i, w + 1, f"{w:.0f}%", ha="center", fontsize=9)
|
||
|
||
ax2.bar(labels, avgs, color=["#16a34a", "#22c55e", "#84cc16", "#eab308", "#f97316"][:len(labels)])
|
||
ax2.axhline(0, color="black", linewidth=0.5)
|
||
ax2.set_ylabel("Average R per trade")
|
||
ax2.set_title("Average R by grade")
|
||
for i, a in enumerate(avgs):
|
||
ax2.text(i, a + (0.02 if a >= 0 else -0.05), f"{a:+.2f}",
|
||
ha="center", fontsize=9)
|
||
|
||
fig.tight_layout()
|
||
fig.savefig(out_dir / "by_grade.png", dpi=120)
|
||
plt.close(fig)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# CLI
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(
|
||
description="Walk-forward backtest of the Forex signal engine.",
|
||
)
|
||
parser.add_argument("--primary", default="M15",
|
||
help="Primary timeframe (default M15)")
|
||
parser.add_argument("--pairs", nargs="+", default=None,
|
||
help="Pairs to test (default: all majors)")
|
||
parser.add_argument("--grades", nargs="+", default=["A+", "A"],
|
||
help='Grades to take (default "A+ A")')
|
||
parser.add_argument("--history-dir", default=str(HISTORY_DIR))
|
||
parser.add_argument("--out-dir", default=str(RESULTS_DIR))
|
||
parser.add_argument("--max-hold-bars", type=int, default=100,
|
||
help="Max bars to hold a trade before timeout (default 100)")
|
||
parser.add_argument("--spread-pips", type=float, default=1.0,
|
||
help="Round-trip spread cost in pips (default 1.0)")
|
||
parser.add_argument("--warmup-bars", type=int, default=250,
|
||
help="Bars to skip at start so indicators stabilize (default 250)")
|
||
parser.add_argument("--plot", action="store_true",
|
||
help="Generate equity / by-grade PNGs (needs matplotlib)")
|
||
parser.add_argument("-v", "--verbose", action="store_true")
|
||
args = parser.parse_args()
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO if args.verbose else logging.WARNING,
|
||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
|
||
primary_tf = _TF_MAP[args.primary.upper()]
|
||
confirmations = _TF_CONFIRMATIONS.get(primary_tf, [])
|
||
needed_tfs = [primary_tf] + confirmations
|
||
|
||
pairs = args.pairs if args.pairs else MAJOR_PAIRS
|
||
pairs = [p if p.endswith("=X") else f"{p}=X" for p in pairs]
|
||
|
||
history_dir = Path(args.history_dir)
|
||
out_dir = Path(args.out_dir)
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
print(f"🔬 Backtest: primary={primary_tf.value} "
|
||
f"confirmations={[t.value for t in confirmations]}")
|
||
print(f" Pairs: {[p.replace('=X','') for p in pairs]}")
|
||
print(f" Grades: {args.grades}")
|
||
print(f" History dir: {history_dir}/")
|
||
print(f" Results dir: {out_dir}/")
|
||
print()
|
||
|
||
history = load_history(pairs, needed_tfs, history_dir=history_dir)
|
||
if not history:
|
||
print(f"❌ No history loaded from {history_dir}/. "
|
||
f"Run: python seed_history.py --tfs {' '.join(t.value for t in needed_tfs)}")
|
||
return 1
|
||
|
||
bt = Backtester(
|
||
primary_tf = primary_tf,
|
||
pairs = pairs,
|
||
history = history,
|
||
target_grades = args.grades,
|
||
max_hold_bars = args.max_hold_bars,
|
||
spread_pips = args.spread_pips,
|
||
warmup_bars = args.warmup_bars,
|
||
)
|
||
|
||
trades = bt.run()
|
||
|
||
if not trades:
|
||
print("\n❌ No trades emitted. Try lowering --grades (e.g., 'B') or "
|
||
"check that primary+confirmation TFs are in history/.")
|
||
return 0
|
||
|
||
df = trades_to_dataframe(trades)
|
||
df.to_csv(out_dir / "trades.csv", index=False)
|
||
|
||
summary = summarize(df)
|
||
(out_dir / "summary.txt").write_text(summary, encoding="utf-8")
|
||
|
||
print("\n" + summary)
|
||
print(f"\n📁 trades.csv → {out_dir/'trades.csv'}")
|
||
print(f"📁 summary.txt → {out_dir/'summary.txt'}")
|
||
|
||
if args.plot:
|
||
plot_results(df, out_dir)
|
||
print(f"📈 equity.png → {out_dir/'equity.png'}")
|
||
print(f"📈 by_grade.png → {out_dir/'by_grade.png'}")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|