""" ForexBot2 — GPU-Accelerated Backtester ====================================== Replays historical OHLCV bar-by-bar through the live SignalAggregator so backtest results are identical to what the production scanner would have generated. Pre-computes all indicator series with CuPy/NumPy on the full dataset (causal, zero look-ahead) and looks up values per bar instead of recomputing each time — giving a 100x+ speedup over a naive replay while preserving the exact signal logic. Key properties: - Uses real SignalAggregator.analyze_from_dataframes (same gates, same XAU strategy hooks, same trade setup math). - News engine disabled (use_news=False) — backtests are pure technical. - Session filter, MTF agreement, regime classification all flow from the bar timestamp, never datetime.now(). - XAU weekend/weekday filter applied here (analyze_from_dataframes skips it; only analyze_pair enforces it in live mode). - Trade simulation is full candle walk on the primary timeframe using High/Low; if both SL and TP1 hit in the same bar, SL is assumed first (conservative). - Ships with CuPy fallback to NumPy so the script runs anywhere. Usage: python backtest/gpu_backtest.py --pair XAUUSD --days 365 python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-01 python backtest/gpu_backtest.py --pair XAUUSD --days 180 --no-cache """ from __future__ import annotations import argparse import logging import os import sys import time from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd # Ensure project root on sys.path so this script can be invoked from anywhere. PROJECT_ROOT = Path(__file__).resolve().parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from config import ( # noqa: E402 EURUSD, PAIR_INFO, PlatformConfig, SignalStrength, Timeframe, XAUUSD, ) from data_provider import ForexDataProvider # noqa: E402 from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine # noqa: E402 from signal_aggregator import SignalAggregator, SignalReport, TradeSetup # noqa: E402 # ───────────────────────────────────────────────────────────────────── # GPU shim — CuPy if available AND CUDA runtime works, else NumPy. # An importable CuPy can still fail at first kernel launch if CUDA libs # are missing (containers, headless boxes), so verify with a tiny op. # ───────────────────────────────────────────────────────────────────── def _detect_gpu(): try: import cupy as cp_mod # type: ignore # Force runtime init: a kernel launch will surface missing libs. _ = cp_mod.asarray(np.zeros(2, dtype=np.float64)) _ = cp_mod.cumsum(_).get() return cp_mod, True except Exception: return np, False cp, _GPU_AVAILABLE = _detect_gpu() def _to_gpu(arr: np.ndarray): """Move a numpy array to GPU (or no-op if CuPy unavailable).""" return cp.asarray(arr) if _GPU_AVAILABLE else arr def _to_cpu(arr) -> np.ndarray: """Pull array back to host.""" if _GPU_AVAILABLE and hasattr(arr, "get"): return arr.get() return np.asarray(arr) logger = logging.getLogger("gpu_backtest") PAIR_ALIASES = { "XAUUSD": XAUUSD, "XAU/USD": XAUUSD, "GOLD": XAUUSD, "EURUSD": EURUSD, "EUR/USD": EURUSD, "EUR": EURUSD, } # Bar duration in seconds — used to compute "fully-closed" cutoffs so # confirmation timeframes can't see into the future at decision time. BAR_SECONDS = { Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, Timeframe.D1: 86400, Timeframe.W1: 604800, } # ═════════════════════════════════════════════════════════════════════ # GPU INDICATOR PRE-COMPUTATION # ═════════════════════════════════════════════════════════════════════ def _gpu_ewm_alpha(x: np.ndarray, alpha: float, min_periods: int = 0) -> np.ndarray: """ Exponential weighted mean matching pandas `ewm(alpha=alpha, min_periods=p, adjust=False).mean()` exactly. Recursion: y[i] = alpha * x[i] + (1-alpha) * y[i-1], seeded at the first non-NaN x. NaN inputs carry the previous y forward without advancing the non-NaN counter (pandas' default `ignore_na=False`). Output positions before min_periods non-NaN observations are NaN. Sequential CPU pass — EMA is recursive so true GPU parallelism requires a scan kernel, which is overkill for a one-shot pre-compute. """ n = x.shape[0] out = np.full(n, np.nan, dtype=np.float64) if n == 0: return out one_minus = 1.0 - alpha seen = 0 last_y = np.nan for i in range(n): xi = x[i] if np.isnan(xi): # NaN input: y stays at last_y, don't advance counter. if not np.isnan(last_y) and seen >= min_periods: out[i] = last_y continue last_y = xi if np.isnan(last_y) else alpha * xi + one_minus * last_y seen += 1 if seen >= min_periods: out[i] = last_y return out def _gpu_ewm_span(x: np.ndarray, span: int) -> np.ndarray: """ewm with span= semantics (alpha = 2/(span+1), no min_periods, seeded at x[0]).""" alpha = 2.0 / (span + 1.0) return _gpu_ewm_alpha(x, alpha, min_periods=0) def _gpu_rolling_mean(x: np.ndarray, window: int) -> np.ndarray: """Rolling mean via cumsum-diff, NaN-aware (matches pandas `.rolling(window).mean()` default min_periods=window — returns NaN if any value in the window is NaN).""" n = x.shape[0] out = np.full(n, np.nan, dtype=np.float64) if n < window: return out xg = _to_gpu(x.astype(np.float64)) nan_mask = cp.isnan(xg) x_filled = cp.where(nan_mask, 0.0, xg) cs = cp.cumsum(x_filled) valid_cnt = cp.cumsum((~nan_mask).astype(cp.int64)) sum_w = cp.empty(n - window + 1, dtype=cp.float64) cnt_w = cp.empty(n - window + 1, dtype=cp.int64) sum_w[0] = cs[window - 1] sum_w[1:] = cs[window:] - cs[:-window] cnt_w[0] = valid_cnt[window - 1] cnt_w[1:] = valid_cnt[window:] - valid_cnt[:-window] # pandas: NaN if any window value is NaN (cnt < window). rolling = cp.where(cnt_w < window, cp.nan, sum_w / window) out[window - 1:] = _to_cpu(rolling) return out def _gpu_rolling_std(x: np.ndarray, window: int, ddof: int = 0) -> np.ndarray: """Rolling std via cumsum of x and x^2, NaN-aware. ddof=0 matches pandas std(ddof=0). NaN if any window value is NaN.""" n = x.shape[0] out = np.full(n, np.nan, dtype=np.float64) if n < window: return out xg = _to_gpu(x.astype(np.float64)) nan_mask = cp.isnan(xg) x_filled = cp.where(nan_mask, 0.0, xg) x2_filled = x_filled * x_filled cs = cp.cumsum(x_filled) cs2 = cp.cumsum(x2_filled) valid_cnt = cp.cumsum((~nan_mask).astype(cp.int64)) sum_x = cp.empty(n - window + 1, dtype=cp.float64) sum_x2 = cp.empty(n - window + 1, dtype=cp.float64) cnt_w = cp.empty(n - window + 1, dtype=cp.int64) sum_x[0] = cs[window - 1] sum_x[1:] = cs[window:] - cs[:-window] sum_x2[0] = cs2[window - 1] sum_x2[1:] = cs2[window:] - cs2[:-window] cnt_w[0] = valid_cnt[window - 1] cnt_w[1:] = valid_cnt[window:] - valid_cnt[:-window] mean = sum_x / window var = sum_x2 / window - mean * mean var = cp.maximum(var, 0.0) # numerical safety against tiny negatives if ddof: var = var * window / max(window - ddof, 1) std = cp.sqrt(var) std = cp.where(cnt_w < window, cp.nan, std) out[window - 1:] = _to_cpu(std) return out @dataclass class IndicatorCache: """Pre-computed indicator series for one (symbol, timeframe). All arrays are aligned to the source df's index (same length). NaN at warmup positions; downstream lookup tolerates it. """ df: pd.DataFrame # full source df (cleaned) index_pos: Dict[pd.Timestamp, int] # timestamp → position lookup # EMAs (4-EMA mode keys; 2-EMA mode reuses ema_slow + ema_trend + ema_medium) ema_fast: np.ndarray ema_medium: np.ndarray ema_slow: np.ndarray ema_trend: Optional[np.ndarray] # None if len(df) < ema_trend period # RSI rsi: np.ndarray rsi_series: pd.Series # pandas series for divergence # MACD macd_hist: np.ndarray macd_hist_series: pd.Series # Bollinger bb_mid: np.ndarray bb_upper: np.ndarray bb_lower: np.ndarray bb_position: np.ndarray bb_band_width: np.ndarray bb_avg_width: np.ndarray # 20-period rolling mean of band width # ATR atr: np.ndarray # ADX adx: np.ndarray plus_di: np.ndarray minus_di: np.ndarray adx_slope: np.ndarray # ROC roc: np.ndarray # Keltner Channels kc_mid: np.ndarray kc_upper: np.ndarray kc_lower: np.ndarray kc_squeeze: np.ndarray def precompute_indicators( df: pd.DataFrame, tech_cfg, gpu_log: bool = False ) -> IndicatorCache: """Compute every causal indicator series once on the full df.""" close = df["Close"].to_numpy(dtype=np.float64) high = df["High"].to_numpy(dtype=np.float64) low = df["Low"].to_numpy(dtype=np.float64) n = len(close) if gpu_log: logger.info( f"Pre-computing indicators on {n} bars " f"({'GPU' if _GPU_AVAILABLE else 'CPU'} backend)" ) # ── EMAs (use ewm span semantics, identical to pandas .ewm(span=, adjust=False).mean()) ─ ema_fast = _gpu_ewm_span(close, tech_cfg.ema_fast) ema_medium = _gpu_ewm_span(close, tech_cfg.ema_medium) ema_slow = _gpu_ewm_span(close, tech_cfg.ema_slow) ema_trend = _gpu_ewm_span(close, tech_cfg.ema_trend) if n >= tech_cfg.ema_trend else None # ── RSI (Wilder via ewm alpha=1/period, min_periods=period) ── # pandas: delta = close.diff() → delta[0] is NaN; gain = delta.where(delta > 0, 0.0) # evaluates NaN > 0 as False so gain[0] becomes 0.0 (NOT NaN). Same for loss. delta = np.empty(n, dtype=np.float64) delta[0] = np.nan delta[1:] = close[1:] - close[:-1] with np.errstate(invalid="ignore"): gain = np.where(delta > 0, delta, 0.0) loss = np.where(delta < 0, -delta, 0.0) avg_gain = _gpu_ewm_alpha(gain, 1.0 / tech_cfg.rsi_period, min_periods=tech_cfg.rsi_period) avg_loss = _gpu_ewm_alpha(loss, 1.0 / tech_cfg.rsi_period, min_periods=tech_cfg.rsi_period) avg_loss_safe = np.where(avg_loss == 0, 1e-10, avg_loss) rs = avg_gain / avg_loss_safe rsi = 100.0 - (100.0 / (1.0 + rs)) rsi_series = pd.Series(rsi, index=df.index, name="rsi") # ── MACD (12/26 EMA spans, signal 9 EMA span) ── ema_macd_fast = _gpu_ewm_span(close, tech_cfg.macd_fast) ema_macd_slow = _gpu_ewm_span(close, tech_cfg.macd_slow) macd_line = ema_macd_fast - ema_macd_slow signal_line = _gpu_ewm_span(macd_line, tech_cfg.macd_signal) macd_hist = macd_line - signal_line macd_hist_series = pd.Series(macd_hist, index=df.index, name="macd_hist") # ── Bollinger ── bb_mid = _gpu_rolling_mean(close, tech_cfg.bb_period) bb_std = _gpu_rolling_std(close, tech_cfg.bb_period, ddof=0) bb_upper = bb_mid + tech_cfg.bb_std * bb_std bb_lower = bb_mid - tech_cfg.bb_std * bb_std bb_range = bb_upper - bb_lower bb_range_safe = np.where(bb_range == 0, np.nan, bb_range) bb_position = (close - bb_lower) / bb_range_safe # pandas behavior: when range is 0 it returns 0.5 (handled by engine fallback) bb_position = np.where(np.isnan(bb_position), 0.5, bb_position) bb_mid_safe = np.where(bb_mid == 0, np.nan, bb_mid) bb_band_width = (bb_upper - bb_lower) / bb_mid_safe bb_avg_width = _gpu_rolling_mean(bb_band_width, 20) # ── ATR ── prev_close = np.empty(n, dtype=np.float64) prev_close[0] = np.nan prev_close[1:] = close[:-1] tr1 = high - low tr2 = np.abs(high - prev_close) tr3 = np.abs(low - prev_close) # NaN handling at index 0: pandas concat+max returns max ignoring NaN # but for our purposes the engine takes the latest value, so propagate NaN here tr2[0] = np.nan tr3[0] = np.nan tr = np.fmax(np.fmax(tr1, tr2), tr3) # fmax ignores NaN — matches pandas behavior # ewm: pandas uses the first non-NaN value to seed atr = _gpu_ewm_alpha(tr, 1.0 / tech_cfg.atr_period, min_periods=tech_cfg.atr_period) # ── ADX ── # pandas: plus_dm = high.diff(); minus_dm = -low.diff(); both NaN at index 0. # plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) — NaN comparisons # are False, so plus_dm[0] becomes 0.0 (NOT NaN). Same for minus_dm. plus_dm_raw = np.empty(n, dtype=np.float64) minus_dm_raw = np.empty(n, dtype=np.float64) plus_dm_raw[0] = np.nan minus_dm_raw[0] = np.nan plus_dm_raw[1:] = high[1:] - high[:-1] minus_dm_raw[1:] = -(low[1:] - low[:-1]) with np.errstate(invalid="ignore"): cond_plus = (plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0) plus_dm = np.where(cond_plus, plus_dm_raw, 0.0) cond_minus = (minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0) minus_dm = np.where(cond_minus, minus_dm_raw, 0.0) atr_adx = _gpu_ewm_alpha(tr, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period) plus_dm_smooth = _gpu_ewm_alpha( plus_dm, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period ) minus_dm_smooth = _gpu_ewm_alpha( minus_dm, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period ) atr_adx_safe = np.where(atr_adx == 0, np.nan, atr_adx) plus_di = 100.0 * plus_dm_smooth / atr_adx_safe minus_di = 100.0 * minus_dm_smooth / atr_adx_safe di_sum = plus_di + minus_di di_sum_safe = np.where(di_sum == 0, np.nan, di_sum) dx = 100.0 * np.abs(plus_di - minus_di) / di_sum_safe adx = _gpu_ewm_alpha(dx, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period) # ── ADX Slope ── adx_slope_lookback = getattr(tech_cfg, 'adx_slope_lookback', 5) adx_slope = np.full(n, np.nan, dtype=np.float64) if n > adx_slope_lookback: adx_slope[adx_slope_lookback:] = adx[adx_slope_lookback:] - adx[:-adx_slope_lookback] # ── ROC (Rate of Change) ── roc_period = getattr(tech_cfg, 'roc_period', 10) roc = np.full(n, np.nan, dtype=np.float64) if n > roc_period: prev_close = close[:-roc_period] prev_safe = np.where(prev_close == 0, np.nan, prev_close) roc[roc_period:] = ((close[roc_period:] - prev_close) / prev_safe) * 100.0 # ── Keltner Channels ── kc_period = getattr(tech_cfg, 'kc_period', 20) kc_atr_period = getattr(tech_cfg, 'kc_atr_period', 10) kc_multiplier = getattr(tech_cfg, 'kc_multiplier', 1.5) kc_mid = _gpu_ewm_span(close, kc_period) kc_atr = _gpu_ewm_alpha(tr, 1.0 / kc_atr_period, min_periods=kc_atr_period) kc_upper = kc_mid + kc_multiplier * kc_atr kc_lower = kc_mid - kc_multiplier * kc_atr with np.errstate(invalid="ignore"): kc_squeeze = (bb_lower > kc_lower) & (bb_upper < kc_upper) kc_squeeze = kc_squeeze.astype(np.float64) index_pos = {ts: i for i, ts in enumerate(df.index)} return IndicatorCache( df=df, index_pos=index_pos, ema_fast=ema_fast, ema_medium=ema_medium, ema_slow=ema_slow, ema_trend=ema_trend, rsi=rsi, rsi_series=rsi_series, macd_hist=macd_hist, macd_hist_series=macd_hist_series, bb_mid=bb_mid, bb_upper=bb_upper, bb_lower=bb_lower, bb_position=bb_position, bb_band_width=bb_band_width, bb_avg_width=bb_avg_width, atr=atr, adx=adx, plus_di=plus_di, minus_di=minus_di, adx_slope=adx_slope, roc=roc, kc_mid=kc_mid, kc_upper=kc_upper, kc_lower=kc_lower, kc_squeeze=kc_squeeze, ) # ═════════════════════════════════════════════════════════════════════ # GPU-AWARE TECHNICAL ENGINE (subclasses live engine, lookup-only) # ═════════════════════════════════════════════════════════════════════ class GPUTechnicalEngine(TechnicalEngine): """ Drop-in TechnicalEngine that pulls indicator values from a pre-computed cache instead of recomputing per call. The conditional/scoring/description logic for each indicator is preserved verbatim by overriding only the `_compute_*` methods — every downstream consumer (regime classifier, composite scorer, divergence, structure_sr, candlestick) sees an IndicatorSignal byte-identical to what the live engine would produce. """ def __init__(self, tech_cfg, caches: Dict[str, IndicatorCache]): super().__init__(tech_cfg) # cache key = timeframe.value (e.g. "15min", "1h", "4h") self.caches = caches self._current_tf: Optional[str] = None def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis: # Latch the timeframe so per-indicator overrides know which cache to use. # Multiple TFs share aligned timestamps (e.g. M15 14:00 == H1 14:00), # so timestamp-only lookup would match the wrong cache. self._current_tf = timeframe try: return super().analyze(df, symbol, timeframe) finally: self._current_tf = None def _resolve_pos(self, df: pd.DataFrame) -> Tuple[Optional[IndicatorCache], int]: """Find this slice's position in its pre-computed cache.""" if df is None or df.empty or self._current_tf is None: return None, -1 cache = self.caches.get(self._current_tf) if cache is None: return None, -1 pos = cache.index_pos.get(df.index[-1]) return (cache, pos) if pos is not None else (None, -1) # ── EMA overrides ────────────────────────────────────────────── def _compute_ema_2(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_ema_2(df, result) c = self.config latest = float(df["Close"].iloc[-1]) s = float(cache.ema_slow[pos]) m = float(cache.ema_medium[pos]) t = float(cache.ema_trend[pos]) if cache.ema_trend is not None and not np.isnan(cache.ema_trend[pos]) else None if t is None or pos < c.ema_trend - 1: result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=0, signal=SignalStrength.NEUTRAL, confidence=30, description="EMA200 data insufficient", weight=1.0, ) result.key_levels["ema_medium"] = m result.key_levels["ema_slow"] = s return if latest > s > t: signal, conf, desc = SignalStrength.STRONG_BUY, 85, "Full bullish: price > EMA50 > EMA200" elif latest > s and s < t: signal, conf, desc = SignalStrength.BUY, 60, "Emerging uptrend: price > EMA50, EMA50 < EMA200" elif latest < s < t: signal, conf, desc = SignalStrength.STRONG_SELL, 85, "Full bearish: price < EMA50 < EMA200" elif latest < s and s > t: signal, conf, desc = SignalStrength.SELL, 60, "Emerging downtrend: price < EMA50, EMA50 > EMA200" else: signal, conf, desc = SignalStrength.NEUTRAL, 40, "EMA alignment mixed" # Crossover detection on prior 10 bars (live: `if len(ema50) >= 10`, # `for lookback in range(2, min(11, len(ema50)))` — len = pos+1). if pos >= 9: for lookback in range(2, min(11, pos + 1)): prev_s = float(cache.ema_slow[pos - lookback + 1]) prev_t = float(cache.ema_trend[pos - lookback + 1]) if prev_s <= prev_t and s > t: desc += " | Golden cross (50/200)" conf = min(conf + 10, 95) break if prev_s >= prev_t and s < t: desc += " | Death cross (50/200)" conf = min(conf + 10, 95) break normalized = { SignalStrength.STRONG_BUY: 1.0, SignalStrength.BUY: 0.5, SignalStrength.NEUTRAL: 0.0, SignalStrength.SELL: -0.5, SignalStrength.STRONG_SELL: -1.0, }[signal] result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=normalized, signal=signal, confidence=conf, description=desc, weight=1.0, ) result.key_levels["ema_trend"] = t result.key_levels["ema_medium"] = m result.key_levels["ema_slow"] = s def _compute_ema_4(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_ema_4(df, result) c = self.config latest = float(df["Close"].iloc[-1]) f = float(cache.ema_fast[pos]) m = float(cache.ema_medium[pos]) s = float(cache.ema_slow[pos]) t_arr = cache.ema_trend t = float(t_arr[pos]) if (t_arr is not None and not np.isnan(t_arr[pos])) else None # Live engine guards `t` with `len(close) >= ema_trend`; replicate. if t is not None and pos < c.ema_trend - 1: t = None alignment_score = 0 checks = [(latest, f), (f, m), (m, s)] if t is not None: checks.append((s, t)) for above, below in checks: if above > below: alignment_score += 1 elif above < below: alignment_score -= 1 max_checks = len(checks) normalized = alignment_score / max_checks if max_checks > 0 else 0 if normalized > 0.6: signal = SignalStrength.STRONG_BUY conf = min(90, 60 + normalized * 30) elif normalized > 0.2: signal = SignalStrength.BUY conf = 55 + normalized * 20 elif normalized < -0.6: signal = SignalStrength.STRONG_SELL conf = min(90, 60 + abs(normalized) * 30) elif normalized < -0.2: signal = SignalStrength.SELL conf = 55 + abs(normalized) * 20 else: signal = SignalStrength.NEUTRAL conf = 40 crossover_desc = "" # Live: `if len(ema_fast) >= 3:` then iloc[-2] — needs len ≥ 2 functionally, # but live's guard requires len ≥ 3 so we match that to stay byte-identical. if pos >= 2: prev_f = float(cache.ema_fast[pos - 1]) prev_m = float(cache.ema_medium[pos - 1]) if prev_f <= prev_m and f > m: crossover_desc = " | Golden cross (fast/medium)" conf = min(conf + 10, 95) elif prev_f >= prev_m and f < m: crossover_desc = " | Death cross (fast/medium)" conf = min(conf + 10, 95) desc = f"EMA alignment: {alignment_score}/{max_checks}{crossover_desc}" result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=normalized, signal=signal, confidence=conf, description=desc, weight=1.0, ) if t is not None: result.key_levels["ema_trend"] = t result.key_levels["ema_fast"] = f result.key_levels["ema_medium"] = m result.key_levels["ema_slow"] = s # ── RSI override ────────────────────────────────────────────── def _compute_rsi(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_rsi(df, result) c = self.config rsi_val = float(cache.rsi[pos]) if np.isnan(rsi_val): rsi_val = 50.0 if rsi_val >= c.rsi_overbought: signal = SignalStrength.SELL conf = min(80, 50 + (rsi_val - c.rsi_overbought)) desc = f"RSI {rsi_val:.1f} — Overbought" elif rsi_val <= c.rsi_oversold: signal = SignalStrength.BUY conf = min(80, 50 + (c.rsi_oversold - rsi_val)) desc = f"RSI {rsi_val:.1f} — Oversold" elif rsi_val >= c.rsi_neutral_upper: signal = SignalStrength.BUY conf = 45 desc = f"RSI {rsi_val:.1f} — Bullish momentum zone" elif rsi_val <= c.rsi_neutral_lower: signal = SignalStrength.SELL conf = 45 desc = f"RSI {rsi_val:.1f} — Bearish momentum zone" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"RSI {rsi_val:.1f} — Neutral" result.indicators["rsi"] = IndicatorSignal( name="RSI", value=rsi_val, signal=signal, confidence=conf, description=desc, weight=1.5, ) # Slice the pre-computed series so divergence detector sees only # bars up to the current decision point (no look-ahead). result._rsi_series = cache.rsi_series.iloc[: pos + 1] # type: ignore[attr-defined] # ── MACD override ───────────────────────────────────────────── def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_macd(df, result) c = self.config result._macd_histogram = cache.macd_hist_series.iloc[: pos + 1] # type: ignore[attr-defined] if not c.macd_as_pillar or pos < 2: return hist_now = float(cache.macd_hist[pos]) hist_prev = float(cache.macd_hist[pos - 1]) hist_prev2 = float(cache.macd_hist[pos - 2]) if np.isnan(hist_now) or np.isnan(hist_prev): return crossed_up = hist_prev <= 0 < hist_now crossed_down = hist_prev >= 0 > hist_now accel_up = hist_now > 0 and hist_now > hist_prev > 0 accel_down = hist_now < 0 and hist_now < hist_prev < 0 fading_up = hist_now > 0 and hist_now < hist_prev fading_down = hist_now < 0 and hist_now > hist_prev if pos >= 19: window = np.abs(cache.macd_hist[pos - 19: pos + 1]) mean_abs = float(np.nanmean(window)) hist_pct = abs(hist_now) / mean_abs if mean_abs > 0 else 1.0 else: hist_pct = 1.0 if crossed_up or accel_up: signal = SignalStrength.STRONG_BUY if accel_up else SignalStrength.BUY conf = min(80, 55 + hist_pct * 10) desc = "MACD histogram" + (" accelerating ↑" if accel_up else " crossed above zero") elif crossed_down or accel_down: signal = SignalStrength.STRONG_SELL if accel_down else SignalStrength.SELL conf = min(80, 55 + hist_pct * 10) desc = "MACD histogram" + (" accelerating ↓" if accel_down else " crossed below zero") elif fading_up: signal, conf, desc = SignalStrength.NEUTRAL, 35, "MACD histogram positive but fading" elif fading_down: signal, conf, desc = SignalStrength.NEUTRAL, 35, "MACD histogram negative but fading" else: signal, conf, desc = SignalStrength.NEUTRAL, 30, "MACD histogram flat" result.indicators["macd"] = IndicatorSignal( name="MACD", value=hist_now, signal=signal, confidence=conf, description=desc, weight=1.5, ) # ── Bollinger override ──────────────────────────────────────── def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_bollinger(df, result) c = self.config latest = float(df["Close"].iloc[-1]) upper_val = float(cache.bb_upper[pos]) lower_val = float(cache.bb_lower[pos]) sma_val = float(cache.bb_mid[pos]) if np.isnan(upper_val) or np.isnan(lower_val) or np.isnan(sma_val) or sma_val == 0: return bb_range = upper_val - lower_val bb_position = (latest - lower_val) / bb_range if bb_range != 0 else 0.5 band_width = float(cache.bb_band_width[pos]) is_squeeze = False if pos >= 19: avg_width = float(cache.bb_avg_width[pos]) if not np.isnan(avg_width) and avg_width > 0: is_squeeze = band_width < avg_width * 0.75 squeeze_note = " | SQUEEZE detected (expect breakout)" if is_squeeze else "" if bb_position > 1.0: signal, conf = SignalStrength.SELL, 65 desc = f"Price above upper BB ({bb_position:.2f}){squeeze_note}" elif bb_position < 0.0: signal, conf = SignalStrength.BUY, 65 desc = f"Price below lower BB ({bb_position:.2f}){squeeze_note}" elif bb_position > 0.8: signal, conf = SignalStrength.SELL, 45 desc = f"Price near upper BB ({bb_position:.2f}){squeeze_note}" elif bb_position < 0.2: signal, conf = SignalStrength.BUY, 45 desc = f"Price near lower BB ({bb_position:.2f}){squeeze_note}" else: signal, conf = SignalStrength.NEUTRAL, 30 desc = f"Price mid-band ({bb_position:.2f}){squeeze_note}" if is_squeeze: conf = min(conf + 10, 85) result.indicators["bollinger"] = IndicatorSignal( name="Bollinger Bands", value=bb_position, signal=signal, confidence=conf, description=desc, weight=1.5, ) result.key_levels["bb_upper"] = upper_val result.key_levels["bb_lower"] = lower_val result.key_levels["bb_mid"] = sma_val # ── ADX override ────────────────────────────────────────────── def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_adx(df, result) c = self.config adx_val = float(cache.adx[pos]) if np.isnan(adx_val): adx_val = 0.0 plus_di_val = float(cache.plus_di[pos]) minus_di_val = float(cache.minus_di[pos]) if np.isnan(plus_di_val): plus_di_val = 0.0 if np.isnan(minus_di_val): minus_di_val = 0.0 if adx_val >= c.adx_very_strong: trend_strength, conf = "Very strong", 80 elif adx_val >= c.adx_strong_trend: trend_strength, conf = "Strong", 65 else: trend_strength, conf = "Weak/Ranging", 40 if plus_di_val > minus_di_val: signal = SignalStrength.BUY if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL direction = "bullish" else: signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL direction = "bearish" adx_slope = float(cache.adx_slope[pos]) if not np.isnan(cache.adx_slope[pos]) else 0.0 slope_desc = "" emerging_threshold = c.adx_slope_emerging_threshold if adx_val < c.adx_strong_trend and adx_slope > emerging_threshold: slope_desc = " | TREND EMERGING (ADX rising fast)" conf = min(conf + 15, 85) elif adx_val >= c.adx_strong_trend and adx_slope > 0: slope_desc = f" | Trend strengthening (slope +{adx_slope:.1f})" conf = min(conf + 5, 85) elif adx_val >= c.adx_strong_trend and adx_slope < -emerging_threshold: slope_desc = f" | Trend weakening (slope {adx_slope:.1f})" conf = max(conf - 10, 30) desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}{slope_desc}" result.indicators["adx"] = IndicatorSignal( name="ADX", value=adx_val, signal=signal, confidence=conf, description=desc, weight=0.5, ) result._adx_slope = adx_slope # ── ATR override ────────────────────────────────────────────── def _compute_atr(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_atr(df, result) latest_close = float(df["Close"].iloc[-1]) atr_val = float(cache.atr[pos]) if np.isnan(atr_val): atr_val = 0.0 atr_pct = (atr_val / latest_close) * 100 if latest_close != 0 else 0 # Volatility regime: replicate live behavior on the 50-bar window. if pos >= 49: window = cache.atr[pos - 49: pos + 1] atr_percentile = float(np.mean(window < atr_val)) * 100 # bars strictly below current # Live code: (atr.iloc[-1] > atr.iloc[-50:]).mean() * 100 → fraction of window strictly less than current else: atr_percentile = 50.0 if atr_percentile > 80: vol_regime = "extreme" elif atr_percentile > 60: vol_regime = "high" elif atr_percentile > 30: vol_regime = "normal" else: vol_regime = "low" result.volatility_regime = vol_regime result.key_levels["atr"] = atr_val result.key_levels["atr_pct"] = atr_pct result.indicators["atr"] = IndicatorSignal( name="ATR", value=atr_val, signal=SignalStrength.NEUTRAL, confidence=50, description=f"ATR {atr_val:.5f} ({atr_pct:.3f}%) | Volatility: {vol_regime} (P{atr_percentile:.0f})", weight=0.0, ) def _compute_roc(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_roc(df, result) c = self.config n = c.roc_period if pos < n + 1: return roc_val = float(cache.roc[pos]) roc_prev = float(cache.roc[pos - 1]) if pos > 0 else np.nan if np.isnan(roc_val) or np.isnan(roc_prev): return roc_rising = roc_val > roc_prev roc_abs = abs(roc_val) threshold = c.roc_threshold if roc_val > 0 and roc_rising: if roc_abs > threshold * 2: signal = SignalStrength.STRONG_BUY conf = min(80, 60 + roc_abs * 5) desc = f"ROC +{roc_val:.2f}% — Strong momentum acceleration" else: signal = SignalStrength.BUY conf = min(70, 50 + roc_abs * 10) desc = f"ROC +{roc_val:.2f}% — Rising momentum" elif roc_val < 0 and not roc_rising: if roc_abs > threshold * 2: signal = SignalStrength.STRONG_SELL conf = min(80, 60 + roc_abs * 5) desc = f"ROC {roc_val:.2f}% — Strong bearish momentum" else: signal = SignalStrength.SELL conf = min(70, 50 + roc_abs * 10) desc = f"ROC {roc_val:.2f}% — Falling momentum" elif roc_prev <= 0 < roc_val: signal = SignalStrength.BUY conf = 65 desc = f"ROC crossed above zero (+{roc_val:.2f}%)" elif roc_prev >= 0 > roc_val: signal = SignalStrength.SELL conf = 65 desc = f"ROC crossed below zero ({roc_val:.2f}%)" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"ROC {roc_val:+.2f}% — No clear momentum" result.indicators["roc"] = IndicatorSignal( name="ROC", value=roc_val, signal=signal, confidence=conf, description=desc, weight=1.5, ) def _compute_keltner(self, df: pd.DataFrame, result: TechnicalAnalysis): cache, pos = self._resolve_pos(df) if cache is None or pos < 0: return super()._compute_keltner(df, result) c = self.config latest = float(df["Close"].iloc[-1]) kc_upper_val = float(cache.kc_upper[pos]) kc_lower_val = float(cache.kc_lower[pos]) kc_mid_val = float(cache.kc_mid[pos]) if np.isnan(kc_upper_val) or np.isnan(kc_lower_val): return is_squeeze = bool(cache.kc_squeeze[pos]) squeeze_released = False if pos >= 1: was_squeeze = bool(cache.kc_squeeze[pos - 1]) squeeze_released = was_squeeze and not is_squeeze macd_direction = 0 if not np.isnan(cache.macd_hist[pos]): macd_direction = 1 if cache.macd_hist[pos] > 0 else -1 if squeeze_released: if macd_direction > 0: signal = SignalStrength.STRONG_BUY desc = "Keltner SQUEEZE RELEASED — bullish breakout" elif macd_direction < 0: signal = SignalStrength.STRONG_SELL desc = "Keltner SQUEEZE RELEASED — bearish breakout" else: signal = SignalStrength.NEUTRAL desc = "Keltner squeeze released — direction unclear" conf = 80 elif is_squeeze: signal = SignalStrength.NEUTRAL conf = 60 desc = "Keltner SQUEEZE active — breakout imminent" elif latest > kc_upper_val: signal = SignalStrength.BUY conf = 65 desc = "Price above Keltner upper — strong uptrend continuation" elif latest < kc_lower_val: signal = SignalStrength.SELL conf = 65 desc = "Price below Keltner lower — strong downtrend continuation" else: signal = SignalStrength.NEUTRAL conf = 30 desc = "Price within Keltner Channels" result.indicators["keltner"] = IndicatorSignal( name="Keltner", value=float(latest - kc_mid_val), signal=signal, confidence=conf, description=desc, weight=1.5, ) result.key_levels["kc_upper"] = kc_upper_val result.key_levels["kc_lower"] = kc_lower_val result.key_levels["kc_mid"] = kc_mid_val result._kc_squeeze = is_squeeze result._kc_squeeze_released = squeeze_released # ═════════════════════════════════════════════════════════════════════ # DATA LOADING (API + parquet cache) # ═════════════════════════════════════════════════════════════════════ CACHE_DIR = Path(__file__).resolve().parent / "cache" def _cache_path(symbol: str, tf: Timeframe, start: datetime, end: datetime) -> Path: safe_sym = symbol.replace("/", "_").replace("=X", "") return CACHE_DIR / f"{safe_sym}_{tf.value}_{start:%Y%m%d}_{end:%Y%m%d}.parquet" def fetch_history( provider: ForexDataProvider, symbol: str, timeframe: Timeframe, start: datetime, end: datetime, use_cache: bool = True, clean: bool = True, ) -> pd.DataFrame: """Fetch OHLCV between start/end with parquet caching and pagination. Twelve Data caps responses at 5000 bars; we paginate by date range so multi-month M5/M15 backtests work. The parquet cache stores the RAW series; `_validate_and_clean` runs on load. Pass clean=False to get the raw price path — trade resolution must use it, because cleaning drops >10σ spike bars (news candles) that hit stops in reality. (Caches written before this change already contain cleaned data; a refetch is needed to restore dropped spike bars.) """ CACHE_DIR.mkdir(parents=True, exist_ok=True) path = _cache_path(symbol, timeframe, start, end) if use_cache and path.exists(): logger.info(f"Loading cached {symbol} {timeframe.value}: {path.name}") df = pd.read_parquet(path) if df.index.tz is None: df.index = df.index.tz_localize("UTC") return provider._validate_and_clean(df, symbol, timeframe) if clean else df logger.info(f"Fetching {symbol} {timeframe.value} from {start.date()} to {end.date()}") # Bars-per-call estimate to size each window safely under the 5000 cap. bars_per_day = ForexDataProvider.BARS_PER_DAY[timeframe] days_per_chunk = max(1, int(4500 / bars_per_day)) chunks: List[pd.DataFrame] = [] cursor_end = end empty_streak = 0 while cursor_end > start: cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk)) try: chunk = provider._fetch_twelve_data( symbol=symbol, timeframe=timeframe, outputsize=5000, start_date=cursor_start.strftime("%Y-%m-%d %H:%M:%S"), end_date=cursor_end.strftime("%Y-%m-%d %H:%M:%S"), ) except Exception as e: # Twelve Data raises BadRequest for windows with no data (weekends, # holidays, asset gaps). Skip the window, step back, retry — but # bail out after a few consecutive empties to avoid spinning. logger.warning( f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}" ) empty_streak += 1 if empty_streak >= 3: break cursor_end = cursor_start - timedelta(seconds=1) continue if chunk is None or chunk.empty: empty_streak += 1 if empty_streak >= 3: break cursor_end = cursor_start - timedelta(seconds=1) continue empty_streak = 0 chunks.append(chunk) # Step back; dedup on concat. cursor_end = chunk.index[0] - timedelta(seconds=1) if len(chunk) < 2: break if not chunks: raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}") df = pd.concat(chunks).sort_index() df = df[~df.index.duplicated(keep="last")] # Fetches ending "now" include the still-forming candle — drop it so the # cache only ever contains closed bars. df = provider._drop_forming_candle(df, timeframe) # Trim to requested window. `start`/`end` are already tz-aware UTC # datetimes — wrapping with pd.Timestamp(..., tz=...) would raise on # already-localized inputs. end_ts = pd.Timestamp(end) start_ts = pd.Timestamp(start) df = df[(df.index >= start_ts) & (df.index <= end_ts)] if use_cache: df.to_parquet(path) logger.info(f"Cached {len(df)} raw bars to {path.name}") return provider._validate_and_clean(df, symbol, timeframe) if clean else df # ═════════════════════════════════════════════════════════════════════ # TRADE SIMULATION (vectorized SL/TP detection) # ═════════════════════════════════════════════════════════════════════ @dataclass class Trade: """One backtested trade — entry through exit + full reasoning.""" pair: str entry_time: pd.Timestamp entry_price: float sl: float tp1: float tp2: float direction: str # "BUY" or "SELL" pip_value: float risk_pips: float reward_pips: float rr_planned: float quality: str # signal grade at entry final_score: float final_confidence: float market_regime: str volatility_regime: str = "" strategy_tag: str = "" # "XAU", "XAU_PB", "EUR_PB", "GBP_PB", "GENERIC" # Per-trade trailing overrides (0 = use the run-level params). trail_arm_r: float = 0.0 trail_dist_r: float = 0.0 # Per-instructions: entry reasoning and indicator breakdown ───── entry_reasons: List[str] = field(default_factory=list) entry_warnings: List[str] = field(default_factory=list) # name -> {value, signal_name, confidence, weight, agrees_with_dir} indicator_breakdown: Dict[str, Dict] = field(default_factory=dict) indicators_agreeing: int = 0 # count agreeing with trade direction indicators_opposing: int = 0 # count opposing indicators_neutral: int = 0 indicators_total: int = 0 # total scored indicators # Filled on exit: exit_time: Optional[pd.Timestamp] = None exit_price: Optional[float] = None exit_reason: str = "" # "TP1", "SL", "TIMEOUT" bars_held: int = 0 r_multiple: float = 0.0 # realized R (negative if loss) mae_r: float = 0.0 # max adverse excursion in R mfe_r: float = 0.0 # max favorable excursion in R def _build_indicator_breakdown( report: SignalReport, direction: str ) -> Tuple[Dict[str, Dict], int, int, int, int]: """Build per-indicator breakdown and tally direction agreement. For each scored indicator on the primary timeframe: - record value/signal/confidence/weight - flag whether its signal agrees with the trade direction """ breakdown: Dict[str, Dict] = {} agree = oppose = neutral = total = 0 if report.primary_analysis is None: return breakdown, 0, 0, 0, 0 is_buy = direction == "BUY" bull = (SignalStrength.BUY, SignalStrength.STRONG_BUY) bear = (SignalStrength.SELL, SignalStrength.STRONG_SELL) for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue # skip private series + context-only indicators (atr) agrees = "neutral" if ind.signal in bull: agrees = "agree" if is_buy else "oppose" elif ind.signal in bear: agrees = "agree" if not is_buy else "oppose" breakdown[name] = { "value": float(ind.value) if ind.value is not None else None, "signal": ind.signal.name, "confidence": float(ind.confidence), "weight": float(ind.weight), "agrees": agrees, "description": ind.description, } total += 1 if agrees == "agree": agree += 1 elif agrees == "oppose": oppose += 1 else: neutral += 1 return breakdown, agree, oppose, neutral, total def _simulate_trailing( trades: List[Trade], opens_arr: np.ndarray, highs_arr: np.ndarray, lows_arr: np.ndarray, closes_arr: np.ndarray, valid_lens: np.ndarray, exit_times: List[List[pd.Timestamp]], spread: float, trail_arm_r: float, trail_dist_r: float, ) -> None: """Resolve trades with a trailing stop (per-trade CPU loop). Exit reasons: "SL" (original stop, full loss), "TRAIL" (trailed stop filled at/above entry region), "TIMEOUT". """ for i, t in enumerate(trades): n = int(valid_lens[i]) if n == 0: t.exit_reason = "NO_DATA" continue o = opens_arr[i, :n] h = highs_arr[i, :n] l = lows_arr[i, :n] c = closes_arr[i, :n] buy = t.direction == "BUY" risk = abs(t.entry_price - t.sl) if risk <= 0: t.exit_reason = "NO_DATA" continue arm = t.trail_arm_r if t.trail_arm_r > 0 else trail_arm_r dist = t.trail_dist_r if t.trail_dist_r > 0 else trail_dist_r cur_sl = t.sl peak = t.entry_price exit_price = None exit_idx = n - 1 reason = "TIMEOUT" for j in range(n): if np.isnan(h[j]): break hit = (l[j] <= cur_sl) if buy else (h[j] >= cur_sl) if hit: exit_price = cur_sl # Gap through the stop → filled at the open. if (buy and o[j] < cur_sl) or (not buy and o[j] > cur_sl): exit_price = o[j] reason = "SL" if cur_sl == t.sl else "TRAIL" exit_idx = j break # Bar closed: advance peak, then ratchet the stop for the # NEXT bar. fav = h[j] if buy else l[j] peak = max(peak, fav) if buy else min(peak, fav) exc = ((peak - t.entry_price) if buy else (t.entry_price - peak)) / risk if exc >= arm: lvl = (peak - dist * risk if buy else peak + dist * risk) cur_sl = max(cur_sl, lvl) if buy else min(cur_sl, lvl) if exit_price is None: exit_price = float(c[exit_idx]) t.exit_reason = reason t.bars_held = exit_idx + 1 t.exit_price = float(exit_price) t.exit_time = exit_times[i][exit_idx] pnl = ((t.exit_price - t.entry_price) if buy else (t.entry_price - t.exit_price)) t.r_multiple = (pnl - spread) / risk h_win = h[:t.bars_held] l_win = l[:t.bars_held] if buy: mfe = (np.nanmax(h_win) - t.entry_price) / risk mae = (t.entry_price - np.nanmin(l_win)) / risk else: mfe = (t.entry_price - np.nanmin(l_win)) / risk mae = (np.nanmax(h_win) - t.entry_price) / risk t.mfe_r = float(max(0.0, mfe)) t.mae_r = float(max(0.0, mae)) def simulate_trades( trades: List[Trade], sim_df: pd.DataFrame, max_bars: int = 200, spread: float = 0.0, trail_arm_r: float = 0.0, trail_dist_r: float = 0.75, ) -> None: """Walk forward bar-by-bar after each entry and resolve the exit. Two exit modes: - Static (trail_arm_r <= 0): resolve SL / TP1, vectorized on GPU when CuPy is available. - Trailing (trail_arm_r > 0): TP1 is advisory. Once favorable excursion reaches trail_arm_r × risk, the stop trails trail_dist_r × risk behind the best price. Stop levels update at bar CLOSE (intra-bar sequence is unknowable), so a new stop can only fill from the next bar on — conservative. SL/TP-same-bar tie: SL assumed first (conservative). Gap-through fills: a bar *opening* beyond SL/TP fills at the open (worse than SL on gaps against, better than TP on gaps in favor). `spread` is a round-trip cost in price units subtracted from every trade's PnL before computing R. Times out (no hit) → marked TIMEOUT, exit at the last simulated bar. """ if not trades: return # Build per-trade post-entry windows (clipped to max_bars). opens_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64) highs_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64) lows_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64) closes_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64) valid_lens = np.zeros(len(trades), dtype=np.int64) exit_times: List[List[pd.Timestamp]] = [] sim_index = sim_df.index sim_open = sim_df["Open"].to_numpy(dtype=np.float64) sim_high = sim_df["High"].to_numpy(dtype=np.float64) sim_low = sim_df["Low"].to_numpy(dtype=np.float64) sim_close = sim_df["Close"].to_numpy(dtype=np.float64) for i, t in enumerate(trades): # entry_time is an exact bar timestamp from sim_df (we choose the # first bar AFTER the signal bar as entry — see backtester loop). try: entry_pos = sim_index.get_loc(t.entry_time) except KeyError: # Snap to next available bar (rare timezone/edge case). entry_pos = sim_index.searchsorted(t.entry_time) if entry_pos >= len(sim_index): exit_times.append([]) continue end_pos = min(entry_pos + max_bars, len(sim_index)) n = end_pos - entry_pos valid_lens[i] = n opens_arr[i, :n] = sim_open[entry_pos:end_pos] highs_arr[i, :n] = sim_high[entry_pos:end_pos] lows_arr[i, :n] = sim_low[entry_pos:end_pos] closes_arr[i, :n] = sim_close[entry_pos:end_pos] exit_times.append(list(sim_index[entry_pos:end_pos])) if trail_arm_r > 0: _simulate_trailing( trades, opens_arr, highs_arr, lows_arr, closes_arr, valid_lens, exit_times, spread, trail_arm_r, trail_dist_r, ) return # GPU vectorized: for each trade, scan its window for first SL/TP hit. sls = np.array([t.sl for t in trades], dtype=np.float64)[:, None] tps = np.array([t.tp1 for t in trades], dtype=np.float64)[:, None] is_buy = np.array([t.direction == "BUY" for t in trades], dtype=bool)[:, None] h_g = _to_gpu(highs_arr) l_g = _to_gpu(lows_arr) c_g = _to_gpu(closes_arr) sls_g = _to_gpu(sls) tps_g = _to_gpu(tps) is_buy_g = _to_gpu(is_buy) # Hit masks. For BUY: SL when low<=sl, TP when high>=tp. # For SELL: SL when high>=sl, TP when low<=tp. sl_hit_buy = l_g <= sls_g tp_hit_buy = h_g >= tps_g sl_hit_sell = h_g >= sls_g tp_hit_sell = l_g <= tps_g sl_hit = cp.where(is_buy_g, sl_hit_buy, sl_hit_sell) tp_hit = cp.where(is_buy_g, tp_hit_buy, tp_hit_sell) # Mask out NaN bars (no data past valid_lens). valid_mask = ~cp.isnan(h_g) sl_hit = sl_hit & valid_mask tp_hit = tp_hit & valid_mask # First-hit indices (or max_bars if never). idx_template = cp.arange(max_bars)[None, :] sl_idx = cp.where(sl_hit, idx_template, max_bars) tp_idx = cp.where(tp_hit, idx_template, max_bars) first_sl = cp.min(sl_idx, axis=1) first_tp = cp.min(tp_idx, axis=1) sl_first_arr = _to_cpu(first_sl) tp_first_arr = _to_cpu(first_tp) # Resolve each trade and compute realized MFE/MAE *up to exit bar* # (a trade closed at bar 5 cannot have meaningful excursions at bar 12). for i, t in enumerate(trades): n_valid = valid_lens[i] if n_valid == 0: t.exit_reason = "NO_DATA" continue sl_hit_idx = int(sl_first_arr[i]) tp_hit_idx = int(tp_first_arr[i]) # SL wins ties (conservative same-bar handling). if sl_hit_idx < n_valid and sl_hit_idx <= tp_hit_idx: t.exit_reason = "SL" t.bars_held = sl_hit_idx + 1 exit_price = t.sl bar_open = opens_arr[i, sl_hit_idx] # Gap through the stop: the bar opened beyond SL → filled at open. if not np.isnan(bar_open): if (t.direction == "BUY" and bar_open < t.sl) or \ (t.direction == "SELL" and bar_open > t.sl): exit_price = bar_open t.exit_price = exit_price t.exit_time = exit_times[i][sl_hit_idx] elif tp_hit_idx < n_valid: t.exit_reason = "TP1" t.bars_held = tp_hit_idx + 1 exit_price = t.tp1 bar_open = opens_arr[i, tp_hit_idx] # Gap beyond the target in our favor → filled at open. if not np.isnan(bar_open): if (t.direction == "BUY" and bar_open > t.tp1) or \ (t.direction == "SELL" and bar_open < t.tp1): exit_price = bar_open t.exit_price = exit_price t.exit_time = exit_times[i][tp_hit_idx] else: # Time out at last simulated bar. t.exit_reason = "TIMEOUT" t.bars_held = int(n_valid) last_idx = int(n_valid) - 1 t.exit_price = float(closes_arr[i, last_idx]) t.exit_time = exit_times[i][last_idx] # Realized R from actual fill prices, net of round-trip spread. risk = abs(t.entry_price - t.sl) if risk > 0 and t.exit_price is not None: pnl = ((t.exit_price - t.entry_price) if t.direction == "BUY" else (t.entry_price - t.exit_price)) t.r_multiple = (pnl - spread) / risk # MFE / MAE measured only over bars [0 .. bars_held). bars = t.bars_held risk = abs(t.entry_price - t.sl) if bars > 0 and risk > 0: h_win = highs_arr[i, :bars] l_win = lows_arr[i, :bars] if t.direction == "BUY": mfe = (np.nanmax(h_win) - t.entry_price) / risk mae = (t.entry_price - np.nanmin(l_win)) / risk else: mfe = (t.entry_price - np.nanmin(l_win)) / risk mae = (np.nanmax(h_win) - t.entry_price) / risk t.mfe_r = float(max(0.0, mfe)) t.mae_r = float(max(0.0, mae)) # ═════════════════════════════════════════════════════════════════════ # BACKTEST RUNNER # ═════════════════════════════════════════════════════════════════════ @dataclass class BacktestConfig: pair: str start: datetime end: datetime primary_tf: Timeframe = Timeframe.M15 confirmation_tfs: List[Timeframe] = field( default_factory=lambda: [Timeframe.H1, Timeframe.H4] ) warmup_bars: int = 250 # min bars before first signal (EMA200 warmup) cooldown_bars: int = 5 # min bars between trades on same pair max_hold_bars: int = 300 # cap simulation length per trade # (300: trailing exits need room to run; # 200→300 adds ~10R on 12mo M5) use_cache: bool = True quality_filter: List[str] = field( default_factory=lambda: ["A+", "A", "B", "C"] ) # Restrict to specific strategy tags (XAU / EUR_PB / GBP_PB / GENERIC). # Empty = take all. Use to A/B test whether GENERIC trades add or destroy edge. strategies_only: List[str] = field(default_factory=list) # Round-trip transaction cost in price units, charged on every trade # (XAU/USD retail spread ≈ $0.30). 0 disables. spread: float = 0.30 def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]: """Apply XAU weekend + allowed-weekday filter matching live `analyze_pair`. Forex market closed: Friday 22:00 UTC -> Sunday 22:00 UTC (Sydney open). Must match `_is_forex_weekend` in signal_aggregator.py exactly. """ xau = cfg.xau_strategy if xau.weekend_filter: wd = ts.weekday() hour = ts.hour if wd == 4 and hour >= 22: return "WEEKEND_FILTER" if wd == 5: return "WEEKEND_FILTER" if wd == 6 and hour < 22: return "WEEKEND_FILTER" if xau.allowed_weekdays and ts.weekday() not in xau.allowed_weekdays: return "DAY_FILTER" return None def run_backtest( bt_cfg: BacktestConfig, platform_cfg: Optional[PlatformConfig] = None, ) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]: """Replay history and collect trades.""" if platform_cfg is None: platform_cfg = PlatformConfig( symbol=bt_cfg.pair, display_name=PAIR_INFO[bt_cfg.pair]["display"], pip_value=PAIR_INFO[bt_cfg.pair]["pip_value"], primary_timeframe=bt_cfg.primary_tf, confirmation_timeframes=list(bt_cfg.confirmation_tfs), ) elif platform_cfg.primary_timeframe != bt_cfg.primary_tf: # A silent mismatch yields 0 trades: the replayer walks # bt_cfg.primary_tf bars while the aggregator looks for # platform_cfg.primary_timeframe in the slices and never finds it. raise ValueError( f"primary_tf mismatch: BacktestConfig={bt_cfg.primary_tf.value} " f"vs PlatformConfig={platform_cfg.primary_timeframe.value}" ) provider = ForexDataProvider(platform_cfg) # ── Load all required timeframes ── timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs) # Pad lookback so earliest TF has enough warmup for EMA200, ADX, etc. pad_days = max(60, bt_cfg.warmup_bars // 4) fetch_start = bt_cfg.start - timedelta(days=pad_days) dfs_full: Dict[Timeframe, pd.DataFrame] = {} for tf in timeframes: df = fetch_history( provider, bt_cfg.pair, tf, start=fetch_start, end=bt_cfg.end, use_cache=bt_cfg.use_cache, ) if df.empty: raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}") dfs_full[tf] = df # ── Pre-compute indicators on each TF ── t0 = time.perf_counter() caches: Dict[str, IndicatorCache] = { tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True) for tf in timeframes } logger.info(f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s") # ── Inject GPU engine into a fresh aggregator ── aggregator = SignalAggregator(platform_cfg) aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches) # ── Walk primary TF bar-by-bar ── primary_df = dfs_full[bt_cfg.primary_tf] # bt_cfg.start/end are already tz-aware UTC datetimes; specifying tz= # again would raise. Just wrap. bt_start_ts = pd.Timestamp(bt_cfg.start) bt_end_ts = pd.Timestamp(bt_cfg.end) # Start at warmup_bars OR first bar inside the bt window (whichever later). first_eligible_pos = max( bt_cfg.warmup_bars, int(primary_df.index.searchsorted(bt_start_ts)), ) last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right")) last_pos = min(last_pos, len(primary_df)) trades: List[Trade] = [] last_entry_pos = -10_000 skipped_reasons: Dict[str, int] = {} logger.info( f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: " f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}" ) primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf]) t_loop = time.perf_counter() for pos in range(first_eligible_pos, last_pos): ts = primary_df.index[pos] # Decision time = close of the current primary bar. decision_time = ts + primary_dur # XAU calendar filter (matches live analyze_pair behavior). # Live evaluates wall-clock time ≈ bar close, so filter on # decision_time, not the bar's open timestamp. if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD": block = _is_xau_blocked_by_calendar(decision_time, platform_cfg) if block: skipped_reasons[block] = skipped_reasons.get(block, 0) + 1 continue # Session window filter (matches live _apply_session_filter_live, # which checks datetime.now() ≈ bar close = decision_time). bt_filter = platform_cfg.backtest if bt_filter.session_filter: bar_hour = decision_time.hour if not (bt_filter.session_start_hour <= bar_hour < bt_filter.session_end_hour): skipped_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1 continue # Blocked hours (live checks datetime.now().hour ≈ bar close). # analyze_from_dataframes step 6b checks bar-open timestamp, # which is one bar behind — misses the last 5min of each hour. _is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD" _blocked = (platform_cfg.xau_strategy.blocked_hours_utc if _is_xau else bt_filter.blocked_hours_utc) if bar_hour in _blocked: skipped_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1 continue # Cooldown. if pos - last_entry_pos < bt_cfg.cooldown_bars: continue # Build sliced dfs per TF — only include bars fully closed by # decision_time. A bar timestamped T closes at T + bar_dur, so we # keep bars where T + bar_dur <= decision_time → T <= decision_time - bar_dur. # Without this, the H1 bar containing the current M15 close would be # included with future high/low/close values: a look-ahead leak. slice_dfs: Dict[Timeframe, pd.DataFrame] = {} for tf in timeframes: full = dfs_full[tf] bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf]) cutoff = decision_time - bar_dur cut = int(full.index.searchsorted(cutoff, side="right")) if cut == 0: continue slice_dfs[tf] = full.iloc[:cut] if bt_cfg.primary_tf not in slice_dfs: continue report: SignalReport = aggregator.analyze_from_dataframes( symbol=bt_cfg.pair, dfs=slice_dfs, context=None, use_news=False, ) if report.trade_setup is None: continue if report.signal_quality not in bt_cfg.quality_filter: continue # Strategy tag (from reasons[0] which the aggregator pre-pends). # Pullback SELLs mark bearish_reasons, so check both lists. tag = "GENERIC" first_bull = report.bullish_reasons[0] if report.bullish_reasons else "" first_bear = report.bearish_reasons[0] if report.bearish_reasons else "" if "XAU PULLBACK" in first_bull or "XAU PULLBACK" in first_bear: tag = "XAU_PB" elif "XAU STRATEGY" in first_bull: tag = "XAU" elif "EUR STRATEGY" in first_bull or "EUR STRATEGY" in first_bear: tag = "EUR" elif "GBP STRATEGY" in first_bull: tag = "GBP_PB" if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only: continue ts_setup: TradeSetup = report.trade_setup # Entry at the OPEN of the next bar — live signals fire at bar close, # execution happens on the next candle open (realistic fill). entry_pos = pos + 1 if entry_pos >= len(primary_df): continue entry_time = primary_df.index[entry_pos] entry_price = float(primary_df["Open"].iloc[entry_pos]) breakdown, agree, oppose, neutral, total = _build_indicator_breakdown( report, ts_setup.direction ) reasons = (report.bullish_reasons if ts_setup.direction == "BUY" else report.bearish_reasons) # Adjust SL/TP relative to actual entry (preserve distance from setup). sl_distance = abs(ts_setup.entry_price - ts_setup.stop_loss) tp1_distance = abs(ts_setup.take_profit_1 - ts_setup.entry_price) tp2_distance = abs(ts_setup.take_profit_2 - ts_setup.entry_price) if ts_setup.direction == "BUY": adj_sl = entry_price - sl_distance adj_tp1 = entry_price + tp1_distance adj_tp2 = entry_price + tp2_distance else: adj_sl = entry_price + sl_distance adj_tp1 = entry_price - tp1_distance adj_tp2 = entry_price - tp2_distance trades.append(Trade( pair=bt_cfg.pair, entry_time=entry_time, entry_price=entry_price, sl=adj_sl, tp1=adj_tp1, tp2=adj_tp2, direction=ts_setup.direction, pip_value=ts_setup.pip_value, risk_pips=sl_distance / ts_setup.pip_value, reward_pips=tp1_distance / ts_setup.pip_value, rr_planned=ts_setup.risk_reward, quality=report.signal_quality, final_score=report.final_score, final_confidence=report.final_confidence, market_regime=report.market_regime, volatility_regime=report.volatility_regime, strategy_tag=tag, entry_reasons=list(reasons), entry_warnings=list(report.warnings), indicator_breakdown=breakdown, indicators_agreeing=agree, indicators_opposing=oppose, indicators_neutral=neutral, indicators_total=total, trail_arm_r=(platform_cfg.signals.pb_trail_arm_r if tag == "XAU_PB" else 0.0), trail_dist_r=(platform_cfg.signals.pb_trail_dist_r if tag == "XAU_PB" else 0.0), )) last_entry_pos = pos logger.info( f"Replay complete: {len(trades)} entries in " f"{time.perf_counter() - t_loop:.1f}s " f"({(last_pos - first_eligible_pos) / max(time.perf_counter() - t_loop, 1e-6):.0f} bars/s)" ) if skipped_reasons: logger.info(f"Skipped bars by reason: {skipped_reasons}") # ── Resolve trade outcomes ── # Resolve against the RAW price path: cleaning removes >10σ spike bars # (news candles) that hit stops in reality. Cache hit → no API call. sim_df = fetch_history( provider, bt_cfg.pair, bt_cfg.primary_tf, start=fetch_start, end=bt_cfg.end, use_cache=bt_cfg.use_cache, clean=False, ) t_sim = time.perf_counter() simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars, spread=bt_cfg.spread, trail_arm_r=platform_cfg.signals.trail_arm_r, trail_dist_r=platform_cfg.signals.trail_dist_r) logger.info( f"Trade simulation took {time.perf_counter() - t_sim:.2f}s " f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})" ) return trades, dfs_full # ═════════════════════════════════════════════════════════════════════ # METRICS & REPORT # ═════════════════════════════════════════════════════════════════════ @dataclass class Metrics: n_trades: int n_wins: int n_losses: int n_timeout: int win_rate: float avg_r: float median_r: float total_r: float profit_factor: float sharpe: float max_drawdown_r: float avg_bars_held: float avg_mfe_r: float avg_mae_r: float by_quality: Dict[str, Dict] by_strategy: Dict[str, Dict] by_regime: Dict[str, Dict] def _summarize(rs: List[float]) -> Dict: if not rs: return {"n": 0, "win_rate": 0.0, "avg_r": 0.0, "total_r": 0.0} arr = np.array(rs) wins = (arr > 0).sum() return { "n": len(rs), "win_rate": float(wins / len(rs)) * 100, "avg_r": float(arr.mean()), "total_r": float(arr.sum()), } def compute_metrics(trades: List[Trade]) -> Metrics: resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"] if not resolved: return Metrics(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, {}, {}, {}) rs = np.array([t.r_multiple for t in resolved]) wins_mask = rs > 0 losses_mask = rs < 0 gross_profit = float(rs[wins_mask].sum()) if wins_mask.any() else 0.0 gross_loss = float(-rs[losses_mask].sum()) if losses_mask.any() else 0.0 pf = gross_profit / gross_loss if gross_loss > 0 else float("inf") cum = np.cumsum(rs) high_water = np.maximum.accumulate(cum) drawdown = high_water - cum max_dd = float(drawdown.max()) if len(drawdown) else 0.0 sharpe = float(rs.mean() / rs.std()) if rs.std() > 0 else 0.0 by_quality = {q: _summarize([t.r_multiple for t in resolved if t.quality == q]) for q in sorted({t.quality for t in resolved})} by_strategy = {s: _summarize([t.r_multiple for t in resolved if t.strategy_tag == s]) for s in sorted({t.strategy_tag for t in resolved})} by_regime = {r: _summarize([t.r_multiple for t in resolved if t.market_regime == r]) for r in sorted({t.market_regime for t in resolved})} return Metrics( n_trades=len(resolved), n_wins=int(wins_mask.sum()), n_losses=int(losses_mask.sum()), n_timeout=sum(1 for t in resolved if t.exit_reason == "TIMEOUT"), win_rate=float(wins_mask.mean()) * 100, avg_r=float(rs.mean()), median_r=float(np.median(rs)), total_r=float(rs.sum()), profit_factor=pf, sharpe=sharpe, max_drawdown_r=max_dd, avg_bars_held=float(np.mean([t.bars_held for t in resolved])), avg_mfe_r=float(np.mean([t.mfe_r for t in resolved])), avg_mae_r=float(np.mean([t.mae_r for t in resolved])), by_quality=by_quality, by_strategy=by_strategy, by_regime=by_regime, ) _DAY_NAMES = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri", 5: "Sat", 6: "Sun"} def _fmt_table(headers: List[str], rows: List[List[str]]) -> str: """Render a Markdown-style ASCII table with auto-sized columns.""" if not rows: return "" cols = len(headers) widths = [len(h) for h in headers] for r in rows: for i in range(cols): widths[i] = max(widths[i], len(str(r[i]))) def fmt_row(r): return "│ " + " │ ".join(str(r[i]).ljust(widths[i]) for i in range(cols)) + " │" sep = "├" + "┼".join("─" * (w + 2) for w in widths) + "┤" top = "┌" + "┬".join("─" * (w + 2) for w in widths) + "┐" bot = "└" + "┴".join("─" * (w + 2) for w in widths) + "┘" out = [top, fmt_row(headers), sep] out.extend(fmt_row(r) for r in rows) out.append(bot) return "\n".join(out) def print_report( pair: str, bt_cfg: BacktestConfig, m: Metrics, trades: Optional[List[Trade]] = None, detail: bool = False, ) -> None: bar = "═" * 78 print(f"\n{bar}") print(f" BACKTEST REPORT — {PAIR_INFO[pair]['display']} " f"({bt_cfg.start.date()} → {bt_cfg.end.date()})") print(f" Primary: {bt_cfg.primary_tf.value} " f"Confirms: {','.join(t.value for t in bt_cfg.confirmation_tfs)}") print(f"{bar}") if m.n_trades == 0: print(" No trades produced — quality filter rejected all signals.") print(f"{bar}\n") return # ── Summary table ── summary_rows = [ ["Trades", f"{m.n_trades}"], ["Wins / Losses / Timeouts", f"{m.n_wins} / {m.n_losses} / {m.n_timeout}"], ["Win rate", f"{m.win_rate:.1f}%"], ["Total R", f"{m.total_r:+.2f}"], ["Avg R / trade", f"{m.avg_r:+.3f}"], ["Median R", f"{m.median_r:+.3f}"], ["Profit factor", f"{m.profit_factor:.2f}"], ["Sharpe (R)", f"{m.sharpe:.2f}"], ["Max DD (R)", f"{m.max_drawdown_r:.2f}"], ["Avg bars held", f"{m.avg_bars_held:.1f}"], ["Avg MFE / MAE", f"{m.avg_mfe_r:.2f}R / {m.avg_mae_r:.2f}R"], ] print("\n ── Summary ──") print(_fmt_table(["Metric", "Value"], summary_rows)) def _grouped_table(title: str, groups: Dict[str, Dict]): if not groups: return rows = [] for k, s in groups.items(): rows.append([k, str(s["n"]), f"{s['win_rate']:5.1f}%", f"{s['avg_r']:+.3f}", f"{s['total_r']:+.2f}"]) print(f"\n ── {title} ──") print(_fmt_table([title.split(" by ")[-1], "n", "WR", "avgR", "totalR"], rows)) _grouped_table("Breakdown by Quality", m.by_quality) _grouped_table("Breakdown by Strategy", m.by_strategy) _grouped_table("Breakdown by Regime", m.by_regime) # ── Per-trade table ── if trades: resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"] if detail or len(resolved) <= 30: print("\n ── Trade Log ──") rows = [] for i, t in enumerate(resolved, 1): rows.append([ str(i), t.entry_time.strftime("%m-%d %a %H:%M"), t.direction, t.strategy_tag, t.quality, f"{t.indicators_agreeing}/{t.indicators_total}", f"{t.final_score:+.1f}", f"{t.final_confidence:.0f}", t.market_regime[:8], t.exit_reason, str(t.bars_held), f"{t.r_multiple:+.2f}", ]) print(_fmt_table( ["#", "Entry (UTC)", "Dir", "Strat", "Q", "Agree", "Score", "Conf", "Regime", "Exit", "Bars", "R"], rows, )) else: print(f"\n Trade log suppressed ({len(resolved)} trades). " f"Use --detail to print all, or open the CSV.") print(f"{bar}\n") def print_trade_reasoning(trades: List[Trade], limit: int = 5) -> None: """Per-instructions: dump the *why* for the first/last few trades — indicators on each TF, agreement count, reasons, warnings.""" resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"] if not resolved: return bar = "─" * 78 print(f"\n{'═' * 78}") print(f" TRADE REASONING (first {min(limit, len(resolved))} + last " f"{min(limit, len(resolved))} of {len(resolved)})") print('═' * 78) sample = resolved[:limit] + ([] if len(resolved) <= 2 * limit else resolved[-limit:]) for t in sample: print(f"\n{bar}") weekday = _DAY_NAMES.get(t.entry_time.weekday(), "?") print(f" {t.direction} {PAIR_INFO[t.pair]['display']} @ {t.entry_price:.5f} " f"({weekday} {t.entry_time:%Y-%m-%d %H:%M UTC})") print(f" Strategy={t.strategy_tag} Quality={t.quality} " f"Score={t.final_score:+.1f} Confidence={t.final_confidence:.0f} " f"Regime={t.market_regime}/{t.volatility_regime}") print(f" SL={t.sl:.5f} TP1={t.tp1:.5f} RR={t.rr_planned:.2f} " f"({t.risk_pips:.0f}pip risk → {t.reward_pips:.0f}pip target)") print(f" Indicator agreement: {t.indicators_agreeing}/{t.indicators_total} agree, " f"{t.indicators_opposing} oppose, {t.indicators_neutral} neutral") if t.indicator_breakdown: ind_rows = [] for name, b in t.indicator_breakdown.items(): tag = {"agree": "✓", "oppose": "✗", "neutral": "·"}.get(b["agrees"], "?") ind_rows.append([ name, b["signal"], f"{b['value']:.3f}" if b["value"] is not None else "—", f"{b['confidence']:.0f}", f"{b['weight']:.1f}", tag, b["description"][:40], ]) print(_fmt_table( ["Indicator", "Signal", "Value", "Conf", "Wt", "vs Dir", "Description"], ind_rows, )) if t.entry_reasons: print(f" Entry reasons:") for r in t.entry_reasons[:5]: print(f" • {r[:120]}") if t.entry_warnings: print(f" Warnings at entry:") for w in t.entry_warnings[:5]: print(f" ⚠ {w[:120]}") weekday_x = _DAY_NAMES.get(t.exit_time.weekday(), "?") if t.exit_time is not None else "?" print(f" → EXIT: {t.exit_reason} @ {t.exit_price:.5f} after {t.bars_held} bars " f"({weekday_x} {t.exit_time:%Y-%m-%d %H:%M UTC}) → R={t.r_multiple:+.2f}") print(f" MFE / MAE: {t.mfe_r:.2f}R / {t.mae_r:.2f}R") def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame: """Export trades for downstream analysis (e.g. parquet/CSV). Includes per-instructions: entry reasoning, indicator-by-indicator signal/value/agreement, and overall agreement count. """ rows = [] for t in trades: row = { "pair": t.pair, "strategy": t.strategy_tag, "quality": t.quality, "regime": t.market_regime, "vol_regime": t.volatility_regime, "direction": t.direction, "entry_time": t.entry_time, "entry_weekday": _DAY_NAMES.get(t.entry_time.weekday(), "?"), "entry": t.entry_price, "sl": t.sl, "tp1": t.tp1, "rr_planned": t.rr_planned, "score": t.final_score, "confidence": t.final_confidence, "indicators_agree": t.indicators_agreeing, "indicators_oppose": t.indicators_opposing, "indicators_neutral": t.indicators_neutral, "indicators_total": t.indicators_total, "entry_reasons": " || ".join(t.entry_reasons[:5]), "warnings": " || ".join(t.entry_warnings[:5]), "exit_time": t.exit_time, "exit": t.exit_price, "exit_reason": t.exit_reason, "bars_held": t.bars_held, "r_multiple": t.r_multiple, "mfe_r": t.mfe_r, "mae_r": t.mae_r, } # Flatten each indicator into columns so spreadsheets can sort/filter. for name, b in t.indicator_breakdown.items(): row[f"ind_{name}_signal"] = b["signal"] row[f"ind_{name}_value"] = b["value"] row[f"ind_{name}_conf"] = b["confidence"] row[f"ind_{name}_agrees"] = b["agrees"] rows.append(row) return pd.DataFrame(rows) # ═════════════════════════════════════════════════════════════════════ # CLI # ═════════════════════════════════════════════════════════════════════ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: p = argparse.ArgumentParser( description="GPU-accelerated backtester for ForexBot2", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) p.add_argument("--pair", required=True, help="XAUUSD | GOLD | EURUSD") p.add_argument("--days", type=int, default=180, help="Days of history (ignored if --start/--end set)") p.add_argument("--start", help="ISO date YYYY-MM-DD (overrides --days)") p.add_argument("--end", help="ISO date YYYY-MM-DD (defaults to today)") p.add_argument("--primary-tf", default="15min", help="Primary TF (1min/5min/15min/1h/4h)") p.add_argument("--confirm-tfs", default="1h,4h", help="Comma-separated confirmation TFs") p.add_argument("--warmup-bars", type=int, default=250) p.add_argument("--cooldown-bars", type=int, default=5) p.add_argument("--max-hold-bars", type=int, default=300) p.add_argument("--swing-sr-lookback", type=int, default=0, help="Override TechnicalConfig.swing_sr_lookback (0 = use config default)") p.add_argument("--quality", default="A+,A,B,C", help="Comma-separated allowed grades") p.add_argument("--strategies", default="", help="Comma-separated strategy tags to include (XAU,XAU_PB,EUR,GENERIC). Empty = all.") p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)") p.add_argument("--spread", type=float, default=None, help="Round-trip cost in price units per trade (default: pair's " "default_spread from PAIR_INFO — XAU $0.30, EUR 0.0001; 0 disables)") p.add_argument("--export-csv", help="Optional path to dump per-trade CSV (incl. indicator breakdown)") p.add_argument("--detail", action="store_true", help="Print full per-trade reasoning (indicators, signals, agreement, why entered)") p.add_argument("--reasoning-samples", type=int, default=3, help="If --detail not set, print this many sample trades' full reasoning") p.add_argument("--log-level", default="INFO") return p.parse_args(argv) def _resolve_pair(s: str) -> str: s_up = s.upper().strip() if s_up in PAIR_ALIASES: return PAIR_ALIASES[s_up] if s_up in PAIR_INFO: return s_up raise SystemExit(f"Unknown pair '{s}'. Try one of: {', '.join(sorted(PAIR_ALIASES))}") def _resolve_tf(s: str) -> Timeframe: for tf in Timeframe: if tf.value == s: return tf raise SystemExit(f"Unknown timeframe '{s}'. Valid: {[t.value for t in Timeframe]}") def main(argv: Optional[List[str]] = None) -> int: args = _parse_args(argv) logging.basicConfig( level=args.log_level.upper(), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) pair = _resolve_pair(args.pair) primary_tf = _resolve_tf(args.primary_tf) confirm_tfs = [_resolve_tf(s.strip()) for s in args.confirm_tfs.split(",") if s.strip()] end = ( datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc) if args.end else datetime.now(timezone.utc) ) if args.start: start = datetime.fromisoformat(args.start).replace(tzinfo=timezone.utc) else: start = end - timedelta(days=args.days) spread = (args.spread if args.spread is not None else PAIR_INFO[pair].get("default_spread", 0.30)) bt_cfg = BacktestConfig( pair=pair, start=start, end=end, primary_tf=primary_tf, confirmation_tfs=confirm_tfs, warmup_bars=args.warmup_bars, cooldown_bars=args.cooldown_bars, max_hold_bars=args.max_hold_bars, use_cache=not args.no_cache, quality_filter=[q.strip() for q in args.quality.split(",") if q.strip()], strategies_only=[s.strip() for s in args.strategies.split(",") if s.strip()], spread=spread, ) logger.info( f"Backend: {'CuPy GPU' if _GPU_AVAILABLE else 'NumPy CPU fallback'} | " f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()} | " f"Spread: {spread}" ) # Per-pair tuned configs come from the pair_signals factories (regime # weights, strategy gates, sessions). Unregistered pairs fall back to # the inline construction. from pair_signals import PAIR_PRODUCERS # noqa: E402 (after sys.path fix) factory = PAIR_PRODUCERS.get(pair) if factory is not None: platform_cfg = factory(primary_tf, confirm_tfs) else: platform_cfg = PlatformConfig( symbol=pair, display_name=PAIR_INFO[pair]["display"], pip_value=PAIR_INFO[pair]["pip_value"], primary_timeframe=primary_tf, confirmation_timeframes=confirm_tfs, ) if args.swing_sr_lookback > 0: platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback trades, _ = run_backtest(bt_cfg, platform_cfg=platform_cfg) metrics = compute_metrics(trades) print_report(pair, bt_cfg, metrics, trades=trades, detail=args.detail) print_trade_reasoning(trades, limit=args.reasoning_samples) if args.export_csv: df = trades_to_dataframe(trades) df.to_csv(args.export_csv, index=False) logger.info(f"Exported {len(df)} trades → {args.export_csv}") return 0 if __name__ == "__main__": sys.exit(main())