From 6d235061b9783d69261bfb8031e08592f8928f6d Mon Sep 17 00:00:00 2001 From: Harwinder Date: Wed, 6 May 2026 13:29:48 -0400 Subject: [PATCH] Add walk-forward backtest harness + paginated history seeder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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/_.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 --- .gitignore | 4 + backtest.py | 712 +++++++++++++++++++++++++++++++++++++++++++++++ data_provider.py | 119 ++++++++ requirements.txt | 3 + seed_history.py | 133 +++++++++ 5 files changed, 971 insertions(+) create mode 100644 backtest.py create mode 100644 seed_history.py diff --git a/.gitignore b/.gitignore index 8e692ec..aa79bfb 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,7 @@ venv/ # OS .DS_Store Thumbs.db + +# Backtest data + outputs (regenerable from seed_history.py / backtest.py) +history/ +backtest_results/ diff --git a/backtest.py b/backtest.py new file mode 100644 index 0000000..0299028 --- /dev/null +++ b/backtest.py @@ -0,0 +1,712 @@ +#!/usr/bin/env python3 +""" +Walk-forward backtest harness for the Forex signal engine. + +Replays historical OHLCV (loaded from history/_.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()) diff --git a/data_provider.py b/data_provider.py index 97677ae..9bc6b35 100644 --- a/data_provider.py +++ b/data_provider.py @@ -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: diff --git a/requirements.txt b/requirements.txt index dd22b83..bed9fac 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/seed_history.py b/seed_history.py new file mode 100644 index 0000000..3cbfac5 --- /dev/null +++ b/seed_history.py @@ -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/_.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())