diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index fc6b6e1..eb8cfbb 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -65,6 +65,22 @@ from pair_signals import PAIR_PRODUCERS # noqa: E402 # An importable CuPy can still fail at first kernel launch if CUDA libs # are missing (containers, headless boxes), so verify with a tiny op. # ───────────────────────────────────────────────────────────────────── +def _auto_detect_cuda_path(): + """Set CUDA_PATH from pip-installed nvidia packages if not already set.""" + if os.environ.get("CUDA_PATH"): + return + for p in sys.path: + nvrtc_lib = os.path.join(p, "nvidia", "cuda_nvrtc", "lib") + if os.path.isdir(nvrtc_lib): + os.environ["CUDA_PATH"] = os.path.join(p, "nvidia") + ld = os.environ.get("LD_LIBRARY_PATH", "") + if nvrtc_lib not in ld: + os.environ["LD_LIBRARY_PATH"] = nvrtc_lib + (":" + ld if ld else "") + return + +_auto_detect_cuda_path() + + def _detect_gpu(): try: import cupy as cp_mod # type: ignore @@ -93,6 +109,147 @@ def _to_cpu(arr) -> np.ndarray: logger = logging.getLogger("gpu_backtest") +# ───────────────────────────────────────────────────────────────────── +# CUDA kernels for parallel EMA (prefix scan on affine transforms) +# +# EMA recurrence: y[i] = α·x[i] + β·y[i-1], β = 1-α +# Represented as affine pair (a, b) where y = a·seed + b. +# Non-NaN element: (β, α·x[i]) +# First non-NaN: (0, x[i]) ← resets/seeds the state +# NaN element: (1, 0) ← carries previous value +# Associative combine: (a1,b1)·(a2,b2) = (a1·a2, a2·b1 + b2) +# After inclusive scan, the b component at position i equals y[i]. +# +# NaN handling & min_periods masking done as CuPy pre/post-processing +# (build a/b arrays before kernel, mask output after). The kernel +# itself is a pure prefix scan on (a, b) pairs — no NaN logic inside. +# ───────────────────────────────────────────────────────────────────── +_EWM_BLOCK_SIZE = 512 + +_ewm_scan_kernel = None +_ewm_batch_kernel = None + +if _GPU_AVAILABLE: + _ewm_scan_kernel = cp.RawKernel(r''' + extern "C" __global__ + void ewm_scan(const double* __restrict__ a_in, + const double* __restrict__ b_in, + double* __restrict__ out, + const int n) { + extern __shared__ double sdata[]; + double* s_a = sdata; + double* s_b = sdata + blockDim.x; + const int tid = threadIdx.x; + + double carry_a = 1.0; + double carry_b = 0.0; + + for (int tile = 0; tile < n; tile += blockDim.x) { + int idx = tile + tid; + s_a[tid] = (idx < n) ? a_in[idx] : 1.0; + s_b[tid] = (idx < n) ? b_in[idx] : 0.0; + __syncthreads(); + + for (int stride = 1; stride < blockDim.x; stride <<= 1) { + double pa = 1.0, pb = 0.0; + if (tid >= stride) { + pa = s_a[tid - stride]; + pb = s_b[tid - stride]; + } + __syncthreads(); + if (tid >= stride) { + s_b[tid] = s_a[tid] * pb + s_b[tid]; + s_a[tid] = pa * s_a[tid]; + } + __syncthreads(); + } + + if (idx < n) + out[idx] = s_a[tid] * carry_b + s_b[tid]; + + __syncthreads(); + if (tid == 0) { + int last = min(blockDim.x, n - tile) - 1; + double la = s_a[last], lb = s_b[last]; + double nb = la * carry_b + lb; + carry_a = la * carry_a; + carry_b = nb; + } + __syncthreads(); + if (tid == 0) { s_a[0] = carry_a; s_b[0] = carry_b; } + __syncthreads(); + carry_a = s_a[0]; + carry_b = s_b[0]; + } + } + ''', 'ewm_scan') + + _ewm_batch_kernel = cp.RawKernel(r''' + extern "C" __global__ + void ewm_batch(const double* __restrict__ a_in, + const double* __restrict__ b_in, + double* __restrict__ out, + const int* __restrict__ offsets, + const int* __restrict__ lengths, + const int num_emas) { + int ema_idx = blockIdx.x; + if (ema_idx >= num_emas) return; + + const int off = offsets[ema_idx]; + const int n = lengths[ema_idx]; + const int tid = threadIdx.x; + + extern __shared__ double sdata[]; + double* s_a = sdata; + double* s_b = sdata + blockDim.x; + + const double* my_a = a_in + off; + const double* my_b = b_in + off; + double* my_out = out + off; + + double carry_a = 1.0; + double carry_b = 0.0; + + for (int tile = 0; tile < n; tile += blockDim.x) { + int idx = tile + tid; + s_a[tid] = (idx < n) ? my_a[idx] : 1.0; + s_b[tid] = (idx < n) ? my_b[idx] : 0.0; + __syncthreads(); + + for (int stride = 1; stride < blockDim.x; stride <<= 1) { + double pa = 1.0, pb = 0.0; + if (tid >= stride) { + pa = s_a[tid - stride]; + pb = s_b[tid - stride]; + } + __syncthreads(); + if (tid >= stride) { + s_b[tid] = s_a[tid] * pb + s_b[tid]; + s_a[tid] = pa * s_a[tid]; + } + __syncthreads(); + } + + if (idx < n) + my_out[idx] = s_a[tid] * carry_b + s_b[tid]; + + __syncthreads(); + if (tid == 0) { + int last = min(blockDim.x, n - tile) - 1; + double la = s_a[last], lb = s_b[last]; + double nb = la * carry_b + lb; + carry_a = la * carry_a; + carry_b = nb; + } + __syncthreads(); + if (tid == 0) { s_a[0] = carry_a; s_b[0] = carry_b; } + __syncthreads(); + carry_a = s_a[0]; + carry_b = s_b[0]; + } + } + ''', 'ewm_batch') + PAIR_ALIASES = { "XAUUSD": XAUUSD, @@ -115,94 +272,167 @@ BAR_SECONDS = { # GPU INDICATOR PRE-COMPUTATION # ═════════════════════════════════════════════════════════════════════ -def _gpu_ewm_alpha(x: np.ndarray, alpha: float, min_periods: int = 0) -> np.ndarray: - """ - Exponential weighted mean matching pandas - `ewm(alpha=alpha, min_periods=p, adjust=False).mean()` exactly. +def _build_ewm_pairs(xg, alpha): + """Build (a, b) affine-transform pairs for EMA prefix scan on GPU. - Recursion: y[i] = alpha * x[i] + (1-alpha) * y[i-1], seeded at the - first non-NaN x. NaN inputs carry the previous y forward without - advancing the non-NaN counter (pandas' default `ignore_na=False`). - Output positions before min_periods non-NaN observations are NaN. - - Sequential CPU pass — EMA is recursive so true GPU parallelism - requires a scan kernel, which is overkill for a one-shot pre-compute. + Returns (a, b, non_nan_cum) — all CuPy arrays. """ - n = x.shape[0] - out = np.full(n, np.nan, dtype=np.float64) - if n == 0: + beta = 1.0 - alpha + nan_mask = cp.isnan(xg) + non_nan_cum = cp.cumsum((~nan_mask).astype(cp.int32)) + + a = cp.where(nan_mask, 1.0, beta) + b = cp.where(nan_mask, 0.0, alpha * xg) + + first_mask = (non_nan_cum == 1) & (~nan_mask) + a = cp.where(first_mask, 0.0, a) + b = cp.where(first_mask, xg, b) + + return a, b, non_nan_cum + + +def _gpu_ewm_alpha(x, alpha: float, min_periods: int = 0): + """EMA via parallel prefix scan (GPU) or sequential loop (CPU). + + Returns CuPy array when GPU available, numpy array otherwise. + NaN positions after the first valid value carry the last computed + EMA value (differs from pandas which outputs NaN at those positions). + """ + if _GPU_AVAILABLE: + xg = cp.asarray(x, dtype=cp.float64) + n = xg.shape[0] + if n == 0: + return cp.empty(0, dtype=cp.float64) + + a, b, non_nan_cum = _build_ewm_pairs(xg, alpha) + + out = cp.empty(n, dtype=cp.float64) + shared = 2 * _EWM_BLOCK_SIZE * 8 + _ewm_scan_kernel((1,), (_EWM_BLOCK_SIZE,), + (a, b, out, np.int32(n)), shared_mem=shared) + + if min_periods > 0: + out = cp.where(non_nan_cum < min_periods, cp.nan, out) + out = cp.where(non_nan_cum == 0, cp.nan, out) + return out + else: + arr = np.asarray(x, dtype=np.float64) + n = arr.shape[0] + out = np.full(n, np.nan, dtype=np.float64) + if n == 0: + return out + one_minus = 1.0 - alpha + seen = 0 + last_y = np.nan + for i in range(n): + xi = arr[i] + if np.isnan(xi): + if not np.isnan(last_y) and seen >= min_periods: + out[i] = last_y + continue + last_y = xi if np.isnan(last_y) else alpha * xi + one_minus * last_y + seen += 1 + if seen >= min_periods: + out[i] = last_y return out - one_minus = 1.0 - alpha - seen = 0 - last_y = np.nan - for i in range(n): - xi = x[i] - if np.isnan(xi): - # NaN input: y stays at last_y, don't advance counter. - if not np.isnan(last_y) and seen >= min_periods: - out[i] = last_y - continue - last_y = xi if np.isnan(last_y) else alpha * xi + one_minus * last_y - seen += 1 - if seen >= min_periods: - out[i] = last_y - - return out - - -def _gpu_ewm_span(x: np.ndarray, span: int) -> np.ndarray: - """ewm with span= semantics (alpha = 2/(span+1), no min_periods, seeded at x[0]).""" +def _gpu_ewm_span(x, span: int): + """ewm with span= semantics (alpha = 2/(span+1), no min_periods).""" alpha = 2.0 / (span + 1.0) return _gpu_ewm_alpha(x, alpha, min_periods=0) -def _gpu_rolling_mean(x: np.ndarray, window: int) -> np.ndarray: - """Rolling mean via cumsum-diff, NaN-aware (matches pandas - `.rolling(window).mean()` default min_periods=window — returns NaN - if any value in the window is NaN).""" - n = x.shape[0] - out = np.full(n, np.nan, dtype=np.float64) +def _gpu_ewm_alpha_batch(items: list): + """Batch multiple EMA computations into a single kernel launch. + + items: list of (x_gpu: cp.ndarray, alpha: float, min_periods: int) + Returns: list of CuPy arrays (one per input EMA). + """ + if not items: + return [] + + all_a, all_b, all_nnc = [], [], [] + offsets, lengths = [], [] + offset = 0 + + for xg, alpha, mp in items: + n = xg.shape[0] + a, b, nnc = _build_ewm_pairs(xg, alpha) + all_a.append(a) + all_b.append(b) + all_nnc.append((nnc, mp)) + offsets.append(offset) + lengths.append(n) + offset += n + + cat_a = cp.concatenate(all_a) + cat_b = cp.concatenate(all_b) + cat_out = cp.empty(offset, dtype=cp.float64) + + d_offsets = cp.array(offsets, dtype=cp.int32) + d_lengths = cp.array(lengths, dtype=cp.int32) + num = len(items) + shared = 2 * _EWM_BLOCK_SIZE * 8 + _ewm_batch_kernel((num,), (_EWM_BLOCK_SIZE,), + (cat_a, cat_b, cat_out, d_offsets, d_lengths, + np.int32(num)), + shared_mem=shared) + + results = [] + for i, (nnc, mp) in enumerate(all_nnc): + o, n = offsets[i], lengths[i] + out = cat_out[o:o + n].copy() + if mp > 0: + out = cp.where(nnc < mp, cp.nan, out) + out = cp.where(nnc == 0, cp.nan, out) + results.append(out) + return results + + +def _gpu_rolling_mean(x, window: int): + """Rolling mean via cumsum-diff, NaN-aware. Returns CuPy array on GPU.""" + xp = cp if _GPU_AVAILABLE else np + xg = xp.asarray(x, dtype=xp.float64) + n = xg.shape[0] + out = xp.full(n, xp.nan, dtype=xp.float64) if n < window: return out - xg = _to_gpu(x.astype(np.float64)) - nan_mask = cp.isnan(xg) - x_filled = cp.where(nan_mask, 0.0, xg) - cs = cp.cumsum(x_filled) - valid_cnt = cp.cumsum((~nan_mask).astype(cp.int64)) + nan_mask = xp.isnan(xg) + x_filled = xp.where(nan_mask, 0.0, xg) + cs = xp.cumsum(x_filled) + valid_cnt = xp.cumsum((~nan_mask).astype(xp.int64)) - sum_w = cp.empty(n - window + 1, dtype=cp.float64) - cnt_w = cp.empty(n - window + 1, dtype=cp.int64) + sum_w = xp.empty(n - window + 1, dtype=xp.float64) + cnt_w = xp.empty(n - window + 1, dtype=xp.int64) sum_w[0] = cs[window - 1] sum_w[1:] = cs[window:] - cs[:-window] cnt_w[0] = valid_cnt[window - 1] cnt_w[1:] = valid_cnt[window:] - valid_cnt[:-window] - # pandas: NaN if any window value is NaN (cnt < window). - rolling = cp.where(cnt_w < window, cp.nan, sum_w / window) - out[window - 1:] = _to_cpu(rolling) + rolling = xp.where(cnt_w < window, xp.nan, sum_w / window) + out[window - 1:] = rolling return out -def _gpu_rolling_std(x: np.ndarray, window: int, ddof: int = 0) -> np.ndarray: - """Rolling std via cumsum of x and x^2, NaN-aware. ddof=0 matches - pandas std(ddof=0). NaN if any window value is NaN.""" - n = x.shape[0] - out = np.full(n, np.nan, dtype=np.float64) +def _gpu_rolling_std(x, window: int, ddof: int = 0): + """Rolling std via cumsum of x and x^2, NaN-aware. Returns CuPy array on GPU.""" + xp = cp if _GPU_AVAILABLE else np + xg = xp.asarray(x, dtype=xp.float64) + n = xg.shape[0] + out = xp.full(n, xp.nan, dtype=xp.float64) if n < window: return out - xg = _to_gpu(x.astype(np.float64)) - nan_mask = cp.isnan(xg) - x_filled = cp.where(nan_mask, 0.0, xg) + nan_mask = xp.isnan(xg) + x_filled = xp.where(nan_mask, 0.0, xg) x2_filled = x_filled * x_filled - cs = cp.cumsum(x_filled) - cs2 = cp.cumsum(x2_filled) - valid_cnt = cp.cumsum((~nan_mask).astype(cp.int64)) + cs = xp.cumsum(x_filled) + cs2 = xp.cumsum(x2_filled) + valid_cnt = xp.cumsum((~nan_mask).astype(xp.int64)) - sum_x = cp.empty(n - window + 1, dtype=cp.float64) - sum_x2 = cp.empty(n - window + 1, dtype=cp.float64) - cnt_w = cp.empty(n - window + 1, dtype=cp.int64) + sum_x = xp.empty(n - window + 1, dtype=xp.float64) + sum_x2 = xp.empty(n - window + 1, dtype=xp.float64) + cnt_w = xp.empty(n - window + 1, dtype=xp.int64) sum_x[0] = cs[window - 1] sum_x[1:] = cs[window:] - cs[:-window] sum_x2[0] = cs2[window - 1] @@ -212,12 +442,12 @@ def _gpu_rolling_std(x: np.ndarray, window: int, ddof: int = 0) -> np.ndarray: mean = sum_x / window var = sum_x2 / window - mean * mean - var = cp.maximum(var, 0.0) # numerical safety against tiny negatives + var = xp.maximum(var, 0.0) if ddof: var = var * window / max(window - ddof, 1) - std = cp.sqrt(var) - std = cp.where(cnt_w < window, cp.nan, std) - out[window - 1:] = _to_cpu(std) + std = xp.sqrt(var) + std = xp.where(cnt_w < window, xp.nan, std) + out[window - 1:] = std return out @@ -275,11 +505,21 @@ 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.""" - 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) + """Compute every causal indicator series once on the full df. + + When GPU available: all intermediate arrays stay on-device (CuPy). + Only converted to numpy at the end for IndicatorCache (scalar lookups). + EMA calls are batched into 3 kernel launches by dependency group. + """ + xp = cp if _GPU_AVAILABLE else np + close_np = df["Close"].to_numpy(dtype=np.float64) + high_np = df["High"].to_numpy(dtype=np.float64) + low_np = df["Low"].to_numpy(dtype=np.float64) + n = len(close_np) + + close = xp.asarray(close_np) + high = xp.asarray(high_np) + low = xp.asarray(low_np) if gpu_log: logger.info( @@ -287,35 +527,138 @@ def precompute_indicators( f"({'GPU' if _GPU_AVAILABLE else 'CPU'} backend)" ) - # ── EMAs (use ewm span semantics, identical to pandas .ewm(span=, adjust=False).mean()) ─ - ema_fast = _gpu_ewm_span(close, tech_cfg.ema_fast) - ema_medium = _gpu_ewm_span(close, tech_cfg.ema_medium) - ema_slow = _gpu_ewm_span(close, tech_cfg.ema_slow) - ema_trend = _gpu_ewm_span(close, tech_cfg.ema_trend) if n >= tech_cfg.ema_trend else None + # ── True Range (shared by ATR, ADX, Keltner) ── + prev_c = xp.empty(n, dtype=xp.float64) + prev_c[0] = xp.nan + prev_c[1:] = close[:-1] + tr1 = high - low + tr2 = xp.abs(high - prev_c) + tr3 = xp.abs(low - prev_c) + tr2[0] = xp.nan + tr3[0] = xp.nan + tr = xp.fmax(xp.fmax(tr1, tr2), tr3) - # ── RSI (Wilder via ewm alpha=1/period, min_periods=period) ── - # pandas: delta = close.diff() → delta[0] is NaN; gain = delta.where(delta > 0, 0.0) - # evaluates NaN > 0 as False so gain[0] becomes 0.0 (NOT NaN). Same for loss. - delta = np.empty(n, dtype=np.float64) - delta[0] = np.nan + # ── RSI delta / gain / loss ── + delta = xp.empty(n, dtype=xp.float64) + delta[0] = xp.nan delta[1:] = close[1:] - close[:-1] - with np.errstate(invalid="ignore"): - gain = np.where(delta > 0, delta, 0.0) - loss = np.where(delta < 0, -delta, 0.0) - avg_gain = _gpu_ewm_alpha(gain, 1.0 / tech_cfg.rsi_period, min_periods=tech_cfg.rsi_period) - avg_loss = _gpu_ewm_alpha(loss, 1.0 / tech_cfg.rsi_period, min_periods=tech_cfg.rsi_period) - avg_loss_safe = np.where(avg_loss == 0, 1e-10, avg_loss) + gain = xp.where(delta > 0, delta, 0.0) + loss = xp.where(delta < 0, -delta, 0.0) + + # ── ADX directional movement ── + plus_dm_raw = xp.empty(n, dtype=xp.float64) + minus_dm_raw = xp.empty(n, dtype=xp.float64) + plus_dm_raw[0] = xp.nan + minus_dm_raw[0] = xp.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 = xp.where(cond_plus, plus_dm_raw, 0.0) + cond_minus = (minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0) + minus_dm = xp.where(cond_minus, minus_dm_raw, 0.0) + + 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) + rsi_alpha = 1.0 / tech_cfg.rsi_period + atr_alpha = 1.0 / tech_cfg.atr_period + adx_alpha = 1.0 / tech_cfg.adx_period + kc_atr_alpha = 1.0 / kc_atr_period + + if _GPU_AVAILABLE: + t_ema = time.perf_counter() + # ── BATCH 1: 10 independent EMAs ── + batch1_items = [ + (close, 2.0 / (tech_cfg.ema_fast + 1), 0), # 0: ema_fast + (close, 2.0 / (tech_cfg.ema_medium + 1), 0), # 1: ema_medium + (close, 2.0 / (tech_cfg.ema_slow + 1), 0), # 2: ema_slow + (gain, rsi_alpha, tech_cfg.rsi_period), # 3: avg_gain + (loss, rsi_alpha, tech_cfg.rsi_period), # 4: avg_loss + (close, 2.0 / (tech_cfg.macd_fast + 1), 0), # 5: ema_macd_fast + (close, 2.0 / (tech_cfg.macd_slow + 1), 0), # 6: ema_macd_slow + (tr, atr_alpha, tech_cfg.atr_period), # 7: atr + (close, 2.0 / (kc_period + 1), 0), # 8: kc_mid + ] + if n >= tech_cfg.ema_trend: + batch1_items.append( + (close, 2.0 / (tech_cfg.ema_trend + 1), 0) # 9: ema_trend + ) + b1 = _gpu_ewm_alpha_batch(batch1_items) + ema_fast = b1[0] + ema_medium = b1[1] + ema_slow = b1[2] + avg_gain = b1[3] + avg_loss = b1[4] + ema_macd_fast = b1[5] + ema_macd_slow = b1[6] + atr = b1[7] + kc_mid = b1[8] + ema_trend = b1[9] if n >= tech_cfg.ema_trend else None + + # ── Intermediate: MACD line, DM inputs ── + macd_line = ema_macd_fast - ema_macd_slow + + # ── BATCH 2: 5 EMAs depending on Batch 1 outputs ── + batch2_items = [ + (macd_line, 2.0 / (tech_cfg.macd_signal + 1), 0), # 0: signal_line + (tr, adx_alpha, tech_cfg.adx_period), # 1: atr_adx + (plus_dm, adx_alpha, tech_cfg.adx_period), # 2: plus_dm_smooth + (minus_dm, adx_alpha, tech_cfg.adx_period), # 3: minus_dm_smooth + (tr, kc_atr_alpha, kc_atr_period), # 4: kc_atr + ] + b2 = _gpu_ewm_alpha_batch(batch2_items) + signal_line = b2[0] + atr_adx = b2[1] + plus_dm_smooth = b2[2] + minus_dm_smooth = b2[3] + kc_atr = b2[4] + + # ── Intermediate: DX for ADX ── + atr_adx_safe = cp.where(atr_adx == 0, cp.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, cp.nan, di_sum) + dx = 100.0 * cp.abs(plus_di - minus_di) / di_sum_safe + + # ── BATCH 3: 1 EMA depending on Batch 2 ── + b3 = _gpu_ewm_alpha_batch([(dx, adx_alpha, tech_cfg.adx_period)]) + adx = b3[0] + + logger.info(f" EMA batches (3 launches, 16 EMAs): {time.perf_counter() - t_ema:.4f}s") + else: + # CPU path — sequential EMA calls (no batching) + ema_fast = _gpu_ewm_span(close, tech_cfg.ema_fast) + ema_medium = _gpu_ewm_span(close, tech_cfg.ema_medium) + ema_slow = _gpu_ewm_span(close, tech_cfg.ema_slow) + ema_trend = _gpu_ewm_span(close, tech_cfg.ema_trend) if n >= tech_cfg.ema_trend else None + avg_gain = _gpu_ewm_alpha(gain, rsi_alpha, min_periods=tech_cfg.rsi_period) + avg_loss = _gpu_ewm_alpha(loss, rsi_alpha, min_periods=tech_cfg.rsi_period) + ema_macd_fast = _gpu_ewm_span(close, tech_cfg.macd_fast) + ema_macd_slow = _gpu_ewm_span(close, tech_cfg.macd_slow) + macd_line = ema_macd_fast - ema_macd_slow + signal_line = _gpu_ewm_span(macd_line, tech_cfg.macd_signal) + atr = _gpu_ewm_alpha(tr, atr_alpha, min_periods=tech_cfg.atr_period) + kc_mid = _gpu_ewm_span(close, kc_period) + atr_adx = _gpu_ewm_alpha(tr, adx_alpha, min_periods=tech_cfg.adx_period) + plus_dm_smooth = _gpu_ewm_alpha(plus_dm, adx_alpha, min_periods=tech_cfg.adx_period) + minus_dm_smooth = _gpu_ewm_alpha(minus_dm, adx_alpha, min_periods=tech_cfg.adx_period) + atr_adx_safe = xp.where(atr_adx == 0, xp.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 = xp.where(di_sum == 0, xp.nan, di_sum) + dx = 100.0 * xp.abs(plus_di - minus_di) / di_sum_safe + adx = _gpu_ewm_alpha(dx, adx_alpha, min_periods=tech_cfg.adx_period) + kc_atr = _gpu_ewm_alpha(tr, kc_atr_alpha, min_periods=kc_atr_period) + + # ── RSI ── + avg_loss_safe = xp.where(avg_loss == 0, 1e-10, avg_loss) rs = avg_gain / avg_loss_safe rsi = 100.0 - (100.0 / (1.0 + rs)) - rsi_series = pd.Series(rsi, index=df.index, name="rsi") - # ── MACD (12/26 EMA spans, signal 9 EMA span) ── - ema_macd_fast = _gpu_ewm_span(close, tech_cfg.macd_fast) - ema_macd_slow = _gpu_ewm_span(close, tech_cfg.macd_slow) - macd_line = ema_macd_fast - ema_macd_slow - signal_line = _gpu_ewm_span(macd_line, tech_cfg.macd_signal) + # ── MACD histogram ── macd_hist = macd_line - signal_line - macd_hist_series = pd.Series(macd_hist, index=df.index, name="macd_hist") # ── Bollinger ── bb_mid = _gpu_rolling_mean(close, tech_cfg.bb_period) @@ -323,115 +666,67 @@ def precompute_indicators( bb_upper = bb_mid + tech_cfg.bb_std * bb_std bb_lower = bb_mid - tech_cfg.bb_std * bb_std bb_range = bb_upper - bb_lower - bb_range_safe = np.where(bb_range == 0, np.nan, bb_range) + bb_range_safe = xp.where(bb_range == 0, xp.nan, bb_range) bb_position = (close - bb_lower) / bb_range_safe - # pandas behavior: when range is 0 it returns 0.5 (handled by engine fallback) - bb_position = np.where(np.isnan(bb_position), 0.5, bb_position) - bb_mid_safe = np.where(bb_mid == 0, np.nan, bb_mid) + bb_position = xp.where(xp.isnan(bb_position), 0.5, bb_position) + bb_mid_safe = xp.where(bb_mid == 0, xp.nan, bb_mid) bb_band_width = (bb_upper - bb_lower) / bb_mid_safe bb_avg_width = _gpu_rolling_mean(bb_band_width, 20) - # ── ATR ── - prev_close = np.empty(n, dtype=np.float64) - prev_close[0] = np.nan - prev_close[1:] = close[:-1] - tr1 = high - low - tr2 = np.abs(high - prev_close) - tr3 = np.abs(low - prev_close) - # NaN handling at index 0: pandas concat+max returns max ignoring NaN - # but for our purposes the engine takes the latest value, so propagate NaN here - tr2[0] = np.nan - tr3[0] = np.nan - tr = np.fmax(np.fmax(tr1, tr2), tr3) # fmax ignores NaN — matches pandas behavior - # ewm: pandas uses the first non-NaN value to seed - atr = _gpu_ewm_alpha(tr, 1.0 / tech_cfg.atr_period, min_periods=tech_cfg.atr_period) - - # ── ADX ── - # pandas: plus_dm = high.diff(); minus_dm = -low.diff(); both NaN at index 0. - # plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) — NaN comparisons - # are False, so plus_dm[0] becomes 0.0 (NOT NaN). Same for minus_dm. - plus_dm_raw = np.empty(n, dtype=np.float64) - minus_dm_raw = np.empty(n, dtype=np.float64) - plus_dm_raw[0] = np.nan - minus_dm_raw[0] = np.nan - plus_dm_raw[1:] = high[1:] - high[:-1] - minus_dm_raw[1:] = -(low[1:] - low[:-1]) - with np.errstate(invalid="ignore"): - cond_plus = (plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0) - plus_dm = np.where(cond_plus, plus_dm_raw, 0.0) - cond_minus = (minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0) - minus_dm = np.where(cond_minus, minus_dm_raw, 0.0) - - atr_adx = _gpu_ewm_alpha(tr, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period) - plus_dm_smooth = _gpu_ewm_alpha( - plus_dm, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period - ) - minus_dm_smooth = _gpu_ewm_alpha( - minus_dm, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period - ) - atr_adx_safe = np.where(atr_adx == 0, np.nan, atr_adx) - plus_di = 100.0 * plus_dm_smooth / atr_adx_safe - minus_di = 100.0 * minus_dm_smooth / atr_adx_safe - di_sum = plus_di + minus_di - di_sum_safe = np.where(di_sum == 0, np.nan, di_sum) - dx = 100.0 * np.abs(plus_di - minus_di) / di_sum_safe - adx = _gpu_ewm_alpha(dx, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period) + # ── Keltner Channels ── + 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(xp.float64) # ── ADX Slope ── adx_slope_lookback = getattr(tech_cfg, 'adx_slope_lookback', 5) - adx_slope = np.full(n, np.nan, dtype=np.float64) + adx_slope = xp.full(n, xp.nan, dtype=xp.float64) if n > adx_slope_lookback: adx_slope[adx_slope_lookback:] = adx[adx_slope_lookback:] - adx[:-adx_slope_lookback] # ── ROC (Rate of Change) ── roc_period = getattr(tech_cfg, 'roc_period', 10) - roc = np.full(n, np.nan, dtype=np.float64) + roc = xp.full(n, xp.nan, dtype=xp.float64) if n > roc_period: - prev_close = close[:-roc_period] - prev_safe = np.where(prev_close == 0, np.nan, prev_close) - roc[roc_period:] = ((close[roc_period:] - prev_close) / prev_safe) * 100.0 + prev_roc = close[:-roc_period] + prev_roc_safe = xp.where(prev_roc == 0, xp.nan, prev_roc) + roc[roc_period:] = ((close[roc_period:] - prev_roc) / prev_roc_safe) * 100.0 - # ── Keltner Channels ── - kc_period = getattr(tech_cfg, 'kc_period', 20) - kc_atr_period = getattr(tech_cfg, 'kc_atr_period', 10) - kc_multiplier = getattr(tech_cfg, 'kc_multiplier', 1.5) - kc_mid = _gpu_ewm_span(close, kc_period) - kc_atr = _gpu_ewm_alpha(tr, 1.0 / kc_atr_period, min_periods=kc_atr_period) - kc_upper = kc_mid + kc_multiplier * kc_atr - kc_lower = kc_mid - kc_multiplier * kc_atr - with np.errstate(invalid="ignore"): - kc_squeeze = (bb_lower > kc_lower) & (bb_upper < kc_upper) - kc_squeeze = kc_squeeze.astype(np.float64) + # ── Transfer to CPU for IndicatorCache (scalar lookups in GPUTechnicalEngine) ── + def cpu(arr): + if arr is None: + return None + return _to_cpu(arr) index_pos = {ts: i for i, ts in enumerate(df.index)} return IndicatorCache( df=df, index_pos=index_pos, - ema_fast=ema_fast, - ema_medium=ema_medium, - ema_slow=ema_slow, - ema_trend=ema_trend, - rsi=rsi, - rsi_series=rsi_series, - macd_hist=macd_hist, - macd_hist_series=macd_hist_series, - bb_mid=bb_mid, - bb_upper=bb_upper, - bb_lower=bb_lower, - bb_position=bb_position, - bb_band_width=bb_band_width, - bb_avg_width=bb_avg_width, - atr=atr, - adx=adx, - plus_di=plus_di, - minus_di=minus_di, - adx_slope=adx_slope, - roc=roc, - kc_mid=kc_mid, - kc_upper=kc_upper, - kc_lower=kc_lower, - kc_squeeze=kc_squeeze, + ema_fast=cpu(ema_fast), + ema_medium=cpu(ema_medium), + ema_slow=cpu(ema_slow), + ema_trend=cpu(ema_trend), + rsi=cpu(rsi), + rsi_series=pd.Series(cpu(rsi), index=df.index, name="rsi"), + macd_hist=cpu(macd_hist), + macd_hist_series=pd.Series(cpu(macd_hist), index=df.index, name="macd_hist"), + bb_mid=cpu(bb_mid), + bb_upper=cpu(bb_upper), + bb_lower=cpu(bb_lower), + bb_position=cpu(bb_position), + bb_band_width=cpu(bb_band_width), + bb_avg_width=cpu(bb_avg_width), + atr=cpu(atr), + adx=cpu(adx), + plus_di=cpu(plus_di), + minus_di=cpu(minus_di), + adx_slope=cpu(adx_slope), + roc=cpu(roc), + kc_mid=cpu(kc_mid), + kc_upper=cpu(kc_upper), + kc_lower=cpu(kc_lower), + kc_squeeze=cpu(kc_squeeze), ) @@ -784,11 +1079,26 @@ class GPUTechnicalEngine(TechnicalEngine): signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL direction = "bearish" - desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}" + adx_slope_val = float(cache.adx_slope[pos]) if not np.isnan(cache.adx_slope[pos]) else 0.0 + + slope_desc = "" + emerging_threshold = c.adx_slope_emerging_threshold + if adx_val < c.adx_strong_trend and adx_slope_val > emerging_threshold: + slope_desc = " | TREND EMERGING (ADX rising fast)" + conf = min(conf + 15, 85) + elif adx_val >= c.adx_strong_trend and adx_slope_val > 0: + slope_desc = f" | Trend strengthening (slope +{adx_slope_val:.1f})" + conf = min(conf + 5, 85) + elif adx_val >= c.adx_strong_trend and adx_slope_val < -emerging_threshold: + slope_desc = f" | Trend weakening (slope {adx_slope_val:.1f})" + conf = max(conf - 10, 30) + + desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}{slope_desc}" result.indicators["adx"] = IndicatorSignal( name="ADX", value=adx_val, signal=signal, confidence=conf, description=desc, weight=0.5, ) + result._adx_slope = adx_slope_val # ── ATR override ────────────────────────────────────────────── @@ -839,7 +1149,7 @@ class GPUTechnicalEngine(TechnicalEngine): c = self.config n = c.roc_period - if pos < n + 1: + if pos < n + 4: return roc_val = float(cache.roc[pos]) @@ -1373,20 +1683,25 @@ class BacktestConfig: def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]: """Apply XAU weekend + allowed-weekday filter (live `analyze_pair` does this; - `analyze_from_dataframes` does not, so we replicate here).""" + `analyze_from_dataframes` does not, so we replicate here). + Matches signal_aggregator._is_forex_weekend: Fri 22:00 → Sun 22:00 UTC.""" xau = cfg.xau_strategy - if xau.weekend_filter and ts.weekday() >= 5: - return "WEEKEND_FILTER" + if xau.weekend_filter: + wd, hour = ts.weekday(), ts.hour + if (wd == 4 and hour >= 22) or wd == 5 or (wd == 6 and hour < 22): + return "WEEKEND_FILTER" if xau.allowed_weekdays and ts.weekday() not in xau.allowed_weekdays: return "DAY_FILTER" return None def _is_pair_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]: - """Generic weekend + weekday filter using FilterConfig (backtest field).""" - bt = cfg.backtest - if ts.weekday() >= 5: + """Generic weekend + weekday filter. Matches signal_aggregator._is_forex_weekend: + Fri 22:00 → Sun 22:00 UTC.""" + wd, hour = ts.weekday(), ts.hour + if (wd == 4 and hour >= 22) or wd == 5 or (wd == 6 and hour < 22): return "WEEKEND_FILTER" + bt = cfg.backtest if bt.allowed_weekdays and ts.weekday() not in bt.allowed_weekdays: return "DAY_FILTER" return None @@ -1917,6 +2232,10 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: help="Print full per-trade reasoning (indicators, signals, agreement, why entered)") p.add_argument("--reasoning-samples", type=int, default=3, help="If --detail not set, print this many sample trades' full reasoning") + p.add_argument("--gpu-only", action="store_true", + help="Require GPU (CuPy/CUDA). Exit with error if unavailable.") + p.add_argument("--quick", action="store_true", + help="Quick test mode: 30-day window for speed verification") p.add_argument("--log-level", default="INFO") return p.parse_args(argv) @@ -1945,6 +2264,18 @@ def main(argv: Optional[List[str]] = None) -> int: format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) + if args.gpu_only and not _GPU_AVAILABLE: + logger.error("--gpu-only specified but CuPy/CUDA is not available") + return 1 + + if args.quick: + if not args.start: + args.days = min(args.days, 30) + args.warmup_bars = min(args.warmup_bars, 200) + logger.info("Quick mode: 30-day window, warmup=200 bars") + + t_total_start = time.perf_counter() + 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()] @@ -1976,6 +2307,14 @@ def main(argv: Optional[List[str]] = None) -> int: f"Backend: {'CuPy GPU' if _GPU_AVAILABLE else 'NumPy CPU fallback'} | " f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()}" ) + if _GPU_AVAILABLE: + import cupy as _cp_info + _dev = _cp_info.cuda.Device() + _mem_free, _mem_total = _dev.mem_info + logger.info( + f"GPU {_dev.id}: compute capability {_dev.compute_capability} | " + f"Memory: {_mem_free / 1e9:.1f} GB free / {_mem_total / 1e9:.1f} GB total" + ) if pair in PAIR_PRODUCERS: platform_cfg = PAIR_PRODUCERS[pair](primary_tf, confirm_tfs) @@ -2000,6 +2339,10 @@ def main(argv: Optional[List[str]] = None) -> int: df.to_csv(args.export_csv, index=False) logger.info(f"Exported {len(df)} trades → {args.export_csv}") + logger.info( + f"Total backtest time: {time.perf_counter() - t_total_start:.2f}s " + f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})" + ) return 0