diff --git a/.gitignore b/.gitignore index dc56bd6..21c7d28 100644 --- a/.gitignore +++ b/.gitignore @@ -21,5 +21,7 @@ Thumbs.db # Logs *.log -# Backtesting (scripts, data, results) -backtest/ +# Backtesting data (keep scripts, ignore generated files) +backtest/cache/ +backtest/*.csv +backtest/__pycache__/ diff --git a/backtest/__init__.py b/backtest/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backtest/diagnose_filters.py b/backtest/diagnose_filters.py new file mode 100644 index 0000000..96b688b --- /dev/null +++ b/backtest/diagnose_filters.py @@ -0,0 +1,609 @@ +""" +ForexBot2 — Filter Rejection Diagnostic +======================================== +Replays the same backtest loop as gpu_backtest.py but instead of +collecting trades, counts how many M15 bars are rejected by each +gate in the signal pipeline. + +Produces a summary table showing where potential trades are being +lost, helping tune filter thresholds. + +Usage: + python backtest/diagnose_filters.py --pair XAUUSD --days 365 + python backtest/diagnose_filters.py --pair XAUUSD --days 365 --sample-every 1 +""" + +from __future__ import annotations + +import argparse +import logging +import sys +import time +from collections import Counter, defaultdict +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. +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 + PAIR_INFO, + PlatformConfig, + SignalStrength, + Timeframe, +) +from signal_aggregator import SignalAggregator, SignalReport # noqa: E402 + +# Reuse GPU backtest infrastructure (data loading, indicator pre-compute, etc.) +from gpu_backtest import ( # noqa: E402 + BAR_SECONDS, + GPUTechnicalEngine, + PAIR_ALIASES, + BacktestConfig, + IndicatorCache, + fetch_history, + precompute_indicators, + _is_xau_blocked_by_calendar, +) +from data_provider import ForexDataProvider # noqa: E402 + +logger = logging.getLogger("diagnose_filters") + + +# ═════════════════════════════════════════════════════════════════════ +# OUTCOME CLASSIFICATION +# ═════════════════════════════════════════════════════════════════════ + +def classify_bar( + report: SignalReport, + is_xau: bool, +) -> Tuple[str, List[str]]: + """Classify a bar's outcome and list which filters killed it. + + Returns: + (outcome, [filter_names]) + + Outcomes: + "TRADE" — quality C+ and trade_setup exists + "QUALITY_OK_NO_SETUP" — quality C+ but trade_setup is None + "LOW_QUALITY" — signal reached BUY/SELL but quality D or F + "NEUTRAL" — signal stayed NEUTRAL (score didn't reach +/-30) + + Filter names are extracted from report.warnings. + """ + filters_hit: List[str] = [] + + # Scan warnings for known filter tags + filter_tags = { + "H4 RSI GATE": "H4_RSI_GATE", + "H1 MACD GATE": "H1_MACD_GATE", + "RSI ZONE GATE": "RSI_ZONE_GATE", + "XAU BUY FILTER": "XAU_BUY_FILTER", + "SESSION FILTER": "SESSION_FILTER", + "DAY FILTER": "DAY_FILTER", + "LOW ATR FILTER": "LOW_ATR_FILTER", + "SELL SUPPRESSED": "SELL_SUPPRESSED", + "EXHAUSTION": "EXHAUSTION_PENALTY", + "TREND EXHAUSTION": "TREND_EXHAUSTION", + "NO PULLBACK": "NO_PULLBACK_PENALTY", + "Weak pullback": "WEAK_PULLBACK_PENALTY", + "Extended entry": "EXTENDED_ENTRY_PENALTY", + "Counter-trend": "COUNTER_TREND_PENALTY", + "EXTREME volatility":"EXTREME_VOL_PENALTY", + "NEAR EMA200": "NEAR_EMA200_PENALTY", + "FAR FROM EMA200": "FAR_EMA200_PENALTY", + "EVENT RISK": "EVENT_RISK", + "REGIME FILTER": "REGIME_FILTER", + "Timeframes CONFLICT": "MTF_CONFLICT", + } + + for w in report.warnings: + for tag, name in filter_tags.items(): + if tag in w: + filters_hit.append(name) + + # Determine outcome + non_neutral = report.final_signal not in (SignalStrength.NEUTRAL,) + quality_ok = report.signal_quality in ("A+", "A", "B", "C") + + if quality_ok and report.trade_setup is not None: + return "TRADE", filters_hit + elif quality_ok and report.trade_setup is None: + return "QUALITY_OK_NO_SETUP", filters_hit + elif non_neutral and not quality_ok: + return "LOW_QUALITY", filters_hit + else: + return "NEUTRAL", filters_hit + + +# ═════════════════════════════════════════════════════════════════════ +# SCORE DISTRIBUTION TRACKER +# ═════════════════════════════════════════════════════════════════════ + +class ScoreTracker: + """Tracks score/confidence distributions for analysis.""" + + def __init__(self): + self.scores: List[float] = [] + self.confidences: List[float] = [] + self.abs_scores: List[float] = [] + # How many bars had score in various ranges + self.score_bins = Counter() + self.conf_bins = Counter() + # Track "near miss" bars — score close to thresholds + self.near_miss_score: int = 0 # score 20-29 (almost BUY/SELL) + self.near_miss_conf: int = 0 # confidence 45-54 (almost passing) + + def record(self, score: float, confidence: float): + self.scores.append(score) + self.confidences.append(confidence) + self.abs_scores.append(abs(score)) + + # Score bins + a = abs(score) + if a < 10: + self.score_bins["0-9"] += 1 + elif a < 20: + self.score_bins["10-19"] += 1 + elif a < 30: + self.score_bins["20-29"] += 1 + self.near_miss_score += 1 + elif a < 40: + self.score_bins["30-39"] += 1 + elif a < 50: + self.score_bins["40-49"] += 1 + else: + self.score_bins["50+"] += 1 + + # Confidence bins + if confidence < 30: + self.conf_bins["0-29"] += 1 + elif confidence < 45: + self.conf_bins["30-44"] += 1 + elif confidence < 55: + self.conf_bins["45-54"] += 1 + self.near_miss_conf += 1 + elif confidence < 65: + self.conf_bins["55-64"] += 1 + elif confidence < 75: + self.conf_bins["65-74"] += 1 + else: + self.conf_bins["75+"] += 1 + + +# ═════════════════════════════════════════════════════════════════════ +# MAIN DIAGNOSTIC LOOP +# ═════════════════════════════════════════════════════════════════════ + +def run_diagnosis( + pair: str, + start: datetime, + end: datetime, + sample_every: int = 4, + use_cache: bool = True, + primary_tf: Timeframe = Timeframe.M15, + confirm_tfs: Optional[List[Timeframe]] = None, + warmup_bars: int = 250, +) -> None: + if confirm_tfs is None: + confirm_tfs = [Timeframe.H1, Timeframe.H4] + + pair_info = PAIR_INFO[pair] + display = pair_info["display"] + is_xau = display == "XAU/USD" + + platform_cfg = PlatformConfig( + symbol=pair, + display_name=display, + pip_value=pair_info["pip_value"], + primary_timeframe=primary_tf, + confirmation_timeframes=list(confirm_tfs), + ) + + provider = ForexDataProvider(platform_cfg) + + # ── Load data ── + timeframes = [primary_tf] + list(confirm_tfs) + pad_days = max(60, warmup_bars // 4) + fetch_start = start - timedelta(days=pad_days) + + dfs_full: Dict[Timeframe, pd.DataFrame] = {} + for tf in timeframes: + df = fetch_history( + provider, pair, tf, + start=fetch_start, end=end, + use_cache=use_cache, + ) + if df.empty: + raise RuntimeError(f"No data for {pair} {tf.value}") + dfs_full[tf] = df + + # ── Pre-compute indicators ── + 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: {time.perf_counter() - t0:.2f}s") + + # ── Build aggregator with GPU engine ── + aggregator = SignalAggregator(platform_cfg) + aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches) + + # ── Walk primary TF ── + primary_df = dfs_full[primary_tf] + bt_start_ts = pd.Timestamp(start) + bt_end_ts = pd.Timestamp(end) + + first_eligible = max( + warmup_bars, + int(primary_df.index.searchsorted(bt_start_ts)), + ) + last_pos = min( + int(primary_df.index.searchsorted(bt_end_ts, side="right")), + len(primary_df), + ) + + primary_dur = pd.Timedelta(seconds=BAR_SECONDS[primary_tf]) + + # Counters + total_bars_in_window = 0 + total_bars_sampled = 0 + outcome_counts = Counter() + filter_counts = Counter() + # Filters that killed bars that were otherwise directional (non-NEUTRAL score) + directional_filter_counts = Counter() + # Pre-pipeline rejections (XAU calendar filters applied before analyze_from_dataframes) + pre_pipeline = Counter() + + # Track signal directions + signal_dir_counts = Counter() + quality_counts = Counter() + + # Score tracker + tracker = ScoreTracker() + + # Track "would-have-been" trades: bars where the primary score was + # directional but got killed by a specific filter + filter_kill_breakdown = defaultdict(Counter) + + logger.info( + f"Diagnosing {display} {primary_tf.value}: " + f"bars {first_eligible}-{last_pos} (every {sample_every}th bar)" + ) + + t_loop = time.perf_counter() + for pos in range(first_eligible, last_pos, sample_every): + total_bars_in_window += 1 + ts = primary_df.index[pos] + decision_time = ts + primary_dur + + # ── XAU calendar pre-filter (same as gpu_backtest.py) ── + if is_xau: + block = _is_xau_blocked_by_calendar(ts, platform_cfg) + if block: + pre_pipeline[block] += 1 + continue + + total_bars_sampled += 1 + + # ── Build sliced dfs (same logic as gpu_backtest.py) ── + 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 primary_tf not in slice_dfs: + continue + + # ── Run the full signal pipeline ── + report: SignalReport = aggregator.analyze_from_dataframes( + symbol=pair, + dfs=slice_dfs, + context=None, + use_news=False, + ) + + # Record score/confidence + tracker.record(report.final_score, report.final_confidence) + + # Signal direction + signal_dir_counts[report.final_signal.name] += 1 + quality_counts[report.signal_quality] += 1 + + # Classify + outcome, filters_hit = classify_bar(report, is_xau) + outcome_counts[outcome] += 1 + + for f in filters_hit: + filter_counts[f] += 1 + + # If the primary score was directional (>= 15 or <= -15), the bar + # had some potential — track what killed it + if abs(report.primary_score) >= 15: + for f in filters_hit: + directional_filter_counts[f] += 1 + + # For bars that got quality D/F but had a directional signal, + # track which filter warnings appeared + if outcome in ("LOW_QUALITY", "QUALITY_OK_NO_SETUP"): + for f in filters_hit: + filter_kill_breakdown[outcome][f] += 1 + + elapsed = time.perf_counter() - t_loop + + # ═════════════════════════════════════════════════════════════════ + # PRINT RESULTS + # ═════════════════════════════════════════════════════════════════ + + bar = "=" * 80 + print(f"\n{bar}") + print(f" FILTER REJECTION DIAGNOSTIC — {display}") + print(f" {start.date()} to {end.date()} | {primary_tf.value} " + f"(sampled every {sample_every}th bar)") + print(f"{bar}") + + # Scale counts if sampling + scale_note = "" + if sample_every > 1: + scale_note = f" (sampled {total_bars_sampled} of ~{total_bars_in_window * sample_every} eligible bars)" + print(f"\n Total bars in backtest window: {total_bars_in_window}") + if pre_pipeline: + for reason, count in sorted(pre_pipeline.items()): + print(f" Pre-pipeline rejection ({reason}): {count}") + print(f" Bars analyzed: {total_bars_sampled}{scale_note}") + print(f" Analysis took: {elapsed:.1f}s " + f"({total_bars_sampled / max(elapsed, 0.001):.0f} bars/s)") + + # ── Outcome breakdown ── + print(f"\n{'-' * 80}") + print(" OUTCOME BREAKDOWN") + print(f"{'-' * 80}") + outcome_order = ["TRADE", "QUALITY_OK_NO_SETUP", "LOW_QUALITY", "NEUTRAL"] + outcome_labels = { + "TRADE": "Actual trades (quality C+ & setup exists)", + "QUALITY_OK_NO_SETUP": "Quality C+ but NO trade setup generated", + "LOW_QUALITY": "Directional signal but quality D/F", + "NEUTRAL": "NEUTRAL (score didn't reach +/-30 or conf < 55)", + } + for o in outcome_order: + count = outcome_counts.get(o, 0) + pct = count / max(total_bars_sampled, 1) * 100 + label = outcome_labels.get(o, o) + print(f" {count:6d} ({pct:5.1f}%) {label}") + + # ── Signal direction breakdown ── + print(f"\n{'-' * 80}") + print(" SIGNAL DIRECTION BREAKDOWN") + print(f"{'-' * 80}") + for name in ["STRONG_BUY", "BUY", "NEUTRAL", "SELL", "STRONG_SELL"]: + count = signal_dir_counts.get(name, 0) + pct = count / max(total_bars_sampled, 1) * 100 + print(f" {count:6d} ({pct:5.1f}%) {name}") + + # ── Quality grade breakdown ── + print(f"\n{'-' * 80}") + print(" QUALITY GRADE BREAKDOWN") + print(f"{'-' * 80}") + for grade in ["A+", "A", "B", "C", "D", "F"]: + count = quality_counts.get(grade, 0) + pct = count / max(total_bars_sampled, 1) * 100 + print(f" {count:6d} ({pct:5.1f}%) Grade {grade}") + + # ── Filter hit counts (all bars) ── + print(f"\n{'-' * 80}") + print(" FILTER HIT COUNTS (all bars, a bar can hit multiple filters)") + print(f"{'-' * 80}") + if filter_counts: + for name, count in filter_counts.most_common(): + pct = count / max(total_bars_sampled, 1) * 100 + print(f" {count:6d} ({pct:5.1f}%) {name}") + else: + print(" (no filters triggered)") + + # ── Filters that killed directional bars (primary_score >= 15) ── + print(f"\n{'-' * 80}") + print(" FILTERS KILLING DIRECTIONAL BARS (|primary_score| >= 15)") + print(f"{'-' * 80}") + if directional_filter_counts: + for name, count in directional_filter_counts.most_common(): + print(f" {count:6d} {name}") + else: + print(" (none)") + + # ── Filter breakdown for LOW_QUALITY bars ── + if filter_kill_breakdown.get("LOW_QUALITY"): + print(f"\n{'-' * 80}") + print(" FILTERS ON LOW_QUALITY BARS (had BUY/SELL but quality D/F)") + print(f"{'-' * 80}") + for name, count in filter_kill_breakdown["LOW_QUALITY"].most_common(): + print(f" {count:6d} {name}") + + # ── Filter breakdown for QUALITY_OK_NO_SETUP bars ── + if filter_kill_breakdown.get("QUALITY_OK_NO_SETUP"): + print(f"\n{'-' * 80}") + print(" FILTERS ON QUALITY_OK_NO_SETUP BARS (quality C+ but no trade setup)") + print(f"{'-' * 80}") + for name, count in filter_kill_breakdown["QUALITY_OK_NO_SETUP"].most_common(): + print(f" {count:6d} {name}") + + # ── Score distribution ── + print(f"\n{'-' * 80}") + print(" SCORE DISTRIBUTION (|score|)") + print(f"{'-' * 80}") + for bin_name in ["0-9", "10-19", "20-29", "30-39", "40-49", "50+"]: + count = tracker.score_bins.get(bin_name, 0) + pct = count / max(total_bars_sampled, 1) * 100 + marker = " <-- near miss (almost BUY/SELL)" if bin_name == "20-29" else "" + print(f" {count:6d} ({pct:5.1f}%) |score| {bin_name}{marker}") + + # ── Confidence distribution ── + print(f"\n{'-' * 80}") + print(" CONFIDENCE DISTRIBUTION") + print(f"{'-' * 80}") + for bin_name in ["0-29", "30-44", "45-54", "55-64", "65-74", "75+"]: + count = tracker.conf_bins.get(bin_name, 0) + pct = count / max(total_bars_sampled, 1) * 100 + marker = " <-- near miss (almost passing)" if bin_name == "45-54" else "" + print(f" {count:6d} ({pct:5.1f}%) confidence {bin_name}{marker}") + + # ── Near misses ── + print(f"\n{'-' * 80}") + print(" NEAR MISSES") + print(f"{'-' * 80}") + print(f" {tracker.near_miss_score:6d} Bars with |score| 20-29 (threshold is 30)") + print(f" {tracker.near_miss_conf:6d} Bars with confidence 45-54 (threshold is 55)") + + # ── Summary statistics ── + if tracker.scores: + scores_arr = np.array(tracker.scores) + conf_arr = np.array(tracker.confidences) + print(f"\n{'-' * 80}") + print(" SCORE/CONFIDENCE STATISTICS") + print(f"{'-' * 80}") + print(f" Score: mean={np.mean(scores_arr):+.1f} " + f"median={np.median(scores_arr):+.1f} " + f"std={np.std(scores_arr):.1f} " + f"min={np.min(scores_arr):+.1f} " + f"max={np.max(scores_arr):+.1f}") + print(f" Confidence: mean={np.mean(conf_arr):.1f} " + f"median={np.median(conf_arr):.1f} " + f"std={np.std(conf_arr):.1f} " + f"min={np.min(conf_arr):.1f} " + f"max={np.max(conf_arr):.1f}") + + # ── Blocked hours analysis (for XAU) ── + if is_xau: + blocked = set(platform_cfg.xau_strategy.blocked_hours_utc) + all_hours = set(range(24)) + open_hours = sorted(all_hours - blocked) + print(f"\n{'-' * 80}") + print(" XAU HOUR/DAY CONFIG") + print(f"{'-' * 80}") + print(f" Blocked hours: {sorted(blocked)}") + print(f" Open hours: {open_hours}") + print(f" Allowed days: {platform_cfg.xau_strategy.allowed_weekdays} " + f"(0=Mon..4=Fri)") + print(f" Open fraction: {len(open_hours)/24:.0%} of hours x " + f"{len(platform_cfg.xau_strategy.allowed_weekdays)}/5 weekdays = " + f"{len(open_hours)/24 * len(platform_cfg.xau_strategy.allowed_weekdays)/5:.0%} of trading time") + + # ── Interpretation / key takeaways ── + print(f"\n{bar}") + print(" KEY TAKEAWAYS") + print(f"{bar}") + + neutral_pct = outcome_counts.get("NEUTRAL", 0) / max(total_bars_sampled, 1) * 100 + low_q_pct = outcome_counts.get("LOW_QUALITY", 0) / max(total_bars_sampled, 1) * 100 + trade_count = outcome_counts.get("TRADE", 0) + + print(f" - {neutral_pct:.1f}% of bars never reach a directional signal (NEUTRAL)") + print(f" - {low_q_pct:.1f}% of bars get a BUY/SELL but confidence/quality too low") + print(f" - {trade_count} bars produce actual trades") + + if filter_counts: + top3 = filter_counts.most_common(3) + print(f" - Top rejection filters:") + for name, count in top3: + print(f" {name}: {count} bars hit") + + if tracker.near_miss_score > 50: + print(f" - SIGNIFICANT: {tracker.near_miss_score} bars had |score| 20-29 " + f"(lowering threshold from 30 to 25 would capture many)") + + if tracker.near_miss_conf > 50: + print(f" - SIGNIFICANT: {tracker.near_miss_conf} bars had confidence 45-54 " + f"(lowering threshold from 55 to 50 would capture many)") + + print(f"\n{bar}\n") + + +# ═════════════════════════════════════════════════════════════════════ +# CLI +# ═════════════════════════════════════════════════════════════════════ + +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}'. Valid: {', '.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}'") + + +def main(argv: Optional[List[str]] = None) -> int: + p = argparse.ArgumentParser( + description="Diagnose which filters reject the most bars in the backtest pipeline", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + p.add_argument("--pair", required=True, help="EURUSD | GBPUSD | USDJPY | XAUUSD") + p.add_argument("--days", type=int, default=365, help="Days of history") + 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") + 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("--sample-every", type=int, default=4, + help="Analyze every Nth bar (1=all, 4=25%% sample for speed)") + p.add_argument("--no-cache", action="store_true", help="Skip parquet cache") + p.add_argument("--log-level", default="INFO") + args = p.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) + + logger.info( + f"Diagnosing {PAIR_INFO[pair]['display']} | " + f"Window: {start.date()} -> {end.date()} | " + f"Sample: every {args.sample_every}th bar" + ) + + run_diagnosis( + pair=pair, + start=start, + end=end, + sample_every=args.sample_every, + use_cache=not args.no_cache, + primary_tf=primary_tf, + confirm_tfs=confirm_tfs, + warmup_bars=args.warmup_bars, + ) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py new file mode 100644 index 0000000..259647e --- /dev/null +++ b/backtest/gpu_backtest.py @@ -0,0 +1,1917 @@ +""" +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 + 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, +} + +# 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" + + desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}" + result.indicators["adx"] = IndicatorSignal( + name="ADX", value=adx_val, signal=signal, confidence=conf, + description=desc, weight=0.5, + ) + + # ── 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, +) -> 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. + """ + 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 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")] + df = provider._validate_and_clean(df, symbol, timeframe) + # Trim to requested window after cleaning. `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)} bars to {path.name}") + return 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", "EUR_PB", "GBP_PB", "GENERIC" + + # 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_trades( + trades: List[Trade], + sim_df: pd.DataFrame, + max_bars: int = 200, +) -> None: + """Walk forward bar-by-bar after each entry and resolve SL / TP1. + + Vectorized on GPU when CuPy is available: builds (n_trades × max_bars) + matrices of post-entry highs/lows and finds the first hit per trade. + + SL/TP-same-bar tie: SL assumed first (conservative). + 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). + 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_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 + 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])) + + # 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 + t.exit_price = t.sl + t.r_multiple = -1.0 + 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 + t.exit_price = t.tp1 + t.r_multiple = t.rr_planned + 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]) + risk = abs(t.entry_price - t.sl) + if risk > 0: + pnl = (t.exit_price - t.entry_price) if t.direction == "BUY" else (t.entry_price - t.exit_price) + t.r_multiple = pnl / risk + t.exit_time = exit_times[i][last_idx] + + # 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 = 200 # cap simulation length per trade + 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) + + +def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]: + """Apply XAU weekend + allowed-weekday filter (live `analyze_pair` does this; + `analyze_from_dataframes` does not, so we replicate here).""" + xau = cfg.xau_strategy + if xau.weekend_filter and ts.weekday() >= 5: + 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), + ) + + 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). + if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD": + block = _is_xau_blocked_by_calendar(ts, platform_cfg) + if block: + skipped_reasons[block] = skipped_reasons.get(block, 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 bullish_reasons[0] which the aggregator pre-pends). + tag = "GENERIC" + if report.bullish_reasons: + first = report.bullish_reasons[0] + if "XAU STRATEGY" in first: + tag = "XAU" + elif "EUR STRATEGY" in first: + tag = "EUR_PB" + elif "GBP STRATEGY" in first: + tag = "GBP_PB" + + if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only: + continue + + ts_setup: TradeSetup = report.trade_setup + # Entry on the NEXT bar's open price would be ideal, but live signals fire + # at bar close and we use that close as entry (matches setup.entry_price). + # For simulation realism we step entry to next bar so bar `pos` is not + # used to resolve the trade outcome (bar `pos` is the signal bar). + entry_pos = pos + 1 + if entry_pos >= len(primary_df): + continue + entry_time = primary_df.index[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) + + trades.append(Trade( + pair=bt_cfg.pair, + entry_time=entry_time, + entry_price=ts_setup.entry_price, + sl=ts_setup.stop_loss, + tp1=ts_setup.take_profit_1, + tp2=ts_setup.take_profit_2, + direction=ts_setup.direction, + pip_value=ts_setup.pip_value, + risk_pips=ts_setup.risk_pips, + reward_pips=ts_setup.reward_pips, + 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, + )) + 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 ── + t_sim = time.perf_counter() + simulate_trades(trades, primary_df, max_bars=bt_cfg.max_hold_bars) + 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 | XAU/USD | GOLD") + 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=200) + 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,EUR_PB,GBP_PB,GENERIC). Empty = all.") + p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)") + 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) + + 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()], + ) + + logger.info( + f"Backend: {'CuPy GPU' if _GPU_AVAILABLE else 'NumPy CPU fallback'} | " + f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()}" + ) + + 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()) diff --git a/backtest/instructions_for_backtest.txt b/backtest/instructions_for_backtest.txt new file mode 100644 index 0000000..6736afd --- /dev/null +++ b/backtest/instructions_for_backtest.txt @@ -0,0 +1,59 @@ +Note down entery and exit Why we took trade? +What were each indicator signal on that specific time when we took trade +How many indicators agreed at trade time +Always backtest using engine.py and signal_aggregator.py +Dont take trades on weekends and Fridays +Keep TP small 1.3 Reward ratio +At the end of test make table of results and print all values you observed. +Trade context + +Time of day and trading session (Asian/London/NY/overlap) +Day of week (track Mon/Tue/Wed/Thu performance separately) +Market regime at entry: trending, ranging, or choppy (use ATR or ADX) +Volatility level (ATR value or ATR percentile vs recent average) +Pair/instrument traded +News proximity (high-impact news within 30 min before/after entry) + +Trade execution details + +Spread at entry in pips +Slippage on entry and exit (planned vs actual fill) +Time in trade (minutes/hours from entry to exit) +Maximum Adverse Excursion (MAE): how far against you it went +Maximum Favorable Excursion (MFE): how far in profit before pullback + +Risk metrics + +Max consecutive losses (streak) +Max consecutive wins (streak) +Expectancy: (Win% x Avg Win) minus (Loss% x Avg Loss) +Expectancy per hour in trade +Recovery factor: Total R divided by Max DD + +Indicator/signal analysis + +Each indicator's standalone win rate when it agreed with the trade +Confluence level performance (2 vs 3 vs 4+ indicators agreeing) +Which indicator disagreed on losing trades (find the "canary") + +Decision quality + +Rate each setup A/B/C at entry, before knowing outcome +Verify A-grade setups actually win more often +Track system trades vs discretionary deviations separately + +Process notes + +Screenshot or chart snapshot at entry and exit +Emotional state (calm, FOMO, revenge, tired) +What would have happened if you held longer or exited sooner + +Aggregate analysis at end of backtest + +Equity curve, not just total R +Performance breakdown by session, day, volatility regime +Win rate over rolling 20-trade windows (check for edge decay) + +Critical extra +"Do not trade" log: setups flagged but skipped, and what happened after +