#!/usr/bin/env python3 """ cudaBasedBacktesting.py — the full-GPU twin of backtest/gpu_backtest.py. Same command line, same cache files, same exported book (byte for byte — the MD5 of the CSV is the acceptance gate), but every stage of the computation runs on the GPU: the indicator precompute (shared with gpu_backtest.py), the per-bar decision (backtest/cuda_signal_kernels.py, a CUDA twin of engine.py + signal_aggregator.py + chart_patterns.py), the entry cooldown, the decision-time features (--export-features), the recorder rows (--record, --record-bars) and the trade resolution (shared). Research tool only (owner decision 2026-09-18): live scans stay on the CPU code and gpu_backtest.py stays in the tree as the base reference. Any change to the signal logic is a parity event for this file — regenerate a book with both scripts and show equal MD5s before using numbers from here as evidence (backtest/instructions_for_backtest.txt). What stays on the host, on purpose: loading and cleaning the parquet cache, the calendar/session flags (pandas time-zone arithmetic over the whole index), turning the kernels' numeric codes into the exact strings the reference script writes (indicator descriptions, warnings, reasons), and the report / CSV / parquet writers, all of which are reused from the reference script unchanged. There is no CPU fallback here by design: the twin IS the live code. """ from __future__ import annotations import logging import sys import time import types from pathlib import Path from typing import Dict, List, Optional, Sequence, Tuple import numpy as np import pandas as pd 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 PAIR_INFO, PlatformConfig, Timeframe # noqa: E402 from providers import make_data_provider # noqa: E402 from providers.base import BaseDataProvider # noqa: E402 from signal_aggregator import SignalAggregator # noqa: E402 (fingerprints + the two static cleanups only) import position_sizing # noqa: E402 from backtest import gpu_kernels as _gk # noqa: E402 from backtest import cuda_signal_kernels as ck # noqa: E402 from backtest import replay_record # noqa: E402 from backtest.gpu_backtest import ( # noqa: E402 BAR_SECONDS, _NY_TZ, BacktestConfig, Trade, _apply_config_overrides, _apply_fixed_rr_tp, _cache_path, _DAY_NAMES, _history_pad_days, _parse_args, _reanchor_entries_to_raw, _resolve_pair, _resolve_tf, compute_metrics, fetch_history, precompute_indicators, print_report, print_trade_reasoning, simulate_trades, trades_to_dataframe, ) logger = logging.getLogger("cudaBasedBacktesting") BACKEND = _gk.BACKEND # ═════════════════════════════════════════════════════════════════════ # Text: the exact templates of engine.py / signal_aggregator.py # ═════════════════════════════════════════════════════════════════════ _RSI_SUFFIX = { 0: "Overbought", 1: "Oversold", 2: "Bullish momentum zone", 3: "Bearish momentum zone", 4: "Neutral", 5: "Pullback zone in uptrend (ideal entry)", 6: "Deep pullback in uptrend", 7: "Momentum ok in uptrend", 8: "Extended in uptrend, wait for pullback", 9: "Oversold in uptrend, possible reversal", 10: "Bounce zone in downtrend (ideal short)", 11: "Deep bounce in downtrend", 12: "Momentum ok in downtrend", 13: "Extended in downtrend, wait for bounce", 14: "Overbought in downtrend, possible reversal", 15: "Overbought while trend forming", 16: "Oversold while trend forming", 17: "No RSI edge while trend forms", 18: "Overbought in range", 19: "Oversold in range", 20: "Upper range zone", 21: "Lower range zone", 22: "Mid-range, no edge", } _MACD_DESC = { 0: "MACD histogram accelerating ↑", 1: "MACD histogram crossed above zero", 2: "MACD histogram accelerating ↓", 3: "MACD histogram crossed below zero", 4: "MACD histogram positive but fading", 5: "MACD histogram negative but fading", 6: "MACD histogram flat", } _BB_DESC = {0: "Price above upper BB", 1: "Price below lower BB", 2: "Price near upper BB", 3: "Price near lower BB", 4: "Price mid-band"} _KC_DESC = { 0: "Keltner SQUEEZE RELEASED — bullish breakout", 1: "Keltner SQUEEZE RELEASED — bearish breakout", 2: "Keltner squeeze released — direction unclear", 3: "Keltner SQUEEZE active — breakout imminent", 4: "Price above Keltner upper — strong uptrend continuation", 5: "Price below Keltner lower — strong downtrend continuation", 6: "Price within Keltner Channels", } _EMA2_DESC = { 0: "EMA200 data insufficient", 1: "Full bullish: price > EMA50 > EMA200", 2: "Emerging uptrend: price > EMA50, EMA50 < EMA200", 3: "Full bearish: price < EMA50 < EMA200", 4: "Emerging downtrend: price < EMA50, EMA50 > EMA200", 5: "EMA alignment mixed", } _DIV_ITEMS = ("Bullish RSI divergence (reversal warning)", "Bearish RSI divergence (reversal warning)", "Bullish MACD divergence (reversal warning)", "Bearish MACD divergence (reversal warning)") _CS_ITEMS = ("Doji (indecision)", "Hammer (bullish reversal)", "Hanging Man (bearish warning)", "Shooting Star (bearish reversal)", "Inverted Hammer (bullish)", "Bullish Engulfing", "Bearish Engulfing", "Bullish Marubozu (strong buyers)", "Bearish Marubozu (strong sellers)", "(at support)", "(at resistance)") _ADX_STRENGTH = ("Very strong", "Strong", "Weak/Ranging") _ADX_DIR = ("bullish", "bearish") class K1Row: """One timeframe bar's K1 outputs as Python scalars (name -> value).""" __slots__ = ("i", "f") def __init__(self, ints: Dict[str, int], floats: Dict[str, float]): self.i, self.f = ints, floats def k1_rows(an: "ck.TFAnalysis", idx: np.ndarray) -> List[K1Row]: """Gather K1 outputs for the given bar indexes (one host transfer).""" idx = np.asarray(idx, dtype=np.int64) if len(idx) == 0: return [] fi = an.OF[:, idx].get() # [K1F_N, m] ii = an.OI[:, idx].get() # [K1I_N, m] rows = [] for m in range(len(idx)): rows.append(K1Row({n: int(ii[k, m]) for k, n in enumerate(ck.K1_INT_FIELDS)}, {n: float(fi[k, m]) for k, n in enumerate(ck.K1_FLOAT_FIELDS)})) return rows def describe_indicators(row: K1Row, ema_mode_2ema: bool) -> Dict[str, Dict]: """The scored indicators of one analysis in engine insertion order: name -> {value, signal (int), confidence, description, weight, effective_weight, weighted_score}. Exactly what result.indicators carries minus the weight-0 ATR entry.""" i, f = row.i, row.f out: Dict[str, Dict] = {} if not i["valid"]: return out if i["ema_present"]: if ema_mode_2ema: desc = _EMA2_DESC[i["ema2_desc"]] desc += {1: " | Golden cross (50/200)", 2: " | Death cross (50/200)"}.get(i["ema_cross"], "") value = 0 if i["ema2_desc"] == 0 else f["ema_value"] else: cross = {1: " | Golden cross (fast/medium)", 2: " | Death cross (fast/medium)"}.get(i["ema_cross"], "") ribbon = {1: " | Ribbon COMPRESSED (trend change imminent)", 2: " | Ribbon EXPANDING (new trend starting)"}.get(i["ema_ribbon"], "") desc = f"EMA alignment: {i['ema_align']}/{i['ema_max_checks']}{cross}{ribbon}" value = f["ema_value"] out["ema_trend"] = dict(value=value, signal=i["ema_signal"], confidence=f["ema_conf"], description=desc, weight=1.0, effective_weight=f["ew_ema"], weighted_score=f["ws_ema"]) if i["rsi_present"]: v = f["rsi_value"] out["rsi"] = dict(value=v, signal=i["rsi_signal"], confidence=f["rsi_conf"], description=f"RSI {v:.1f} — {_RSI_SUFFIX[i['rsi_desc']]}", weight=1.5, effective_weight=f["ew_rsi"], weighted_score=f["ws_rsi"]) if i["bb_present"]: v = f["bb_value"] note = " | SQUEEZE detected (expect breakout)" if i["bb_squeeze"] else "" out["bollinger"] = dict(value=v, signal=i["bb_signal"], confidence=f["bb_conf"], description=f"{_BB_DESC[i['bb_desc']]} ({v:.2f}){note}", weight=1.5, effective_weight=f["ew_bb"], weighted_score=f["ws_bb"]) if i["adx_present"]: slope = f["adx_slope"] sd = {1: " | TREND EMERGING (ADX rising fast)", 2: f" | Trend strengthening (slope +{slope:.1f})", 3: f" | Trend weakening (slope {slope:.1f})"}.get(i["adx_slope_desc"], "") desc = (f"ADX {f['adx_value']:.1f} ({_ADX_STRENGTH[i['adx_strength']]} {_ADX_DIR[i['adx_dir']]}) | " f"+DI={f['plus_di']:.1f} -DI={f['minus_di']:.1f}{sd}") out["adx"] = dict(value=f["adx_value"], signal=i["adx_signal"], confidence=f["adx_conf"], description=desc, weight=0.5, effective_weight=f["ew_adx"], weighted_score=f["ws_adx"]) if i["macd_present"]: out["macd"] = dict(value=f["macd_value"], signal=i["macd_signal"], confidence=f["macd_conf"], description=_MACD_DESC[i["macd_desc"]], weight=1.5, effective_weight=f["ew_macd"], weighted_score=f["ws_macd"]) if i["roc_present"]: v = f["roc_value"] d = i["roc_desc"] desc = {0: f"ROC +{v:.2f}% — Strong momentum acceleration", 1: f"ROC +{v:.2f}% — Rising momentum", 2: f"ROC {v:.2f}% — Strong bearish momentum", 3: f"ROC {v:.2f}% — Falling momentum", 4: f"ROC {v:+.2f}% — No clear momentum"}[d] out["roc"] = dict(value=v, signal=i["roc_signal"], confidence=f["roc_conf"], description=desc, weight=1.5, effective_weight=f["ew_roc"], weighted_score=f["ws_roc"]) if i["kc_present"]: out["keltner"] = dict(value=f["kc_value"], signal=i["kc_signal"], confidence=f["kc_conf"], description=_KC_DESC[i["kc_desc"]], weight=1.5, effective_weight=f["ew_kc"], weighted_score=f["ws_kc"]) if i["div_present"]: items = [s for b, s in enumerate(_DIV_ITEMS) if i["div_mask"] & (1 << b)] out["divergence"] = dict(value=len(items), signal=i["div_signal"], confidence=70, description=" | ".join(items), weight=1.5, effective_weight=f["ew_div"], weighted_score=f["ws_div"]) if i["cs_present"]: items = [s for b, s in enumerate(_CS_ITEMS) if i["cs_mask"] & (1 << b)] out["candlestick"] = dict(value=f["cs_value"], signal=i["cs_signal"], confidence=f["cs_conf"], description=" | ".join(items), weight=1.0, effective_weight=f["ew_cs"], weighted_score=f["ws_cs"]) return out def pattern_signals(row: K1Row) -> List[str]: if not row.i["valid"] or not row.i["cs_present"]: return [] return [s for b, s in enumerate(_CS_ITEMS) if row.i["cs_mask"] & (1 << b)] def key_levels(row: K1Row, ema_mode_2ema: bool) -> Dict[str, float]: """result.key_levels in insertion order.""" i, f = row.i, row.f kl: Dict[str, float] = {} if not i["valid"]: return kl if i["atr_present"]: kl["atr"] = f["atr"] kl["atr_pct"] = f["atr_pct"] if i["ema_present"]: if i["ema_trend_present"]: kl["ema_trend"] = f["ema_trend"] if not ema_mode_2ema: kl["ema_fast"] = f["ema_fast"] kl["ema_medium"] = f["ema_medium"] kl["ema_slow"] = f["ema_slow"] if i["bb_present"]: kl["bb_upper"] = f["bb_upper"]; kl["bb_lower"] = f["bb_lower"]; kl["bb_mid"] = f["bb_mid"] if i["kc_present"]: kl["kc_upper"] = f["kc_upper"]; kl["kc_lower"] = f["kc_lower"]; kl["kc_mid"] = f["kc_mid"] if not i["sh_fallback"]: for a in range(3): if i["sh_mask"] & (1 << a): kl[f"swing_high_{a + 1}"] = f[f"sh{a + 1}"] if not i["sl_fallback"]: for a in range(3): if i["sl_mask"] & (1 << a): kl[f"swing_low_{a + 1}"] = f[f"sl{a + 1}"] if i["sh_fallback"]: kl["swing_high_1"] = f["sh1"] if i["sl_fallback"]: kl["swing_low_1"] = f["sl1"] return kl def format_warning(code: int, arg: float, ctx: Dict) -> str: """signal_aggregator's warning text for one K2 warning code.""" n = ck.WARN_NAMES[code] if n == "conflict": return "Timeframes CONFLICT — mixed signals across timeframes" if n == "counter_trend": return "Counter-trend signal — reduced confidence" if n == "extreme_vol": return "EXTREME volatility — wider stops needed" if n == "low_vol": return "Low volatility — potential for breakout" if n == "ema200_near": return f"NEAR EMA200 ({arg:.2f}%) — chop risk" if n == "ema200_far": return f"FAR FROM EMA200 ({arg:.2f}%) — over-extended" if n == "exhaustion_low": return "TREND EXHAUSTION — price extended >3.5 ATR past swing low" if n == "exhaustion_high": return "TREND EXHAUSTION — price extended >3.5 ATR past swing high" if n == "no_pullback": return f"NO PULLBACK — price {arg:.1f} ATR from EMA21" if n == "weak_pullback": return f"Weak pullback — price {arg:.1f} ATR from EMA21" if n == "rsi_ext_up": return f"Extended — RSI {arg:.0f} in uptrend" if n == "rsi_ext_down": return f"Extended — RSI {arg:.0f} in downtrend" if n == "exhaustion_block": return f"EXHAUSTION BLOCK — {int(arg)} exhaustion signals fired, trade rejected" if n == "session_filter": return f"SESSION FILTER: {int(arg):02}:00 UTC blocked" if n == "day_filter": day_names = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri", 5: "Sat", 6: "Sun"} return f"DAY FILTER: {day_names.get(int(arg), '?')} not in allowed days" if n == "exhausted_trend": return "EXHAUSTED TREND FILTER: C-grade blocked in exhausted trend" if n == "sell_quality": return f"SELL FILTER: quality {ck.QUALITY_NAMES[int(arg)]} below minimum for SELL" if n == "mtf_opposition": return "MTF OPPOSITION: majority of timeframes point against signal — no trade" if n == "sell_vol": return f"SELL FILTER: volatility '{ck.VOL_NAMES[int(arg)]}' too quiet for shorts" if n == "regime_blocked": return f"REGIME FILTER: '{ck.REGIME_NAMES[int(arg)]}' blocked for XAU" if n == "sell_regime": return f"SELL FILTER: regime '{ck.REGIME_NAMES[int(arg)]}' blocked for shorts" if n == "sell_hour": return f"SELL FILTER: {int(arg):02}:00 UTC entry blocked for shorts" if n == "big_candle": return (f"BIG CANDLE FILTER: {arg:.1f}x ATR bar at {ctx['bc_ts']} within " f"{ctx['big_candle_lookback_hours']:g}h — no entries") if n == "m5_score": return f"M5 FILTER: |score| {arg:.1f} < {ctx['m5_min_abs_score']:.0f} minimum" if n == "m5_mtf": return "M5 FILTER: no MTF agreement — M5 signals require confirmation" if n == "m5_conviction": return "M5 FILTER: no confirmation TF shows directional conviction" if n == "m5_sell_quality": return (f"M5 FILTER: SELL quality {ck.QUALITY_NAMES[int(arg)]} capped — " f"high-conviction M5 shorts enter at extremes") if n == "m5_vol": return (f"M5 FILTER: '{ck.VOL_NAMES[int(arg)]}' volatility — " f"chop-transition regime, stops get swept") if n == "m5_sell_rsi": return (f"M5 FILTER: SELL blocked — RSI {arg:.0f} < " f"{ctx['m5_sell_min_rsi']:.0f}, dip already extended") if n == "spread_gate_atr" or n == "spread_gate_generic": return (f"SPREAD GATE: target ${arg:.2f} < ${ctx['min_tp']:.2f} minimum — " f"costs would eat the edge") if n == "low_atr": return f"LOW ATR FILTER: ATR ${arg:.2f} < ${ctx['min_atr_threshold']:.2f} minimum" if n == "rf_unavailable": return f"Regime filter requested but {ctx['rf_tf']} data unavailable" if n == "rf_no_adx": return f"Regime filter: ADX could not be computed on {ctx['rf_tf']}" if n == "rf_blocked": return (f"REGIME FILTER: {ctx['rf_tf']} ADX={arg:.1f} < " f"{ctx['regime_filter_min_adx']:.0f} — market choppy, skipping signal") raise ValueError(f"unknown warning code {code}") def collect_reasons(inds: Dict[str, Dict]) -> Tuple[List[str], List[str]]: """signal_aggregator._collect_reasons (news never contributes here).""" bullish: List[str] = [] bearish: List[str] = [] for name, ind in inds.items(): sig = ind["signal"] label = ck.INDICATOR_LABELS[name] if sig > 0: prefix = "STRONG: " if sig == 2 else "" bullish.append(f"{prefix}{label}: {ind['description']}") elif sig < 0: prefix = "STRONG: " if sig == -2 else "" bearish.append(f"{prefix}{label}: {ind['description']}") return bullish, bearish # ═════════════════════════════════════════════════════════════════════ # Host flags: the calendar / session rules of _replay_span, vectorised # ═════════════════════════════════════════════════════════════════════ def host_flags(primary_index: pd.DatetimeIndex, platform_cfg: PlatformConfig, bt_cfg: BacktestConfig) -> Dict[str, np.ndarray]: """Per primary bar: the replay skip reason (1 WEEKEND_FILTER, 2 DAY_FILTER, 3 SESSION_WINDOW, 0 analysed), the bar-label hour/weekday (gate 6b) and the bar-close hour/weekday/minute/day (gate 6g3, XAU_CP, features).""" n = len(primary_index) primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf]) decision = primary_index + primary_dur skip = np.zeros(n, dtype=np.int8) is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD" xau = platform_cfg.xau_strategy if is_xau: if xau.weekend_filter: ny = decision.tz_convert(_NY_TZ) wd = np.asarray(ny.weekday) hr = np.asarray(ny.hour) weekend = ((wd == 4) & (hr >= 17)) | (wd == 5) | ((wd == 6) & (hr < 18)) skip[weekend] = 1 if xau.allowed_weekdays: day = ~np.isin(np.asarray(decision.weekday), list(xau.allowed_weekdays)) skip[day & (skip == 0)] = 2 flt = platform_cfg.backtest if flt.session_filter: h = np.asarray(decision.hour) blocked = xau.blocked_hours_utc if is_xau else flt.blocked_hours_utc sess = ~((h >= flt.session_start_hour) & (h < flt.session_end_hour)) | np.isin(h, list(blocked)) skip[sess & (skip == 0)] = 3 return { "skip_reason": skip, "label_hour": np.asarray(primary_index.hour, dtype=np.int8), "label_wd": np.asarray(primary_index.weekday, dtype=np.int8), "close_hour": np.asarray(decision.hour, dtype=np.int8), "close_wd": np.asarray(decision.weekday, dtype=np.int8), "close_minute": np.asarray(decision.minute, dtype=np.int8), "close_day": np.asarray(decision.day, dtype=np.int8), } def replay_bounds(primary_df: pd.DataFrame, bt_cfg: BacktestConfig) -> Tuple[int, int]: bt_start_ts = pd.Timestamp(bt_cfg.start) bt_end_ts = pd.Timestamp(bt_cfg.end) 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)) return first_eligible_pos, last_pos # ═════════════════════════════════════════════════════════════════════ # The pipeline # ═════════════════════════════════════════════════════════════════════ class _Run: """Everything one replay produces on the device plus the host handles the builders need.""" def refuse_unsupported(platform_cfg: PlatformConfig, bt_cfg: BacktestConfig) -> None: t = platform_cfg.technical x = platform_cfg.xau_strategy problems = [] if len(platform_cfg.confirmation_timeframes) > ck.MAX_CONFIRM: problems.append(f"{len(platform_cfg.confirmation_timeframes)} confirmation timeframes " f"(the CUDA aggregate gathers at most {ck.MAX_CONFIRM})") if t.divergence_lookback >= ck.MAX_DIV_LOOKBACK: problems.append(f"technical.divergence_lookback={t.divergence_lookback} " f"(CUDA limit {ck.MAX_DIV_LOOKBACK - 1})") if max(t.swing_sr_lookback, 120) >= ck.MAX_SR_LOOKBACK: problems.append(f"technical.swing_sr_lookback={t.swing_sr_lookback} " f"(CUDA limit {ck.MAX_SR_LOOKBACK - 1})") if x.cp_lookback + 1 > ck.MAX_CP_LOOKBACK: problems.append(f"xau_strategy.cp_lookback={x.cp_lookback} (CUDA limit {ck.MAX_CP_LOOKBACK - 1})") if x.big_candle_atr_multiple > 0: bar_secs = BaseDataProvider.CANDLE_SECONDS[platform_cfg.primary_timeframe] n_bc = max(1, int(x.big_candle_lookback_hours * 3600 // bar_secs)) if n_bc + 3 * t.atr_period > ck.FEAT_MAX_P: problems.append(f"big-candle window {n_bc + 3 * t.atr_period} bars (CUDA limit {ck.FEAT_MAX_P})") if t.ema_mode not in ("4ema", "2ema"): problems.append(f"technical.ema_mode={t.ema_mode!r}") if platform_cfg.signals.grading_mode not in ck.GRADING_MODE_CODES: problems.append(f"signals.grading_mode={platform_cfg.signals.grading_mode!r}") if problems: raise SystemExit("cudaBasedBacktesting.py cannot run this configuration: " + "; ".join(problems) + ". Use backtest/gpu_backtest.py.") def analyze_frames(bt_cfg: BacktestConfig, platform_cfg: PlatformConfig, dfs_full: Dict[Timeframe, pd.DataFrame], timeframes: Sequence[Timeframe]) -> _Run: """Precompute -> K1 per timeframe -> alignment -> host flags -> K2 -> cooldown, on already loaded (cleaned) frames. Returns the run context the builders read. Shared by run_backtest_cuda and the parity tests.""" cp = _gk._require_cp() flt = platform_cfg.backtest t0 = time.perf_counter() caches = {tf: 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 (CuPy GPU)") pf, pi, rm = ck.pack_params(platform_cfg, bt_cfg) t1 = time.perf_counter() dev = {tf: ck.upload_cache(caches[tf], dfs_full[tf]) for tf in timeframes} an = {tf: ck.analyze_tf(dev[tf], pf, pi, rm, tf.value == "5min") for tf in timeframes} logger.info(f"K1: {sum(dev[tf]['n'] for tf in timeframes)} bars of " f"{len(timeframes)} timeframes analysed on the device in {time.perf_counter() - t1:.3f}s") primary_df = dfs_full[bt_cfg.primary_tf] primary_dev = dev[bt_cfg.primary_tf] primary_secs = BAR_SECONDS[bt_cfg.primary_tf] cuts = {tf: ck.align_cut(dev[tf]["index_ns"], primary_dev["index_ns"], primary_secs, BAR_SECONDS[tf]) for tf in timeframes if tf != bt_cfg.primary_tf} confirms = [(an[tf], cuts[tf]) for tf in bt_cfg.confirmation_tfs] rf = None if flt.regime_filter: rtf = flt.regime_filter_timeframe rf_cut = cuts[rtf] if rtf != bt_cfg.primary_tf else cp.arange(1, primary_dev["n"] + 1, dtype=cp.int64) rf = (dev[rtf]["adx"], rf_cut) flags = host_flags(primary_df.index, platform_cfg, bt_cfg) first_eligible_pos, last_pos = replay_bounds(primary_df, bt_cfg) logger.info(f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: " f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}") n_bars = last_pos - first_eligible_pos t_loop = time.perf_counter() k2 = ck.aggregate(primary_dev, an[bt_cfg.primary_tf], confirms, rf, flags, pf, pi, first_eligible_pos, last_pos) logger.info(f"K2: {n_bars} primary bars aggregated on the device in {time.perf_counter() - t_loop:.3f}s") logger.info("GPU cache fallbacks: 0 (no per-bar recompute path in cudaBasedBacktesting)") status = k2.i("status").get()[:n_bars] if n_bars > 0 else np.zeros(0, dtype=np.int32) cand_pos = (np.flatnonzero(status == 2) + first_eligible_pos).astype(np.int64) accepted_mask = ck.cooldown(cand_pos, bt_cfg.cooldown_bars) acc_pos = cand_pos[accepted_mask.astype(bool)] skipped_reasons: Dict[str, int] = {} skip = flags["skip_reason"][first_eligible_pos:last_pos] order: List[str] = [] for code in skip: if code: name = ck.SKIP_NAMES[int(code)] if name not in skipped_reasons: order.append(name) skipped_reasons[name] = skipped_reasons.get(name, 0) + 1 skipped_reasons = {name: skipped_reasons[name] for name in order} ctx = _Run() ctx.bt_cfg, ctx.platform_cfg, ctx.timeframes = bt_cfg, platform_cfg, list(timeframes) ctx.dfs_full, ctx.caches, ctx.dev, ctx.an, ctx.cuts, ctx.k2, ctx.flags = dfs_full, caches, dev, an, cuts, k2, flags ctx.pf, ctx.pi, ctx.rm = pf, pi, rm ctx.first_pos, ctx.last_pos, ctx.n_bars = first_eligible_pos, last_pos, n_bars ctx.status, ctx.cand_pos, ctx.acc_pos, ctx.skipped_reasons = status, cand_pos, acc_pos, skipped_reasons ctx.t_loop = t_loop return ctx def run_backtest_cuda(bt_cfg: BacktestConfig, platform_cfg: PlatformConfig) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]: """gpu_backtest.run_backtest, with the replay on the device.""" cp = _gk._require_cp() if platform_cfg.primary_timeframe != bt_cfg.primary_tf: raise ValueError( f"primary_tf mismatch: BacktestConfig={bt_cfg.primary_tf.value} " f"vs PlatformConfig={platform_cfg.primary_timeframe.value}") refuse_unsupported(platform_cfg, bt_cfg) provider = make_data_provider(platform_cfg) # ── Load all required timeframes (same files as the reference script) ── timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs) flt = platform_cfg.backtest if flt.regime_filter and flt.regime_filter_timeframe not in timeframes: timeframes.append(flt.regime_filter_timeframe) dfs_full: Dict[Timeframe, pd.DataFrame] = {} cache_names: Dict[str, str] = {} tf_starts = {} from datetime import timedelta for tf in timeframes: tf_start = bt_cfg.start - timedelta(days=_history_pad_days(tf, bt_cfg.warmup_bars)) tf_starts[tf] = tf_start cache_names[tf.value] = _cache_path(bt_cfg.pair, tf, tf_start, bt_cfg.end, provider.name).name df = fetch_history(provider, bt_cfg.pair, tf, start=tf_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}") ohlc = df[["Open", "High", "Low", "Close"]].to_numpy(dtype=np.float64) if not np.isfinite(ohlc).all(): raise RuntimeError(f"{tf.value}: the cleaned frame carries non-finite OHLC — " f"the device replay assumes clean bars") dfs_full[tf] = df ctx = analyze_frames(bt_cfg, platform_cfg, dfs_full, timeframes) primary_df = dfs_full[bt_cfg.primary_tf] first_eligible_pos, last_pos, n_bars = ctx.first_pos, ctx.last_pos, ctx.n_bars acc_pos, cand_pos, skipped_reasons, t_loop = ctx.acc_pos, ctx.cand_pos, ctx.skipped_reasons, ctx.t_loop trades = build_trades(ctx, acc_pos) signal_ts_list = [primary_df.index[int(p)] for p in acc_pos] replay_seconds = time.perf_counter() - t_loop logger.info( f"Replay complete: {len(trades)} entries in {replay_seconds:.1f}s " f"({n_bars / max(replay_seconds, 1e-6):.0f} bars/s, device)") if skipped_reasons: logger.info(f"Skipped bars by reason: {skipped_reasons}") # ── Resolve trade outcomes on the RAW price path (shared GPU resolver) ── sim_df = fetch_history(provider, bt_cfg.pair, bt_cfg.primary_tf, start=tf_starts[bt_cfg.primary_tf], end=bt_cfg.end, use_cache=bt_cfg.use_cache, clean=False) _reanchor_entries_to_raw(trades, signal_ts_list, sim_df) if bt_cfg.fixed_rr_tp > 0: _apply_fixed_rr_tp(trades, bt_cfg.fixed_rr_tp) logger.info( f"RESEARCH override: fixed-RR TP = {bt_cfg.fixed_rr_tp} x risk; " f"trailing exit DISABLED for all {len(trades)} trades (incl. XAU_PB)") run_arm = 0.0 if bt_cfg.fixed_rr_tp > 0 else platform_cfg.signals.trail_arm_r t_sim = time.perf_counter() simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars, spread=bt_cfg.spread, trail_arm_r=run_arm, trail_dist_r=platform_cfg.signals.trail_dist_r) logger.info(f"Trade simulation took {time.perf_counter() - t_sim:.2f}s (CuPy GPU)") # ── research flags ── if bt_cfg.export_features and trades: t_f = time.perf_counter() attach_features(ctx, acc_pos, trades) logger.info(f"K3: {len(trades)} feature vectors on the device in {time.perf_counter() - t_f:.3f}s") if bt_cfg.record_dir: t_r = time.perf_counter() attach_snapshots(ctx, acc_pos, trades) bar_chunks = build_bar_chunks(ctx) if bt_cfg.record_bars else None logger.info(f"--record: snapshots{' + bar rows' if bt_cfg.record_bars else ''} built in " f"{time.perf_counter() - t_r:.2f}s") replay_record.write_run( bt_cfg.record_dir, trades=trades, book_df=trades_to_dataframe(trades), bt_cfg=bt_cfg, platform_cfg=platform_cfg, aggregator=SignalAggregator(platform_cfg), run_info=dict(first_pos=first_eligible_pos, last_pos=last_pos, n_primary_bars=len(primary_df), n_bars_in_window=n_bars, skipped=skipped_reasons, n_candidates=int(len(cand_pos)), workers=0, backend=BACKEND.describe(), cache_files=cache_names, replay_seconds=replay_seconds, cache_fallbacks=0), sim_df=sim_df, bar_chunks=bar_chunks) logger.info(f"--record: wrote run to {bt_cfg.record_dir}") return trades, dfs_full # ═════════════════════════════════════════════════════════════════════ # Host builders: Trade objects, warnings, snapshots, features, bar rows # ═════════════════════════════════════════════════════════════════════ def _k2_rows(k2: "ck.K2Out", positions: np.ndarray): """K2 fields + warning codes/args for the given absolute bar positions.""" idx = np.asarray(positions, dtype=np.int64) - k2.first_pos if len(idx) == 0: return [], [], np.zeros((0, ck.MAX_WARN), np.int16), np.zeros((0, ck.MAX_WARN), np.float64) fi = k2.OF2[:, idx].get() ii = k2.OI2[:, idx].get() wc = k2.warn_code[idx].get() wa = k2.warn_arg[idx].get() ints = [{n: int(ii[k, m]) for k, n in enumerate(ck.K2_INT_FIELDS)} for m in range(len(idx))] floats = [{n: float(fi[k, m]) for k, n in enumerate(ck.K2_FLOAT_FIELDS)} for m in range(len(idx))] return ints, floats, wc, wa def _warning_ctx(ctx, pos: int, r2i: Dict, r2f: Dict) -> Dict: s = ctx.platform_cfg.signals x = ctx.platform_cfg.xau_strategy primary_df = ctx.dfs_full[ctx.bt_cfg.primary_tf] return { "min_tp": s.round_trip_cost * s.min_tp_cost_multiple, "min_atr_threshold": x.min_atr_threshold, "m5_min_abs_score": x.m5_min_abs_score, "m5_sell_min_rsi": x.m5_sell_min_rsi, "big_candle_lookback_hours": x.big_candle_lookback_hours, "bc_ts": primary_df.index[r2i["bc_idx"]] if r2i["bc_idx"] >= 0 else None, "rf_tf": ctx.platform_cfg.backtest.regime_filter_timeframe.value, "regime_filter_min_adx": ctx.platform_cfg.backtest.regime_filter_min_adx, } _RF_CODES = {ck.WARN_CODES["rf_unavailable"], ck.WARN_CODES["rf_no_adx"], ck.WARN_CODES["rf_blocked"]} def build_warnings(ctx, pos: int, r2i: Dict, r2f: Dict, codes, args) -> Tuple[List[str], List[str]]: """(warnings, cleaned_warnings) of one analysed bar, after the XAU_CP cleanup (when a pattern setup was accepted) and the stale spread-gate step (when a setup existed at step 11), with the regime-filter lines appended last like the replay does.""" wctx = _warning_ctx(ctx, pos, r2i, r2f) n = r2i["warn_n"] main: List[str] = [] rf: List[str] = [] for q in range(n): code = int(codes[q]) text = format_warning(code, float(args[q]), wctx) (rf if code in _RF_CODES else main).append(text) ns = types.SimpleNamespace(warnings=main, cleaned_warnings=[], trade_setup=object() if r2i["strategy_pre_rf"] != 0 else None) if r2i["cp_accepted"]: SignalAggregator._clean_accepted_cp_warnings(ns) SignalAggregator._drop_stale_spread_gate_vetoes(ns) return list(ns.warnings) + rf, list(ns.cleaned_warnings) def build_reasons(ctx, r2i: Dict, r2f: Dict, inds: Dict[str, Dict]) -> Tuple[List[str], List[str]]: bullish, bearish = collect_reasons(inds) strat = r2i["strategy"] if r2i["strategy"] else r2i["strategy_pre_rf"] if r2i["pb_accepted"]: marker = (f"XAU PULLBACK STRATEGY: retrace {r2f['pb_retrace_atr']:.1f} " f"ATR to EMA21/KC-mid, trend resumed") (bullish if r2i["direction"] > 0 else bearish).insert(0, marker) if r2i["cp_accepted"]: atr = ctx_atr = None pattern = ck.PATTERN_ORDER[r2i["cp_pattern"]] direction = "BUY" if r2i["direction"] > 0 else "SELL" marker = (f"XAU CHART PATTERN: {pattern} ({direction}) break of {r2f['cp_trigger']:.2f}, " f"height {r2f['cp_height'] / r2f['_atr']:.1f} ATR") (bullish if r2i["direction"] > 0 else bearish).insert(0, marker) if r2i["rf_state"] == 4: rtf = ctx.platform_cfg.backtest.regime_filter_timeframe.value line = f"Regime: {rtf} ADX={r2f['rf_adx']:.1f} confirms trending market" (bullish if r2f["final_score"] > 0 else bearish).append(line) return bullish, bearish def build_trades(ctx, acc_pos: np.ndarray) -> List[Trade]: bt_cfg, platform_cfg = ctx.bt_cfg, ctx.platform_cfg primary_df = ctx.dfs_full[bt_cfg.primary_tf] pan = ctx.an[bt_cfg.primary_tf] two = platform_cfg.technical.ema_mode == "2ema" ints, floats, wc, wa = _k2_rows(ctx.k2, acc_pos) prows = k1_rows(pan, acc_pos) pip_value = PAIR_INFO[bt_cfg.pair]["pip_value"] trades: List[Trade] = [] for m, pos in enumerate(acc_pos): pos = int(pos) r2i, r2f, prow = ints[m], floats[m], prows[m] r2f["_atr"] = prow.f["atr"] if prow.i["atr_present"] else 0.0 direction = "BUY" if r2i["direction"] > 0 else "SELL" tag = ck.STRATEGY_NAMES[r2i["strategy"]] or "GENERIC" inds = describe_indicators(prow, two) # gpu_backtest._build_indicator_breakdown breakdown: Dict[str, Dict] = {} agree = oppose = neutral = 0 is_buy = direction == "BUY" for name, ind in inds.items(): sig = ind["signal"] agrees = "neutral" if sig > 0: agrees = "agree" if is_buy else "oppose" elif sig < 0: agrees = "agree" if not is_buy else "oppose" breakdown[name] = { "value": float(ind["value"]) if ind["value"] is not None else None, "signal": ck.SIGNAL_NAMES[sig], "confidence": float(ind["confidence"]), "weight": float(ind["weight"]), "agrees": agrees, "description": ind["description"], } if agrees == "agree": agree += 1 elif agrees == "oppose": oppose += 1 else: neutral += 1 total = len(breakdown) bullish, bearish = build_reasons(ctx, r2i, r2f, inds) reasons = bullish if direction == "BUY" else bearish warnings, _cleaned = build_warnings(ctx, pos, r2i, r2f, wc[m], wa[m]) # the setup's own pip value: 0.10 hard-coded by the XAU ATR generator, PAIR_INFO otherwise setup_pip = 0.10 if tag == "XAU_ATR" else pip_value # the setup's own trailing parameters (XAU_PB carries the pullback pair) trail_arm = platform_cfg.signals.pb_trail_arm_r if tag == "XAU_PB" else 0.0 trail_dist = platform_cfg.signals.pb_trail_dist_r if tag == "XAU_PB" else 0.0 trades.append(Trade( pair=bt_cfg.pair, entry_time=primary_df.index[pos + 1], entry_price=r2f["entry_price"], sl=r2f["adj_sl"], tp1=r2f["adj_tp1"], tp2=r2f["adj_tp2"], direction=direction, pip_value=setup_pip, risk_pips=r2f["sl_dist"] / setup_pip, reward_pips=r2f["tp1_dist"] / setup_pip, rr_planned=r2f["rr_rounded"], quality=ck.QUALITY_NAMES[r2i["quality"]], final_score=r2f["final_score"], final_confidence=r2f["final_conf"], market_regime=ck.REGIME_NAMES[prow.i["regime"]], volatility_regime=ck.VOL_NAMES[prow.i["vol"]], strategy_tag=tag, entry_reasons=list(reasons), entry_warnings=list(warnings), indicator_breakdown=breakdown, indicators_agreeing=agree, indicators_opposing=oppose, indicators_neutral=neutral, indicators_total=total, stake_multiplier=1.0, trail_arm_r=trail_arm, trail_dist_r=trail_dist, )) return trades def attach_features(ctx, acc_pos: np.ndarray, trades: List[Trade]) -> None: bt_cfg = ctx.bt_cfg confirms = [(ctx.an[tf], ctx.cuts[tf]) for tf in bt_cfg.confirmation_tfs] # f_h1_ext_atr: the H1 frame wherever it sits; f_c_other_ext_atr: the first # non-H1 confirmation with >= 21 bars (evaluated per bar on the device). if bt_cfg.primary_tf == Timeframe.H1: h1 = (1, None, None) elif Timeframe.H1 in ctx.dfs_full and Timeframe.H1 in ctx.cuts: h1 = (2, ctx.dev[Timeframe.H1], ctx.cuts[Timeframe.H1]) else: h1 = (0, None, None) others = [(ctx.dev[tf], ctx.cuts[tf]) for tf in bt_cfg.confirmation_tfs if tf is not Timeframe.H1][:2] F = ck.features(acc_pos, ctx.dev[bt_cfg.primary_tf], ctx.an[bt_cfg.primary_tf], ctx.k2, len(bt_cfg.confirmation_tfs), confirms, h1, others, ctx.flags, ctx.pf, ctx.pi) for t, row in zip(trades, F): t.features = {name: float(row[k]) for k, name in enumerate(position_sizing.FEATURE_ORDER)} def _tf_block(row: K1Row, tf_name: str, role: str, direction: str, two: bool) -> Dict: inds = describe_indicators(row, two) is_buy = direction == "BUY" out_inds: Dict[str, Dict] = {} for name, ind in inds.items(): sig = ind["signal"] base = float(ind["weight"]) eff = replay_record._finite(ind["effective_weight"]) out_inds[name] = { "signal": sig, "signal_name": ck.SIGNAL_NAMES[sig], "value": replay_record._finite(ind["value"]), "confidence": replay_record._finite(ind["confidence"]), "base_weight": base, "regime_mult": (eff / base) if (eff is not None and base) and base > 0 else None, "eff_weight": eff, "wscore": replay_record._finite(ind["weighted_score"]), "agrees": replay_record._agrees(sig, direction), "description": str(ind["description"] or ""), } kl = key_levels(row, two) return { "tf": tf_name, "role": role, "regime": ck.REGIME_NAMES[row.i["regime"]], "vol": ck.VOL_NAMES[row.i["vol"]], "composite_score": replay_record._finite(row.f["composite_score"]), "composite_signal": int(row.i["composite_signal"]), "confidence": replay_record._finite(row.f["composite_conf"]), "atr": replay_record._finite(kl.get("atr")), "atr_pct": replay_record._finite(kl.get("atr_pct")), "pattern_signals": pattern_signals(row), "indicators": out_inds, } def attach_snapshots(ctx, acc_pos: np.ndarray, trades: List[Trade]) -> None: """replay_record.snapshot_report for every accepted trade, from the device outputs.""" bt_cfg, platform_cfg = ctx.bt_cfg, ctx.platform_cfg primary_df = ctx.dfs_full[bt_cfg.primary_tf] primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf]) two = platform_cfg.technical.ema_mode == "2ema" pdev = ctx.dev[bt_cfg.primary_tf] medians = ck.prefix_median(pdev["high"], pdev["low"], pdev["n"])[acc_pos].get() if len(acc_pos) else [] ints, floats, wc, wa = _k2_rows(ctx.k2, acc_pos) prows = k1_rows(ctx.an[bt_cfg.primary_tf], acc_pos) conf_rows = {} conf_cuts = {} for tf in bt_cfg.confirmation_tfs: cut = ctx.cuts[tf][acc_pos].get() if len(acc_pos) else np.zeros(0, np.int64) conf_cuts[tf] = cut present = cut > 0 rows = k1_rows(ctx.an[tf], np.where(present, cut - 1, 0)) conf_rows[tf] = rows s = platform_cfg.signals for m, (pos, t) in enumerate(zip(acc_pos, trades)): pos = int(pos) r2i, r2f, prow = ints[m], floats[m], prows[m] r2f["_atr"] = prow.f["atr"] if prow.i["atr_present"] else 0.0 direction = t.direction inds = describe_indicators(prow, two) bullish, bearish = build_reasons(ctx, r2i, r2f, inds) warnings, cleaned = build_warnings(ctx, pos, r2i, r2f, wc[m], wa[m]) tag = t.strategy_tag pmask = r2i["penalty_mask"] penalty_tags = {tagname: float(ck.PENALTY_AMOUNT[tagname]) for tagname in ck.PENALTY_SORTED if pmask & (1 << ck.PENALTY_TAGS.index(tagname))} mtf_signals = {bt_cfg.primary_tf.value: int(prow.i["composite_signal"])} mtf_scores = {bt_cfg.primary_tf.value: replay_record._finite(prow.f["composite_score"])} tfs = [_tf_block(prow, bt_cfg.primary_tf.value, "primary", direction, two)] for k, tf in enumerate(bt_cfg.confirmation_tfs): if conf_cuts[tf][m] == 0: continue crow = conf_rows[tf][m] mtf_signals[tf.value] = int(crow.i["composite_signal"]) mtf_scores[tf.value] = replay_record._finite(crow.f["composite_score"]) if k + 1 < len(replay_record.ROLES): tfs.append(_tf_block(crow, tf.value, replay_record.ROLES[k + 1], direction, two)) trail_arm = s.pb_trail_arm_r if tag == "XAU_PB" else s.trail_arm_r trail_dist = s.pb_trail_dist_r if tag == "XAU_PB" else s.trail_dist_r t.snapshot = { "bar_ts": str(primary_df.index[pos]), "decision_time": pd.Timestamp(primary_df.index[pos] + primary_dur).isoformat(), "final_signal": int(r2i["final_signal"]), "final_score": replay_record._finite(r2f["final_score"]), "final_confidence": replay_record._finite(r2f["final_conf"]), "quality": ck.QUALITY_NAMES[r2i["quality"]], "mtf_agreement": bool(r2i["mtf_agreement"]), "mtf_signals": mtf_signals, "mtf_scores": mtf_scores, "penalty_tags": penalty_tags, "penalty_total": replay_record._finite(r2f["penalty_total"]), "momentum_fading": bool(r2i["momentum_fading"]), "exhaustion_count": int(r2i["exhaustion_count"]), "spread_estimate": replay_record._finite(float(medians[m])), "current_price": replay_record._finite(float(primary_df["Close"].iloc[pos])), "warnings": warnings, "cleaned_warnings": cleaned, "bullish_reasons": bullish, "bearish_reasons": bearish, "setup": { "strategy": tag, "setup_pattern": ck.PATTERN_ORDER[r2i["cp_pattern"]] if tag == "XAU_CP" else None, "round_trip_cost": replay_record._finite(s.round_trip_cost), "trail_arm_r": replay_record._finite(trail_arm), "trail_dist_r": replay_record._finite(trail_dist), "entry_price": replay_record._finite(r2f["setup_entry"]), "stop_loss": replay_record._finite(r2f["setup_sl"]), "take_profit_1": replay_record._finite(r2f["setup_tp1"]), "take_profit_2": replay_record._finite(r2f["setup_tp2"]), "risk_reward": replay_record._finite(r2f["rr_rounded"]), }, "tfs": tfs, } def build_bar_chunks(ctx): """replay_record's (matrix float32 [n,160], ts_ns, pos, warning0) for every analysed primary bar (the K6 rows plus the host warning text).""" bt_cfg = ctx.bt_cfg k2 = ctx.k2 n_bars = k2.n_bars if n_bars <= 0: return None status = k2.i("status").get()[:n_bars] bar_pos = (np.flatnonzero(status >= 1) + k2.first_pos).astype(np.int64) if len(bar_pos) == 0: return None confirms = [(ctx.an[tf], ctx.cuts[tf]) for tf in bt_cfg.confirmation_tfs] mat = ck.bar_rows(bar_pos, ctx.dev[bt_cfg.primary_tf], ctx.an[bt_cfg.primary_tf], k2, len(bt_cfg.confirmation_tfs), confirms) ints, floats, wc, wa = _k2_rows(k2, bar_pos) prows_atr = None warning0: List[str] = [] prim = ctx.an[bt_cfg.primary_tf] atr_present = prim.i("atr_present")[bar_pos].get() atr_vals = prim.f("atr")[bar_pos].get() for m, pos in enumerate(bar_pos): r2i, r2f = ints[m], floats[m] if r2i["warn_n"] == 0: warning0.append("") continue r2f["_atr"] = float(atr_vals[m]) if atr_present[m] else 0.0 warnings, _ = build_warnings(ctx, int(pos), r2i, r2f, wc[m], wa[m]) warning0.append(warnings[0] if warnings else "") primary_index = ctx.dfs_full[bt_cfg.primary_tf].index return [(mat, primary_index.as_unit("ns").asi8[bar_pos].astype(np.int64), bar_pos.astype(np.int64), warning0)] # ═════════════════════════════════════════════════════════════════════ # CLI (the reference script's parser, config block and output, verbatim) # ═════════════════════════════════════════════════════════════════════ def require_gpu_or_exit(requested_backend: Optional[str]) -> None: if requested_backend == "cpu" or BACKEND.mode == "cpu": raise SystemExit( "cudaBasedBacktesting.py runs the decision stage on CUDA and has no CPU path " "(--backend cpu / FB2_BACKEND=cpu requested). Use backtest/gpu_backtest.py for the CPU twins.") if requested_backend: BACKEND.set_mode(requested_backend) if not BACKEND.available: raise SystemExit( f"cudaBasedBacktesting.py needs a working CUDA backend: {BACKEND.describe()}. " f"Use backtest/gpu_backtest.py.") 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", ) require_gpu_or_exit(args.backend) 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()] if args.confirm_tfs is not None else None ) from pair_signals import PAIR_PRODUCERS # noqa: E402 factory = PAIR_PRODUCERS.get(pair) if factory is not None: platform_cfg = factory(primary_tf, confirm_tfs) else: platform_cfg = PlatformConfig( symbol=pair, display_name=PAIR_INFO[pair]["display"], primary_timeframe=primary_tf, confirmation_timeframes=confirm_tfs or [Timeframe.H1, Timeframe.H4], ) from app_config import apply_pair_overrides # noqa: E402 apply_pair_overrides(platform_cfg) if args.fixed_rr_tp is not None: if args.fixed_rr_tp <= 0: raise SystemExit("--fixed-rr-tp must be > 0") if (args.trail_arm is not None or args.trail_dist is not None or any(s.split("=", 1)[0].strip().startswith(("signals.trail", "signals.pb_trail")) for s in (args.set or []))): raise SystemExit( "--fixed-rr-tp conflicts with --trail-arm/--trail-dist/" "--set signals.*trail* (the flag disables trailing entirely)") if args.swing_sr_lookback > 0: platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback if args.trail_arm is not None: platform_cfg.signals.trail_arm_r = args.trail_arm if args.trail_dist is not None: platform_cfg.signals.trail_dist_r = args.trail_dist if args.cp: platform_cfg.xau_strategy.cp_enabled = True if args.cp_patterns: platform_cfg.xau_strategy.cp_patterns = [ s.strip() for s in args.cp_patterns.split(",") if s.strip() ] if args.big_candle is not None: platform_cfg.xau_strategy.big_candle_atr_multiple = args.big_candle if args.big_candle_hours is not None: platform_cfg.xau_strategy.big_candle_lookback_hours = args.big_candle_hours if args.set: _apply_config_overrides(platform_cfg, args.set) confirm_tfs = list(platform_cfg.confirmation_timeframes) from datetime import datetime, timedelta, timezone end = ( datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc) if args.end else datetime.now(timezone.utc) ) if args.start: start = datetime.fromisoformat(args.start).replace(tzinfo=timezone.utc) else: start = end - timedelta(days=args.days) spread = (args.spread if args.spread is not None else PAIR_INFO[pair].get("default_spread", 0.30)) bt_cfg = BacktestConfig( pair=pair, start=start, end=end, primary_tf=primary_tf, confirmation_tfs=confirm_tfs, warmup_bars=args.warmup_bars, cooldown_bars=args.cooldown_bars, max_hold_bars=args.max_hold_bars, use_cache=not args.no_cache, quality_filter=[q.strip() for q in args.quality.split(",") if q.strip()], strategies_only=[("XAU_ATR" if s.strip() == "XAU" else s.strip()) for s in args.strategies.split(",") if s.strip()], spread=spread, workers=args.workers, fixed_rr_tp=args.fixed_rr_tp or 0.0, export_features=args.export_features, record_dir=args.record, record_bars=args.record_bars, ) if args.workers: logger.info(f"--workers {args.workers} is ignored: the replay runs on the device") if args.record: logger.info( f"--record ON: decision snapshots → {args.record} " f"(trades/indicators/events{'/bars' if args.record_bars else ''} parquet + manifest.json; " f"the trade book is unchanged)") if args.export_features: logger.info( f"--export-features ON: appending stake_multiplier, r_sized and " f"{len(position_sizing.FEATURE_ORDER)} f_* feature columns to the " f"CSV export (research flag; book itself is unchanged)") logger.info( f"Backend: {BACKEND.describe()} | " f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()} | " f"Confirm TFs: {','.join(tf.value for tf in confirm_tfs)} | Spread: {spread}" ) trades, _ = run_backtest_cuda(bt_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, include_features=args.export_features) 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())