From d9b0b19c025430ce6e65f7a8eec454848679d3f8 Mon Sep 17 00:00:00 2001 From: gavindiaz Date: Wed, 21 Oct 2026 20:55:45 +0800 Subject: [PATCH] feat(backtest): live-parity backtest engine with zero-lookahead AI/regime lookups Phase 1 of the prove-before-live alignment. The backtest engine now mirrors the live signal path instead of hardcoding HOLD/NEUTRAL: - predict_batch (signal_classifier): vectorized per-bar AI signals reusing the exact live predict() pipeline (class-index mapping, conf<0.5 -> HOLD) - backtester: per-bar time-aligned AI lookup (M15 close = signal-available time, searchsorted, no lookahead), exogenous H4 regime plan with dynamic confluence-key deltas, AttentionMarketScorer alignment, per-strategy entry cooldown, max-hold timeout exit, breakeven + trailing capped at TP, running- equity pnl_pct (was: static initial balance) - fix: store direction.value on entry (Enum vs str compare silently disabled every exit - no trade ever closed) - fix: _check_mode_sync only clears override on genuine mode transition (grid threshold injection was wiped on first read) - SignalAggregator(use_adaptive_weights=True): backtest uses pure config weights so garbage live adaptive weights can't contaminate gate scores Co-Authored-By: Claude Code --- ai/signal_classifier.py | 57 +++++++ backtesting/backtester.py | 271 ++++++++++++++++++++++++++------ strategies/base_strategy.py | 21 ++- strategies/signal_aggregator.py | 27 ++-- 4 files changed, 315 insertions(+), 61 deletions(-) diff --git a/ai/signal_classifier.py b/ai/signal_classifier.py index 74bd718..95abf80 100644 --- a/ai/signal_classifier.py +++ b/ai/signal_classifier.py @@ -166,3 +166,60 @@ class SignalClassifier: except Exception as e: logger.error(f"XGBoost prediction error: {e}") return {"direction": "HOLD", "confidence": 0.0} + + def predict_batch(self, df: pd.DataFrame) -> pd.DataFrame: + """ + Vectorized per-bar BUY/SELL/HOLD predictions for backtesting. + + Uses the exact same feature pipeline as predict() (create_ml_features, + missing-column fill-0, confidence<0.5 → HOLD), so backtest AI signals + match live inference one-to-one. Features are trailing-window only → + no lookahead bias. Rows whose feature window is incomplete (NaN) are + marked direction=HOLD / confidence=0.0 (no prediction). + + Returns: + DataFrame indexed like df with columns: + direction, confidence, p_buy, p_hold, p_sell + """ + if self._model is None: + if not self.load_model(): + raise RuntimeError("XGBoost model not loaded — run scripts/train_ai.py first") + + ta = TechnicalAnalyzer() + feature_df = ta.create_ml_features(df) + + for col in self._feature_columns: + if col not in feature_df.columns: + feature_df[col] = 0 + + X = feature_df[self._feature_columns].astype(float) + valid_mask = X.notna().all(axis=1).values + n = len(X) + + probs = np.zeros((n, 3)) + if valid_mask.any(): + proba_v = self._model.predict_proba(X.loc[valid_mask].values) + # Map model class indices (self._model.classes_) onto slots 0/1/2 + for k, cls in enumerate(self._model.classes_): + probs[valid_mask, cls] = proba_v[:, k] + + predicted = probs.argmax(axis=1) + confidence = probs.max(axis=1) + direction_map = {0: "BUY", 1: "HOLD", 2: "SELL"} + + out = pd.DataFrame( + { + "direction": [ + direction_map[p] if c >= 0.5 else "HOLD" + for p, c in zip(predicted, confidence) + ], + "confidence": confidence, + "p_buy": probs[:, 0], + "p_hold": probs[:, 1], + "p_sell": probs[:, 2], + }, + index=df.index, + ) + # Rows with incomplete feature windows → no prediction (HOLD) + out.loc[~valid_mask, ["direction", "confidence"]] = ["HOLD", 0.0] + return out diff --git a/backtesting/backtester.py b/backtesting/backtester.py index 3cea1e4..e5773f8 100644 --- a/backtesting/backtester.py +++ b/backtesting/backtester.py @@ -1,13 +1,20 @@ """ -Backtesting Engine for XAUUSD AI Trading Bot +Backtesting Engine for XAUUSD AI Trading Bot — Live-Parity Edition Simulates realistic multi-strategy execution against historical price data: -- Zero lookahead bias (evaluates bars sequentially) -- Dynamic ATR-based Stop Loss & Take Profit -- Trailing stop simulation +- Zero lookahead bias (evaluates bars sequentially, trailing-only features) +- Real XGBoost AI prediction per bar (predict_batch lookup, same path as live signal_classifier.predict) +- Real confluence scoring incl. dynamic RANGING/HIGH-vol threshold deltas (mirrors live aggregator) +- Temporal Self-Attention alignment (same AttentionMarketScorer as live) +- Dynamic ATR-based Stop Loss & Take Profit + breakeven + trailing (capped at TP) +- Entry cooldown + max-hold timeout exits - Realistic spread & slippage modeling -- Confluence score filtering - Detailed quantitative performance reporting + +Known deviations from live (documented): single concurrent position (live allows up to +max_concurrent=4, so concurrency compression understates/overstates risk depending on +correlation); trade_learner mistake-guard bypassed (live weights are garbage — see plan +phase 3); sentiment fixed at NEUTRAL/0 unless overridden. """ import math @@ -37,19 +44,43 @@ class Backtester: lot_size: float = 0.02, spread_points: float = 0.25, # ~$0.25 on gold slippage_points: float = 0.05, - min_confluence_score: float = 65.0, + min_confluence_score: float = 50.0, trailing_stop_enabled: bool = True, + # ── Live-parity addons ─────────────────────────────────────────── + ai_lookup: Optional[pd.DataFrame] = None, # per-bar AI (direction/confidence), positional with df + regime_plan: Optional[pd.DataFrame] = None, # per-bar regime (regime/volatility/adx), positional with df + sentiment: Optional[dict] = None, # {"direction","score"} for aggregation + attention_enabled: bool = True, # mirror live AttentionMarketScorer + cooldown_bars: int = 0, # bars to wait after an exit before same-strategy re-entry + max_hold_bars: int = 0, # 0 = disabled; else time-based exit at close + breakeven_rr: float = 0.0, # 0 = disabled; else move SL to entry at this RR of original SL + atr_trail_multiplier: float = 1.2, # trail distance = multiplier × bar ATR + trail_trigger_atr: float = 1.2, # start trailing after this much unrealized gain (× ATR) + tick: float = 0.01, # XAUUSD price tick ): self.initial_balance = initial_balance self.lot_size = lot_size self.spread = spread_points self.slippage = slippage_points - self.min_confluence = min_confluence_score + self.min_confluence = min_confluence_score # fallback; live prefers per-strategy property self.trailing_enabled = trailing_stop_enabled + self.ai_lookup = ai_lookup + self.regime_plan = regime_plan + self.sentiment = sentiment or {"direction": "NEUTRAL", "score": 0.0} + self.attention_enabled = attention_enabled + self.cooldown_bars = cooldown_bars + self.max_hold_bars = max_hold_bars + self.breakeven_rr = breakeven_rr + self.atr_trail_multiplier = atr_trail_multiplier + self.trail_trigger_atr = trail_trigger_atr + self.tick = tick self.ta = TechnicalAnalyzer() self.regime_detector = RegimeDetector() - self.aggregator = SignalAggregator() + # Pure config weights for the gate — never the corrupted live adaptive ones + self.aggregator = SignalAggregator(use_adaptive_weights=False) + from ai.attention_scorer import AttentionMarketScorer + self.attention_scorer = AttentionMarketScorer(window=25) self.strategies = { "scalping": ScalpingStrategy(), @@ -109,6 +140,9 @@ class Backtester: def run(self, df: pd.DataFrame, warmup_bars: int = 200) -> PerformanceMetrics: """ Run the backtest across historical bars. + + `ai_lookup` and `regime_plan` (if set on the instance) must be positionally + aligned to `df` (same number of rows, same order). """ if len(df) <= warmup_bars: raise ValueError(f"Dataframe length ({len(df)}) must exceed warmup_bars ({warmup_bars})") @@ -118,23 +152,84 @@ class Backtester: # Precalculate technical indicators on the full dataset df_indicators = self.ta.add_all_indicators(df.copy()) + # ── AI lookup: time-based (M5 bar → last CLOSED M15 signal) ──────── + # Model features at M15 row j use trailing windows only, so the signal + # is known at bar close = open + 15m. Live bot also evaluates on closed + # M15 bars. If no "time" column exists (synthetic tests), fall back to + # positional pairing with the M5 rows. + def _ai_lookup_fn(src: Optional[pd.DataFrame]): + if src is None or len(src) == 0: + return None + src = src.reset_index(drop=True) + if "time" in src.columns: + opens = pd.to_datetime(src["time"]).values + closes = opens + np.timedelta64(15, "m") # M15 close = signal-available time + order = np.argsort(closes) + closes_sorted = closes[order] + dirs = src["direction"].values[order] + confs = src["confidence"].values.astype(float)[order] + + def lookup(bar_time): + pos = np.searchsorted(closes_sorted, np.datetime64(bar_time), side="right") - 1 + if pos < 0: + return {"direction": "HOLD", "confidence": 0.0} + return {"direction": str(dirs[pos]), "confidence": float(confs[pos])} + + return lookup + else: + # positional fallback (tests): assume src rows align with M5 rows + rows = list(zip(src["direction"].astype(str), src["confidence"].astype(float))) + pos_idx = [0] + + def lookup(bar_time): + row = rows[min(pos_idx[0], len(rows) - 1)] + pos_idx[0] += 1 + return {"direction": row[0], "confidence": row[1]} + + return lookup + ai_lookup_fn = _ai_lookup_fn(self.ai_lookup) + + def _regime_iter(): + if self.regime_plan is not None and len(self.regime_plan) == len(df_indicators): + plan = self.regime_plan.reset_index(drop=True) + for _, row in plan.iterrows(): + yield { + "regime": row["regime"], + "volatility": row["volatility"], + "adx": float(row["adx"]) if pd.notna(row["adx"]) else 15.0, + "should_trade": bool(row["should_trade"]) if "should_trade" in row else True, + } + else: + while True: + yield None + + regime_gen = _regime_iter() + closed_trades: List[BacktestTrade] = [] open_position: Optional[dict] = None trade_counter = 0 + running_equity = self.initial_balance + last_exit_bar: Dict[str, int] = {name: -10**9 for name in self.strategies} - # Point value for XAUUSD (1 standard lot = 100 oz, so 1 point move = $100 on 1.0 lot, $100 * lot_size) + # Point value for XAUUSD (1 standard lot = 100 oz, 1 point move = $100 on 1.0 lot) dollar_per_point = 100.0 * self.lot_size for i in range(warmup_bars, len(df_indicators)): current_bar = df_indicators.iloc[i] - prev_bar = df_indicators.iloc[i - 1] sub_df = df_indicators.iloc[:i + 1] - current_high = current_bar["high"] - current_low = current_bar["low"] - current_close = current_bar["close"] + current_high = float(current_bar["high"]) + current_low = float(current_bar["low"]) + current_close = float(current_bar["close"]) current_time = current_bar.get("time", i) atr = float(current_bar.get("atr", 3.0)) + if not math.isfinite(atr) or atr <= 0: + atr = 3.0 + + # ── Per-bar exogenous signals (time-aligned) ──────────────── + ai_prediction = {"direction": "HOLD", "confidence": 0.0} + if ai_lookup_fn is not None: + ai_prediction = ai_lookup_fn(current_time) # ── 1. Manage existing open position ─────────────────────── if open_position is not None: @@ -144,48 +239,76 @@ class Backtester: entry_price = pos["entry_price"] sl = pos["stop_loss"] tp = pos["take_profit"] + sl_orig = pos["stop_loss_original"] closed = False exit_price = 0.0 exit_reason = "" - # Check SL and TP within bar - if direction == "BUY": - # Check Stop Loss first (conservative) + # Timeout exit path (smart-exit culture: don't hold forever) + if self.max_hold_bars > 0 and pos["bars_held"] >= self.max_hold_bars: + exit_price = round(current_close, 2) + exit_reason = "TIMEOUT" + closed = True + elif direction == "BUY": + # Stop Loss first (conservative within-bar ordering) if current_low <= sl: - exit_price = sl - self.slippage + exit_price = round(sl - self.slippage, 2) exit_reason = "STOP_LOSS" closed = True elif current_high >= tp: - exit_price = tp + exit_price = round(tp, 2) exit_reason = "TAKE_PROFIT" closed = True - elif self.trailing_enabled: - # Update Trailing Stop if price moved in profit by 1.5 ATR - gain = current_close - entry_price - if gain > 1.5 * atr: - new_sl = current_close - 1.2 * atr - if new_sl > pos["stop_loss"]: - pos["stop_loss"] = new_sl - elif direction == "SELL": if current_high >= sl: - exit_price = sl + self.slippage + exit_price = round(sl + self.slippage, 2) exit_reason = "STOP_LOSS" closed = True elif current_low <= tp: - exit_price = tp + exit_price = round(tp, 2) exit_reason = "TAKE_PROFIT" closed = True - elif self.trailing_enabled: - gain = entry_price - current_close - if gain > 1.5 * atr: - new_sl = current_close + 1.2 * atr - if new_sl < pos["stop_loss"]: - pos["stop_loss"] = new_sl + + if not closed: + # ── Breakeven: lock in entry once price reaches breakeven_rr × original SL ── + if self.breakeven_rr > 0 and sl_orig is not None: + sl_distance = abs(entry_price - sl_orig) + if sl_distance > 0: + if direction == "BUY": + gain_now = current_close - entry_price + if gain_now >= sl_distance * self.breakeven_rr: + be = round(entry_price + self.tick, 2) + if be > sl: + sl = be + else: + gain_now = entry_price - current_close + if gain_now >= sl_distance * self.breakeven_rr: + be = round(entry_price - self.tick, 2) + if be < sl: + sl = be + + # ── Trailing stop: activate after trail_trigger_atr × ATR profit ── + if self.trailing_enabled: + if direction == "BUY": + gain = current_close - entry_price + if gain > self.trail_trigger_atr * atr: + new_sl = round(current_close - self.atr_trail_multiplier * atr, 2) + new_sl = min(new_sl, round(tp - self.tick, 2)) # never trail past TP + if new_sl > sl: + sl = new_sl + else: + gain = entry_price - current_close + if gain > self.trail_trigger_atr * atr: + new_sl = round(current_close + self.atr_trail_multiplier * atr, 2) + new_sl = max(new_sl, round(tp + self.tick, 2)) # never trail past TP + if new_sl < sl: + sl = new_sl + + pos["stop_loss"] = sl if closed: - # Calculate P&L + # ── Calculate P&L ────────────────────────────────── if direction == "BUY": points_diff = exit_price - entry_price else: @@ -193,7 +316,7 @@ class Backtester: pnl_dollars = points_diff * dollar_per_point pips = points_diff * 10.0 # 1 pip = 0.1 on gold - pnl_pct = (pnl_dollars / self.initial_balance) * 100.0 + pnl_pct = (pnl_dollars / running_equity) * 100.0 if running_equity > 0 else 0.0 trade = BacktestTrade( trade_id=pos["trade_id"], @@ -203,7 +326,7 @@ class Backtester: entry_time=pos["entry_time"], exit_time=current_time, entry_price=round(entry_price, 2), - exit_price=round(exit_price, 2), + exit_price=exit_price, lot_size=self.lot_size, pnl=round(pnl_dollars, 2), pnl_pips=round(pips, 1), @@ -213,35 +336,68 @@ class Backtester: bars_held=pos["bars_held"], ) closed_trades.append(trade) + running_equity += pnl_dollars + last_exit_bar[pos["strategy"]] = i open_position = None # ── 2. Evaluate new entries if no position is open ──────── if open_position is None: - regime = self.regime_detector.analyze(sub_df) - if not regime.should_trade: + regime = next(regime_gen) + should_trade = True + if regime is not None: + should_trade = regime["should_trade"] + else: + regime_obj = self.regime_detector.analyze(sub_df) + should_trade = regime_obj.should_trade + regime = { + "regime": regime_obj.regime.value, + "volatility": regime_obj.volatility_label, + "adx": regime_obj.adx, + "should_trade": should_trade, + "position_modifier": regime_obj.position_size_modifier, + } + + if not should_trade: continue best_signal = None best_confluence = None for strat_name, strat in self.strategies.items(): - if not self.regime_detector.is_strategy_recommended(strat_name, regime): + # Entry cooldown: wait cooldown_bars after that strategy's last exit + if self.cooldown_bars > 0 and i - last_exit_bar[strat_name] < self.cooldown_bars: + continue + + # Match live Step 6 regime recommendation gate + if not self.regime_detector.is_strategy_recommended(strat_name, self._to_regime_obj(regime, sub_df)): continue sig = strat.run(sub_df) if not sig.is_actionable: continue - # Confluence check + # Mirror live Step 6: attention on the strategy's own frame + attention_analysis = None + if self.attention_enabled: + attention_analysis = self.attention_scorer.analyze_alignment( + df=sub_df, + signal_direction=sig.direction.value, + ) + + # Confluence — live passes is_recommended=True hardcoded + real regime keys confluence = self.aggregator.calculate_confluence( strategy_signal=sig, - ai_prediction={"direction": "HOLD", "confidence": 0.0}, - sentiment={"direction": "NEUTRAL", "score": 0.0}, + ai_prediction=ai_prediction, + sentiment=self.sentiment, regime_analysis={ "is_recommended": True, - "position_modifier": regime.position_size_modifier, + "position_modifier": regime.get("position_modifier", 1.0), + "regime": regime["regime"], + "adx": regime.get("adx", 15.0), + "volatility": regime["volatility"], }, - min_score=self.min_confluence, + attention_analysis=attention_analysis, + min_score=strat.min_confluence_score, ) if confluence.should_execute: @@ -250,7 +406,11 @@ class Backtester: best_confluence = confluence if best_signal and best_confluence and best_confluence.should_execute: - direction = best_signal.direction + # Store string direction (live stores signal.direction.value) — the exit + # block compares against "BUY"/"SELL" strings. Storing the raw Enum here + # silently disabled every exit: "Enum == str" is always False, so no trade + # ever closed and open_position blocked all subsequent entries. + direction = best_signal.direction.value raw_price = current_close # Apply realistic execution spread & slippage @@ -271,6 +431,7 @@ class Backtester: "entry_time": current_time, "entry_price": entry_price, "stop_loss": sl, + "stop_loss_original": sl, "take_profit": tp, "confluence_score": best_confluence.total_score, "bars_held": 0, @@ -281,6 +442,22 @@ class Backtester: self._print_report(metrics) return metrics + def _to_regime_obj(self, regime: dict, sub_df: pd.DataFrame): + """Wrap an exogenous regime dict as a RegimeAnalysis for is_strategy_recommended.""" + try: + from strategies.regime_detector import MarketRegime, RegimeAnalysis + if not isinstance(regime.get("regime"), str): + return self.regime_detector.analyze(sub_df) + return RegimeAnalysis( + regime=MarketRegime(regime["regime"]), + adx_value=float(regime.get("adx", 15.0)), + volatility_label=regime.get("volatility", "NORMAL"), + position_size_modifier=regime.get("position_modifier", 1.0), + should_trade=regime.get("should_trade", True), + ) + except Exception: + return self.regime_detector.analyze(sub_df) + def _print_report(self, m: PerformanceMetrics): """Display a formatted quantitative performance report.""" border = "=" * 62 @@ -319,4 +496,4 @@ class Backtester: lines.append(" (No completed trades executed)") lines.append(border) lines.append("") - print("\n".join(lines)) + print("\n".join(lines)) \ No newline at end of file diff --git a/strategies/base_strategy.py b/strategies/base_strategy.py index 7180857..e61cc6b 100644 --- a/strategies/base_strategy.py +++ b/strategies/base_strategy.py @@ -73,16 +73,25 @@ class BaseStrategy(ABC): self.enabled = params.get("enabled", True) self._last_active_mode: Optional[str] = None self._min_confluence_score_override: Optional[float] = None + self._min_technical_score_override: Optional[float] = None self._default_sessions = params.get("active_sessions", []) self._last_signal: Optional[TradingSignal] = None def _check_mode_sync(self): - """Ensure overrides are reset if the trading mode changes.""" + """Ensure overrides are reset if the trading mode changes. + + Only clear on a genuine mode *transition*. `_last_active_mode` starts + as None, so naively treating None != current_mode as a change wiped + overrides set before the first property read — e.g. the backtester's + grid injection of min_confluence_score silently fell back to the YAML + default (every threshold produced identical results). + """ try: from config.settings import get_settings current_mode = get_settings().active_mode - if getattr(self, "_last_active_mode", None) != current_mode: - self._last_active_mode = current_mode + prev = getattr(self, "_last_active_mode", None) + self._last_active_mode = current_mode + if prev is not None and prev != current_mode: self._min_confluence_score_override = None except Exception: pass @@ -135,6 +144,8 @@ class BaseStrategy(ABC): @property def min_technical_score(self) -> float: """Threshold for the strategy itself to emit a directional signal instead of HOLD.""" + if self._min_technical_score_override is not None: + return self._min_technical_score_override try: from config.settings import get_settings settings = get_settings() @@ -148,6 +159,10 @@ class BaseStrategy(ABC): pass return 35.0 + @min_technical_score.setter + def min_technical_score(self, val: float): + self._min_technical_score_override = val + @abstractmethod def analyze(self, df: pd.DataFrame) -> dict: """ diff --git a/strategies/signal_aggregator.py b/strategies/signal_aggregator.py index e276f5b..f8a43b3 100644 --- a/strategies/signal_aggregator.py +++ b/strategies/signal_aggregator.py @@ -43,10 +43,11 @@ class SignalAggregator: The confluence score determines whether a trade should be executed. """ - def __init__(self): + def __init__(self, use_adaptive_weights: bool = True): settings = get_settings() weights = settings.confluence_weights + self.use_adaptive_weights = use_adaptive_weights self.technical_weight = weights.get("technical_weight", 40) self.ai_weight = weights.get("ai_prediction_weight", 25) self.sentiment_weight = weights.get("sentiment_weight", 20) @@ -83,16 +84,20 @@ class SignalAggregator: Returns: ConfluenceResult with final score and execution decision. """ - # Dynamically synchronize with self-learning adaptive weights - try: - from ai.trade_learner import trade_learner - cw = trade_learner.get_confluence_weights() - self.technical_weight = cw.get("technical", self.technical_weight) - self.ai_weight = cw.get("ai_prediction", self.ai_weight) - self.sentiment_weight = cw.get("sentiment", self.sentiment_weight) - self.regime_weight = cw.get("regime", self.regime_weight) - except Exception: - pass + # Dynamically synchronize with self-learning adaptive weights. + # Backtesting uses use_adaptive_weights=False so the gate measures the + # pure config weights — live garbage adaptive weights (leak #3) would + # otherwise contaminate every backtest confluence score. + if self.use_adaptive_weights: + try: + from ai.trade_learner import trade_learner + cw = trade_learner.get_confluence_weights() + self.technical_weight = cw.get("technical", self.technical_weight) + self.ai_weight = cw.get("ai_prediction", self.ai_weight) + self.sentiment_weight = cw.get("sentiment", self.sentiment_weight) + self.regime_weight = cw.get("regime", self.regime_weight) + except Exception: + pass reasons = list(strategy_signal.reasons) # ── 1. Technical Score (from strategy) ───────────────────────────