harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Audit fixes: mixed-regime bias, live==backtest session parity, parallel replay

Browse Source 
Fix 1 — mixed-"trending" regime (ADX strong, EMAs unaligned) was
hard-coded bullish in 5 sites: RSI reinterpretation used the uptrend
playbook (RSI 38-55 -> STRONG_BUY conf 75), BUY-only confidence
bonuses, BUY-only extension penalties, no counter-trend penalty.
Now: extremes-only RSI (as transitioning), no directional bonuses,
symmetric extension penalties. 6yr validation: H1 +67.2->+61.9R
(mixed skew 16B/11S -> 4B/7S), H4 +31.6->+27.5R (the 4 bias BUYs),
M15 +152.6->+154.7R, M5 +207.1->+206.1R with maxDD 28.7->24.3R.
The ~8R aggregate give-back is bull-market subsidy from the bug.

Fix 2 — live session filter now applies the EXACT replay-loop
predicate (session window + blocked hours on the signal bar's CLOSE
time, from the bar timestamp not wall clock). Replaces a wall-clock
filter gated on a "XAU STRATEGY" marker that no code writes, which
made live diverge from every validated book.

Fix 3 — delete dead _check_xau_oversold_bounce and
_generate_xau_trade_setup (unreachable; the latter hardcoded BUY and
skipped the spread/RR gates).

Backtester: parallel bar replay via forked workers (--workers,
default auto = cores-2 capped 12). Per-bar evaluation is pure; the
entry cooldown is applied in an ordered phase-2 pass. Proven
byte-identical to sequential on the same window; ~300 -> 1030-1680
bars/s (6yr M5: ~26min -> 5-8min). --workers 1 = sequential path.

Tests: 179 passed (new mixed-regime RSI regression test).
This commit is contained in:
Harwinder Sekhon
2026-07-11 22:57:37 -04:00
parent 6f69b9dd90
commit dde41fbeb8
5 changed files with 282 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtest/gpu_backtest.py
+212 -142
View File
@@ -31,6 +31,7 @@ from __future__ import annotations
import argparse
import logging
import multiprocessing as mp
import os
import sys
import time
@@ -1458,6 +1459,13 @@ class BacktestConfig:
# Round-trip transaction cost in price units, charged on every trade
# (XAU/USD retail spread ≈ $0.30). 0 disables.
spread: float = 0.30
# Replay worker processes for the bar loop. 0 = auto (cpu_count - 2,
# capped at 12), 1 = sequential in-process. Signal evaluation per bar
# is pure (past data only), so it fans out across forked workers; the
# one piece of sequential state — the entry cooldown — is applied
# afterwards in a fast ordered pass. Results are identical to the
# sequential path by construction.
workers: int = 0
def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
@@ -1481,6 +1489,164 @@ def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Option
return None
# Shared context for replay workers. Populated in run_backtest before the
# fork; child processes inherit it copy-on-write, so nothing large is pickled.
_REPLAY_CTX: Dict = {}
def _replay_span(span: Tuple[int, int]):
"""Evaluate primary bars [start, stop) — everything except the cooldown.
Per-bar signal evaluation is pure: it sees only data up to the bar's
close, never prior trades. The single stateful rule (entry cooldown)
is applied by the caller in a sequential pass over the returned
candidates, so parallel and sequential runs produce identical books.
Must stay behaviorally identical to the original sequential loop.
"""
ctx = _REPLAY_CTX
primary_df = ctx["primary_df"]
dfs_full = ctx["dfs_full"]
timeframes = ctx["timeframes"]
aggregator = ctx["aggregator"]
platform_cfg = ctx["platform_cfg"]
bt_cfg = ctx["bt_cfg"]
primary_dur = ctx["primary_dur"]
skipped: Dict[str, int] = {}
candidates: List[Tuple[int, Trade]] = []
is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
for pos in range(span[0], span[1]):
ts = primary_df.index[pos]
# Decision time = close of the current primary bar.
decision_time = ts + primary_dur
# XAU calendar filter (matches live analyze_pair behavior).
if is_xau:
block = _is_xau_blocked_by_calendar(decision_time, platform_cfg)
if block:
skipped[block] = skipped.get(block, 0) + 1
continue
# Session window filter (matches the live session check in
# analyze_pair, which evaluates the signal bar's close time).
bt_filter = platform_cfg.backtest
if bt_filter.session_filter:
bar_hour = decision_time.hour
if not (bt_filter.session_start_hour <= bar_hour < bt_filter.session_end_hour):
skipped["SESSION_WINDOW"] = skipped.get("SESSION_WINDOW", 0) + 1
continue
_blocked = (platform_cfg.xau_strategy.blocked_hours_utc
if is_xau else bt_filter.blocked_hours_utc)
if bar_hour in _blocked:
skipped["SESSION_WINDOW"] = skipped.get("SESSION_WINDOW", 0) + 1
continue
# (Cooldown is applied by the caller — it is the only cross-bar state.)
# Build sliced dfs per TF — only bars fully closed by decision_time.
slice_dfs: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
full = dfs_full[tf]
bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf])
cutoff = decision_time - bar_dur
cut = int(full.index.searchsorted(cutoff, side="right"))
if cut == 0:
continue
slice_dfs[tf] = full.iloc[:cut]
if bt_cfg.primary_tf not in slice_dfs:
continue
report: SignalReport = aggregator.analyze_from_dataframes(
symbol=bt_cfg.pair,
dfs=slice_dfs,
context=None,
use_news=False,
)
if report.trade_setup is None:
continue
if report.signal_quality not in bt_cfg.quality_filter:
continue
# Strategy tag (from reasons[0] which the aggregator pre-pends).
tag = "GENERIC"
first_bull = report.bullish_reasons[0] if report.bullish_reasons else ""
first_bear = report.bearish_reasons[0] if report.bearish_reasons else ""
if "XAU PULLBACK" in first_bull or "XAU PULLBACK" in first_bear:
tag = "XAU_PB"
elif "XAU STRATEGY" in first_bull:
tag = "XAU"
elif "EUR STRATEGY" in first_bull or "EUR STRATEGY" in first_bear:
tag = "EUR"
elif "GBP STRATEGY" in first_bull:
tag = "GBP_PB"
if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only:
continue
ts_setup: TradeSetup = report.trade_setup
# Entry at the OPEN of the next bar (realistic fill).
entry_pos = pos + 1
if entry_pos >= len(primary_df):
continue
entry_time = primary_df.index[entry_pos]
entry_price = float(primary_df["Open"].iloc[entry_pos])
breakdown, agree, oppose, neutral, total = _build_indicator_breakdown(
report, ts_setup.direction
)
reasons = (report.bullish_reasons if ts_setup.direction == "BUY"
else report.bearish_reasons)
# Adjust SL/TP relative to actual entry (preserve setup distances).
sl_distance = abs(ts_setup.entry_price - ts_setup.stop_loss)
tp1_distance = abs(ts_setup.take_profit_1 - ts_setup.entry_price)
tp2_distance = abs(ts_setup.take_profit_2 - ts_setup.entry_price)
if ts_setup.direction == "BUY":
adj_sl = entry_price - sl_distance
adj_tp1 = entry_price + tp1_distance
adj_tp2 = entry_price + tp2_distance
else:
adj_sl = entry_price + sl_distance
adj_tp1 = entry_price - tp1_distance
adj_tp2 = entry_price - tp2_distance
candidates.append((pos, Trade(
pair=bt_cfg.pair,
entry_time=entry_time,
entry_price=entry_price,
sl=adj_sl,
tp1=adj_tp1,
tp2=adj_tp2,
direction=ts_setup.direction,
pip_value=ts_setup.pip_value,
risk_pips=sl_distance / ts_setup.pip_value,
reward_pips=tp1_distance / ts_setup.pip_value,
rr_planned=ts_setup.risk_reward,
quality=report.signal_quality,
final_score=report.final_score,
final_confidence=report.final_confidence,
market_regime=report.market_regime,
volatility_regime=report.volatility_regime,
strategy_tag=tag,
entry_reasons=list(reasons),
entry_warnings=list(report.warnings),
indicator_breakdown=breakdown,
indicators_agreeing=agree,
indicators_opposing=oppose,
indicators_neutral=neutral,
indicators_total=total,
trail_arm_r=(platform_cfg.signals.pb_trail_arm_r
if tag == "XAU_PB" else 0.0),
trail_dist_r=(platform_cfg.signals.pb_trail_dist_r
if tag == "XAU_PB" else 0.0),
)))
return skipped, candidates
def run_backtest(
bt_cfg: BacktestConfig,
platform_cfg: Optional[PlatformConfig] = None,
@@ -1549,10 +1715,6 @@ def run_backtest(
last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right"))
last_pos = min(last_pos, len(primary_df))
trades: List[Trade] = []
last_entry_pos = -10_000
skipped_reasons: Dict[str, int] = {}
logger.info(
f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: "
f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}"
@@ -1560,154 +1722,59 @@ def run_backtest(
primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf])
# ── Phase 1: evaluate every bar (parallelizable — no cross-bar state).
_REPLAY_CTX.update(
primary_df=primary_df, dfs_full=dfs_full, timeframes=timeframes,
aggregator=aggregator, platform_cfg=platform_cfg, bt_cfg=bt_cfg,
primary_dur=primary_dur,
)
n_bars = last_pos - first_eligible_pos
workers = (bt_cfg.workers if bt_cfg.workers > 0
else max(1, min((os.cpu_count() or 2) - 2, 12)))
if n_bars < 5_000:
workers = 1 # fork + merge overhead outweighs the gain
try:
fork_ctx = mp.get_context("fork") # COW-shares dfs/caches with workers
except ValueError:
workers = 1
fork_ctx = None
t_loop = time.perf_counter()
for pos in range(first_eligible_pos, last_pos):
ts = primary_df.index[pos]
# Decision time = close of the current primary bar.
decision_time = ts + primary_dur
if workers > 1:
n_chunks = workers * 4
bounds = np.linspace(first_eligible_pos, last_pos, n_chunks + 1, dtype=int)
spans = [(int(bounds[i]), int(bounds[i + 1]))
for i in range(n_chunks) if bounds[i] < bounds[i + 1]]
logger.info(f"Replay: {workers} worker processes, {len(spans)} spans")
with fork_ctx.Pool(processes=workers) as pool:
results = pool.map(_replay_span, spans)
else:
results = [_replay_span((first_eligible_pos, last_pos))]
# XAU calendar filter (matches live analyze_pair behavior).
# Live evaluates wall-clock time ≈ bar close, so filter on
# decision_time, not the bar's open timestamp.
if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD":
block = _is_xau_blocked_by_calendar(decision_time, platform_cfg)
if block:
skipped_reasons[block] = skipped_reasons.get(block, 0) + 1
continue
skipped_reasons: Dict[str, int] = {}
candidates: List[Tuple[int, Trade]] = []
for skips, cands in results:
for reason, count in skips.items():
skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
candidates.extend(cands)
candidates.sort(key=lambda pc: pc[0])
# Session window filter (matches live _apply_session_filter_live,
# which checks datetime.now() ≈ bar close = decision_time).
bt_filter = platform_cfg.backtest
if bt_filter.session_filter:
bar_hour = decision_time.hour
if not (bt_filter.session_start_hour <= bar_hour < bt_filter.session_end_hour):
skipped_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1
continue
# Blocked hours (live checks datetime.now().hour ≈ bar close).
# analyze_from_dataframes step 6b checks bar-open timestamp,
# which is one bar behind — misses the last 5min of each hour.
_is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
_blocked = (platform_cfg.xau_strategy.blocked_hours_utc
if _is_xau else bt_filter.blocked_hours_utc)
if bar_hour in _blocked:
skipped_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1
continue
# Cooldown.
# ── Phase 2: entry cooldown — the only sequential rule. A candidate
# inside the cooldown window of the previously ACCEPTED entry is
# dropped, exactly as the sequential loop skipped those bars.
trades: List[Trade] = []
last_entry_pos = -10_000
for pos, trade in candidates:
if pos - last_entry_pos < bt_cfg.cooldown_bars:
continue
# Build sliced dfs per TF — only include bars fully closed by
# decision_time. A bar timestamped T closes at T + bar_dur, so we
# keep bars where T + bar_dur <= decision_time → T <= decision_time - bar_dur.
# Without this, the H1 bar containing the current M15 close would be
# included with future high/low/close values: a look-ahead leak.
slice_dfs: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
full = dfs_full[tf]
bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf])
cutoff = decision_time - bar_dur
cut = int(full.index.searchsorted(cutoff, side="right"))
if cut == 0:
continue
slice_dfs[tf] = full.iloc[:cut]
if bt_cfg.primary_tf not in slice_dfs:
continue
report: SignalReport = aggregator.analyze_from_dataframes(
symbol=bt_cfg.pair,
dfs=slice_dfs,
context=None,
use_news=False,
)
if report.trade_setup is None:
continue
if report.signal_quality not in bt_cfg.quality_filter:
continue
# Strategy tag (from reasons[0] which the aggregator pre-pends).
# Pullback SELLs mark bearish_reasons, so check both lists.
tag = "GENERIC"
first_bull = report.bullish_reasons[0] if report.bullish_reasons else ""
first_bear = report.bearish_reasons[0] if report.bearish_reasons else ""
if "XAU PULLBACK" in first_bull or "XAU PULLBACK" in first_bear:
tag = "XAU_PB"
elif "XAU STRATEGY" in first_bull:
tag = "XAU"
elif "EUR STRATEGY" in first_bull or "EUR STRATEGY" in first_bear:
tag = "EUR"
elif "GBP STRATEGY" in first_bull:
tag = "GBP_PB"
if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only:
continue
ts_setup: TradeSetup = report.trade_setup
# Entry at the OPEN of the next bar — live signals fire at bar close,
# execution happens on the next candle open (realistic fill).
entry_pos = pos + 1
if entry_pos >= len(primary_df):
continue
entry_time = primary_df.index[entry_pos]
entry_price = float(primary_df["Open"].iloc[entry_pos])
breakdown, agree, oppose, neutral, total = _build_indicator_breakdown(
report, ts_setup.direction
)
reasons = (report.bullish_reasons if ts_setup.direction == "BUY"
else report.bearish_reasons)
# Adjust SL/TP relative to actual entry (preserve distance from setup).
sl_distance = abs(ts_setup.entry_price - ts_setup.stop_loss)
tp1_distance = abs(ts_setup.take_profit_1 - ts_setup.entry_price)
tp2_distance = abs(ts_setup.take_profit_2 - ts_setup.entry_price)
if ts_setup.direction == "BUY":
adj_sl = entry_price - sl_distance
adj_tp1 = entry_price + tp1_distance
adj_tp2 = entry_price + tp2_distance
else:
adj_sl = entry_price + sl_distance
adj_tp1 = entry_price - tp1_distance
adj_tp2 = entry_price - tp2_distance
trades.append(Trade(
pair=bt_cfg.pair,
entry_time=entry_time,
entry_price=entry_price,
sl=adj_sl,
tp1=adj_tp1,
tp2=adj_tp2,
direction=ts_setup.direction,
pip_value=ts_setup.pip_value,
risk_pips=sl_distance / ts_setup.pip_value,
reward_pips=tp1_distance / ts_setup.pip_value,
rr_planned=ts_setup.risk_reward,
quality=report.signal_quality,
final_score=report.final_score,
final_confidence=report.final_confidence,
market_regime=report.market_regime,
volatility_regime=report.volatility_regime,
strategy_tag=tag,
entry_reasons=list(reasons),
entry_warnings=list(report.warnings),
indicator_breakdown=breakdown,
indicators_agreeing=agree,
indicators_opposing=oppose,
indicators_neutral=neutral,
indicators_total=total,
trail_arm_r=(platform_cfg.signals.pb_trail_arm_r
if tag == "XAU_PB" else 0.0),
trail_dist_r=(platform_cfg.signals.pb_trail_dist_r
if tag == "XAU_PB" else 0.0),
))
trades.append(trade)
last_entry_pos = pos
logger.info(
f"Replay complete: {len(trades)} entries in "
f"{time.perf_counter() - t_loop:.1f}s "
f"({(last_pos - first_eligible_pos) / max(time.perf_counter() - t_loop, 1e-6):.0f} bars/s)"
f"({(last_pos - first_eligible_pos) / max(time.perf_counter() - t_loop, 1e-6):.0f} bars/s, "
f"{workers} worker{'s' if workers != 1 else ''})"
)
if skipped_reasons:
logger.info(f"Skipped bars by reason: {skipped_reasons}")
@@ -2059,6 +2126,8 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
p.add_argument("--export-csv", help="Optional path to dump per-trade CSV (incl. indicator breakdown)")
p.add_argument("--detail", action="store_true",
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("--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")
@@ -2118,6 +2187,7 @@ def main(argv: Optional[List[str]] = None) -> int:
quality_filter=[q.strip() for q in args.quality.split(",") if q.strip()],
strategies_only=[s.strip() for s in args.strategies.split(",") if s.strip()],
spread=spread,
workers=args.workers,
)
logger.info(