Harwinder
17cb63e86e
## 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.
1300 lines
51 KiB
Python
1300 lines
51 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ForexBot2 — GPU-Accelerated Backtester (standalone)
|
||
=====================================================
|
||
Computes ALL indicators in ONE vectorised pass over the full price series
|
||
(O(N) total work), then replays positions bar-by-bar with a sequential
|
||
state machine. Completely separate from backtest.py — no shared engine code.
|
||
|
||
Compute backend
|
||
GPU — CuPy (NVIDIA CUDA): pip install cupy-cuda12x
|
||
CPU — NumPy fallback when CuPy is absent (auto-detected)
|
||
|
||
Why O(N) beats O(N²)
|
||
backtest.py slices the DataFrame and runs all indicators fresh each bar
|
||
→ N × O(N) indicator work = O(N²)
|
||
backtest_gpu.py computes every indicator series end-to-end once, then reads
|
||
bar T's value in O(1) → O(N) total
|
||
|
||
Usage
|
||
python backtest_gpu.py full history, H1, no filters
|
||
python backtest_gpu.py --start 2024-01-01 date-windowed
|
||
python backtest_gpu.py --regime-filter ADX gate on H4
|
||
python backtest_gpu.py --session-filter UTC 07-16 gate
|
||
python backtest_gpu.py --tf H4 --journal H4 with trade log
|
||
python backtest_gpu.py --help
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import logging
|
||
import os
|
||
import sys
|
||
from collections import deque
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
# Force UTF-8 on Windows
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
# ── GPU backend ───────────────────────────────────────────────────
|
||
try:
|
||
import cupy as cp # type: ignore
|
||
_GPU = True
|
||
_BACKEND = "CuPy (GPU)"
|
||
except ImportError:
|
||
cp = np # type: ignore[assignment]
|
||
_GPU = False
|
||
_BACKEND = "NumPy (CPU — install cupy-cuda12x for GPU)"
|
||
|
||
from config import PlatformConfig, Timeframe, USDJPY, PIP_VALUE
|
||
from data_provider import ForexDataProvider
|
||
from metrics import BacktestResult, Trade, compute_metrics, trades_to_dataframe
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
TF_MINUTES: Dict[Timeframe, int] = {
|
||
Timeframe.M1: 1, Timeframe.M5: 5, Timeframe.M15: 15,
|
||
Timeframe.M30: 30, Timeframe.H1: 60, Timeframe.H4: 240,
|
||
Timeframe.D1: 1440, Timeframe.W1: 10080,
|
||
}
|
||
|
||
GRADE_ORDER = ["A+", "A", "B", "C", "D", "F"]
|
||
|
||
|
||
def _grade_passes(grade: str, min_grade: str) -> bool:
|
||
if grade not in GRADE_ORDER or min_grade not in GRADE_ORDER:
|
||
return False
|
||
return GRADE_ORDER.index(grade) <= GRADE_ORDER.index(min_grade)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# VECTORISED INDICATOR ENGINE
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
class VectorIndicators:
|
||
"""
|
||
Compute every indicator series across the full OHLCV dataset in one pass.
|
||
Uses pandas ewm/rolling (C-optimised) for EMA/rolling math.
|
||
CuPy is applied to the scoring arrays where element-wise GPU math helps.
|
||
"""
|
||
|
||
def __init__(self, config: PlatformConfig):
|
||
self.tc = config.technical
|
||
|
||
# ── Low-level helpers ─────────────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _ema(s: pd.Series, span: int) -> pd.Series:
|
||
return s.ewm(span=span, adjust=False).mean()
|
||
|
||
@staticmethod
|
||
def _wilder(s: pd.Series, period: int) -> pd.Series:
|
||
return s.ewm(alpha=1.0 / period, min_periods=period).mean()
|
||
|
||
# ── EMA alignment ─────────────────────────────────────────────
|
||
|
||
def ema_alignment(
|
||
self, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series, pd.Series]:
|
||
"""
|
||
Returns (norm [-1..+1], confidence, ema200).
|
||
norm > 0.6 → STRONG_BUY, > 0.2 → BUY, < -0.2 → SELL, < -0.6 → STRONG_SELL.
|
||
"""
|
||
tc = self.tc
|
||
f = self._ema(close, tc.ema_fast)
|
||
m = self._ema(close, tc.ema_medium)
|
||
s = self._ema(close, tc.ema_slow)
|
||
t200 = self._ema(close, tc.ema_trend)
|
||
|
||
cv, fv, mv, sv, tv = (
|
||
close.values, f.values, m.values, s.values, t200.values
|
||
)
|
||
raw = (
|
||
np.sign(cv - fv) + np.sign(fv - mv) +
|
||
np.sign(mv - sv) + np.sign(sv - tv)
|
||
) / 4.0 # -1 .. +1
|
||
|
||
conf = np.where(
|
||
np.abs(raw) > 0.6,
|
||
np.minimum(90.0, 60.0 + np.abs(raw) * 30.0),
|
||
np.where(np.abs(raw) > 0.2, 55.0 + np.abs(raw) * 20.0, 40.0),
|
||
)
|
||
return (
|
||
pd.Series(raw, index=close.index),
|
||
pd.Series(conf, index=close.index),
|
||
t200,
|
||
)
|
||
|
||
# ── RSI ───────────────────────────────────────────────────────
|
||
|
||
def rsi(
|
||
self, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series, pd.Series]:
|
||
"""Returns (rsi_values, signal [-1..+1], confidence)."""
|
||
tc = self.tc
|
||
delta = close.diff()
|
||
gain = delta.where(delta > 0, 0.0)
|
||
loss = (-delta).where(delta < 0, 0.0)
|
||
avg_g = self._wilder(gain, tc.rsi_period)
|
||
avg_l = self._wilder(loss, tc.rsi_period)
|
||
rs = avg_g / avg_l.replace(0, np.nan)
|
||
rsi_v = (100 - 100 / (1 + rs)).fillna(50.0)
|
||
|
||
v = rsi_v.values
|
||
ob, os_, nu, nl = (
|
||
tc.rsi_overbought, tc.rsi_oversold,
|
||
tc.rsi_neutral_upper, tc.rsi_neutral_lower,
|
||
)
|
||
sig = np.where(v >= ob, -1,
|
||
np.where(v <= os_, 1,
|
||
np.where(v >= nu, 1,
|
||
np.where(v <= nl, -1, 0)))).astype(float)
|
||
|
||
conf = np.where(v >= ob, np.minimum(80.0, 50.0 + (v - ob)),
|
||
np.where(v <= os_, np.minimum(80.0, 50.0 + (os_ - v)),
|
||
np.where(np.abs(v - 50) > 10, 45.0, 30.0)))
|
||
|
||
return rsi_v, pd.Series(sig, index=close.index), pd.Series(conf, index=close.index)
|
||
|
||
# ── MACD ──────────────────────────────────────────────────────
|
||
|
||
def macd(
|
||
self, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series, pd.Series]:
|
||
"""Returns (histogram, signal [-2..+2], confidence)."""
|
||
tc = self.tc
|
||
ema_f = self._ema(close, tc.macd_fast)
|
||
ema_s = self._ema(close, tc.macd_slow)
|
||
line = ema_f - ema_s
|
||
sig_l = self._ema(line, tc.macd_signal)
|
||
hist = line - sig_l
|
||
|
||
above = (line.values > sig_l.values)
|
||
prev_ab = np.roll(above, 1); prev_ab[0] = above[0]
|
||
bull_x = above & ~prev_ab
|
||
bear_x = ~above & prev_ab
|
||
hist_v = hist.values
|
||
|
||
sig = np.where(bull_x, 2.0,
|
||
np.where(bear_x, -2.0,
|
||
np.where(hist_v > 0, 1.0,
|
||
np.where(hist_v < 0, -1.0, 0.0))))
|
||
|
||
conf = np.where(bull_x | bear_x, 75.0,
|
||
np.where(np.abs(hist_v) > 0, 55.0, 35.0))
|
||
|
||
return hist, pd.Series(sig, index=close.index), pd.Series(conf, index=close.index)
|
||
|
||
# ── Bollinger Bands ───────────────────────────────────────────
|
||
|
||
def bollinger(
|
||
self, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series, pd.Series]:
|
||
"""Returns (bb_position [0..1], signal [-1..+1], confidence)."""
|
||
tc = self.tc
|
||
sma = close.rolling(tc.bb_period).mean()
|
||
std = close.rolling(tc.bb_period).std()
|
||
upper = sma + tc.bb_std * std
|
||
lower = sma - tc.bb_std * std
|
||
bb_range = (upper - lower).replace(0, np.nan)
|
||
bb_pos = (close - lower) / bb_range
|
||
|
||
v = bb_pos.values
|
||
sig = np.where(v > 1.0, -1.0,
|
||
np.where(v < 0.0, 1.0,
|
||
np.where(v > 0.8, -1.0,
|
||
np.where(v < 0.2, 1.0, 0.0))))
|
||
|
||
conf = np.where((v > 1.0) | (v < 0.0), 65.0,
|
||
np.where((v > 0.8) | (v < 0.2), 45.0, 30.0))
|
||
|
||
return bb_pos, pd.Series(sig, index=close.index), pd.Series(conf, index=close.index)
|
||
|
||
# ── Stochastic ────────────────────────────────────────────────
|
||
|
||
def stochastic(
|
||
self, high: pd.Series, low: pd.Series, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series, pd.Series]:
|
||
"""Returns (k_line, signal [-1..+1], confidence)."""
|
||
tc = self.tc
|
||
ll = low.rolling(tc.stoch_k).min()
|
||
hh = high.rolling(tc.stoch_k).max()
|
||
denom = (hh - ll).replace(0, np.nan)
|
||
k = (100 * (close - ll) / denom).fillna(50.0)
|
||
d = k.rolling(tc.stoch_d).mean().fillna(50.0)
|
||
|
||
kv, dv = k.values, d.values
|
||
ob, os_ = tc.stoch_overbought, tc.stoch_oversold
|
||
|
||
sig = np.where((kv >= ob) & (dv >= ob), -1.0,
|
||
np.where((kv <= os_) & (dv <= os_), 1.0, 0.0))
|
||
|
||
conf = np.where((kv >= ob) & (dv >= ob), 60.0,
|
||
np.where((kv <= os_) & (dv <= os_), 60.0, 30.0))
|
||
|
||
return k, pd.Series(sig, index=close.index), pd.Series(conf, index=close.index)
|
||
|
||
# ── ADX ───────────────────────────────────────────────────────
|
||
|
||
def adx(
|
||
self, high: pd.Series, low: pd.Series, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series, pd.Series, pd.Series]:
|
||
"""Returns (adx, plus_di, minus_di, signal [-1..0..+1])."""
|
||
tc = self.tc
|
||
tr = pd.concat([
|
||
high - low,
|
||
(high - close.shift()).abs(),
|
||
(low - close.shift()).abs(),
|
||
], axis=1).max(axis=1)
|
||
|
||
p_dm = high.diff()
|
||
m_dm = -low.diff()
|
||
p_dm = p_dm.where((p_dm > m_dm) & (p_dm > 0), 0.0)
|
||
m_dm = m_dm.where((m_dm > p_dm) & (m_dm > 0), 0.0)
|
||
|
||
atr = self._wilder(tr, tc.adx_period)
|
||
pdi = 100 * self._wilder(p_dm, tc.adx_period) / atr
|
||
mdi = 100 * self._wilder(m_dm, tc.adx_period) / atr
|
||
dx = 100 * (pdi - mdi).abs() / (pdi + mdi).replace(0, np.nan)
|
||
adx_s = self._wilder(dx.fillna(0), tc.adx_period)
|
||
|
||
av, pv, mv = adx_s.values, pdi.values, mdi.values
|
||
trending = av >= tc.adx_strong_trend
|
||
sig = np.where(trending & (pv > mv), 1.0,
|
||
np.where(trending & (mv > pv), -1.0, 0.0))
|
||
|
||
adx_conf = np.where(av >= tc.adx_very_strong, 80.0,
|
||
np.where(av >= tc.adx_strong_trend, 65.0, 40.0))
|
||
|
||
return adx_s, pdi, mdi, pd.Series(sig, index=close.index), pd.Series(adx_conf, index=close.index)
|
||
|
||
# ── ATR ───────────────────────────────────────────────────────
|
||
|
||
def atr(self, high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series:
|
||
tc = self.tc
|
||
tr = pd.concat([
|
||
high - low,
|
||
(high - close.shift()).abs(),
|
||
(low - close.shift()).abs(),
|
||
], axis=1).max(axis=1)
|
||
return self._wilder(tr, tc.atr_period)
|
||
|
||
# ── Ichimoku ──────────────────────────────────────────────────
|
||
|
||
def ichimoku(
|
||
self, high: pd.Series, low: pd.Series, close: pd.Series
|
||
) -> Tuple[pd.Series, pd.Series]:
|
||
"""Returns (score [-4..+4], confidence)."""
|
||
tc = self.tc
|
||
tenkan = (high.rolling(tc.ichimoku_tenkan).max() +
|
||
low.rolling(tc.ichimoku_tenkan).min()) / 2
|
||
kijun = (high.rolling(tc.ichimoku_kijun).max() +
|
||
low.rolling(tc.ichimoku_kijun).min()) / 2
|
||
sa = (tenkan + kijun) / 2
|
||
sb = (high.rolling(tc.ichimoku_senkou_b).max() +
|
||
low.rolling(tc.ichimoku_senkou_b).min()) / 2
|
||
|
||
cloud_top = np.maximum(sa.values, sb.values)
|
||
cloud_bot = np.minimum(sa.values, sb.values)
|
||
cv, tv, kv = close.values, tenkan.values, kijun.values
|
||
|
||
price_score = np.where(cv > cloud_top, 2.0,
|
||
np.where(cv < cloud_bot, -2.0, 0.0))
|
||
tk_score = np.sign(tv - kv).astype(float)
|
||
cloud_score = np.sign(sa.values - sb.values).astype(float)
|
||
total = price_score + tk_score + cloud_score
|
||
|
||
conf = np.where(np.abs(total) >= 3, 75.0,
|
||
np.where(np.abs(total) >= 1, 55.0, 35.0))
|
||
|
||
return (
|
||
pd.Series(total, index=close.index),
|
||
pd.Series(conf, index=close.index),
|
||
)
|
||
|
||
# ── Master compute ────────────────────────────────────────────
|
||
|
||
def compute_all(
|
||
self, df: pd.DataFrame, disabled: Optional[List[str]] = None
|
||
) -> Dict[str, pd.Series]:
|
||
"""
|
||
Run all indicators on the full DataFrame, return a flat series dict.
|
||
Keys: ema_norm, ema_conf, ema200, rsi_val, rsi_sig, rsi_conf,
|
||
macd_hist, macd_sig, macd_conf, bb_pos, bb_sig, bb_conf,
|
||
stoch_k, stoch_sig, stoch_conf, adx_val, adx_conf, pdi, mdi,
|
||
adx_sig, atr, ichi_score, ichi_conf.
|
||
"""
|
||
disabled = set(disabled or [])
|
||
c, h, lo = df["Close"], df["High"], df["Low"]
|
||
out: Dict[str, pd.Series] = {}
|
||
|
||
if "ema_trend" not in disabled:
|
||
norm, ema_c, ema200 = self.ema_alignment(c)
|
||
out["ema_norm"] = norm
|
||
out["ema_conf"] = ema_c
|
||
out["ema200"] = ema200
|
||
|
||
if "rsi" not in disabled:
|
||
rv, rs, rc = self.rsi(c)
|
||
out["rsi_val"] = rv
|
||
out["rsi_sig"] = rs
|
||
out["rsi_conf"] = rc
|
||
|
||
if "macd" not in disabled:
|
||
mh, ms, mc = self.macd(c)
|
||
out["macd_hist"] = mh
|
||
out["macd_sig"] = ms
|
||
out["macd_conf"] = mc
|
||
|
||
if "bollinger" not in disabled:
|
||
bp, bs, bc = self.bollinger(c)
|
||
out["bb_pos"] = bp
|
||
out["bb_sig"] = bs
|
||
out["bb_conf"] = bc
|
||
|
||
if "stochastic" not in disabled:
|
||
sk, ss, sc = self.stochastic(h, lo, c)
|
||
out["stoch_k"] = sk
|
||
out["stoch_sig"] = ss
|
||
out["stoch_conf"] = sc
|
||
|
||
if "adx" not in disabled:
|
||
av, pdi, mdi, as_, ac = self.adx(h, lo, c)
|
||
out["adx_val"] = av
|
||
out["adx_conf"] = ac
|
||
out["pdi"] = pdi
|
||
out["mdi"] = mdi
|
||
out["adx_sig"] = as_
|
||
|
||
out["atr"] = self.atr(h, lo, c)
|
||
|
||
if "ichimoku" not in disabled:
|
||
is_, ic = self.ichimoku(h, lo, c)
|
||
out["ichi_score"] = is_
|
||
out["ichi_conf"] = ic
|
||
|
||
return out
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# VECTORISED COMPOSITE SCORING (GPU or NumPy)
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
# Per-indicator base weights
|
||
# Optimised 2026-05-07 via 36-combo sweep on H4 2022-2026 local data:
|
||
# ema_trend ×3, macd ×3, stochastic ×2, adx ×0.5, ichimoku ×0.5, rsi=0, bollinger=0
|
||
# Combined result: +21.30% / -8.52% DD / Sharpe 0.47 vs baseline +7.12% / -9.05% DD / 0.23
|
||
# Original weights (engine.py defaults, pre-optimisation):
|
||
# ema_trend=1.2, rsi=0.3, macd=1.1, bollinger=0.9, stochastic=0.8, adx=1.0, ichimoku=1.1
|
||
_BASE_W = {
|
||
"ema_trend": 3.6,
|
||
"macd": 3.3,
|
||
"stochastic": 1.6,
|
||
"adx": 0.5,
|
||
"ichimoku": 0.55,
|
||
"rsi": 0.0, # ablated — net drag on H4 2022-2026
|
||
"bollinger": 0.0, # ablated — counter-trend noise on trending pair
|
||
}
|
||
|
||
# Regime-based weight multipliers — matches engine._compute_composite
|
||
_TREND_MULTS = {
|
||
"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "ichimoku": 1.3,
|
||
"rsi": 0.7, "bollinger": 0.6, "stochastic": 1.0,
|
||
}
|
||
_RANGE_MULTS = {
|
||
"rsi": 1.4, "stochastic": 1.3, "bollinger": 1.4,
|
||
"ema_trend": 0.6, "adx": 0.5, "ichimoku": 0.2, "macd": 1.0,
|
||
}
|
||
|
||
|
||
def compute_bar_scores(
|
||
indicators: Dict[str, pd.Series],
|
||
disabled: Optional[List[str]] = None,
|
||
weights: Optional[Dict[str, float]] = None,
|
||
) -> Tuple[np.ndarray, np.ndarray]:
|
||
"""
|
||
Vectorised composite scoring — same formula as engine._compute_composite.
|
||
|
||
composite_score = (Σ signal × weight × regime_mult × conf/100) / Σ weight × 50
|
||
Clipped to [-100, +100].
|
||
|
||
weights: override per-indicator base weights (default: _BASE_W).
|
||
Returns (score_arr, confidence_arr) shape (N,).
|
||
"""
|
||
disabled = set(disabled or [])
|
||
base_w = weights if weights is not None else _BASE_W
|
||
n = len(next(iter(indicators.values())))
|
||
xp = cp # CuPy or NumPy alias
|
||
|
||
# Determine regime per bar from ADX + EMA direction
|
||
if "adx_val" in indicators and "ema_norm" in indicators:
|
||
adx_v = xp.asarray(indicators["adx_val"].values)
|
||
is_trending = adx_v >= 25.0
|
||
else:
|
||
is_trending = xp.zeros(n, dtype=bool)
|
||
|
||
total_score = xp.zeros(n, dtype=float)
|
||
total_weight = xp.zeros(n, dtype=float)
|
||
total_conf_w = xp.zeros(n, dtype=float)
|
||
|
||
def _add(name: str, sig_arr: np.ndarray, conf_arr: np.ndarray):
|
||
nonlocal total_score, total_weight, total_conf_w
|
||
if name in disabled:
|
||
return
|
||
bw = base_w.get(name, 1.0)
|
||
if bw == 0.0:
|
||
return
|
||
tm = _TREND_MULTS.get(name, 1.0)
|
||
rm = _RANGE_MULTS.get(name, 1.0)
|
||
regime_mult = xp.where(is_trending, float(tm), float(rm))
|
||
w = bw * regime_mult
|
||
s = xp.asarray(sig_arr)
|
||
c = xp.asarray(conf_arr)
|
||
total_score = total_score + xp.nan_to_num(s * w * (c / 100.0))
|
||
total_weight = total_weight + xp.nan_to_num(w)
|
||
total_conf_w = total_conf_w + xp.nan_to_num(c * w)
|
||
|
||
# EMA trend: map norm → signal integer matching engine thresholds
|
||
if "ema_norm" in indicators:
|
||
en = xp.asarray(indicators["ema_norm"].values)
|
||
ema_sig = xp.where(en > 0.6, 2.0,
|
||
xp.where(en > 0.2, 1.0,
|
||
xp.where(en < -0.6, -2.0,
|
||
xp.where(en < -0.2, -1.0, 0.0))))
|
||
_add("ema_trend", ema_sig, xp.asarray(indicators["ema_conf"].values))
|
||
|
||
if "rsi_sig" in indicators:
|
||
_add("rsi", indicators["rsi_sig"].values, indicators["rsi_conf"].values)
|
||
|
||
if "macd_sig" in indicators:
|
||
_add("macd", indicators["macd_sig"].values, indicators["macd_conf"].values)
|
||
|
||
if "bb_sig" in indicators:
|
||
_add("bollinger", indicators["bb_sig"].values, indicators["bb_conf"].values)
|
||
|
||
if "stoch_sig" in indicators:
|
||
_add("stochastic", indicators["stoch_sig"].values, indicators["stoch_conf"].values)
|
||
|
||
if "adx_sig" in indicators:
|
||
_add("adx", indicators["adx_sig"].values, indicators["adx_conf"].values)
|
||
|
||
if "ichi_score" in indicators:
|
||
ichi_s = xp.asarray(indicators["ichi_score"].values)
|
||
ichi_sig = xp.clip(ichi_s / 2.0, -2.0, 2.0)
|
||
_add("ichimoku", ichi_sig, xp.asarray(indicators["ichi_conf"].values))
|
||
|
||
safe_w = xp.where(total_weight > 0, total_weight, 1.0)
|
||
score = xp.clip((total_score / safe_w) * 50.0, -100.0, 100.0)
|
||
conf = xp.where(total_weight > 0, total_conf_w / total_weight, 0.0)
|
||
|
||
# Bring results back to NumPy if we used CuPy
|
||
if _GPU:
|
||
score = cp.asnumpy(score)
|
||
conf = cp.asnumpy(conf)
|
||
return score.astype(float), conf.astype(float)
|
||
|
||
|
||
def align_to_primary(
|
||
secondary_series: pd.Series,
|
||
secondary_tf_min: int,
|
||
primary_index: pd.DatetimeIndex,
|
||
primary_tf_min: int,
|
||
) -> pd.Series:
|
||
"""
|
||
Forward-fill a lower-frequency signal to the primary TF index without
|
||
lookahead. At primary bar T (close = T + primary_tf_min), we use the
|
||
most recent secondary bar whose close time ≤ T + primary_tf_min.
|
||
"""
|
||
sec_close_idx = secondary_series.index + pd.Timedelta(minutes=secondary_tf_min)
|
||
pri_close_idx = primary_index + pd.Timedelta(minutes=primary_tf_min)
|
||
sec_at_close = pd.Series(secondary_series.values, index=sec_close_idx)
|
||
aligned = sec_at_close.reindex(pri_close_idx, method="ffill")
|
||
aligned.index = primary_index
|
||
return aligned
|
||
|
||
|
||
def compute_final_scores(
|
||
primary_score: np.ndarray,
|
||
primary_conf: np.ndarray,
|
||
m15_score: Optional[np.ndarray],
|
||
h4_score: Optional[np.ndarray],
|
||
ema200: Optional[pd.Series],
|
||
close: pd.Series,
|
||
ema_lo: float = 0.5,
|
||
ema_hi: float = 1.5,
|
||
ema_mult: float = 0.55,
|
||
) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
|
||
"""
|
||
Apply MTF blending, MTF agreement bonus/penalty, and EMA200 distance gate.
|
||
Mirrors signal_aggregator._compute_final_signal exactly.
|
||
|
||
Returns (final_score, final_conf, mtf_agree [bool], signal_int [-2..+2]).
|
||
"""
|
||
score = primary_score.copy()
|
||
conf = primary_conf.copy()
|
||
n = len(score)
|
||
mtf_agree = np.zeros(n, dtype=bool)
|
||
|
||
if m15_score is not None and h4_score is not None:
|
||
pri_bull = primary_score > 10
|
||
pri_bear = primary_score < -10
|
||
m15_bull = m15_score > 10
|
||
m15_bear = m15_score < -10
|
||
h4_bull = h4_score > 10
|
||
h4_bear = h4_score < -10
|
||
|
||
bull_ct = pri_bull.astype(int) + m15_bull.astype(int) + h4_bull.astype(int)
|
||
bear_ct = pri_bear.astype(int) + m15_bear.astype(int) + h4_bear.astype(int)
|
||
mtf_agree = (bull_ct >= 2) | (bear_ct >= 2)
|
||
|
||
# MTF blending (same weights as signal_aggregator)
|
||
all_scores = np.stack([primary_score, m15_score, h4_score], axis=1)
|
||
mtf_avg = all_scores.mean(axis=1)
|
||
score = np.where(mtf_agree,
|
||
0.6 * primary_score + 0.4 * mtf_avg,
|
||
primary_score)
|
||
conf = np.where(mtf_agree,
|
||
np.minimum(conf + 15.0, 95.0),
|
||
conf * 0.7)
|
||
|
||
# EMA200 distance gate
|
||
if ema200 is not None:
|
||
e200 = ema200.values
|
||
cv = close.values
|
||
valid = e200 > 0
|
||
ema_dist = np.where(valid, np.abs(cv - e200) / e200 * 100.0, 999.0)
|
||
gate_on = (ema_dist < ema_lo) | (ema_dist > ema_hi)
|
||
conf = np.where(gate_on, conf * ema_mult, conf)
|
||
|
||
# Signal integer from thresholds (matches signal_aggregator._compute_final_signal)
|
||
sig = np.where((score >= 40) & (conf >= 65), 2,
|
||
np.where((score >= 15) & (conf >= 50), 1,
|
||
np.where((score <= -40) & (conf >= 65), -2,
|
||
np.where((score <= -15) & (conf >= 50), -1, 0)))).astype(float)
|
||
|
||
return score, conf, mtf_agree, sig
|
||
|
||
|
||
def compute_grades(
|
||
score: np.ndarray,
|
||
conf: np.ndarray,
|
||
mtf_agree: np.ndarray,
|
||
) -> np.ndarray:
|
||
"""Assign quality grades — same thresholds as signal_aggregator."""
|
||
abs_s = np.abs(score)
|
||
grades = np.full(len(score), "F", dtype=object)
|
||
# Apply lowest-to-highest priority (each overwrites the one below)
|
||
grades = np.where(abs_s >= 10, "D", grades)
|
||
grades = np.where((abs_s >= 15) & (conf >= 45), "C", grades)
|
||
grades = np.where((abs_s >= 25) & (conf >= 55), "B", grades)
|
||
grades = np.where((abs_s >= 35) & (conf >= 65) & mtf_agree, "A", grades)
|
||
grades = np.where((abs_s >= 50) & (conf >= 75) & mtf_agree, "A+", grades)
|
||
return grades
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# SEQUENTIAL STATE MACHINE
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def _bar_close(open_time: pd.Timestamp, tf_min: int) -> pd.Timestamp:
|
||
return open_time + pd.Timedelta(minutes=tf_min)
|
||
|
||
|
||
def _check_exit(pos: dict, bar: pd.Series, bt) -> Optional[dict]:
|
||
"""Mirror of WalkForwardBacktester._check_exit."""
|
||
high = float(bar["High"])
|
||
low = float(bar["Low"])
|
||
sl, tp = pos["sl"], pos["tp"]
|
||
slip = bt.slippage_pips * PIP_VALUE
|
||
half = (0.5 if bt.use_spread_cost else 0.0) * PIP_VALUE
|
||
|
||
if pos["direction"] == "BUY":
|
||
sl_hit = low <= sl
|
||
tp_hit = high >= tp
|
||
if sl_hit and tp_hit:
|
||
return {"price": sl - slip - half, "reason": "sl"} if bt.sl_first_on_tie \
|
||
else {"price": tp - slip - half, "reason": "tp"}
|
||
if sl_hit: return {"price": sl - slip - half, "reason": "sl"}
|
||
if tp_hit: return {"price": tp - slip - half, "reason": "tp"}
|
||
else:
|
||
sl_hit = high >= sl
|
||
tp_hit = low <= tp
|
||
if sl_hit and tp_hit:
|
||
return {"price": sl + slip + half, "reason": "sl"} if bt.sl_first_on_tie \
|
||
else {"price": tp + slip + half, "reason": "tp"}
|
||
if sl_hit: return {"price": sl + slip + half, "reason": "sl"}
|
||
if tp_hit: return {"price": tp + slip + half, "reason": "tp"}
|
||
return None
|
||
|
||
|
||
def _open_trade(
|
||
sig: int, score: float, conf: float, grade: str,
|
||
atr_v: float, next_bar: pd.Series, balance: float,
|
||
config: PlatformConfig,
|
||
) -> dict:
|
||
bt = config.backtest
|
||
sc = config.signals
|
||
|
||
raw_entry = float(next_bar["Open"])
|
||
direction = "BUY" if sig > 0 else "SELL"
|
||
if bt.invert_signal:
|
||
direction = "SELL" if direction == "BUY" else "BUY"
|
||
|
||
cost_price = (0.5 if bt.use_spread_cost else 0.0 + bt.slippage_pips) * PIP_VALUE
|
||
entry = raw_entry + cost_price if direction == "BUY" else raw_entry - cost_price
|
||
|
||
sl_dist = atr_v * sc.atr_sl_multiplier
|
||
tp_dist = sl_dist * sc.default_rr_ratio
|
||
risk_pips = max(sl_dist / PIP_VALUE, 1e-9)
|
||
|
||
if direction == "BUY":
|
||
sl = entry - sl_dist
|
||
tp = entry + tp_dist
|
||
else:
|
||
sl = entry + sl_dist
|
||
tp = entry - tp_dist
|
||
|
||
risk_amt = balance * (bt.risk_per_trade_pct / 100.0)
|
||
pip_val = risk_amt / risk_pips
|
||
|
||
entry_ts = next_bar.name
|
||
entry_hr = int(entry_ts.hour) if hasattr(entry_ts, "hour") else -1
|
||
entry_dow = int(entry_ts.dayofweek) if hasattr(entry_ts, "dayofweek") else -1
|
||
|
||
return {
|
||
"direction": direction,
|
||
"entry_time": next_bar.name,
|
||
"entry_price": entry,
|
||
"sl": sl, "tp": tp,
|
||
"risk_pips": risk_pips,
|
||
"pip_currency_value": pip_val,
|
||
"balance_at_open": balance,
|
||
"grade": grade, "score": score, "conf": conf,
|
||
"entry_hr": entry_hr, "entry_dow": entry_dow,
|
||
}
|
||
|
||
|
||
def _close_trade(
|
||
pos: dict, exit_time: pd.Timestamp, exit_info: dict,
|
||
balance: float, config: PlatformConfig,
|
||
) -> Trade:
|
||
exit_price = exit_info["price"]
|
||
reason = exit_info["reason"]
|
||
|
||
if pos["direction"] == "BUY":
|
||
pnl_pips = (exit_price - pos["entry_price"]) / PIP_VALUE
|
||
else:
|
||
pnl_pips = (pos["entry_price"] - exit_price) / PIP_VALUE
|
||
|
||
pnl_currency = pnl_pips * pos["pip_currency_value"]
|
||
pnl_pct = pnl_currency / pos["balance_at_open"] * 100.0
|
||
|
||
tf_min = TF_MINUTES[config.primary_timeframe]
|
||
dur_min = (exit_time - pos["entry_time"]).total_seconds() / 60.0
|
||
dur_bars = int(round(dur_min / tf_min))
|
||
|
||
return Trade(
|
||
entry_time=pos["entry_time"],
|
||
exit_time=exit_time,
|
||
direction=pos["direction"],
|
||
entry_price=pos["entry_price"],
|
||
exit_price=exit_price,
|
||
stop_loss=pos["sl"],
|
||
take_profit=pos["tp"],
|
||
exit_reason=reason,
|
||
risk_pips=pos["risk_pips"],
|
||
pnl_pips=pnl_pips,
|
||
pnl_pct=pnl_pct,
|
||
pnl_currency=pnl_currency,
|
||
duration_bars=dur_bars,
|
||
grade=pos["grade"],
|
||
score_at_entry=pos["score"],
|
||
confidence_at_entry=pos["conf"],
|
||
entry_hour=pos["entry_hr"],
|
||
entry_dow=pos["entry_dow"],
|
||
)
|
||
|
||
|
||
def _regime_passes(
|
||
decision_time: pd.Timestamp,
|
||
regime_adx: Optional[pd.Series],
|
||
bt,
|
||
regime_tf_min: int,
|
||
) -> bool:
|
||
if not bt.regime_filter:
|
||
return True
|
||
if regime_adx is None or regime_adx.empty:
|
||
return False
|
||
close_times = regime_adx.index + pd.Timedelta(minutes=regime_tf_min)
|
||
valid = regime_adx[close_times <= decision_time]
|
||
if valid.empty:
|
||
return False
|
||
last = float(valid.iloc[-1])
|
||
return not np.isnan(last) and last >= bt.regime_filter_min_adx
|
||
|
||
|
||
def _session_passes(
|
||
decision_time: pd.Timestamp,
|
||
recent_pnl: deque,
|
||
bt,
|
||
) -> bool:
|
||
active = bt.session_filter
|
||
if not active and bt.auto_session_filter:
|
||
if len(recent_pnl) >= bt.auto_session_lookback_trades:
|
||
mean_pnl = sum(recent_pnl) / len(recent_pnl)
|
||
active = mean_pnl < bt.auto_session_threshold_pct
|
||
if not active:
|
||
return True
|
||
return bt.session_start_hour <= int(decision_time.hour) < bt.session_end_hour
|
||
|
||
|
||
def run_state_machine(
|
||
primary_df: pd.DataFrame,
|
||
final_score: np.ndarray,
|
||
final_conf: np.ndarray,
|
||
final_sig: np.ndarray,
|
||
grades: np.ndarray,
|
||
atr_arr: np.ndarray,
|
||
regime_adx: Optional[pd.Series],
|
||
config: PlatformConfig,
|
||
warmup: int,
|
||
progress: bool = True,
|
||
) -> BacktestResult:
|
||
"""Sequential position management over pre-computed signal arrays."""
|
||
bt = config.backtest
|
||
ptf = config.primary_timeframe
|
||
ptf_min = TF_MINUTES[ptf]
|
||
rtf_min = TF_MINUTES[bt.regime_filter_timeframe]
|
||
|
||
balance = bt.initial_balance
|
||
equity_points = [(primary_df.index[warmup], balance)]
|
||
trades: List[Trade] = []
|
||
recent_pnl = deque(maxlen=max(1, bt.auto_session_lookback_trades))
|
||
open_pos: Optional[dict] = None
|
||
|
||
iterator = range(warmup, len(primary_df) - 1)
|
||
if progress:
|
||
try:
|
||
from tqdm import tqdm # type: ignore
|
||
iterator = tqdm(iterator, desc="Replaying bars", unit="bar", leave=False)
|
||
except ImportError:
|
||
pass
|
||
|
||
for t in iterator:
|
||
cur_bar = primary_df.iloc[t]
|
||
decision_time = _bar_close(primary_df.index[t], ptf_min)
|
||
|
||
# ── Manage open position ─────────────────────────────────
|
||
if open_pos is not None:
|
||
if cur_bar.name != open_pos["entry_time"]:
|
||
exit_info = _check_exit(open_pos, cur_bar, bt)
|
||
if exit_info:
|
||
closed = _close_trade(open_pos, cur_bar.name, exit_info, balance, config)
|
||
trades.append(closed)
|
||
balance += closed.pnl_currency
|
||
equity_points.append((cur_bar.name, balance))
|
||
recent_pnl.append(float(closed.pnl_pct))
|
||
open_pos = None
|
||
|
||
# ── Open new position ────────────────────────────────────
|
||
if open_pos is None:
|
||
grade = str(grades[t])
|
||
sig = int(final_sig[t])
|
||
score = float(final_score[t])
|
||
conf = float(final_conf[t])
|
||
atr_v = float(atr_arr[t])
|
||
|
||
if not _grade_passes(grade, bt.min_trade_grade): continue
|
||
if abs(score) < bt.min_abs_score: continue
|
||
if sig == 0: continue
|
||
if atr_v == 0 or np.isnan(atr_v): continue
|
||
if t + 1 >= len(primary_df): continue
|
||
if not _regime_passes(decision_time, regime_adx, bt, rtf_min): continue
|
||
if not _session_passes(decision_time, recent_pnl, bt): continue
|
||
|
||
next_bar = primary_df.iloc[t + 1]
|
||
open_pos = _open_trade(sig, score, conf, grade, atr_v, next_bar, balance, config)
|
||
|
||
# Force-close remaining position at end of data
|
||
if open_pos is not None:
|
||
last = primary_df.iloc[-1]
|
||
closed = _close_trade(
|
||
open_pos, last.name,
|
||
{"price": float(last["Close"]), "reason": "end_of_data"},
|
||
balance, config,
|
||
)
|
||
trades.append(closed)
|
||
balance += closed.pnl_currency
|
||
equity_points.append((last.name, balance))
|
||
|
||
eq_idx = pd.DatetimeIndex([p[0] for p in equity_points])
|
||
eq_val = [p[1] for p in equity_points]
|
||
equity_curve = pd.Series(eq_val, index=eq_idx)
|
||
|
||
elapsed = None
|
||
if len(equity_curve) >= 2:
|
||
elapsed = (equity_curve.index[-1] - equity_curve.index[0]).total_seconds() / 86400
|
||
|
||
metrics = compute_metrics(trades, equity_curve, bt.initial_balance, elapsed_days=elapsed)
|
||
return BacktestResult(
|
||
trades=trades,
|
||
equity_curve=equity_curve,
|
||
initial_balance=bt.initial_balance,
|
||
final_balance=balance,
|
||
metrics=metrics,
|
||
)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# GPU BACKTESTER (orchestrator)
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
class GPUBacktester:
|
||
"""
|
||
Phase 1 — Vectorised indicator computation (GPU/NumPy, O(N))
|
||
Phase 2 — Sequential state machine (CPU, O(N), unavoidable)
|
||
"""
|
||
|
||
def __init__(self, config: PlatformConfig):
|
||
self.config = config
|
||
self.dp = ForexDataProvider(config)
|
||
self.vingine = VectorIndicators(config)
|
||
|
||
def load_data(
|
||
self,
|
||
use_local: bool = False,
|
||
data_dir: str = "market_data",
|
||
) -> Dict[Timeframe, pd.DataFrame]:
|
||
bt = self.config.backtest
|
||
timeframes = [self.config.primary_timeframe] + list(self.config.confirmation_timeframes)
|
||
if bt.regime_filter and bt.regime_filter_timeframe not in timeframes:
|
||
timeframes.append(bt.regime_filter_timeframe)
|
||
|
||
dfs: Dict[Timeframe, pd.DataFrame] = {}
|
||
for tf in timeframes:
|
||
if use_local:
|
||
df = self.dp.load_local_csv(
|
||
timeframe=tf, data_dir=data_dir,
|
||
start_date=bt.start_date, end_date=bt.end_date,
|
||
)
|
||
if df is None or df.empty:
|
||
logger.warning(f"No local CSV for {tf.value} — skipping")
|
||
continue
|
||
else:
|
||
df = self.dp.fetch_full_history(
|
||
symbol=self.config.symbol, timeframe=tf,
|
||
max_bars=bt.max_bars,
|
||
start_date=bt.start_date,
|
||
end_date=bt.end_date,
|
||
)
|
||
if bt.start_date:
|
||
df = df[df.index >= pd.Timestamp(bt.start_date, tz="UTC")]
|
||
if bt.end_date:
|
||
df = df[df.index <= pd.Timestamp(bt.end_date, tz="UTC")]
|
||
dfs[tf] = df
|
||
return dfs
|
||
|
||
def prepare_arrays(self, dfs: Dict[Timeframe, pd.DataFrame]) -> dict:
|
||
"""
|
||
Phase 1 only — compute all indicator arrays and return them as a
|
||
reusable bundle. Call this once, then pass the bundle to
|
||
score_and_run() many times with different weights/disabled sets.
|
||
"""
|
||
bt = self.config.backtest
|
||
primary_tf = self.config.primary_timeframe
|
||
primary_df = dfs.get(primary_tf)
|
||
|
||
if primary_df is None or primary_df.empty:
|
||
return {}
|
||
|
||
warmup = max(
|
||
self.config.technical.ema_trend,
|
||
self.config.technical.ichimoku_senkou_b,
|
||
200,
|
||
) + 30
|
||
|
||
disabled = list(self.config.technical.disabled_indicators or [])
|
||
ptf_min = TF_MINUTES[primary_tf]
|
||
|
||
print(f" Phase 1 — Indicators [{_BACKEND}] on {len(primary_df):,} bars...")
|
||
primary_inds = self.vingine.compute_all(primary_df, disabled=[]) # compute all, disable at score time
|
||
|
||
# MTF raw indicator arrays (scored independently)
|
||
cf_inds_map: Dict[Timeframe, Dict[str, pd.Series]] = {}
|
||
for cf_tf in self.config.confirmation_timeframes:
|
||
cf_df = dfs.get(cf_tf)
|
||
if cf_df is not None and not cf_df.empty:
|
||
cf_inds_map[cf_tf] = self.vingine.compute_all(cf_df, disabled=[])
|
||
|
||
# Regime ADX
|
||
regime_adx: Optional[pd.Series] = None
|
||
if bt.regime_filter:
|
||
rf_df = dfs.get(bt.regime_filter_timeframe)
|
||
if rf_df is not None and not rf_df.empty:
|
||
rf_inds = self.vingine.compute_all(rf_df, disabled=[])
|
||
regime_adx = rf_inds.get("adx_val")
|
||
|
||
ema_lo = float(os.environ.get("FB2_EMA_LO", "0.5"))
|
||
ema_hi = float(os.environ.get("FB2_EMA_HI", "1.5"))
|
||
ema_mult = float(os.environ.get("FB2_EMA_MULT", "0.55"))
|
||
|
||
return {
|
||
"primary_df": primary_df,
|
||
"primary_inds": primary_inds,
|
||
"cf_inds_map": cf_inds_map,
|
||
"regime_adx": regime_adx,
|
||
"warmup": warmup,
|
||
"ptf_min": ptf_min,
|
||
"disabled_base": disabled,
|
||
"ema_lo": ema_lo,
|
||
"ema_hi": ema_hi,
|
||
"ema_mult": ema_mult,
|
||
}
|
||
|
||
def score_and_run(
|
||
self,
|
||
bundle: dict,
|
||
weights: Optional[Dict[str, float]] = None,
|
||
disabled: Optional[List[str]] = None,
|
||
progress: bool = False,
|
||
) -> BacktestResult:
|
||
"""
|
||
Phase 2 only — re-score bars with given weights/disabled set and
|
||
replay the state machine. Reuses pre-computed indicator arrays.
|
||
"""
|
||
if not bundle:
|
||
return BacktestResult(initial_balance=self.config.backtest.initial_balance)
|
||
|
||
primary_df = bundle["primary_df"]
|
||
primary_inds = bundle["primary_inds"]
|
||
cf_inds_map = bundle["cf_inds_map"]
|
||
regime_adx = bundle["regime_adx"]
|
||
warmup = bundle["warmup"]
|
||
ptf_min = bundle["ptf_min"]
|
||
ema_lo = bundle["ema_lo"]
|
||
ema_hi = bundle["ema_hi"]
|
||
ema_mult = bundle["ema_mult"]
|
||
|
||
dis = set(disabled if disabled is not None else bundle["disabled_base"])
|
||
|
||
primary_score, primary_conf = compute_bar_scores(
|
||
primary_inds, disabled=list(dis), weights=weights
|
||
)
|
||
|
||
# MTF scores with same weights/disabled
|
||
m15_score_arr: Optional[np.ndarray] = None
|
||
h4_score_arr: Optional[np.ndarray] = None
|
||
for cf_tf, cf_inds in cf_inds_map.items():
|
||
cf_score, _ = compute_bar_scores(cf_inds, disabled=list(dis), weights=weights)
|
||
cf_score_s = pd.Series(cf_score, index=cf_inds[next(iter(cf_inds))].index)
|
||
aligned = align_to_primary(cf_score_s, TF_MINUTES[cf_tf], primary_df.index, ptf_min)
|
||
if cf_tf == Timeframe.M15:
|
||
m15_score_arr = aligned.values.astype(float)
|
||
elif cf_tf == Timeframe.H4:
|
||
h4_score_arr = aligned.values.astype(float)
|
||
|
||
ema200_s = primary_inds.get("ema200")
|
||
final_score, final_conf, mtf_agree, final_sig = compute_final_scores(
|
||
primary_score, primary_conf,
|
||
m15_score_arr, h4_score_arr,
|
||
ema200_s, primary_df["Close"],
|
||
ema_lo, ema_hi, ema_mult,
|
||
)
|
||
grades = compute_grades(final_score, final_conf, mtf_agree)
|
||
atr_arr = primary_inds["atr"].values
|
||
|
||
return run_state_machine(
|
||
primary_df, final_score, final_conf, final_sig,
|
||
grades, atr_arr, regime_adx,
|
||
self.config, warmup, progress=progress,
|
||
)
|
||
|
||
def run(
|
||
self,
|
||
dfs=None,
|
||
progress: bool = True,
|
||
use_local: bool = False,
|
||
data_dir: str = "market_data",
|
||
) -> BacktestResult:
|
||
if dfs is None:
|
||
dfs = self.load_data(use_local=use_local, data_dir=data_dir)
|
||
|
||
bundle = self.prepare_arrays(dfs)
|
||
if not bundle:
|
||
return BacktestResult(initial_balance=self.config.backtest.initial_balance)
|
||
|
||
primary_df = bundle["primary_df"]
|
||
warmup = bundle["warmup"]
|
||
if len(primary_df) <= warmup + 1:
|
||
logger.warning(f"Insufficient data: {len(primary_df)} bars, need >{warmup}")
|
||
return BacktestResult(initial_balance=self.config.backtest.initial_balance)
|
||
|
||
replay_bars = len(primary_df) - warmup - 1
|
||
print(f" Phase 2 — Replaying {replay_bars:,} bars sequentially...")
|
||
|
||
return self.score_and_run(bundle, progress=progress)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# OPTIMIZATION SWEEP
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def _build_opt_combos() -> List[Tuple[str, Dict[str, float], List[str]]]:
|
||
"""
|
||
Build the full list of (label, weights_dict, disabled_list) tuples:
|
||
- 1 baseline run
|
||
- 7 ablation runs (each indicator off one at a time)
|
||
- 28 weight-multiplier runs (each indicator at ×0.5, ×1.5, ×2.0, ×3.0)
|
||
= 36 total combos.
|
||
"""
|
||
INDICATORS = list(_BASE_W.keys())
|
||
MULTS = [0.5, 1.5, 2.0, 3.0]
|
||
combos: List[Tuple[str, Dict[str, float], List[str]]] = []
|
||
|
||
combos.append(("baseline", _BASE_W.copy(), []))
|
||
|
||
for name in INDICATORS:
|
||
combos.append((f"ablate {name}", _BASE_W.copy(), [name]))
|
||
|
||
for name in INDICATORS:
|
||
for m in MULTS:
|
||
w = _BASE_W.copy()
|
||
w[name] = round(w[name] * m, 4)
|
||
combos.append((f"{name} x{m}", w, []))
|
||
|
||
return combos
|
||
|
||
|
||
def run_optimization(bt: GPUBacktester, dfs: Dict[Timeframe, pd.DataFrame]) -> None:
|
||
"""
|
||
Call prepare_arrays() once, then score_and_run() for every combo.
|
||
Prints a progress line per combo and a final sorted results table.
|
||
"""
|
||
combos = _build_opt_combos()
|
||
|
||
print(f"\n Optimization sweep — {len(combos)} combos")
|
||
print(f" Computing indicator arrays once (Phase 1)...")
|
||
bundle = bt.prepare_arrays(dfs)
|
||
if not bundle:
|
||
print(" ERROR: No data loaded.")
|
||
return
|
||
|
||
primary_df = bundle["primary_df"]
|
||
warmup = bundle["warmup"]
|
||
if len(primary_df) <= warmup + 1:
|
||
print(f" ERROR: Only {len(primary_df)} bars — need >{warmup+1}")
|
||
return
|
||
|
||
print(f" Replaying {len(combos)} × {len(primary_df)-warmup-1:,} bars...\n")
|
||
|
||
rows = []
|
||
for i, (label, w, dis) in enumerate(combos, 1):
|
||
r = bt.score_and_run(bundle, weights=w, disabled=dis, progress=False)
|
||
m = r.metrics
|
||
ret = m.get("total_return_pct", 0.0)
|
||
rows.append({
|
||
"label": label,
|
||
"trades": m.get("total_trades", 0),
|
||
"win_rate": m.get("win_rate", 0.0),
|
||
"pf": m.get("profit_factor", 0.0),
|
||
"return": ret,
|
||
"max_dd": m.get("max_drawdown_pct", 0.0),
|
||
"sharpe": m.get("sharpe", 0.0),
|
||
})
|
||
sign = "+" if ret >= 0 else ""
|
||
print(f" [{i:2d}/{len(combos)}] {label:<30s} "
|
||
f"trades={rows[-1]['trades']:3d} "
|
||
f"WR={rows[-1]['win_rate']:5.1f}% "
|
||
f"ret={sign}{ret:.2f}%")
|
||
|
||
rows.sort(key=lambda r: r["return"], reverse=True)
|
||
|
||
W = max(max(len(r["label"]) for r in rows), 30)
|
||
sep = "─" * (W + 55)
|
||
|
||
print(f"\n {sep}")
|
||
print(f" {'Label':<{W}} {'Trades':>6} {'WR%':>6} {'PF':>5} "
|
||
f"{'Return%':>8} {'MaxDD%':>7} {'Sharpe':>7}")
|
||
print(f" {sep}")
|
||
for r in rows:
|
||
pf_s = " inf" if r["pf"] == float("inf") else f"{r['pf']:5.2f}"
|
||
print(f" {r['label']:<{W}} {r['trades']:>6} "
|
||
f"{r['win_rate']:>6.1f} {pf_s} "
|
||
f"{r['return']:>+8.2f} {r['max_dd']:>7.2f} {r['sharpe']:>7.2f}")
|
||
print(f" {sep}")
|
||
print(f" Best: {rows[0]['label']} return={rows[0]['return']:+.2f}%\n")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# CLI
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def parse_args():
|
||
p = argparse.ArgumentParser(
|
||
description="ForexBot2 GPU Backtester — vectorised O(N) indicator engine",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
Examples:
|
||
python backtest_gpu.py --start 2024-01-01
|
||
python backtest_gpu.py --start 2024-01-01 --regime-filter
|
||
python backtest_gpu.py --tf H4 --journal
|
||
python backtest_gpu.py --session-filter --journal-csv trades.csv
|
||
""",
|
||
)
|
||
p.add_argument("--tf", default="H1",
|
||
choices=["M5", "M15", "M30", "H1", "H4", "D1"])
|
||
p.add_argument("--bars", type=int, default=5000,
|
||
help="Max bars of history (default 5000)")
|
||
p.add_argument("--start", default=None, help="Start date YYYY-MM-DD")
|
||
p.add_argument("--end", default=None, help="End date YYYY-MM-DD")
|
||
p.add_argument("--min-grade", default="B",
|
||
choices=["A+", "A", "B", "C", "D", "F"])
|
||
p.add_argument("--regime-filter", action="store_true",
|
||
help="Gate signals on a higher-TF ADX trend confirmation")
|
||
p.add_argument("--regime-tf", default=None,
|
||
choices=["M15", "M30", "H1", "H4", "D1"],
|
||
help="Timeframe for regime ADX gate (default: one step above --tf)")
|
||
p.add_argument("--min-adx", type=float, default=25.0,
|
||
help="Min ADX for regime filter (default 25)")
|
||
p.add_argument("--session-filter", action="store_true",
|
||
help="Only trade UTC 07-16 (London + NY)")
|
||
p.add_argument("--session-start", type=int, default=7)
|
||
p.add_argument("--session-end", type=int, default=16)
|
||
p.add_argument("--journal", action="store_true",
|
||
help="Print last 10 trades")
|
||
p.add_argument("--journal-csv", default=None,
|
||
help="Write full trade journal to CSV")
|
||
p.add_argument("--local", action="store_true",
|
||
help="Load from local market_data/ CSVs instead of Twelve Data API")
|
||
p.add_argument("--data-dir", default="market_data",
|
||
help="Directory for local CSV files (default: market_data)")
|
||
p.add_argument("--optimize", action="store_true",
|
||
help="Sweep indicator weights and ablations (36 combos), print ranked table")
|
||
p.add_argument("--verbose", "-v", action="store_true")
|
||
return p.parse_args()
|
||
|
||
|
||
def build_config(args) -> PlatformConfig:
|
||
config = PlatformConfig()
|
||
tf_map = {
|
||
"M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30,
|
||
"H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1,
|
||
}
|
||
tf_conf = {
|
||
"M5": [Timeframe.M1, Timeframe.M15],
|
||
"M15": [Timeframe.M5, Timeframe.H1],
|
||
"M30": [Timeframe.M15, Timeframe.H1],
|
||
"H1": [Timeframe.M15, Timeframe.H4],
|
||
"H4": [Timeframe.H1, Timeframe.D1],
|
||
"D1": [Timeframe.H4, Timeframe.W1],
|
||
}
|
||
config.primary_timeframe = tf_map[args.tf]
|
||
config.confirmation_timeframes = tf_conf.get(args.tf, [Timeframe.M15, Timeframe.H4])
|
||
config.backtest.start_date = args.start
|
||
config.backtest.end_date = args.end
|
||
config.backtest.max_bars = args.bars
|
||
config.backtest.min_trade_grade = args.min_grade
|
||
config.backtest.regime_filter = args.regime_filter
|
||
config.backtest.regime_filter_min_adx = args.min_adx
|
||
# Regime TF: explicit flag > sensible default (one step above primary)
|
||
_default_regime_tf = {
|
||
"M5": "M15", "M15": "H1", "M30": "H1",
|
||
"H1": "H4", "H4": "D1", "D1": "D1",
|
||
}
|
||
regime_tf_str = args.regime_tf or _default_regime_tf.get(args.tf, "H4")
|
||
config.backtest.regime_filter_timeframe = tf_map[regime_tf_str]
|
||
config.backtest.session_filter = args.session_filter
|
||
config.backtest.session_start_hour = args.session_start
|
||
config.backtest.session_end_hour = args.session_end
|
||
return config
|
||
|
||
|
||
def print_results(result: BacktestResult, verbose_journal: bool = False):
|
||
m = result.metrics
|
||
ret = m.get("total_return_pct", 0)
|
||
pf = m.get("profit_factor", 0)
|
||
pf_s = "inf" if pf == float("inf") else f"{pf:.2f}"
|
||
|
||
print(f"\n {'─' * 76}")
|
||
print(f" Backtest Summary — USD/JPY [GPU Backtester]")
|
||
print(f" {'─' * 76}")
|
||
print(f" Trades: {m.get('total_trades', 0)}"
|
||
f" ({m.get('wins', 0)}W / {m.get('losses', 0)}L)")
|
||
print(f" Win Rate: {m.get('win_rate', 0):.1f}%")
|
||
print(f" Profit Factor: {pf_s}")
|
||
print(f" Expectancy: {m.get('expectancy_pips', 0):+.2f} pips/trade"
|
||
f" ({m.get('expectancy_per_trade', 0):+.2f} per trade)")
|
||
print(f" Initial Balance: ${result.initial_balance:,.2f}")
|
||
print(f" Final Balance: ${result.final_balance:,.2f}")
|
||
print(f" Total Return: {ret:+.2f}%")
|
||
print(f" CAGR: {m.get('cagr', 0):+.2f}%")
|
||
print(f" Max Drawdown: {m.get('max_drawdown_pct', 0):.2f}%")
|
||
print(f" Sharpe Ratio: {m.get('sharpe', 0):.2f}")
|
||
print(f" Sortino Ratio: {m.get('sortino', 0):.2f}")
|
||
print(f" Avg Duration: {m.get('avg_trade_duration_bars', 0):.1f} bars")
|
||
print(f" Win Streak: {m.get('longest_winning_streak', 0)}"
|
||
f" | Loss Streak: {m.get('longest_losing_streak', 0)}")
|
||
print(f" {'─' * 76}")
|
||
|
||
if verbose_journal and result.trades:
|
||
print(f"\n Trade Journal (last 10):")
|
||
print(f" {'#':<4} {'Entry Time':<22} {'Dir':<5} {'Entry':<9}"
|
||
f" {'Exit':<9} {'Reason':<6} {'Pips':>8} {'P&L%':>7}")
|
||
start = max(1, len(result.trades) - 9)
|
||
for i, t in enumerate(result.trades[-10:], start=start):
|
||
sign = "+" if t.pnl_currency > 0 else ""
|
||
print(f" {i:<4} {str(t.entry_time)[:19]:<22} {t.direction:<5}"
|
||
f" {t.entry_price:<9.3f} {t.exit_price:<9.3f} {t.exit_reason:<6}"
|
||
f" {t.pnl_pips:>+8.1f} {t.pnl_pct:>+6.2f}%")
|
||
|
||
|
||
def main():
|
||
args = parse_args()
|
||
logging.basicConfig(
|
||
level=logging.DEBUG if args.verbose else logging.WARNING,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
|
||
config = build_config(args)
|
||
bt = GPUBacktester(config)
|
||
|
||
print(f"\n ForexBot2 — GPU Backtester")
|
||
print(f" Backend : {_BACKEND}")
|
||
print(f" Data : {'Local CSV (' + args.data_dir + ')' if args.local else 'Twelve Data API'}")
|
||
print(f" Symbol : USD/JPY | TF: {args.tf}")
|
||
filters = []
|
||
if args.regime_filter:
|
||
rtf = config.backtest.regime_filter_timeframe.value
|
||
filters.append(f"regime {rtf} ADX≥{args.min_adx}")
|
||
if args.session_filter: filters.append(f"session ({args.session_start:02}-{args.session_end:02} UTC)")
|
||
if filters: print(f" Filters : {', '.join(filters)}")
|
||
print(f" {'─' * 40}")
|
||
|
||
_load_kw = dict(use_local=args.local, data_dir=args.data_dir)
|
||
|
||
if args.optimize:
|
||
dfs = bt.load_data(**_load_kw)
|
||
run_optimization(bt, dfs)
|
||
print()
|
||
return
|
||
|
||
result = bt.run(**_load_kw)
|
||
print_results(result, verbose_journal=args.journal)
|
||
|
||
if args.journal_csv and result.trades:
|
||
df_out = trades_to_dataframe(result.trades)
|
||
df_out.to_csv(args.journal_csv, index=False)
|
||
print(f"\n Trade journal written to {args.journal_csv}")
|
||
|
||
print()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|