Add walk-forward backtest harness + paginated history seeder
- data_provider: new fetch_historical() that paginates /time_series in 5000-bar chunks (1 credit/chunk) for arbitrary date ranges. Tolerates the API's "no data on specified dates" boundary error gracefully when prior chunks succeeded — that error is just the marker that we walked past the symbol's earliest available bar. - seed_history.py: one-shot CLI that downloads OHLCV for the configured pairs/TFs into history/<PAIR>_<TF>.csv. Skips existing files unless --force. - backtest.py: walk-forward replay through SignalAggregator with strict no-look-ahead semantics. Each bar is only visible to the engine after it has fully closed (`df.index + interval <= as_of`). Trade simulation is first-touch (SL vs TP1) with worst-case (SL first if both touched same bar), spread cost deducted from R, max-hold timeout, no-overlap. Reports per-grade win-rate / avg-R / profit-factor / max-DD, plus optional matplotlib equity + by-grade plots. - requirements: add matplotlib (lazy-imported, only needed for --plot). - .gitignore: history/ and backtest_results/ regenerable, kept untracked. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Harwinder
@@ -15,3 +15,7 @@ venv/
|
||||
# OS
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
|
||||
# Backtest data + outputs (regenerable from seed_history.py / backtest.py)
|
||||
history/
|
||||
backtest_results/
|
||||
|
||||
+712
@@ -0,0 +1,712 @@
|
||||
#!/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)
|
||||
|
||||
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())
|
||||
@@ -456,6 +456,125 @@ class ForexDataProvider:
|
||||
f"Failed to fetch {td_symbol}: {e}"
|
||||
) from e
|
||||
|
||||
def fetch_historical(
|
||||
self,
|
||||
symbol: str,
|
||||
timeframe: Timeframe,
|
||||
start: datetime,
|
||||
end: datetime,
|
||||
) -> pd.DataFrame:
|
||||
"""
|
||||
Paginated historical OHLCV fetch for backtesting seed.
|
||||
|
||||
Walks backward from `end` to `start` in 5000-bar chunks (the free-tier
|
||||
per-call max), 1 credit per chunk. Honours the shared rate limiter so
|
||||
it cooperates with any concurrent live scans.
|
||||
|
||||
Returns a single concatenated DataFrame indexed by UTC timestamp.
|
||||
"""
|
||||
if not self._api_key:
|
||||
raise DataUnavailableError("No Twelve Data API key set.")
|
||||
|
||||
if start.tzinfo is None:
|
||||
start = start.replace(tzinfo=timezone.utc)
|
||||
if end.tzinfo is None:
|
||||
end = end.replace(tzinfo=timezone.utc)
|
||||
|
||||
bars_per_day = self.BARS_PER_DAY[timeframe]
|
||||
# Stay well clear of the 5000-bar cap — leave a small margin for
|
||||
# weekend gaps that don't decrement bars on calendar-day arithmetic.
|
||||
chunk_days = max(1, int(4500 / bars_per_day))
|
||||
|
||||
td_symbol = self._to_td_symbol(symbol)
|
||||
interval = self.TD_INTERVALS[timeframe]
|
||||
|
||||
chunks: List[pd.DataFrame] = []
|
||||
cursor_end = end
|
||||
|
||||
while cursor_end > start:
|
||||
cursor_start = max(start, cursor_end - timedelta(days=chunk_days))
|
||||
|
||||
self._throttle(credits=1)
|
||||
|
||||
resp = self._session.get(
|
||||
f"{self.TD_BASE_URL}/time_series",
|
||||
params={
|
||||
"symbol": td_symbol,
|
||||
"interval": interval,
|
||||
"start_date": cursor_start.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"end_date": cursor_end.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"outputsize": 5000,
|
||||
"timezone": "UTC",
|
||||
"format": "json",
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
if isinstance(data, dict) and data.get("status") == "error":
|
||||
msg = data.get("message", "unknown")
|
||||
# If we already have chunks, an error on a deeper-history call
|
||||
# almost always means "you've walked past the available start."
|
||||
# Treat as graceful end-of-history instead of fatal.
|
||||
if chunks:
|
||||
logger.info(
|
||||
f" {td_symbol} {interval}: end of available history "
|
||||
f"reached ({msg})"
|
||||
)
|
||||
break
|
||||
raise DataUnavailableError(
|
||||
f"Twelve Data error for {td_symbol}: {msg}"
|
||||
)
|
||||
|
||||
values = data.get("values") or []
|
||||
if not values:
|
||||
logger.info(
|
||||
f" {td_symbol} {interval}: empty range "
|
||||
f"{cursor_start.date()}..{cursor_end.date()}, stopping"
|
||||
)
|
||||
break
|
||||
|
||||
records = [
|
||||
{
|
||||
"Datetime": pd.Timestamp(v["datetime"], tz="UTC"),
|
||||
"Open": float(v["open"]),
|
||||
"High": float(v["high"]),
|
||||
"Low": float(v["low"]),
|
||||
"Close": float(v["close"]),
|
||||
"Volume": float(v.get("volume") or 0),
|
||||
}
|
||||
for v in values if v.get("datetime")
|
||||
]
|
||||
df_chunk = pd.DataFrame(records).set_index("Datetime").sort_index()
|
||||
df_chunk.index.name = "Datetime"
|
||||
chunks.append(df_chunk)
|
||||
|
||||
earliest = df_chunk.index.min()
|
||||
logger.info(
|
||||
f" {td_symbol} {interval}: chunk "
|
||||
f"{earliest.date()} → {df_chunk.index.max().date()} "
|
||||
f"({len(df_chunk):,} bars)"
|
||||
)
|
||||
|
||||
# Stop when:
|
||||
# 1. The current chunk's start window already hit the requested
|
||||
# `start` boundary — nothing earlier was asked for.
|
||||
# 2. The chunk is unusually thin (< 100 bars), suggesting we've
|
||||
# reached the edge of the symbol's available history.
|
||||
if cursor_start <= start or len(df_chunk) < 100:
|
||||
break
|
||||
|
||||
cursor_end = earliest - timedelta(seconds=1)
|
||||
|
||||
if not chunks:
|
||||
raise DataUnavailableError(f"No historical data returned for {td_symbol}")
|
||||
|
||||
df = pd.concat(chunks).sort_index()
|
||||
df = df[~df.index.duplicated(keep="last")]
|
||||
df = df[(df.index >= start) & (df.index <= end)]
|
||||
return df
|
||||
|
||||
def _validate_and_clean(
|
||||
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
||||
) -> pd.DataFrame:
|
||||
|
||||
@@ -11,3 +11,6 @@ python-dotenv>=1.0 # loads GEMINI_API_KEY and TELEGRAM_BOT_TOKEN from .env
|
||||
|
||||
# Telegram bot
|
||||
python-telegram-bot[job-queue]>=20.0
|
||||
|
||||
# Backtest analysis (only needed for `python backtest.py --plot`)
|
||||
matplotlib>=3.7
|
||||
|
||||
+133
@@ -0,0 +1,133 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
One-shot historical OHLCV seeder for backtesting.
|
||||
|
||||
Downloads bars from Twelve Data for the configured pairs/timeframes,
|
||||
paginated to stay within the 5000-bar per-call limit, and writes each
|
||||
(symbol, timeframe) to history/<PAIR>_<TF>.csv.
|
||||
|
||||
Usage:
|
||||
python seed_history.py --years 1
|
||||
python seed_history.py --years 2 --pairs EURUSD GBPUSD USDJPY
|
||||
python seed_history.py --tfs M15 H1 D1 --years 1 --force
|
||||
|
||||
Cost (free tier, 800 credits/day):
|
||||
1 year M15 + H1 + D1 × 7 majors ≈ 56 credits
|
||||
5 years same ≈ 231 credits
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
from config import MAJOR_PAIRS, PlatformConfig, Timeframe
|
||||
from data_provider import DataUnavailableError, ForexDataProvider
|
||||
|
||||
DEFAULT_TFS = ["M15", "H1", "D1"]
|
||||
HISTORY_DIR = Path("history")
|
||||
|
||||
_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,
|
||||
}
|
||||
|
||||
|
||||
def csv_path(out_dir: Path, symbol: str, tf: Timeframe) -> Path:
|
||||
return out_dir / f"{symbol.replace('=X', '')}_{tf.value}.csv"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Seed historical OHLCV for backtesting (Twelve Data, free tier).",
|
||||
)
|
||||
parser.add_argument("--years", type=float, default=1.0,
|
||||
help="Years of history to download (default 1.0)")
|
||||
parser.add_argument("--pairs", nargs="+", default=None,
|
||||
help="Pair symbols, e.g. EURUSD GBPUSD (default all 7 majors)")
|
||||
parser.add_argument("--tfs", nargs="+", default=DEFAULT_TFS,
|
||||
help=f"Timeframes (default {' '.join(DEFAULT_TFS)})")
|
||||
parser.add_argument("--out", default=str(HISTORY_DIR),
|
||||
help="Output directory (default history/)")
|
||||
parser.add_argument("--force", action="store_true",
|
||||
help="Re-download even if a CSV already exists")
|
||||
args = parser.parse_args()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
datefmt="%H:%M:%S",
|
||||
)
|
||||
|
||||
out_dir = Path(args.out)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
pairs = args.pairs if args.pairs else MAJOR_PAIRS
|
||||
pairs = [p if p.endswith("=X") else f"{p}=X" for p in pairs]
|
||||
|
||||
try:
|
||||
tfs = [_TF_MAP[t.upper()] for t in args.tfs]
|
||||
except KeyError as e:
|
||||
print(f"❌ Unknown timeframe: {e}. Valid: {list(_TF_MAP.keys())}")
|
||||
return 2
|
||||
|
||||
end = datetime.now(timezone.utc)
|
||||
start = end - timedelta(days=int(args.years * 365))
|
||||
|
||||
provider = ForexDataProvider(PlatformConfig())
|
||||
|
||||
total = len(pairs) * len(tfs)
|
||||
print(f"📥 Seeding {total} pair × tf combinations from "
|
||||
f"{start.date()} to {end.date()} ({args.years} year(s))")
|
||||
print(f" Pairs: {[p.replace('=X', '') for p in pairs]}")
|
||||
print(f" TFs: {[tf.value for tf in tfs]}")
|
||||
print(f" Output: {out_dir}/\n")
|
||||
|
||||
done = 0
|
||||
failed: list = []
|
||||
for symbol in pairs:
|
||||
for tf in tfs:
|
||||
done += 1
|
||||
out_path = csv_path(out_dir, symbol, tf)
|
||||
|
||||
if out_path.exists() and not args.force:
|
||||
print(f"[{done}/{total}] ⏭ {symbol} {tf.value} — exists, "
|
||||
f"skipping (--force to overwrite)")
|
||||
continue
|
||||
|
||||
print(f"[{done}/{total}] 📥 {symbol} {tf.value}…", flush=True)
|
||||
try:
|
||||
df = provider.fetch_historical(symbol, tf, start, end)
|
||||
df.to_csv(out_path)
|
||||
print(f" ✅ {len(df):,} bars "
|
||||
f"({df.index.min().date()} → {df.index.max().date()}) "
|
||||
f"→ {out_path}")
|
||||
except DataUnavailableError as e:
|
||||
print(f" ❌ {e}")
|
||||
failed.append((symbol, tf.value, str(e)))
|
||||
except Exception as e:
|
||||
print(f" ❌ unexpected: {e}")
|
||||
failed.append((symbol, tf.value, str(e)))
|
||||
|
||||
print()
|
||||
if failed:
|
||||
print(f"⚠️ {len(failed)} fetch(es) failed:")
|
||||
for sym, tfn, msg in failed:
|
||||
print(f" {sym} {tfn}: {msg}")
|
||||
return 1
|
||||
|
||||
print("✅ Seeding complete. Run: python backtest.py --primary M15")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user