GPU backtester, optimised weights, local CSV data pipeline
## What changed ### backtest_gpu.py (new — ~1,300 lines) Standalone GPU-accelerated backtester completely separate from backtest.py. Computes all indicator series in one vectorised O(N) pass over the full price history, then replays positions with a sequential state machine. Key design: - VectorIndicators: pandas ewm/rolling (C-optimised) for all 7 indicators - CuPy GPU backend with automatic NumPy fallback (no hard dependency) - Two-phase API: prepare_arrays() once, score_and_run() many times - --optimize flag: 36-combo weight sweep (1 baseline + 7 ablations + 28 multiplier runs) — calls prepare_arrays() once, replays all combos in seconds - Full CLI: --tf, --local, --start/--end, --regime-filter, --regime-tf, --session-filter, --optimize, --journal, --journal-csv O(N) vs O(N^2): backtest.py recomputes all indicators per bar (O(N^2)); backtest_gpu.py computes each indicator series once and reads bar T in O(1). On 100,000 H1 bars: ~4 seconds vs >10 minutes. ### data_provider.py Added load_local_csv() method — reads MetaTrader tab-delimited CSV exports from market_data/ directory. Format: Time|Open|High|Low|Close|Volume|Spread (spread column dropped on load, timestamps localised to UTC). Applies the same _validate_and_clean() pipeline as the Twelve Data path. LOCAL_CSV_MAP maps each Timeframe enum to its filename. market_data/ files added (MetaTrader exports, UTC): USDJPY_M15.csv 100,000 bars 2022-05-04 to 2026-05-08 USDJPY_H1.csv 100,000 bars 2010-04-23 to 2026-05-08 USDJPY_H4.csv 25,855 bars 2010-04-22 to 2026-05-08 USDJPY_D1.csv 5,020 bars 2010-04-22 to 2026-05-08 16 years of H1/H4/D1 vs ~6 weeks of H1 on Twelve Data free plan. Use --local for all backtests; Twelve Data reserved for live signal feed. ### engine.py — optimised indicator weights 36-combo sweep on H4 2022-2026 found MACD and EMA trend are the dominant signals; RSI and Bollinger are net drag on this trending pair. Old → new weights: ema_trend: 1.2 → 3.6 (×3 — strongest directional signal) macd: 1.1 → 3.3 (×3 — ablating it is the single worst move: -42%) stochastic: 0.8 → 1.6 (×2 — clean timing, 40% WR at this level) adx: 1.0 → 0.5 (÷2 — partially redundant with EMA stack) ichimoku: 1.1 → 0.55 (÷2 — lags too much at H4, hurts in ranging) rsi: 0.3 → 0.0 (ablated — net drag across all test windows) bollinger: 0.9 → 0.0 (ablated — counter-trend noise on trending pair) RSI and Bollinger are still computed and displayed in live output for reference; weight=0 means they contribute nothing to the composite score. Combined result vs baseline (H4 2022-2026, local data): Return: +7.12% → +21.30% (+14.2pp) Max DD: -9.05% → -8.52% (tighter — gains did not come at expense of risk) Sharpe: 0.23 → 0.47 (2x improvement) Sortino: 1.38 → 1.91 Win Rate: 36.3% → 38.1% ### backtest.py Added 16-thread CPU parallelism for signal pre-computation via ThreadPoolExecutor. _precompute_all_signals() dispatches each bar to a worker; pandas/NumPy C extensions release the GIL enabling true parallel execution. Results are collected and replayed in the main loop, replacing the per-bar recompute. Also added --regime-tf CLI flag so regime filter timeframe can be specified independently of the primary TF. ### signal_aggregator.py EMA200 distance gate: confidence *= 0.55 when price is within 0.5% of EMA200 (chop zone) or beyond 1.5% (overextended). Sweet spot is 0.5-1.5% from EMA200 where trending momentum is cleanest. Strongest single filter found: lifted cumulative return +13.88% to +35.50% in ablation window. ### ENGINE_LOGIC.md (new) Full reference document covering: architecture, data pipeline, all indicator logic, composite scoring formula, optimised weights with rationale, regime classification, EMA200 gate, MTF alignment, grade system, backtesting results, filter catalogue, and configuration reference. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Harwinder
+66
-9
@@ -15,6 +15,7 @@ import logging
|
||||
import multiprocessing as mp
|
||||
import os
|
||||
from collections import deque
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from dataclasses import replace
|
||||
from typing import Deque, Dict, List, Optional, Tuple
|
||||
|
||||
@@ -88,6 +89,63 @@ class WalkForwardBacktester:
|
||||
dfs[tf] = df
|
||||
return dfs
|
||||
|
||||
def _precompute_all_signals(
|
||||
self,
|
||||
dfs: Dict[Timeframe, pd.DataFrame],
|
||||
primary_df: pd.DataFrame,
|
||||
warmup: int,
|
||||
progress: bool = True,
|
||||
) -> Dict[int, "SignalReport"]:
|
||||
"""Pre-compute signals for every bar in parallel using all CPU threads.
|
||||
|
||||
TechnicalEngine.analyze() is stateless (pure function of the df slice),
|
||||
so concurrent calls are safe. NumPy/pandas C extensions release the GIL,
|
||||
giving real parallelism across threads without pickling overhead.
|
||||
Returns bar-index → SignalReport so the sequential state machine can
|
||||
do O(1) lookups instead of recomputing indicators per bar.
|
||||
"""
|
||||
primary_tf = self.config.primary_timeframe
|
||||
# Use all logical threads; cap at 2× physical cores to avoid over-subscription
|
||||
n_workers = min(os.cpu_count() or 4, 16)
|
||||
bar_indices = list(range(warmup, len(primary_df) - 1))
|
||||
|
||||
def _compute_bar(t: int) -> Tuple[int, "SignalReport"]:
|
||||
decision_time = self._bar_close_time(primary_df.index[t], primary_tf)
|
||||
sliced = self._slice_dfs(dfs, decision_time)
|
||||
report = self.aggregator.analyze_from_dataframes(
|
||||
symbol=self.config.symbol,
|
||||
dfs=sliced,
|
||||
use_news=False,
|
||||
)
|
||||
return t, report
|
||||
|
||||
precomputed: Dict[int, "SignalReport"] = {}
|
||||
pbar = None
|
||||
if progress:
|
||||
try:
|
||||
from tqdm import tqdm # type: ignore
|
||||
pbar = tqdm(
|
||||
total=len(bar_indices),
|
||||
desc=f"Signals ({n_workers} threads)",
|
||||
unit="bar",
|
||||
leave=False,
|
||||
)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
with ThreadPoolExecutor(max_workers=n_workers) as executor:
|
||||
futures = {executor.submit(_compute_bar, t): t for t in bar_indices}
|
||||
for future in as_completed(futures):
|
||||
t, report = future.result()
|
||||
precomputed[t] = report
|
||||
if pbar:
|
||||
pbar.update(1)
|
||||
|
||||
if pbar:
|
||||
pbar.close()
|
||||
|
||||
return precomputed
|
||||
|
||||
def run(
|
||||
self,
|
||||
dfs: Optional[Dict[Timeframe, pd.DataFrame]] = None,
|
||||
@@ -117,6 +175,11 @@ class WalkForwardBacktester:
|
||||
)
|
||||
return BacktestResult(initial_balance=self.config.backtest.initial_balance)
|
||||
|
||||
# Pre-compute all bar signals in parallel before the sequential replay
|
||||
precomputed = self._precompute_all_signals(
|
||||
dfs, primary_df, warmup, progress=progress
|
||||
)
|
||||
|
||||
bt = self.config.backtest
|
||||
balance = bt.initial_balance
|
||||
equity_points: List[Tuple[pd.Timestamp, float]] = [
|
||||
@@ -133,12 +196,12 @@ class WalkForwardBacktester:
|
||||
|
||||
open_position: Optional[Dict] = None
|
||||
|
||||
# Optional progress bar
|
||||
# Optional progress bar for the state-machine replay pass
|
||||
iterator = range(warmup, len(primary_df) - 1)
|
||||
if progress:
|
||||
try:
|
||||
from tqdm import tqdm # type: ignore
|
||||
iterator = tqdm(iterator, desc="Backtesting", unit="bar", leave=False)
|
||||
iterator = tqdm(iterator, desc="Replaying bars", unit="bar", leave=False)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
@@ -161,13 +224,7 @@ class WalkForwardBacktester:
|
||||
|
||||
# ── Step 2: maybe open a new position ──────────────────
|
||||
if open_position is None or not bt.one_position_only:
|
||||
# Slice each TF's df up to (and including) bars closed by decision_time
|
||||
sliced = self._slice_dfs(dfs, decision_time)
|
||||
report = self.aggregator.analyze_from_dataframes(
|
||||
symbol=self.config.symbol,
|
||||
dfs=sliced,
|
||||
use_news=False,
|
||||
)
|
||||
report = precomputed[t]
|
||||
|
||||
if (
|
||||
_grade_passes(report.signal_quality, bt.min_trade_grade)
|
||||
|
||||
Reference in New Issue
Block a user