diff --git a/CLAUDE.md b/CLAUDE.md index e565aa5..26766ce 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,8 +20,10 @@ you must follow before changing anything that affects signals. Section 10 lists python -m venv venv && venv/bin/pip install -r requirements.txt # Python 3.12; a venv/ already exists on the dev box cp config.json.example config.json # then fill in the keys (see section 5) +venv/bin/pip install cupy-cuda12x "nvidia-cuda-runtime-cu12==12.9.*" # optional, dev box: backtester GPU backend + # (CuPy from pip needs that header wheel to compile kernels) venv/bin/python test_all.py # QA suite, no network; prints its own count - # (529 tests / 34 modules as of 2026-09-17) + # (536 tests / 35 modules as of 2026-09-18) venv/bin/python main.py --pair XAUUSD --timeframe H1 --detail # one signal report in the terminal # --json full dump, --panel adds the advisory LLM panel, # --timeframe M5 | M15 | H1 | H4 @@ -41,7 +43,10 @@ realistic Oanda figure is 0.65 and must be passed explicitly), `--quality A+,A,B `--export-csv out.csv`, `--detail`, `--set PATH=JSON` (research override of any config value; an unknown path is a hard exit), `--export-features` (appends the 65 decision-time columns; never use it on a parity run because it changes the file), `--fixed-rr-tp N` (research only, not shipped), -`--record DIR [--record-bars]` (research recorder, see section 9). +`--record DIR [--record-bars]` (research recorder, see section 9), `--backend auto|gpu|cpu` (`auto` +uses the CuPy GPU backend when CUDA works and falls back to NumPy with the reason in the `Backend:` +line; `gpu` refuses to fall back; `cpu` forces the NumPy twins for a parity run; the env var +`FB2_BACKEND` sets the default). Both backends produce byte-identical books (section 9). `test_all.py` is **not pytest**. It uses its own `@test("name")` decorator that runs each test the moment the file is imported. Run the whole file; it prints per-module results and exits non-zero @@ -266,7 +271,8 @@ Arrows show who imports whom (generated from the code on 2026-09-15). | File | Role | |---|---| -| `backtest/gpu_backtest.py` | The backtester (CuPy on a GPU, NumPy otherwise; check the `Backend:` log line) | +| `backtest/gpu_backtest.py` | The backtester. The indicator precompute over the full history and the resolution of every trade run on the GPU when CUDA works (`--backend`, `Backend:` log line names the device or the fallback reason); the per-bar replay through `analyze_from_dataframes` is the live Python signal code in forked CPU workers by design and is 80–95% of wall time | +| `backtest/gpu_kernels.py` | CuPy backend state (lazy probe, never silent, sets `CUDA_PATH` to the pip header wheel) and the three CUDA kernels: EWM batch, pandas rolling mean/var replica, trade resolver. Each is a bit-for-bit twin of the NumPy/pandas code kept in `gpu_backtest.py` (`--fmad=false`, one thread per series or trade); the twins are pinned by Module 36 and by the reference MD5s | | `backtest/replay_record.py` | Opt-in `--record` recorder: why every accepted trade was taken, plus calendar events. Book-neutral by contract | | `backtest/replay_scorecard.py` | Read-only reader of recorded runs (`DIR --check` verifies its invariants) | | `scripts/ff_calendar_scraper.py` | Builds `backtest/cache/calendar.parquet` (Forex Factory events) for the recorder | @@ -281,7 +287,7 @@ Arrows show who imports whom (generated from the code on 2026-09-15). | `app_config.py` | Loads `config.json` once per process; env var → config.json → refuse; validates keys | | `config.json.example` | The schema of `config.json` (which is gitignored) | | `deploy/README-dashboard.md`, `deploy/gold-dashboard.service` | Live-server checklist and systemd unit for the dashboard daemon | -| `test_all.py` | The QA suite (34 modules; Module 35 is the agent-panel v2 module) | +| `test_all.py` | The QA suite (35 modules; Module 36 is the GPU backend parity module) | --- @@ -359,7 +365,9 @@ day, 30 kept), `logs/dashboard.log`. Read the databases with `mode=ro`, never `i ## 6. Testing - `venv/bin/python test_all.py` runs everything with mocked network. Expect - `RESULTS: 529 passed, 0 failed, 529 total` (2026-09-17). The suite's hermetic app config points + `RESULTS: 536 passed, 0 failed, 536 total` (2026-09-18). On a box with a visible CUDA device + Module 36 fails if the GPU backend falls back (the reason is in the message); with + `FB2_BACKEND=cpu` or no CUDA the GPU-path tests print a notice and the NumPy twins are tested. The suite's hermetic app config points both database writers at a suite-private temp dir; before 2026-09-17 the RSS fixture tests wrote fixture headlines into the real `logs/newsdata.db`, and a Module 32 test now pins the isolation. One Twelve Data interval-map test reaches the network and can time out on a slow connection; a rerun clears it. @@ -406,8 +414,9 @@ day, 30 kept), `logs/dashboard.log`. Read the databases with `mode=ro`, never `i Process rules that go with them: every backtest report follows `backtest/instructions_for_backtest.txt`; every candidate is compared with a control run on the -same code and cache, never with an old book directly; the `Backend:` line and the "GPU cache -fallbacks: 0" line are part of the evidence; keep `docs/` in sync when behaviour changes. +same code and cache, never with an old book directly; the `Backend:` line (since 2026-09-18 it +names the CUDA device or the fallback reason) and the "GPU cache fallbacks: 0" line are part of +the evidence; keep `docs/` in sync when behaviour changes. --- @@ -446,7 +455,9 @@ trade-level diff is needed; each run takes seconds to a few minutes. Harness for every reference book: `venv/bin/python backtest/gpu_backtest.py --pair XAUUSD --spread 0.65 --workers 12 --primary-tf --start --end --export-csv `, factory config, no `--set`, `--quality`, `--strategies`, `--confirm-tfs` or `--export-features`. -All of these run offline from the kept cache windows (NumPy CPU backend, 0 cache fallbacks). +All of these run offline from the kept cache windows (0 cache fallbacks). Since 2026-09-18 the +CuPy GPU backend and `--backend cpu` both reproduce every MD5 below; earlier reports say +`NumPy CPU fallback` because CuPy could not compile kernels on the dev box until then. | Book | Window | Trades / R / PF | MD5 of the CSV | |---|---|---|---| @@ -466,7 +477,8 @@ live names (`XAU` still means `XAU_ATR`). Other current-HEAD figures at $0.65: M5 W3 (3-year) 870 / +153.6R / PF 1.50; M15 W3 1273 / +186.6R / PF 1.39; M5 12-month (2025-08-14 → 2026-08-14) 560 / +108.81R / PF 1.54; H1 12-month (2025-08-04 → 2026-08-04) 91 / +22.2R / PF 1.79. All five MD5s above were -reproduced on 2026-09-15 after the cache prune. +reproduced on 2026-09-15 after the cache prune and on 2026-09-18 on both backends +(`backtest/research/gpu_backend_2026-09-18.md`). Kept cache windows (`backtest/cache/oanda_XAU_USD___.parquet`; the padded start is the window start minus 62 days for M5/M15/H1, 234 for H4 and 1,400 for D1, so the D1 @@ -515,6 +527,20 @@ MD5s. Scorecards from it are in-book statistics, never decisions. `NewsEngine`. - Twelve Data candles are not a drop-in replacement for Oanda (section 4). - `pyarrow` (Parquet) is installed in the dev venv only; live code never touches Parquet. +- CuPy from pip (`cupy-cuda12x`) cannot compile a kernel without the CUDA headers; the + `nvidia-cuda-runtime-cu12` wheel provides them and `backtest/gpu_kernels.py` points `CUDA_PATH` + at it. Until 2026-09-18 that failure was swallowed: every book was built on NumPy while the file + was called a GPU backtester. Now the fallback reason is in the `Backend:` line and Module 36 + fails on a box that has a visible GPU but no working backend. `cupy` is banned from the live + roots by the same AST guard as scikit-learn. +- The CUDA kernels are bit-for-bit replicas of the CPU code: one thread per series (EWM, rolling) + or per trade, compiled with `--fmad=false`. Do not "optimise" them with a parallel scan or a + two-pass variance: the last bits change and a threshold can flip somewhere in 500k scans. A + single sequential series is not faster on the GPU than pandas' C loop; the win is the batch + across series and trades, and the replay itself stays on the CPU (same code as live). +- The replay workers are forked after the CUDA context exists and never touch the device (every + cache array is host NumPy, asserted before the fork). Python 3.12 may warn about forking a + multi-threaded process under `-X dev`; that is expected. - The dashboard's due-timeframe check uses the raw wake time; flooring `wake − buffer` skipped every H1 and H4 scan once (2026-08-02). - The agent panel's price context reads the frames `analyze_pair` stashes on `report._frames`. @@ -535,6 +561,7 @@ One line per validated change or verdict, newest first. Full evidence in the nam | Date | Decision | Where the evidence is | |---|---|---| +| 2026-09-18 | GPU backend made real: CuPy had never compiled a kernel on the dev box (`CUDA_PATH` missing for the pip layout), so every book to date was NumPy. SHIPPED: `backtest/gpu_kernels.py` (lazy, loud backend probe; `nvidia-cuda-runtime-cu12` header wheel; CUDA kernels for the EWM batch, the pandas rolling mean/var replica and the trade resolver, each a bit-for-bit twin of the CPU code), `--backend auto/gpu/cpu` + `FB2_BACKEND`, Module 36. All five reference MD5s reproduced on the GPU and with `--backend cpu`; precompute GPU == CPU on all 42 cached windows (2.34 M bars). Wall time roughly unchanged: the replay is CPU by design | `backtest/research/gpu_backend_2026-09-18.md` | | 2026-09-17 | Local news scorer evaluated (the owner's scikit TF-IDF model in `~/Development/ML-NEWS`, trained on 606 DeepSeek gold labels): golden set 39/78 vs DeepSeek 78/78 vs keywords 24/78; out-of-sample agreement with DeepSeek 64% (DeepSeek agrees with itself 90%); cleaning / negation marking / label weighting move nothing; backfilling the 961 missing gold labels lifts macro-headline agreement 38% → 52% but only reshuffles golden-set errors. Verdict: a fallback, not a replacement. SHIPPED to grow a complete corpus: `xau_score` archived for every headline, no-currency headlines scored and archived (never shown), every new headline sent in its cycle (25 per call). Dashboard process made database-read-only (`set_db_writes_for_process`); the bot is the single writer. Books unaffected (news never enters signals) | `~/Development/ML-NEWS/eval/RESULTS_2026-09-17.md` (evaluation scripts live outside the repo by owner instruction) | | 2026-09-17 | Live review of two weeks of `signals.db`: live == backtest bar for bar (26/29 trades matched, R within 0.14); the raw advisory −7.15R is cluster re-entries the validated books never take (cooldown view −1.6R, backtester +1.1R on the window, an ordinary fortnight); composite score inversely related to the 24 h move on all timeframes (two-week bear reading, test in-book with `--record-bars`); big-candle gate re-tested and REJECTED again (removes +32R / +60R of net-positive trades). Agent panel v2 SHIPPED: probabilities anchored to book base rates, code-computed decision context, no news doctrine, mean aggregator (`verdict_source: "mean"`), prompt version logged, offline replay harness; owner decision the same day: only the macro persona runs live (`personas: ["macro"]`, 1 call per panel; technical / risk / contrarian carried no information in either evaluation, kept for research); backtester `strategy` tag fixed (label-only, books re-pinned). Delivery findings: dashboard daemon on stale code since Sep 10, bot outage Sep 9 | `backtest/research/signal_log_review_2026-09-17.md`, `backtest/research/panel_v2_2026-09-17.md` | | 2026-09-15 | Backtest folder cleanup: `diagnose_filters.py`, `shadow_scan.py`, and the two one-time news scripts removed (last commit with them `931b265`); every pinned CSV, the research output folders, the BTC/ETH cache snapshots, the Tiingo caches and 13 uncited Oanda windows moved to `~/Development/ForexBot2_trash_2026-09-15/` (346 MB) pending the owner's delete; all five reference MD5s reproduced afterwards. Suite 506 / 33 | commit `b0b71e1` | diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 686f290..c108f07 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -19,7 +19,9 @@ Key properties: - 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. +- Ships with CuPy fallback to NumPy so the script runs anywhere. The CUDA + kernels (backtest/gpu_kernels.py) are bit-for-bit replicas of the NumPy + path, so both backends produce the same book; --backend cpu forces NumPy. Usage: python backtest/gpu_backtest.py --pair XAUUSD --days 365 @@ -64,36 +66,22 @@ from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine # noqa: from signal_aggregator import SignalAggregator, SignalReport, TradeSetup # noqa: E402 import position_sizing # noqa: E402 from backtest import replay_record # noqa: E402 +from backtest import gpu_kernels as _gk # 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. +# GPU backend — backtest/gpu_kernels.py owns the CuPy state (probed +# lazily and never silently: a fallback carries its reason into the +# "Backend:" line) and the CUDA kernels. Every kernel is a bit-for-bit +# replica of the NumPy/pandas code kept in this file as the CPU path, so +# `--backend cpu` and the GPU produce the same book (proof: the five +# reference MD5s, CLAUDE.md section 9). The forked replay workers never +# touch the device: every precomputed series is moved to host first. # ───────────────────────────────────────────────────────────────────── -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 +BACKEND = _gk.BACKEND -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) +def _gpu_on() -> bool: + return BACKEND.available logger = logging.getLogger("gpu_backtest") @@ -215,18 +203,29 @@ class IndicatorCache: def precompute_indicators( df: pd.DataFrame, tech_cfg, gpu_log: bool = False ) -> IndicatorCache: - """Compute every causal indicator series once on the full df.""" + """Compute every causal indicator series once on the full df. + + Dispatches to the CUDA kernels when the backend is available and to + the NumPy/pandas code otherwise. The two are bit-identical (Module 36 + and the 42-window check in backtest/research/gpu_backend_2026-09-18.md). + """ + if gpu_log: + logger.info( + f"Pre-computing indicators on {len(df)} bars " + f"({'CuPy GPU' if _gpu_on() else 'NumPy CPU'} backend)" + ) + if _gpu_on(): + return _precompute_indicators_gpu(df, tech_cfg) + return _precompute_indicators_cpu(df, tech_cfg) + + +def _precompute_indicators_cpu(df: pd.DataFrame, tech_cfg) -> IndicatorCache: + """The NumPy/pandas path — the reference the CUDA twin replicates.""" 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) @@ -388,6 +387,187 @@ def precompute_indicators( ) +def _assert_host_cache(cache: IndicatorCache) -> None: + """Every array the forked workers will read must be plain NumPy.""" + for name, val in vars(cache).items(): + if name in ("df", "index_pos", "rsi_series", "macd_hist_series") or val is None: + continue + if type(val) is not np.ndarray: + raise TypeError(f"IndicatorCache.{name} is {type(val).__name__}, not np.ndarray") + + +def _precompute_indicators_gpu(df: pd.DataFrame, tech_cfg) -> IndicatorCache: + """CUDA twin of `_precompute_indicators_cpu`. + + Same expressions in the same order (single-op CuPy ufuncs cannot fuse + into an FMA and double add/mul/div/sqrt are correctly rounded on both + sides); the recurrences and the pandas rolling statistics run in the + kernels of backtest/gpu_kernels.py, one thread per series. Everything + comes back to the host once at the end: the replay workers read NumPy. + """ + cp = BACKEND.cp + nan = float("nan") + close_h = df["Close"].to_numpy(dtype=np.float64) + n = len(close_h) + close = cp.asarray(close_h) + high = cp.asarray(df["High"].to_numpy(dtype=np.float64)) + low = cp.asarray(df["Low"].to_numpy(dtype=np.float64)) + + # ── RSI inputs (delta[0] is NaN; NaN > 0 is False so gain[0] = 0.0) ── + delta = cp.empty(n, dtype=cp.float64) + delta[0] = nan + delta[1:] = close[1:] - close[:-1] + gain = cp.where(delta > 0, delta, 0.0) + loss = cp.where(delta < 0, -delta, 0.0) + + # ── True range (fmax ignores NaN, like the CPU path) ── + prev_close = cp.empty(n, dtype=cp.float64) + prev_close[0] = nan + prev_close[1:] = close[:-1] + tr1 = high - low + tr2 = cp.abs(high - prev_close) + tr3 = cp.abs(low - prev_close) + tr2[0] = nan + tr3[0] = nan + tr = cp.fmax(cp.fmax(tr1, tr2), tr3) + + # ── Directional movement ── + plus_dm_raw = cp.empty(n, dtype=cp.float64) + minus_dm_raw = cp.empty(n, dtype=cp.float64) + plus_dm_raw[0] = nan + minus_dm_raw[0] = nan + plus_dm_raw[1:] = high[1:] - high[:-1] + minus_dm_raw[1:] = -(low[1:] - low[:-1]) + cond_plus = (plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0) + plus_dm = cp.where(cond_plus, plus_dm_raw, 0.0) + cond_minus = (minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0) + minus_dm = cp.where(cond_minus, minus_dm_raw, 0.0) + + # ── Stage 1: every recurrence that depends only on price, one launch ── + def span_alpha(span: int) -> float: + return 2.0 / (span + 1.0) # identical to _gpu_ewm_span + + rsi_p = tech_cfg.rsi_period + atr_p = tech_cfg.atr_period + adx_p = tech_cfg.adx_period + 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) + have_trend = n >= tech_cfg.ema_trend + jobs = [ + (close, span_alpha(tech_cfg.ema_fast), 0), + (close, span_alpha(tech_cfg.ema_medium), 0), + (close, span_alpha(tech_cfg.ema_slow), 0), + (close, span_alpha(tech_cfg.macd_fast), 0), + (close, span_alpha(tech_cfg.macd_slow), 0), + (close, span_alpha(kc_period), 0), + (gain, 1.0 / rsi_p, rsi_p), + (loss, 1.0 / rsi_p, rsi_p), + (tr, 1.0 / atr_p, atr_p), + (tr, 1.0 / adx_p, adx_p), + (plus_dm, 1.0 / adx_p, adx_p), + (minus_dm, 1.0 / adx_p, adx_p), + (tr, 1.0 / kc_atr_period, kc_atr_period), + ] + if have_trend: + jobs.append((close, span_alpha(tech_cfg.ema_trend), 0)) + out = _gk.ewm_batch_dev([j[0] for j in jobs], [j[1] for j in jobs], + [j[2] for j in jobs]) + (ema_fast, ema_medium, ema_slow, ema_macd_fast, ema_macd_slow, kc_mid, + avg_gain, avg_loss, atr, atr_adx, plus_dm_smooth, minus_dm_smooth, + kc_atr) = out[:13] + ema_trend = out[13] if have_trend else None + + # ── RSI ── + avg_loss_safe = cp.where(avg_loss == 0, 1e-10, avg_loss) + rs = avg_gain / avg_loss_safe + rsi = 100.0 - (100.0 / (1.0 + rs)) + + # ── MACD line; DI / DX ── + macd_line = ema_macd_fast - ema_macd_slow + atr_adx_safe = cp.where(atr_adx == 0, 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 = cp.where(di_sum == 0, nan, di_sum) + dx = 100.0 * cp.abs(plus_di - minus_di) / di_sum_safe + + # ── Stage 2: recurrences on stage-1 outputs, one launch ── + signal_line, adx = _gk.ewm_batch_dev( + [macd_line, dx], [span_alpha(tech_cfg.macd_signal), 1.0 / adx_p], [0, adx_p]) + macd_hist = macd_line - signal_line + + # ── Bollinger: the pandas rolling replica (mean + var, then zsqrt) ── + (bb_mid,), (bb_var,) = _gk.roll_mean_var_dev([close], tech_cfg.bb_period, ddof=0) + bb_std = _gk.roll_std_from_var_dev(bb_var) + 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 = cp.where(bb_range == 0, nan, bb_range) + bb_position = (close - bb_lower) / bb_range_safe + bb_position = cp.where(cp.isnan(bb_position), 0.5, bb_position) + bb_mid_safe = cp.where(bb_mid == 0, nan, bb_mid) + bb_band_width = (bb_upper - bb_lower) / bb_mid_safe + # 20-bar mean of the band width: the CPU path uses the literal 20. + (bb_avg_width,), _unused = _gk.roll_mean_var_dev([bb_band_width], 20, ddof=0) + + # ── ADX slope ── + adx_slope_lookback = getattr(tech_cfg, 'adx_slope_lookback', 5) + adx_slope = cp.full(n, nan, dtype=cp.float64) + if n > adx_slope_lookback: + adx_slope[adx_slope_lookback:] = adx[adx_slope_lookback:] - adx[:-adx_slope_lookback] + + # ── ROC ── + roc_period = getattr(tech_cfg, 'roc_period', 10) + roc = cp.full(n, nan, dtype=cp.float64) + if n > roc_period: + prev_close = close[:-roc_period] + prev_safe = cp.where(prev_close == 0, nan, prev_close) + roc[roc_period:] = ((close[roc_period:] - prev_close) / prev_safe) * 100.0 + + # ── Keltner ── + kc_upper = kc_mid + kc_multiplier * kc_atr + kc_lower = kc_mid - kc_multiplier * kc_atr + kc_squeeze = ((bb_lower > kc_lower) & (bb_upper < kc_upper)).astype(cp.float64) + + # ── One transfer to the host; the workers read NumPy only ── + def h(a): + return a.get() + + rsi_h = h(rsi) + macd_hist_h = h(macd_hist) + cache = IndicatorCache( + df=df, + index_pos={ts: i for i, ts in enumerate(df.index)}, + ema_fast=h(ema_fast), + ema_medium=h(ema_medium), + ema_slow=h(ema_slow), + ema_trend=h(ema_trend) if ema_trend is not None else None, + rsi=rsi_h, + rsi_series=pd.Series(rsi_h, index=df.index, name="rsi"), + macd_hist=macd_hist_h, + macd_hist_series=pd.Series(macd_hist_h, index=df.index, name="macd_hist"), + bb_mid=h(bb_mid), + bb_upper=h(bb_upper), + bb_lower=h(bb_lower), + bb_position=h(bb_position), + bb_band_width=h(bb_band_width), + bb_avg_width=h(bb_avg_width), + atr=h(atr), + adx=h(adx), + plus_di=h(plus_di), + minus_di=h(minus_di), + adx_slope=h(adx_slope), + roc=h(roc), + kc_mid=h(kc_mid), + kc_upper=h(kc_upper), + kc_lower=h(kc_lower), + kc_squeeze=h(kc_squeeze), + ) + _assert_host_cache(cache) + return cache + + # ═════════════════════════════════════════════════════════════════════ # GPU-AWARE TECHNICAL ENGINE (subclasses live engine, lookup-only) # ═════════════════════════════════════════════════════════════════════ @@ -1330,6 +1510,15 @@ def simulate_trades( closes_arr[i, :n] = sim_close[entry_pos:end_pos] exit_times.append(list(sim_index[entry_pos:end_pos])) + if _gpu_on(): + # One CUDA thread per trade, both exit modes (bit-identical twin of + # the two CPU paths below; Module 36 and the reference MD5s). + _resolve_trades_gpu( + trades, opens_arr, highs_arr, lows_arr, closes_arr, + valid_lens, exit_times, spread, trail_arm_r, trail_dist_r, + ) + return + if trail_arm_r > 0: _simulate_trailing( trades, opens_arr, highs_arr, lows_arr, closes_arr, @@ -1337,42 +1526,33 @@ def simulate_trades( ) return - # GPU vectorized: for each trade, scan its window for first SL/TP hit. + # CPU static path, vectorized: for each trade, scan its window for the + # 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_buy = lows_arr <= sls + tp_hit_buy = highs_arr >= tps + sl_hit_sell = highs_arr >= sls + tp_hit_sell = lows_arr <= tps - 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) + sl_hit = np.where(is_buy, sl_hit_buy, sl_hit_sell) + tp_hit = np.where(is_buy, tp_hit_buy, tp_hit_sell) # Mask out NaN bars (no data past valid_lens). - valid_mask = ~cp.isnan(h_g) + valid_mask = ~np.isnan(highs_arr) 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) + idx_template = np.arange(max_bars)[None, :] + sl_idx = np.where(sl_hit, idx_template, max_bars) + tp_idx = np.where(tp_hit, idx_template, max_bars) + sl_first_arr = np.min(sl_idx, axis=1) + tp_first_arr = np.min(tp_idx, axis=1) # 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). @@ -1447,6 +1627,60 @@ def simulate_trades( t.mae_r = float(max(0.0, mae)) +def _resolve_trades_gpu( + trades: List[Trade], + opens_arr: np.ndarray, + highs_arr: np.ndarray, + lows_arr: np.ndarray, + closes_arr: np.ndarray, + valid_lens: np.ndarray, + exit_times: List[List[pd.Timestamp]], + spread: float, + trail_arm_r: float, + trail_dist_r: float, +) -> None: + """CUDA twin of `_simulate_trailing` (trail_arm_r > 0) and of the static + SL/TP1 branch of `simulate_trades`: the kernel walks every trade's window + in its own thread and returns exit index/price/reason, bars held, R and + the excursions; this host loop only writes them onto the Trade objects, + with the same 'not set' cases as the CPU code.""" + trailing = trail_arm_r > 0 + entries = np.array([t.entry_price for t in trades], dtype=np.float64) + sls = np.array([t.sl for t in trades], dtype=np.float64) + tps = np.array([t.tp1 for t in trades], dtype=np.float64) + is_buy = np.array([t.direction == "BUY" for t in trades], dtype=np.int8) + arms = np.array([t.trail_arm_r if t.trail_arm_r > 0 else trail_arm_r + for t in trades], dtype=np.float64) + dists = np.array([t.trail_dist_r if t.trail_dist_r > 0 else trail_dist_r + for t in trades], dtype=np.float64) + res = _gk.resolve_trades( + opens_arr, highs_arr, lows_arr, closes_arr, valid_lens, + entries, sls, tps, is_buy, arms, dists, spread, trailing) + for i, t in enumerate(trades): + code = int(res["reason"][i]) + if code == _gk.REASON_NO_DATA: + t.exit_reason = "NO_DATA" + continue + bars = int(res["bars_held"][i]) + exit_idx = int(res["exit_idx"][i]) + if code == _gk.REASON_NO_DATA_AFTER_TIMEOUT: + # Static path: the TIMEOUT bookkeeping ran before the NaN close + # was seen, so bars_held and exit_time are kept and the price + # is None (never a NaN R in the book). + t.exit_reason = "NO_DATA" + t.bars_held = bars + t.exit_price = None + t.exit_time = exit_times[i][exit_idx] + continue + t.exit_reason = _gk.REASON_NAMES[code] + t.bars_held = bars + t.exit_price = float(res["exit_price"][i]) + t.exit_time = exit_times[i][exit_idx] + t.r_multiple = float(res["r"][i]) + t.mfe_r = float(res["mfe"][i]) + t.mae_r = float(res["mae"][i]) + + # ═════════════════════════════════════════════════════════════════════ # BACKTEST RUNNER # ═════════════════════════════════════════════════════════════════════ @@ -1878,7 +2112,12 @@ def run_backtest( 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") + logger.info( + f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s " + f"({'CuPy GPU' if _gpu_on() else 'NumPy CPU'})") + # The replay workers are forked below and must never touch the device: + # every cache array is already host NumPy; hand pooled memory back too. + BACKEND.release_device_memory() # ── Inject GPU engine into a fresh aggregator ── aggregator = SignalAggregator(platform_cfg) @@ -1998,7 +2237,7 @@ def run_backtest( trail_dist_r=platform_cfg.signals.trail_dist_r) logger.info( f"Trade simulation took {time.perf_counter() - t_sim:.2f}s " - f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})" + f"({'CuPy GPU' if _gpu_on() else 'NumPy CPU'})" ) if bt_cfg.record_dir: @@ -2009,7 +2248,7 @@ def run_backtest( n_primary_bars=len(primary_df), n_bars_in_window=n_bars, skipped=skipped_reasons, n_candidates=len(candidates), workers=workers, - backend="CuPy GPU" if _GPU_AVAILABLE else "NumPy CPU fallback", + backend=BACKEND.describe(), cache_files=cache_names, replay_seconds=replay_seconds, cache_fallbacks=cache_fallbacks), sim_df=sim_df, bar_chunks=bar_chunks or None) @@ -2421,6 +2660,12 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: help="Print full per-trade reasoning (indicators, signals, agreement, why entered)") p.add_argument("--workers", type=int, default=0, help="Replay worker processes (0=auto: cpu-2 capped at 12, 1=sequential)") + p.add_argument("--backend", choices=("auto", "gpu", "cpu"), default=None, + help="Compute backend for the indicator pre-compute and the " + "trade resolution: auto = CuPy when CUDA works, else " + "NumPy (the reason is logged); gpu = fail loudly without " + "CUDA; cpu = force the NumPy twins (parity runs). The " + "env var FB2_BACKEND sets the default.") 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") @@ -2486,6 +2731,8 @@ def main(argv: Optional[List[str]] = None) -> int: level=args.log_level.upper(), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) + if args.backend: + BACKEND.set_mode(args.backend) pair = _resolve_pair(args.pair) primary_tf = _resolve_tf(args.primary_tf) @@ -2592,7 +2839,7 @@ def main(argv: Optional[List[str]] = None) -> int: ) logger.info( - f"Backend: {'CuPy GPU' if _GPU_AVAILABLE else 'NumPy CPU fallback'} | " + 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}" ) diff --git a/backtest/gpu_kernels.py b/backtest/gpu_kernels.py new file mode 100644 index 0000000..0a24551 --- /dev/null +++ b/backtest/gpu_kernels.py @@ -0,0 +1,699 @@ +""" +backtest/gpu_kernels.py — the CUDA side of the backtester. + +Owns two things: + +1. ``BACKEND`` — the one CuPy/CUDA state object the backtester consults. + It is probed lazily (never at import: test_all.py, analyze_signal_log.py + and replay_record import the backtester without wanting a CUDA context) + and it never fails silently: when CUDA is unusable the reason is kept in + ``BACKEND.reason`` and printed in the ``Backend:`` log line. Before + 2026-09-18 the probe swallowed "Failed to auto-detect CUDA root + directory" and every book was built on the NumPy path while the file + was called a GPU backtester. + + CuPy from pip (cupy-cuda12x) needs the CUDA headers to JIT kernels. They + come from the ``nvidia-cuda-runtime-cu12`` wheel, whose splayed layout + has no single root, so CuPy cannot find them on its own: this module + points ``CUDA_PATH`` at that wheel when the variable is unset. + +2. Three CUDA kernels, each a bit-for-bit replica of the CPU code it + replaces (proven on every cached window, see + backtest/research/gpu_backend_2026-09-18.md): + + - ``ewm_batch`` — pandas ``ewm(alpha, min_periods, adjust=False)`` + as implemented by ``gpu_backtest._gpu_ewm_alpha``; + one CUDA thread per series (the recurrence is + order-dependent, so a parallel scan would change + the last bits — that is not acceptable here). + - ``roll_mean_var`` — pandas 3.0.2 ``rolling(window).mean()`` and + ``.var(ddof)`` (Kahan-compensated add/remove and + Welford with the numerically-unstable recompute), + one thread per series. + - ``resolve_trades``— ``gpu_backtest._simulate_trailing`` and the static + SL/TP1 branch of ``simulate_trades``, one thread per + trade. + + Every kernel is compiled with ``--fmad=false`` so ``a*b + c`` rounds + twice exactly like the NumPy/Python code, and every kernel object lives + in a module global (a forked worker must never drop the last reference + and unload a CUDA module it did not create). The backtester's forked + replay workers never touch the device: precompute results are moved to + host NumPy arrays before the fork. + +Import as ``from backtest import gpu_kernels`` (package form — test_all.py +puts backtest/ on sys.path and imports the backtester under two names; the +backend state must live in one module object). +""" + +from __future__ import annotations + +import contextlib +import importlib.util +import logging +import os +import sys +from typing import Dict, List, Optional, Sequence + +import numpy as np + +logger = logging.getLogger("gpu_backtest.kernels") + +ENV_BACKEND = "FB2_BACKEND" # auto | gpu | cpu +KERNEL_OPTIONS = ("--fmad=false",) # no FMA contraction: two roundings, like the CPU +_MODES = ("auto", "gpu", "cpu") + +# Exit-reason codes written by resolve_trades (see the host mapping in +# gpu_backtest._resolve_trades_gpu). +REASON_NO_DATA = 0 +REASON_SL = 1 +REASON_TRAIL = 2 +REASON_TP1 = 3 +REASON_TIMEOUT = 4 +REASON_NO_DATA_AFTER_TIMEOUT = 5 # static path: bars_held/exit_time set, price None +REASON_NAMES = { + REASON_NO_DATA: "NO_DATA", REASON_SL: "SL", REASON_TRAIL: "TRAIL", + REASON_TP1: "TP1", REASON_TIMEOUT: "TIMEOUT", + REASON_NO_DATA_AFTER_TIMEOUT: "NO_DATA", +} + + +# ═════════════════════════════════════════════════════════════════════ +# CUDA root discovery for pip-installed CuPy +# ═════════════════════════════════════════════════════════════════════ + +def _cuda_runtime_wheel_dir() -> Optional[str]: + """Directory of the nvidia-cuda-runtime-cu12 wheel (holds include/).""" + try: + spec = importlib.util.find_spec("nvidia.cuda_runtime") + except (ImportError, ValueError): + spec = None + candidates: List[str] = [] + if spec is not None and spec.submodule_search_locations: + candidates.extend(spec.submodule_search_locations) + try: + import site + for sp in list(site.getsitepackages()) + [site.getusersitepackages()]: + candidates.append(os.path.join(sp, "nvidia", "cuda_runtime")) + except Exception: + pass + for d in candidates: + if os.path.isfile(os.path.join(d, "include", "cuda_runtime.h")): + return d + return None + + +def ensure_cuda_path() -> str: + """Point CUDA_PATH at the runtime-header wheel when it is unset. + + Returns a short note for the log ("" when nothing was changed). Runs at + import (cheap: no CuPy import, no CUDA context) so that any later kernel + compile — including one triggered by a test that imports cupy itself — + finds the headers. + """ + current = os.environ.get("CUDA_PATH", "") + if current and os.path.isdir(current): + return "" + wheel = _cuda_runtime_wheel_dir() + if wheel is None: + return "" + os.environ["CUDA_PATH"] = wheel + # CuPy caches a failed lookup; if it was imported before this ran, drop + # that cache so the new CUDA_PATH is seen (private, hence guarded). + env_mod = sys.modules.get("cupy._environment") + if env_mod is not None and getattr(env_mod, "_cuda_path", "") is None: + env_mod._cuda_path = "" + return f"CUDA_PATH set to the nvidia-cuda-runtime wheel: {wheel}" + + +_CUDA_PATH_NOTE = ensure_cuda_path() + + +# ═════════════════════════════════════════════════════════════════════ +# Backend state +# ═════════════════════════════════════════════════════════════════════ + +_PROBE_SRC = r''' +extern "C" __global__ +void fb2_probe(const double* __restrict__ x, double* __restrict__ y, const int n) { + int i = blockIdx.x * blockDim.x + threadIdx.x; + if (i < n) y[i] = x[i] * 2.0 + 1.0; +} +''' + + +class Backend: + """CuPy availability, selection mode and the device description.""" + + def __init__(self) -> None: + mode = os.environ.get(ENV_BACKEND, "auto").strip().lower() or "auto" + if mode not in _MODES: + logger.warning("%s=%r is not one of %s; using auto", ENV_BACKEND, mode, _MODES) + mode = "auto" + self.mode = mode + self._probed = False + self._ok = False + self.reason = "" + self.device_name = "" + self.cupy_version = "" + self.cp = None # the cupy module once probed OK + + # ── selection ──────────────────────────────────────────────────── + def set_mode(self, mode: str) -> None: + mode = (mode or "auto").strip().lower() + if mode not in _MODES: + raise ValueError(f"backend mode must be one of {_MODES}, got {mode!r}") + self.mode = mode + if mode == "gpu" and not self.available: + raise RuntimeError( + f"--backend gpu requested but CUDA is unavailable: {self.reason}") + + def force_cpu(self) -> None: + self.mode = "cpu" + + @contextlib.contextmanager + def cpu_only(self): + """Temporarily route everything to the NumPy twins (tests).""" + old = self.mode + self.mode = "cpu" + try: + yield + finally: + self.mode = old + + @property + def available(self) -> bool: + if self.mode == "cpu": + return False + self._probe() + return self._ok + + def describe(self) -> str: + """The text after ``Backend:`` in the run log and the --record manifest. + + Prefixes are load-bearing: "CuPy GPU" / "NumPy CPU fallback" are what + every report and the Module 33 manifest fixture use.""" + if self.available: + return f"CuPy GPU ({self.device_name}, cupy {self.cupy_version})" + if self.mode == "cpu": + return f"NumPy CPU fallback (forced: --backend cpu / {ENV_BACKEND}=cpu)" + return f"NumPy CPU fallback ({self.reason})" + + # ── probe ──────────────────────────────────────────────────────── + def _probe(self) -> None: + if self._probed: + return + self._probed = True + try: + import cupy as cp # type: ignore + a = cp.asarray(np.zeros(4, dtype=np.float64)) + _ = cp.cumsum(a).get() + # The compile probe is the step that used to fail: an importable + # CuPy still cannot JIT without the CUDA headers. + k = cp.RawKernel(_PROBE_SRC, "fb2_probe", options=KERNEL_OPTIONS) + out = cp.empty(4, dtype=np.float64) + k((1,), (4,), (a, out, np.int32(4))) + cp.cuda.Device().synchronize() + if not np.array_equal(out.get(), np.ones(4)): + raise RuntimeError("probe kernel returned wrong values") + props = cp.cuda.runtime.getDeviceProperties(cp.cuda.Device().id) + name = props.get("name", b"") + if isinstance(name, bytes): + name = name.decode("utf-8", "replace") + self.cp = cp + self._ok = True + self.device_name = str(name) + self.cupy_version = str(cp.__version__) + logger.info("CUDA backend ready: %s, cupy %s%s", self.device_name, + self.cupy_version, + f" ({_CUDA_PATH_NOTE})" if _CUDA_PATH_NOTE else "") + except Exception as e: # noqa: BLE001 — any failure means CPU, loudly + first = str(e).strip().splitlines()[0] if str(e).strip() else "" + self.reason = f"{type(e).__name__}: {first}"[:240] + self._ok = False + logger.warning("CUDA backend unavailable, NumPy CPU fallback: %s", self.reason) + + def release_device_memory(self) -> None: + """Synchronise and return pooled device/pinned memory before a fork.""" + if not self._ok or self.cp is None: + return + cp = self.cp + cp.cuda.Device().synchronize() + cp.get_default_memory_pool().free_all_blocks() + cp.get_default_pinned_memory_pool().free_all_blocks() + + +BACKEND = Backend() + + +def _require_cp(): + if not BACKEND.available: + raise RuntimeError(f"CUDA backend unavailable: {BACKEND.reason or BACKEND.mode}") + return BACKEND.cp + + +# ═════════════════════════════════════════════════════════════════════ +# Kernel sources +# ═════════════════════════════════════════════════════════════════════ + +# Replica of gpu_backtest._gpu_ewm_alpha: y[i] = alpha*x[i] + (1-alpha)*y[i-1], +# seeded at the first non-NaN x; a NaN input carries y forward without +# advancing the non-NaN counter; output is NaN until min_periods non-NaN +# inputs were seen. Rounding intrinsics pin the two-multiply-one-add order. +_EWM_SRC = r''' +extern "C" __global__ +void fb2_ewm_batch(const double* __restrict__ x, double* __restrict__ y, + const long long* __restrict__ offs, const long long* __restrict__ lens, + const double* __restrict__ alphas, const int* __restrict__ minp, + const int n_series) { + int s = blockIdx.x * blockDim.x + threadIdx.x; + if (s >= n_series) return; + const double alpha = alphas[s]; + const double one_minus = __dsub_rn(1.0, alpha); + const int mp = minp[s]; + const double* xs = x + offs[s]; + double* ys = y + offs[s]; + const long long n = lens[s]; + double last_y = nan(""); + int seen = 0; + for (long long i = 0; i < n; ++i) { + const double xi = xs[i]; + if (isnan(xi)) { + ys[i] = (!isnan(last_y) && seen >= mp) ? last_y : nan(""); + continue; + } + if (isnan(last_y)) last_y = xi; + else last_y = __dadd_rn(__dmul_rn(alpha, xi), __dmul_rn(one_minus, last_y)); + seen += 1; + ys[i] = (seen >= mp) ? last_y : nan(""); + } +} +''' + +# Replica of pandas 3.0.2 pandas/_libs/window/aggregations.pyx roll_mean and +# roll_var for a fixed window, closed='right', min_periods=window (the +# defaults of Series.rolling(window)). Kahan-compensated running sum with +# the sign count and consecutive-same-value guard (GH#42064), Welford's +# online variance with Kahan compensation and the InvCondTol recompute. +_ROLL_SRC = r''' +#define FB2_EPS 2.220446049250313e-16 +#define FB2_INVCONDTOL (FB2_EPS * 1e3) + +__device__ __forceinline__ void fb2_add_mean(double val, long long* nobs, double* sum_x, + long long* neg_ct, double* comp, long long* ncsv, double* prev_value) { + if (val == val) { + *nobs += 1; + double y = val - *comp; + double t = *sum_x + y; + *comp = t - *sum_x - y; + *sum_x = t; + if (signbit(val)) *neg_ct += 1; + if (val == *prev_value) *ncsv += 1; else *ncsv = 1; + *prev_value = val; + } +} + +__device__ __forceinline__ void fb2_remove_mean(double val, long long* nobs, double* sum_x, + long long* neg_ct, double* comp) { + if (val == val) { + *nobs -= 1; + double y = -val - *comp; + double t = *sum_x + y; + *comp = t - *sum_x - y; + *sum_x = t; + if (signbit(val)) *neg_ct -= 1; + } +} + +__device__ __forceinline__ void fb2_add_var(double val, double* nobs, double* mean_x, + double* ssqdm_x, double* comp, bool* unstable) { + if (val != val) return; + double prev_m2 = *ssqdm_x; + *nobs += 1; + double prev_mean = *mean_x - *comp; + double y = val - *comp; + double t = y - *mean_x; + *comp = t + *mean_x - y; + double delta = t; + if (*nobs != 0) *mean_x = *mean_x + delta / *nobs; else *mean_x = 0; + *ssqdm_x = *ssqdm_x + (val - prev_mean) * (val - *mean_x); + if (prev_m2 * FB2_INVCONDTOL > *ssqdm_x) *unstable = true; +} + +__device__ __forceinline__ void fb2_remove_var(double val, double* nobs, double* mean_x, + double* ssqdm_x, double* comp, bool* unstable) { + if (val == val) { + double prev_m2 = *ssqdm_x; + *nobs -= 1; + if (*nobs != 0) { + double prev_mean = *mean_x - *comp; + double y = val - *comp; + double t = y - *mean_x; + *comp = t + *mean_x - y; + double delta = t; + *mean_x = *mean_x - delta / *nobs; + *ssqdm_x = *ssqdm_x - (val - prev_mean) * (val - *mean_x); + if (prev_m2 * FB2_INVCONDTOL > *ssqdm_x) *unstable = true; + } else { + *mean_x = 0; *ssqdm_x = 0; *unstable = false; + } + } +} + +extern "C" __global__ +void fb2_roll_mean_var(const double* __restrict__ x, double* __restrict__ out_mean, + double* __restrict__ out_var, + const long long* __restrict__ offs, const long long* __restrict__ lens, + const int n_series, const int window, const int ddof) { + int s = blockIdx.x * blockDim.x + threadIdx.x; + if (s >= n_series) return; + const double* xs = x + offs[s]; + double* om = out_mean + offs[s]; + double* ov = out_var + offs[s]; + const long long n = lens[s]; + if (n <= 0) return; + const long long minp = window; // min_periods defaults to the window + // roll_mean state + double comp_add = 0, comp_rem = 0, sum_x = 0, prev_value = xs[0]; + long long nobs = 0, neg_ct = 0, ncsv = 0; + // roll_var state (nobs is a double in pandas' roll_var) + double vnobs = 0, mean_x = 0, ssqdm_x = 0, vcomp_add = 0, vcomp_rem = 0; + bool unstable = false; + for (long long i = 0; i < n; ++i) { + long long st = i - window + 1; if (st < 0) st = 0; // start[i] + long long en = i + 1; // end[i] + long long pst = (i - 1) - window + 1; if (pst < 0) pst = 0; // start[i-1] + long long pen = i; // end[i-1] + // ---- mean + if (i == 0) { + comp_add = comp_rem = sum_x = 0; nobs = neg_ct = 0; prev_value = xs[st]; ncsv = 0; + for (long long j = st; j < en; ++j) + fb2_add_mean(xs[j], &nobs, &sum_x, &neg_ct, &comp_add, &ncsv, &prev_value); + } else { + for (long long j = pst; j < st; ++j) + fb2_remove_mean(xs[j], &nobs, &sum_x, &neg_ct, &comp_rem); + for (long long j = pen; j < en; ++j) + fb2_add_mean(xs[j], &nobs, &sum_x, &neg_ct, &comp_add, &ncsv, &prev_value); + } + double rm; + if (nobs >= minp && nobs > 0) { + rm = sum_x / (double)nobs; + if (ncsv >= nobs) rm = prev_value; + else if (neg_ct == 0 && rm < 0) rm = 0; + else if (neg_ct == nobs && rm > 0) rm = 0; + } else { + rm = nan(""); + } + om[i] = rm; + // ---- var + bool recompute = (i == 0); + if (!recompute) { + for (long long j = pst; j < st; ++j) + fb2_remove_var(xs[j], &vnobs, &mean_x, &ssqdm_x, &vcomp_rem, &unstable); + for (long long j = pen; j < en; ++j) + fb2_add_var(xs[j], &vnobs, &mean_x, &ssqdm_x, &vcomp_add, &unstable); + } + if (recompute || unstable) { + mean_x = ssqdm_x = vnobs = vcomp_add = vcomp_rem = 0; + for (long long j = st; j < en; ++j) + fb2_add_var(xs[j], &vnobs, &mean_x, &ssqdm_x, &vcomp_add, &unstable); + unstable = false; + } + ov[i] = (vnobs >= (double)minp && vnobs > (double)ddof) + ? ssqdm_x / (vnobs - (double)ddof) : nan(""); + } +} +''' + +# Replica of gpu_backtest._simulate_trailing (trailing != 0) and of the +# static SL/TP1 branch of gpu_backtest.simulate_trades (trailing == 0). +# Python max/min are written out as (b > a) ? b : a so NaN handling matches. +_TRADES_SRC = r''' +#define R_NO_DATA 0 +#define R_SL 1 +#define R_TRAIL 2 +#define R_TP1 3 +#define R_TIMEOUT 4 +#define R_NO_DATA_AFTER_TIMEOUT 5 + +__device__ __forceinline__ void fb2_excursions(const double* h, const double* l, long long bars, + bool buy, double entry, double risk, double* mfe_out, double* mae_out) { + // np.nanmax(h[:bars]) / np.nanmin(l[:bars]); all-NaN gives NaN. + double hmax = nan(""), lmin = nan(""); + for (long long j = 0; j < bars; ++j) { + double hv = h[j], lv = l[j]; + if (!isnan(hv) && (isnan(hmax) || hv > hmax)) hmax = hv; + if (!isnan(lv) && (isnan(lmin) || lv < lmin)) lmin = lv; + } + double mfe, mae; + if (buy) { mfe = (hmax - entry) / risk; mae = (entry - lmin) / risk; } + else { mfe = (entry - lmin) / risk; mae = (hmax - entry) / risk; } + *mfe_out = (mfe > 0.0) ? mfe : 0.0; // Python max(0.0, x): NaN -> 0.0 + *mae_out = (mae > 0.0) ? mae : 0.0; +} + +extern "C" __global__ +void fb2_resolve_trades(const double* __restrict__ opens, const double* __restrict__ highs, + const double* __restrict__ lows, const double* __restrict__ closes, + const long long* __restrict__ valid_lens, + const double* __restrict__ entries, const double* __restrict__ sls, + const double* __restrict__ tps, const signed char* __restrict__ is_buy, + const double* __restrict__ arms, const double* __restrict__ dists, + const double spread, const int trailing, const int max_bars, + const int n_trades, + long long* __restrict__ exit_idx_out, double* __restrict__ exit_price_out, + int* __restrict__ reason_out, long long* __restrict__ bars_held_out, + double* __restrict__ r_out, double* __restrict__ mfe_out, + double* __restrict__ mae_out) { + int i = blockIdx.x * blockDim.x + threadIdx.x; + if (i >= n_trades) return; + const long long base = (long long)i * max_bars; + const double* o = opens + base; + const double* h = highs + base; + const double* l = lows + base; + const double* c = closes + base; + const long long n = valid_lens[i]; + const bool buy = is_buy[i] != 0; + const double entry = entries[i]; + const double sl = sls[i]; + const double tp = tps[i]; + // defaults (Trade dataclass defaults / "not set") + exit_idx_out[i] = -1; exit_price_out[i] = nan(""); reason_out[i] = R_NO_DATA; + bars_held_out[i] = 0; r_out[i] = 0.0; mfe_out[i] = 0.0; mae_out[i] = 0.0; + if (n == 0) return; + + if (trailing) { + const double risk = fabs(entry - sl); + if (risk <= 0) return; // NO_DATA + const double arm = arms[i], dist = dists[i]; + double cur_sl = sl, peak = entry; + bool have_price = false; double exit_price = nan(""); + long long exit_idx = n - 1; int reason = R_TIMEOUT; + for (long long j = 0; j < n; ++j) { + if (isnan(h[j])) { + if (j == 0) { exit_idx = 0; reason = R_NO_DATA; } + else { exit_idx = j - 1; exit_price = c[exit_idx]; have_price = true; } + break; + } + const bool hit = buy ? (l[j] <= cur_sl) : (h[j] >= cur_sl); + if (hit) { + exit_price = cur_sl; have_price = true; + if ((buy && o[j] < cur_sl) || (!buy && o[j] > cur_sl)) exit_price = o[j]; + reason = (cur_sl == sl) ? R_SL : R_TRAIL; + exit_idx = j; + break; + } + const double fav = buy ? h[j] : l[j]; + peak = buy ? ((fav > peak) ? fav : peak) : ((fav < peak) ? fav : peak); + const double exc = (buy ? (peak - entry) : (entry - peak)) / risk; + if (exc >= arm) { + const double lvl = buy ? (peak - dist * risk) : (peak + dist * risk); + cur_sl = buy ? ((lvl > cur_sl) ? lvl : cur_sl) : ((lvl < cur_sl) ? lvl : cur_sl); + } + } + if (reason == R_NO_DATA) return; + if (!have_price) exit_price = c[exit_idx]; + if (!isfinite(exit_price)) return; // NO_DATA, nothing else set + const long long bars = exit_idx + 1; + reason_out[i] = reason; bars_held_out[i] = bars; exit_idx_out[i] = exit_idx; + exit_price_out[i] = exit_price; + const double pnl = buy ? (exit_price - entry) : (entry - exit_price); + r_out[i] = (pnl - spread) / risk; + fb2_excursions(h, l, bars, buy, entry, risk, &mfe_out[i], &mae_out[i]); + return; + } + + // ---- static SL / TP1: first-hit scan over the whole window, NaN highs masked + long long sl_idx = max_bars, tp_idx = max_bars; + for (long long j = 0; j < max_bars; ++j) { + if (isnan(h[j])) continue; + const bool sl_hit = buy ? (l[j] <= sl) : (h[j] >= sl); + const bool tp_hit = buy ? (h[j] >= tp) : (l[j] <= tp); + if (sl_hit && j < sl_idx) sl_idx = j; + if (tp_hit && j < tp_idx) tp_idx = j; + } + long long exit_idx; double exit_price; int reason; long long bars; + if (sl_idx < n && sl_idx <= tp_idx) { + reason = R_SL; bars = sl_idx + 1; exit_price = sl; exit_idx = sl_idx; + const double bar_open = o[sl_idx]; + if (!isnan(bar_open)) { + if ((buy && bar_open < sl) || (!buy && bar_open > sl)) exit_price = bar_open; + } + } else if (tp_idx < n) { + reason = R_TP1; bars = tp_idx + 1; exit_price = tp; exit_idx = tp_idx; + const double bar_open = o[tp_idx]; + if (!isnan(bar_open)) { + if ((buy && bar_open > tp) || (!buy && bar_open < tp)) exit_price = bar_open; + } + } else { + reason = R_TIMEOUT; bars = n; exit_idx = n - 1; exit_price = c[exit_idx]; + if (!isfinite(exit_price)) { + reason_out[i] = R_NO_DATA_AFTER_TIMEOUT; bars_held_out[i] = bars; + exit_idx_out[i] = exit_idx; + return; // exit_price None, R untouched + } + } + reason_out[i] = reason; bars_held_out[i] = bars; exit_idx_out[i] = exit_idx; + exit_price_out[i] = exit_price; + const double risk = fabs(entry - sl); + if (risk > 0) { + const double pnl = buy ? (exit_price - entry) : (entry - exit_price); + r_out[i] = (pnl - spread) / risk; + } + if (bars > 0 && risk > 0) + fb2_excursions(h, l, bars, buy, entry, risk, &mfe_out[i], &mae_out[i]); +} +''' + +_KERNELS: Dict[str, object] = {} # module-level: never collected in a child + + +def _kernel(name: str): + cp = _require_cp() + k = _KERNELS.get(name) + if k is None: + src = {"ewm": _EWM_SRC, "roll": _ROLL_SRC, "trades": _TRADES_SRC}[name] + fn = {"ewm": "fb2_ewm_batch", "roll": "fb2_roll_mean_var", + "trades": "fb2_resolve_trades"}[name] + k = cp.RawKernel(src, fn, options=KERNEL_OPTIONS) + _KERNELS[name] = k + return k + + +def _ragged(cp, series: Sequence) -> tuple: + """Concatenate 1-D float64 series on the device; return (flat, offs, lens).""" + arrs = [cp.asarray(s, dtype=cp.float64).ravel() for s in series] + lens = np.array([int(a.shape[0]) for a in arrs], dtype=np.int64) + offs = np.zeros(len(arrs), dtype=np.int64) + if len(arrs) > 1: + offs[1:] = np.cumsum(lens)[:-1] + flat = cp.concatenate(arrs) if arrs else cp.empty(0, dtype=cp.float64) + return flat, cp.asarray(offs), cp.asarray(lens), offs, lens + + +def _launch_1d(n: int, threads: int = 128) -> tuple: + return ((max(1, (n + threads - 1) // threads),), (threads,)) + + +# ═════════════════════════════════════════════════════════════════════ +# Public wrappers (device in/out; callers .get() once) +# ═════════════════════════════════════════════════════════════════════ + +def ewm_batch_dev(series: Sequence, alphas: Sequence[float], + min_periods: Sequence[int]) -> List: + """EWM of every series in one launch. Inputs may be NumPy or CuPy; + returns CuPy views (one per series).""" + cp = _require_cp() + if len(series) != len(alphas) or len(series) != len(min_periods): + raise ValueError("series, alphas and min_periods must have equal length") + if not series: + return [] + flat, d_offs, d_lens, offs, lens = _ragged(cp, series) + out = cp.empty_like(flat) + k = _kernel("ewm") + grid, block = _launch_1d(len(series)) + k(grid, block, (flat, out, d_offs, d_lens, + cp.asarray(np.asarray(alphas, dtype=np.float64)), + cp.asarray(np.asarray(min_periods, dtype=np.int32)), + np.int32(len(series)))) + return [out[o:o + n] for o, n in zip(offs, lens)] + + +def ewm_batch(series: Sequence[np.ndarray], alphas: Sequence[float], + min_periods: Sequence[int]) -> List[np.ndarray]: + """Host convenience: NumPy in, NumPy out.""" + return [y.get() for y in ewm_batch_dev(series, alphas, min_periods)] + + +def roll_mean_var_dev(series: Sequence, window: int, ddof: int = 0) -> tuple: + """pandas rolling(window).mean() and .var(ddof) of every series in one + launch; returns (means, vars) as lists of CuPy views.""" + cp = _require_cp() + if not series: + return [], [] + flat, d_offs, d_lens, offs, lens = _ragged(cp, series) + out_mean = cp.empty_like(flat) + out_var = cp.empty_like(flat) + k = _kernel("roll") + grid, block = _launch_1d(len(series)) + k(grid, block, (flat, out_mean, out_var, d_offs, d_lens, + np.int32(len(series)), np.int32(window), np.int32(ddof))) + means = [out_mean[o:o + n] for o, n in zip(offs, lens)] + vars_ = [out_var[o:o + n] for o, n in zip(offs, lens)] + return means, vars_ + + +def roll_std_from_var_dev(var): + """pandas zsqrt: sqrt(var) with a negative variance clipped to 0.""" + cp = _require_cp() + return cp.where(var < 0, 0.0, cp.sqrt(var)) + + +def roll_mean_var(series: Sequence[np.ndarray], window: int, ddof: int = 0) -> tuple: + """Host convenience: (means, vars, stds) as NumPy arrays.""" + means, vars_ = roll_mean_var_dev(series, window, ddof) + stds = [roll_std_from_var_dev(v) for v in vars_] + return ([m.get() for m in means], [v.get() for v in vars_], + [s.get() for s in stds]) + + +def resolve_trades(opens: np.ndarray, highs: np.ndarray, lows: np.ndarray, + closes: np.ndarray, valid_lens: np.ndarray, + entries: np.ndarray, sls: np.ndarray, tps: np.ndarray, + is_buy: np.ndarray, arms: np.ndarray, dists: np.ndarray, + spread: float, trailing: bool) -> Dict[str, np.ndarray]: + """Resolve every trade's exit on the device; one thread per trade. + + ``opens/highs/lows/closes`` are (n_trades, max_bars) float64 windows + padded with NaN, ``valid_lens`` the real length of each. Returns host + arrays: exit_idx, exit_price, reason (REASON_* codes), bars_held, r, + mfe, mae. + """ + cp = _require_cp() + n_trades, max_bars = opens.shape + if n_trades == 0: + empty = np.zeros(0) + return {k: empty for k in ("exit_idx", "exit_price", "reason", + "bars_held", "r", "mfe", "mae")} + d = lambda a, dt: cp.asarray(np.ascontiguousarray(a, dtype=dt)) # noqa: E731 + d_o, d_h, d_l, d_c = (d(opens, np.float64), d(highs, np.float64), + d(lows, np.float64), d(closes, np.float64)) + o_exit_idx = cp.empty(n_trades, dtype=np.int64) + o_exit_price = cp.empty(n_trades, dtype=np.float64) + o_reason = cp.empty(n_trades, dtype=np.int32) + o_bars = cp.empty(n_trades, dtype=np.int64) + o_r = cp.empty(n_trades, dtype=np.float64) + o_mfe = cp.empty(n_trades, dtype=np.float64) + o_mae = cp.empty(n_trades, dtype=np.float64) + k = _kernel("trades") + grid, block = _launch_1d(n_trades) + k(grid, block, (d_o, d_h, d_l, d_c, d(valid_lens, np.int64), + d(entries, np.float64), d(sls, np.float64), d(tps, np.float64), + d(is_buy, np.int8), d(arms, np.float64), d(dists, np.float64), + np.float64(spread), np.int32(1 if trailing else 0), + np.int32(max_bars), np.int32(n_trades), + o_exit_idx, o_exit_price, o_reason, o_bars, o_r, o_mfe, o_mae)) + cp.cuda.Device().synchronize() + return dict(exit_idx=o_exit_idx.get(), exit_price=o_exit_price.get(), + reason=o_reason.get(), bars_held=o_bars.get(), r=o_r.get(), + mfe=o_mfe.get(), mae=o_mae.get()) diff --git a/backtest/instructions_for_backtest.txt b/backtest/instructions_for_backtest.txt index 04a333d..30e9146 100644 --- a/backtest/instructions_for_backtest.txt +++ b/backtest/instructions_for_backtest.txt @@ -38,3 +38,11 @@ re-implementation. - Scorecards and "what-if" tables from these files (backtest/replay_scorecard.py) are IN-BOOK descriptive statistics: they cannot add trades a change would create and ignore the gates. Any weight / indicator change they suggest is a hypothesis for a full rule-1 rerun, never a decision. + +--- 2026-09-18: GPU backend (backtest/gpu_kernels.py) --- +- The "Backend:" line now names the CUDA device ("CuPy GPU (, cupy )") or the + reason for the NumPy fallback; quote it verbatim. `--backend cpu` forces the NumPy twins, `gpu` + refuses to fall back. Books are byte-identical on both backends (reference MD5s, CLAUDE.md §9). +- Any change to a kernel or to its CPU twin is a parity event: re-run Module 36, regenerate the + five reference books on BOTH backends and show the ten MD5s unchanged. Never replace a kernel + with a differently ordered algorithm (parallel scan, two-pass variance) even if it is faster. diff --git a/backtest/research/gpu_backend_2026-09-18.md b/backtest/research/gpu_backend_2026-09-18.md new file mode 100644 index 0000000..6e61ec1 --- /dev/null +++ b/backtest/research/gpu_backend_2026-09-18.md @@ -0,0 +1,137 @@ +# GPU backend made real — 2026-09-18 + +Behaviour-neutral change. No signal logic, weight, gate or exit rule moved; the proof is the ten +MD5s below (five reference books on the GPU, five with `--backend cpu`), all equal to the pins in +CLAUDE.md section 9. The books therefore carry the same trades, R, win rate, profit factor and +drawdown, in both halves and at both cost conventions, as the pinned books do. + +## 1. What was wrong + +`backtest/gpu_backtest.py` probed CuPy at import with a `cumsum` and swallowed any exception. +On the dev box the first kernel compile raised + +``` +RuntimeError: Failed to auto-detect CUDA root directory. Please specify `CUDA_PATH` environment +variable if you are using CUDA versions not yet supported by CuPy. +``` + +CuPy 14.0.1 from pip (`cupy-cuda12x`) was installed with only the `nvidia-cuda-nvrtc-cu12` +wheel; the CUDA-12 pip layout is splayed (no `/usr/local/cuda`, no `nvcc`), so CuPy's +`_get_cuda_path()` returned `None` and NVRTC could not find the headers. Every run since the +Oanda re-pin logged `Backend: NumPy CPU fallback`; every research report in this folder records +it. Independently of that, almost nothing in the file was written for the GPU: the EWM +precompute was a pure-Python loop, Bollinger was pandas `rolling`, and the trailing-stop +resolver (the exit mode of every reference book) was a per-trade per-bar Python loop. The only +CuPy code was a first-hit reduction in the static SL/TP branch, which the reference books never +execute. + +## 2. What shipped + +- `nvidia-cuda-runtime-cu12==12.9.79` installed into the dev venv (headers only, not + `requirements.txt`; same status as `cupy` and `pyarrow`). +- `backtest/gpu_kernels.py`: the backend state (`BACKEND`, probed lazily so importing the + backtester or `analyze_signal_log.py` creates no CUDA context; the probe compiles a kernel, and + a failure keeps its reason), `CUDA_PATH` pointed at the header wheel when unset, and three CUDA + kernels compiled with `--fmad=false`: + - `ewm_batch`: one thread per series, the exact recurrence of `_gpu_ewm_alpha` (seed at the + first non-NaN value, NaN carries the state without advancing the `min_periods` counter). + - `roll_mean_var`: pandas 3.0.2 `roll_mean` (Kahan add/remove, sign count, consecutive-same- + value guard) and `roll_var` (Welford + Kahan, `InvCondTol` unstable recompute); std as + pandas' `zsqrt`. + - `resolve_trades`: one thread per trade; trailing mode mirrors `_simulate_trailing` line by + line, static mode mirrors the SL-first/TP1/TIMEOUT branch, including the "NaN close after a + TIMEOUT" bookkeeping; R and MFE/MAE on the device. +- `backtest/gpu_backtest.py`: `precompute_indicators` dispatches to `_precompute_indicators_gpu` + (CuPy elementwise ops in the original order, two EWM launches and one rolling launch per + timeframe, one transfer to host at the end, host-only cache asserted before the fork) or to the + untouched NumPy path; `simulate_trades` dispatches to `_resolve_trades_gpu` or to the untouched + CPU paths; pooled device memory is released before the workers are forked; `--backend + auto|gpu|cpu` (env `FB2_BACKEND`); the `Backend:` line names the device or the fallback reason. +- `test_all.py` Module 36 (7 tests): a visible CUDA device must never fall back silently; the + three kernels against their CPU references on synthetic edge cases (NaN head, interior NaN, + flat window, cancellation spike, gaps, NaN bars, zero risk, per-trade trail overrides); + `--backend` / `FB2_BACKEND` / `cpu_only()`; import guards (`cupy` is now banned from the live + roots by the Module 32 AST guard as well). + +Why the kernels are sequential per series: EWM and rolling statistics are order-dependent +floating point. A parallel scan or a two-pass variance changes the last bits, and the file's own +history shows what that does (a cumsum Bollinger shortcut moved `bb_position` by up to 8e-3 and +crossed signal thresholds). The GPU's gain here is the batch across series and across trades, +not per-series speed: on a single very long series the sequential kernel is slower than pandas' +C loop. + +## 3. Evidence + +Environment: NVIDIA GeForce RTX 4070 SUPER, driver 615.71.09 (CUDA 13.4 UMD), CuPy 14.0.1 +(`cupy-cuda12x`), `nvidia-cuda-nvrtc-cu12` 12.9.86, `nvidia-cuda-runtime-cu12` 12.9.79, NumPy +2.2.6, pandas 3.0.2, Python 3.12, 16 CPUs. Code: this working tree on branch `sql-integration` +on top of `1f4741b`. + +### 3.1 Precompute equality on every cached window + +GPU vs CPU `precompute_indicators`, every `IndicatorCache` field compared with +`np.array_equal(..., equal_nan=True)`: + +``` +precompute GPU == CPU on 42/42 windows, 2342009 bars total; gpu 13.71s (incl. cold JIT) vs cpu 40.18s +``` + +### 3.2 Test suite, three ways + +| Run | Result | Module 36 note | +|---|---|---| +| default (GPU) | `RESULTS: 536 passed, 0 failed, 536 total` | kernels exercised | +| `CUDA_VISIBLE_DEVICES=""` | `RESULTS: 536 passed, 0 failed, 536 total` | `NumPy CPU fallback (CUDARuntimeError: cudaErrorNoDevice: ...)` printed | +| `FB2_BACKEND=cpu` | `RESULTS: 536 passed, 0 failed, 536 total` | `NumPy CPU fallback (forced: ...)` printed | + +### 3.3 Reference books on both backends + +Harness: `venv/bin/python backtest/gpu_backtest.py --pair XAUUSD --spread 0.65 --workers 12 +--primary-tf --start --end --export-csv ` (factory config; no `--set`, +`--quality`, `--strategies`, `--confirm-tfs` or `--export-features`), plus `--backend cpu` for +the CPU column. `GPU cache fallbacks: 0` on every run. + +| Book | Window | Backend | Precompute | Replay | Trade sim | Wall | MD5 | Pinned | +|---|---|---|---|---|---|---|---|---| +| M5 W2 | 2024-09-05 → 2026-09-05 | CuPy GPU | 1.31 s | 112.0 s | 0.31 s | 114.8 s | `49f8e602fc6ae769713c94ef7b257810` | ✓ | +| M5 W2 | | NumPy CPU (`--backend cpu`) | 3.74 s | 112.6 s | 0.22 s | 117.0 s | `49f8e602fc6ae769713c94ef7b257810` | ✓ | +| M15 W2 | 2024-09-05 → 2026-09-05 | CuPy GPU | 1.29 s | 34.6 s | 0.33 s | 37.2 s | `b4aa07e33e17e23774c486a6900d673b` | ✓ | +| M15 W2 | | NumPy CPU | 3.69 s | 34.6 s | 0.28 s | 39.0 s | `b4aa07e33e17e23774c486a6900d673b` | ✓ | +| M15 OOS | 2024-06-01 → 2025-03-30 | CuPy GPU | 0.63 s | 14.0 s | 0.05 s | 15.6 s | `9fd624fb387c36052122bd1830cf9c32` | ✓ | +| M15 OOS | | NumPy CPU | 1.70 s | 13.9 s | 0.09 s | 16.1 s | `9fd624fb387c36052122bd1830cf9c32` | ✓ | +| H1 W3 | 2023-09-05 → 2026-09-05 | CuPy GPU | 0.61 s | 11.3 s | 0.07 s | 13.0 s | `7b848c917dcb5c371ecbb9f8f7292278` | ✓ | +| H1 W3 | | NumPy CPU | 1.68 s | 11.1 s | 0.07 s | 13.3 s | `7b848c917dcb5c371ecbb9f8f7292278` | ✓ | +| H4 W3 | 2023-09-05 → 2026-09-05 | CuPy GPU | 0.20 s | 20.0 s (1 worker) | 0.13 s | 21.2 s | `186e677fbbd53cca735c94094c23ce05` | ✓ | +| H4 W3 | | NumPy CPU | 0.46 s | 19.9 s (1 worker) | 0.02 s | 20.7 s | `186e677fbbd53cca735c94094c23ce05` | ✓ | + +`Backend:` lines verbatim: `CuPy GPU (NVIDIA GeForce RTX 4070 SUPER, cupy 14.0.1)` and +`NumPy CPU fallback (forced: --backend cpu / FB2_BACKEND=cpu)`. The GPU precompute figures are +warm (the kernel cache under `~/.cupy/kernel_cache` was already populated; a cold first run adds +roughly a second of JIT once per machine). The trade-simulation figures on the GPU include the +host↔device transfers and the launch of a kernel that runs a few hundred threads; on these trade +counts it is not faster than the CPU loop, which is expected. The before/after picture is exactly +as planned: the precompute is about 3× faster, everything else is unchanged, and the replay +through the live signal code (the same bars/s on both rows) is where the wall time lives. + +### 3.4 Other checks + +- Single worker on the GPU (`--workers 1`, H1 W3): MD5 `7b848c917dcb5c371ecbb9f8f7292278`, equal + to the 12-worker book (the CUDA context exists in the parent before the fork; the workers never + touch it). +- `--record` pair on the GPU (H4 W3): control `186e677f…` == recorded `186e677f…` == + `manifest.book_csv_md5`; the manifest records `backend = CuPy GPU (NVIDIA GeForce RTX 4070 SUPER, + cupy 14.0.1)` and `cache_fallbacks = 0`. +- `CUDA_VISIBLE_DEVICES="" ... --backend gpu` exits with + `RuntimeError: --backend gpu requested but CUDA is unavailable: CUDARuntimeError: cudaErrorNoDevice ...`; + the same run in `auto` mode logs the warning and + `Backend: NumPy CPU fallback (CUDARuntimeError: cudaErrorNoDevice: no CUDA-capable device is detected)`. +- `analyze_signal_log.py --stored-candles` (which reuses `simulate_trades`) on the imported + `logs/signals.db`: the report is line-for-line identical on the GPU and with `FB2_BACKEND=cpu`. +- `nvidia-smi --query-compute-apps` during the M5 precompute listed the backtester process on the + device (184 MiB). +- Scratchpad experiments that preceded the implementation: the EWM kernel bit-identical to + `_gpu_ewm_alpha` on 9 series over the 154k-bar M5 Close (95 ms incl. JIT vs 1.29 s Python); + the rolling kernel bit-identical to pandas on all 84 series of the 42 cached windows. +- `git diff --stat` touches `backtest/gpu_backtest.py`, `test_all.py`, `CLAUDE.md`, + `backtest/instructions_for_backtest.txt` and adds `backtest/gpu_kernels.py` and this note; + `engine.py`, `signal_aggregator.py` and `requirements.txt` are untouched. diff --git a/test_all.py b/test_all.py index 2d10782..f99a1d5 100644 --- a/test_all.py +++ b/test_all.py @@ -10998,7 +10998,7 @@ def _(): def _(): """The corpus is trained on offline in venv/; anything the live signal path can reach must stay numpy-only (requirements.txt is the contract).""" - banned = {"sklearn", "scipy", "lightgbm", "torch", "transformers"} + banned = {"sklearn", "scipy", "lightgbm", "torch", "transformers", "cupy"} roots = [Path(".")] + [Path(p) for p in ("providers", "pair_signals")] offenders = [] for root in roots: @@ -13490,6 +13490,305 @@ def _(): finally: os.remove(path) +# ══════════════════════════════════════════════════════════════════════ +# MODULE 36: GPU backend parity (backtest/gpu_kernels.py, 2026-09-18) +# The CUDA kernels must be bit-for-bit twins of the NumPy/pandas code; +# a fallback to the CPU must never be silent (it was, for months). +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 36: GPU backend parity ═══") + +import importlib +import importlib.util as _ilu36 +from backtest import gpu_kernels as _gk36 +from backtest import gpu_backtest as _gb36 + +_GPU36 = _gk36.BACKEND.available +if not _GPU36: + print(f" (CUDA unavailable here: {_gk36.BACKEND.describe()} — GPU-path " + f"assertions below print a notice instead of running)") + + +def _same36(a, b) -> bool: + """Equality that treats NaN == NaN (floats, Timestamps, strings, None).""" + if isinstance(a, float) and isinstance(b, float) and np.isnan(a) and np.isnan(b): + return True + return a == b + + +@test("GPU backend: a visible CUDA device must never fall back silently") +def _(): + if _ilu36.find_spec("cupy") is None: + print(" (cupy not installed: the NumPy path is the only path on this box)") + return + try: + _cupy36 = importlib.import_module("cupy") # not a literal import: the AST guard scans this file + n_dev = int(_cupy36.cuda.runtime.getDeviceCount()) + except Exception as e: # noqa: BLE001 + print(f" (cupy present but no usable CUDA device: {type(e).__name__})") + return + if n_dev < 1: + print(" (no CUDA device visible)") + return + if _gk36.BACKEND.mode == "cpu": + print(" (backend forced to cpu by FB2_BACKEND)") + return + assert _gk36.BACKEND.available, ( + f"cupy sees {n_dev} device(s) but the backend fell back: {_gk36.BACKEND.reason}") + assert _gk36.BACKEND.describe().startswith("CuPy GPU ("), _gk36.BACKEND.describe() + assert _gk36.BACKEND.device_name and _gk36.BACKEND.cupy_version + + +@test("GPU backend: ewm_batch is bit-identical to _gpu_ewm_alpha (NaN head, interior NaN, flat run, short)") +def _(): + if not _GPU36: + print(" (no CUDA: kernel not exercised)") + return + rng = np.random.default_rng(36) + n = 700 + gold = 3300.0 + np.cumsum(rng.normal(0, 3.0, n)) + nan_head = gold.copy(); nan_head[:15] = np.nan + interior = gold.copy(); interior[100:107] = np.nan; interior[400] = np.nan + flat = gold.copy(); flat[200:260] = flat[199] + short = gold[:9] + gainlike = np.where(rng.normal(size=n) > 0, np.abs(rng.normal(size=n)), 0.0) + gainlike[0] = 0.0 + series, alphas, minps, refs = [], [], [], [] + for x in (gold, nan_head, interior, flat, short, gainlike): + for alpha, mp in ((2.0 / 10, 0), (2.0 / 201, 0), (1.0 / 14, 14), (1.0 / 10, 10), (0.5, 3)): + series.append(x); alphas.append(alpha); minps.append(mp) + refs.append(_gb36._gpu_ewm_alpha(x, alpha, mp)) + outs = _gk36.ewm_batch(series, alphas, minps) + assert len(outs) == len(refs) == 30 + for i, (o, r) in enumerate(zip(outs, refs)): + assert o.dtype == np.float64 and o.shape == r.shape, i + assert np.array_equal(o, r, equal_nan=True), ( + f"case {i}: {int(np.sum(~((o == r) | (np.isnan(o) & np.isnan(r)))))} bits differ") + # and the CPU twin itself still equals pandas on a clean series (live-engine tie) + assert np.array_equal( + pd.Series(gold).ewm(span=200, adjust=False).mean().to_numpy(), + _gk36.ewm_batch([gold], [2.0 / 201], [0])[0], equal_nan=True) + assert _gk36.ewm_batch([], [], []) == [] + + +@test("GPU backend: roll_mean_var is bit-identical to pandas rolling mean/var/std (NaN prefix, mid NaN, flat window, spike)") +def _(): + if not _GPU36: + print(" (no CUDA: kernel not exercised)") + return + rng = np.random.default_rng(37) + n = 900 + gold = 3300.0 + np.cumsum(rng.normal(0, 3.0, n)) + nan_prefix = gold.copy(); nan_prefix[:19] = np.nan + mid_nan = gold.copy(); mid_nan[300:305] = np.nan # nobs dips below min_periods, recovers + flat = gold.copy(); flat[100:120] = 3311.25; flat[120] = 3311.75 # exactly one window of equal values + spike = gold.copy(); spike[500] = 1e6 # no recompute trigger + cancel = gold.copy(); cancel[600] = 1e12 # trips the InvCondTol recompute + tiny = gold[:7] # n < window: all NaN + centred = gold - gold.mean() # sign changes exercise neg_ct + series = [gold, nan_prefix, mid_nan, flat, spike, cancel, tiny, centred, -gold] + means, vars_, stds = _gk36.roll_mean_var(series, 20, ddof=0) + for i, x in enumerate(series): + s = pd.Series(x) + pm = s.rolling(20).mean().to_numpy() + pv = s.rolling(20).var(ddof=0).to_numpy() + ps = s.rolling(20).std(ddof=0).to_numpy() + assert np.array_equal(means[i], pm, equal_nan=True), f"mean case {i}" + assert np.array_equal(vars_[i], pv, equal_nan=True), f"var case {i}" + assert np.array_equal(stds[i], ps, equal_nan=True), f"std case {i}" + assert np.isnan(means[6]).all() and np.isnan(vars_[6]).all() + # ddof=1 too (pandas' default), on the plain series + (m1,), (v1,), _ = _gk36.roll_mean_var([gold], 20, ddof=1) + assert np.array_equal(v1, pd.Series(gold).rolling(20).var(ddof=1).to_numpy(), equal_nan=True) + + +@test("GPU backend: precompute_indicators GPU == CPU on every IndicatorCache field; arrays are host NumPy") +def _(): + df = make_xau_ohlcv(600, seed=36) + cols = df.columns.get_indexer(["Open", "High", "Low", "Close"]) + df.iloc[250:262, cols] = float(df["Close"].iloc[249]) # flat stretch: zero TR / DM divisors + tech = TechnicalConfig() + with _gk36.BACKEND.cpu_only(): + c = _gb36.precompute_indicators(df, tech) + _gb36._assert_host_cache(c) + if not _GPU36: + print(" (no CUDA: CPU twin only)") + return + g = _gb36.precompute_indicators(df, tech) + _gb36._assert_host_cache(g) + checked = 0 + for name in vars(g): + if name in ("df", "index_pos"): + continue + a, b = getattr(g, name), getattr(c, name) + if a is None or b is None: + assert a is None and b is None, name + continue + if isinstance(a, pd.Series): + assert a.index.equals(b.index), name + a, b = a.to_numpy(), b.to_numpy() + assert type(a) is np.ndarray and a.dtype == np.float64, name + assert np.array_equal(a, b, equal_nan=True), ( + f"{name}: {int(np.sum(~((a == b) | (np.isnan(a) & np.isnan(b)))))} values differ") + checked += 1 + assert checked >= 24, checked + assert g.index_pos == c.index_pos + # the short-frame branch (no EMA200) is identical too + short = make_xau_ohlcv(150, seed=5) + gs = _gb36.precompute_indicators(short, tech) + with _gk36.BACKEND.cpu_only(): + cs = _gb36.precompute_indicators(short, tech) + assert gs.ema_trend is None and cs.ema_trend is None + assert np.array_equal(gs.adx, cs.adx, equal_nan=True) + + +def _sim_fixture36(): + rng = np.random.default_rng(38) + n = 1500 + idx = pd.date_range("2025-01-06", periods=n, freq="15min", tz="UTC") + close = 3000 + np.cumsum(rng.normal(0, 2.0, n)) + op = close + rng.normal(0, 0.5, n) + hi = np.maximum(op, close) + np.abs(rng.normal(0, 1.5, n)) + lo = np.minimum(op, close) - np.abs(rng.normal(0, 1.5, n)) + for j in (500,): # a NaN bar + op[j] = hi[j] = lo[j] = close[j] = np.nan + op[900] = close[899] - 25.0; lo[900] = min(lo[900], op[900]) # gap through a BUY stop + op[950] = close[949] + 25.0; hi[950] = max(hi[950], op[950]) # gap through a SELL stop / beyond a BUY target + close[n - 1] = np.nan # NaN close on a final window bar + sim = pd.DataFrame({"Open": op, "High": hi, "Low": lo, "Close": close}, index=idx) + + def mk(i, buy, arm=0.0, dist=0.0, risk=6.0, tp_dist=9.0): + e = float(op[i]) if not np.isnan(op[i]) else float(close[i - 1]) + sl = e - risk if buy else e + risk + tp = e + tp_dist if buy else e - tp_dist + return _gb36.Trade(pair=XAUUSD, entry_time=idx[i], entry_price=e, sl=sl, tp1=tp, + tp2=tp + 3, direction="BUY" if buy else "SELL", pip_value=0.1, + risk_pips=60, reward_pips=90, rr_planned=1.5, quality="A", + final_score=40, final_confidence=60, market_regime="trending_up", + trail_arm_r=arm, trail_dist_r=dist) + + trades = [mk(int(i), bool(rng.integers(0, 2)), *((0.7, 0.85) if k % 5 == 0 else (0.0, 0.0))) + for k, i in enumerate(rng.integers(1, n - 1, 300))] + trades += [mk(n - 1, True), mk(n - 1, False)] # window of one NaN-close bar + trades += [mk(n - 3, True, risk=500.0, tp_dist=500.0), + mk(n - 3, False, risk=500.0, tp_dist=500.0)] # TIMEOUT onto the NaN close -> NO_DATA + trades += [mk(n - 400, True, risk=500.0, tp_dist=500.0), + mk(n - 400, False, risk=500.0, tp_dist=500.0)] # a clean TIMEOUT at max_bars + trades += [mk(500, True), mk(500, False)] # NaN bar first (NO_DATA) + trades += [mk(499, True), mk(499, False)] # NaN bar second + trades += [mk(899, True), mk(949, False), mk(949, True)] # gap fills + trades += [mk(200, True, risk=0.0), mk(200, False, risk=0.0)] # zero risk + return sim, trades + + +@test("GPU backend: simulate_trades GPU == CPU (trailing and static; NaN windows, gaps, zero risk, per-trade trail)") +def _(): + if not _GPU36: + print(" (no CUDA: kernel not exercised)") + return + sim, trades = _sim_fixture36() + fields = ("exit_reason", "exit_price", "exit_time", "bars_held", "r_multiple", "mfe_r", "mae_r") + for arm in (0.6, 0.0): + a = copy.deepcopy(trades) + b = copy.deepcopy(trades) + _gb36.simulate_trades(a, sim, max_bars=300, spread=0.65, trail_arm_r=arm, trail_dist_r=0.75) + with _gk36.BACKEND.cpu_only(): + _gb36.simulate_trades(b, sim, max_bars=300, spread=0.65, trail_arm_r=arm, trail_dist_r=0.75) + for i, (x, y) in enumerate(zip(a, b)): + for f in fields: + assert _same36(getattr(x, f), getattr(y, f)), ( + f"arm={arm} trade {i} {f}: gpu={getattr(x, f)!r} cpu={getattr(y, f)!r}") + reasons = {t.exit_reason for t in a} + assert "NO_DATA" in reasons and "TIMEOUT" in reasons, reasons + assert ("TRAIL" in reasons) == (arm > 0) + assert ("TP1" in reasons) == (arm == 0) + # the NaN-close TIMEOUT never leaks a NaN R into the book + assert all(np.isfinite(t.r_multiple) for t in a) + # empty and no-data-after-entry cases return without touching the kernel + _gb36.simulate_trades([], sim, max_bars=300) + + +@test("GPU backend: --backend parses, FB2_BACKEND is honoured, cpu_only() routes to the NumPy twins, gpu fails loudly") +def _(): + assert _gb36._parse_args(["--pair", "XAUUSD", "--backend", "cpu"]).backend == "cpu" + assert _gb36._parse_args(["--pair", "XAUUSD"]).backend is None + try: + _gb36._parse_args(["--pair", "XAUUSD", "--backend", "tpu"]) + assert False, "invalid backend must be rejected" + except SystemExit: + pass + with patch.dict(os.environ, {_gk36.ENV_BACKEND: "cpu"}): + b = _gk36.Backend() + assert b.mode == "cpu" and b.available is False + assert b.describe().startswith("NumPy CPU fallback (forced") + with patch.dict(os.environ, {_gk36.ENV_BACKEND: "nonsense"}): + assert _gk36.Backend().mode == "auto" + # cpu_only(): the GPU twin must not be called + with _gk36.BACKEND.cpu_only(): + assert _gk36.BACKEND.available is False + with patch.object(_gb36, "_precompute_indicators_gpu", + side_effect=AssertionError("GPU twin called under cpu_only")): + _gb36.precompute_indicators(make_xau_ohlcv(120, seed=1), TechnicalConfig()) + with patch.object(_gb36, "_resolve_trades_gpu", + side_effect=AssertionError("GPU resolver called under cpu_only")): + sim, trades = _sim_fixture36() + _gb36.simulate_trades(trades[:5], sim, max_bars=50, trail_arm_r=0.6) + assert _gk36.BACKEND.describe().startswith(("CuPy GPU (", "NumPy CPU fallback (")) + # --backend gpu without CUDA raises instead of falling back + b3 = _gk36.Backend() + b3._probed, b3._ok, b3.reason = True, False, "synthetic probe failure" + try: + b3.set_mode("gpu") + assert False, "must raise" + except RuntimeError as e: + assert "synthetic probe failure" in str(e) + assert b3.describe() == "NumPy CPU fallback (synthetic probe failure)" + try: + b3.set_mode("tpu") + assert False + except ValueError: + pass + # release_device_memory is a no-op without CUDA + b3.release_device_memory() + + +@test("GPU backend: gpu_kernels imports none of the live writers; cupy never enters the live roots or the backtester") +def _(): + banned = {"signal_log", "news_engine", "providers", "telegram_bot", "news_archive", + "agent_panel", "violent_watch", "gpu_backtest", "engine", "signal_aggregator", + "replay_record", "sklearn", "scipy", "lightgbm", "torch"} + kpath = Path("backtest/gpu_kernels.py") + seen = set() + for node in ast.walk(ast.parse(kpath.read_text(encoding="utf-8"), str(kpath))): + if isinstance(node, ast.Import): + seen.update(a.name.split(".")[0] for a in node.names) + elif isinstance(node, ast.ImportFrom): + seen.add((node.module or "").split(".")[0]) + assert not (seen & banned), seen & banned + assert "numpy" in seen and "cupy" in seen # cupy only inside the probe / kernels + ksrc = kpath.read_text(encoding="utf-8") + assert 'KERNEL_OPTIONS = ("--fmad=false",)' in ksrc # no FMA contraction, ever + assert "options=KERNEL_OPTIONS" in ksrc + # the backtester never imports cupy itself; it reads backend state through gpu_kernels + bsrc = Path("backtest/gpu_backtest.py").read_text(encoding="utf-8") + assert "import cupy" not in bsrc and "_GPU_AVAILABLE" not in bsrc + assert "from backtest import gpu_kernels" in bsrc + assert "BACKEND.release_device_memory()" in bsrc # before the fork + assert bsrc.index("BACKEND.release_device_memory()") < bsrc.index('mp.get_context("fork")') + # live roots stay free of cupy (Module 32's guard bans it too) + offenders = [] + for root in (Path("."), Path("providers"), Path("pair_signals")): + for path in sorted(root.glob("*.py")): + if "venv" in path.parts: + continue + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"), str(path))): + names = ([a.name for a in node.names] if isinstance(node, ast.Import) + else [node.module or ""] if isinstance(node, ast.ImportFrom) else []) + offenders += [f"{path}:{node.lineno}" for nm in names if nm.split(".")[0] == "cupy"] + assert not offenders, offenders + assert "cupy" not in open("requirements.txt").read() + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════