Files
ForexBot2/backtest/gpu_backtest.py
T
Harwinder Sekhon 1793d9167a Daily-trend veto on M15 + lot-size / already-open lines (campaign 2026-09-30), and the 2026-10-01 whole-codebase bug review
Two bodies of work that touch the same files, so one commit.

Direction and setup-size campaign (2026-09-30 → 10-01, owner's RISK-FIRST rule,
written before the runs):
- SIGNAL: M15 daily-trend veto (factory override 7, xau_strategy.daily_trend_veto
  "both"). A finished setup against a clear daily trend (closed D1 bars, Sunday
  stubs dropped; close vs its 50-bar EMA AND vs the close 60 kept days earlier)
  is removed as the last step of the core. One definition
  (signal_aggregator.daily_trend_series) for live, the bar replay and the twin's
  host side; D1 is a required frame of the M15 scan; mirrored in
  cuda_signal_kernels.py K2. W3 824 / +205.12R / PF 1.73 / max DD 12.4R vs
  1105 / +195.79R / 1.49 / 20.9R; 2019 → 2026 1473 / +223.33R / 1.42 / 41.5R vs
  2091 / +192.53R / 1.24 / 64.8R; unseen holdout −4.62R vs −6.63R.
- FAILED, not shipped: the veto on M5 (W3 drawdown 10.1R → 15.2R) and on H1
  (−9.7R), buy-only and sell-only on M15, the wide-stop cap
  (xau_strategy.wide_stop_cap_atr stays 0.0, a research knob).
- New M15 pins: W2 dfc647ee…, OOS e87b6de1…, W3 ad3b0dbd…. M5 / H1 / H4 pins
  unchanged. --set xau_strategy.daily_trend_veto=off reproduces the previous M15
  pins byte for byte. Twin and reference script MD5-equal on every window;
  recorder and feature-export parity re-proved; base-rate table rebuilt.
- DISPLAY (books byte-identical): exposure.py — the lot size for
  trade_size.risk_per_trade (new config.json block, off by default, private to
  the owner) and the count of alerted same-direction plans still open
  (signals.alerted_at, read-only open_alerted_setups); the daily trend in words
  on Telegram, the CLI and the dashboard; signals.daily_trend logged.
- providers/oanda: a date window with no bars is an empty page, not an error
  (a D1 pad starting on a Saturday could not be fetched).
- Reports only, nothing shipped: seven context screens (no state nominated) and
  eight outside data sources (none adds a forecast to gold's own daily trend).
- Independent verification (report section 5.13): every figure reproduced on
  both scripts, scoring identical to the previous commit on all four
  timeframes, no look-ahead. Caveats recorded there: this is a risk filter, not
  a more accurate direction call; nine tenths of the 2019 → 2026 gain is 2022;
  the neighbour clause is thin (a simple 100-day average misses the R line by
  1.2R); "never adds a trade" is an observation, not a guarantee.

Bug review of the whole codebase (2026-10-01). Signal path clean. 17 defects
elsewhere, plus 4 found by an independent review of the fixes; all fixed, each
with a test; no trade book changed:
- telegram_bot: one _deliver_alert for the candle scan and the level break. The
  row is stamped alerted_at BEFORE the send, on a one-worker alert-db pool, and
  un-stamped when no chat received it or the task is cancelled mid-stamp, so
  two alerts at one candle close count each other and no SQLite write runs on
  the event loop. The 5-bar cooldown counts bars of the scan's own frame
  (== gpu_backtest._apply_cooldown; clock time released the 22:00 UTC bar after
  the settlement pause) and is restored from signals.db at start-up. The
  lot-size line goes to the owner's own chat only (a group the owner ran
  /autoscan in used to receive it). /status gains a news-model line and escapes
  failure details.
- signal_log: mark_alerted returns the stamp it wrote; unmark_alerted;
  last_alerted_bars (read-only).
- advisory_outcome: candle continuity is judged bar by bar inside the walk; a
  hole after a trade's exit no longer voids it. The stored-candle replay
  resolves 108 of 111 logged setups instead of 99, all equal to the live
  resolver's R.
- providers/oanda: an empty date window is accepted only when it is 5 days or
  shorter and no malformed candle was skipped; a longer one is retried, then
  refused, instead of being cached as history with a hole.
- news_engine: blank RSS titles skipped; one unreadable headline no longer
  costs the local model its whole batch (checked on the real bundle); model
  health recorded; no DeepSeek audit call in a process that does not archive;
  event-direction cache keyed per release; peek_week_events never fetches.
- llm_client: chat_json never raises and never logs the API key (requests
  quotes the Authorization value in InvalidHeader; error bodies are redacted).
- signal_aggregator: the scan's news block is guarded; the replay path
  (use_news=False) assigns exactly the values it did before.
- agent_panel: PANEL_PROMPT_VERSION v2.2 — same prompt text, corrected input
  (event timing over the cached calendar week, headlines newest first).
  scripts/panel_replay.py replays the logged payload per persona;
  scripts/label_scan_outcomes.py counts with an ISO cutoff.
- test_all.py: no ALTER TABLE … DROP COLUMN (the live box has SQLite 3.34);
  tests that write Parquet print a notice and return without pyarrow.

Verification after the last code change: suite 632 passed, 0 failed (also with
both Parquet engines blocked and FB2_BACKEND=cpu, the live box's view); the
five reference MD5s reproduce on both backtesters, 0 cache fallbacks; Module 37
green; static Python 3.9 / SQLite 3.34 check clean on every touched file.

Evidence: backtest/research/daily_trend_campaign_2026-09-30.md,
context_screens_2026-10.md, new_inputs_2026-10.md; CLAUDE.md sections 2, 4, 6,
8, 9, 10 and 11; pair_signals/xauusd.py docstring (M15 override 7).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-01 10:28:28 -04:00

2869 lines
122 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
ForexBot2 — GPU-Accelerated Backtester
======================================
Replays historical OHLCV bar-by-bar through the live SignalAggregator so
backtest results are identical to what the production scanner would have
generated. Pre-computes all indicator series with CuPy/NumPy on the full
dataset (causal, zero look-ahead) and looks up values per bar instead of
recomputing each time — giving a 100x+ speedup over a naive replay while
preserving the exact signal logic.
Key properties:
- Uses real SignalAggregator.analyze_from_dataframes (same gates,
same XAU strategy hooks, same trade setup math).
- News engine disabled (use_news=False) — backtests are pure technical.
- Session filter, MTF agreement, regime classification all flow from the
bar timestamp, never datetime.now().
- XAU weekend/weekday filter applied here (analyze_from_dataframes skips
it; only analyze_pair enforces it in live mode).
- Trade simulation is full candle walk on the primary timeframe using
High/Low; if both SL and TP1 hit in the same bar, SL is assumed first
(conservative).
- Ships with CuPy fallback to NumPy so the script runs anywhere. The CUDA
kernels (backtest/gpu_kernels.py) are bit-for-bit replicas of the NumPy
path, so both backends produce the same book; --backend cpu forces NumPy.
Usage:
python backtest/gpu_backtest.py --pair XAUUSD --days 365
python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-01
python backtest/gpu_backtest.py --pair XAUUSD --days 180 --no-cache
"""
from __future__ import annotations
import argparse
import json
import logging
import multiprocessing as mp
import os
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from zoneinfo import ZoneInfo
import numpy as np
import pandas as pd
_NY_TZ = ZoneInfo("America/New_York")
# Ensure project root on sys.path so this script can be invoked from anywhere.
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from config import ( # noqa: E402
PAIR_INFO,
PlatformConfig,
SignalStrength,
Timeframe,
XAUUSD,
)
from providers import BaseDataProvider, make_data_provider # noqa: E402
from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine # noqa: E402
from signal_aggregator import ( # noqa: E402
SignalAggregator, SignalReport, TradeSetup, daily_trend_needed,
)
import position_sizing # noqa: E402
from backtest import replay_record # noqa: E402
from backtest import gpu_kernels as _gk # noqa: E402
# ─────────────────────────────────────────────────────────────────────
# GPU backend — backtest/gpu_kernels.py owns the CuPy state (probed
# lazily and never silently: a fallback carries its reason into the
# "Backend:" line) and the CUDA kernels. Every kernel is a bit-for-bit
# replica of the NumPy/pandas code kept in this file as the CPU path, so
# `--backend cpu` and the GPU produce the same book (proof: the five
# reference MD5s, CLAUDE.md section 9). The forked replay workers never
# touch the device: every precomputed series is moved to host first.
# ─────────────────────────────────────────────────────────────────────
BACKEND = _gk.BACKEND
def _gpu_on() -> bool:
return BACKEND.available
logger = logging.getLogger("gpu_backtest")
PAIR_ALIASES = {
"XAUUSD": XAUUSD,
"XAU/USD": XAUUSD,
"GOLD": XAUUSD,
}
# Bar duration in seconds — used to compute "fully-closed" cutoffs so
# confirmation timeframes can't see into the future at decision time.
BAR_SECONDS = {
Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900,
Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400,
Timeframe.D1: 86400, Timeframe.W1: 604800,
}
# ═════════════════════════════════════════════════════════════════════
# 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.
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.
"""
n = x.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 = 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])."""
alpha = 2.0 / (span + 1.0)
return _gpu_ewm_alpha(x, alpha, min_periods=0)
@dataclass
class IndicatorCache:
"""Pre-computed indicator series for one (symbol, timeframe).
All arrays are aligned to the source df's index (same length).
NaN at warmup positions; downstream lookup tolerates it.
"""
df: pd.DataFrame # full source df (cleaned)
index_pos: Dict[pd.Timestamp, int] # timestamp → position lookup
# EMAs (4-EMA mode keys; 2-EMA mode reuses ema_slow + ema_trend + ema_medium)
ema_fast: np.ndarray
ema_medium: np.ndarray
ema_slow: np.ndarray
ema_trend: Optional[np.ndarray] # None if len(df) < ema_trend period
# RSI
rsi: np.ndarray
rsi_series: pd.Series # pandas series for divergence
# MACD
macd_hist: np.ndarray
macd_hist_series: pd.Series
# Bollinger
bb_mid: np.ndarray
bb_upper: np.ndarray
bb_lower: np.ndarray
bb_position: np.ndarray
bb_band_width: np.ndarray
bb_avg_width: np.ndarray # 20-period rolling mean of band width
# ATR
atr: np.ndarray
# ADX
adx: np.ndarray
plus_di: np.ndarray
minus_di: np.ndarray
adx_slope: np.ndarray
# ROC
roc: np.ndarray
# Keltner Channels
kc_mid: np.ndarray
kc_upper: np.ndarray
kc_lower: np.ndarray
kc_squeeze: np.ndarray
def precompute_indicators(
df: pd.DataFrame, tech_cfg, gpu_log: bool = False
) -> IndicatorCache:
"""Compute every causal indicator series once on the full df.
Dispatches to the CUDA kernels when the backend is available and to
the NumPy/pandas code otherwise. The two are bit-identical (Module 36
and the 42-window check in backtest/research/gpu_backend_2026-09-18.md).
"""
if gpu_log:
logger.info(
f"Pre-computing indicators on {len(df)} bars "
f"({'CuPy GPU' if _gpu_on() else 'NumPy CPU'} backend)"
)
if _gpu_on():
return _precompute_indicators_gpu(df, tech_cfg)
return _precompute_indicators_cpu(df, tech_cfg)
def _precompute_indicators_cpu(df: pd.DataFrame, tech_cfg) -> IndicatorCache:
"""The NumPy/pandas path — the reference the CUDA twin replicates."""
close = df["Close"].to_numpy(dtype=np.float64)
high = df["High"].to_numpy(dtype=np.float64)
low = df["Low"].to_numpy(dtype=np.float64)
n = len(close)
# ── 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
# ── 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
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)
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_hist = macd_line - signal_line
macd_hist_series = pd.Series(macd_hist, index=df.index, name="macd_hist")
# ── Bollinger ──
# Computed with the SAME pandas rolling calls as the live engine
# (engine._compute_bollinger). The previous cumsum E[x²]−E[x]² shortcut
# loses precision at gold's price level (bb_position error up to ~8e-3
# — enough to cross the 0.0/0.2/0.8/1.0 signal thresholds, worst during
# squeezes). Replay slices are prefixes of this df, so pandas rolling
# values here are bit-identical to what the live engine computes per
# slice. Rolling on the full series runs once — cost is negligible.
close_s = df["Close"]
bb_mid_s = close_s.rolling(tech_cfg.bb_period).mean()
bb_std_s = close_s.rolling(tech_cfg.bb_period).std(ddof=0)
bb_mid = bb_mid_s.to_numpy(dtype=np.float64)
bb_std = bb_std_s.to_numpy(dtype=np.float64)
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_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_band_width = (bb_upper - bb_lower) / bb_mid_safe
bb_avg_width = (
pd.Series(bb_band_width, index=df.index).rolling(20).mean()
.to_numpy(dtype=np.float64)
)
# ── 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)
# ── ADX Slope ──
adx_slope_lookback = getattr(tech_cfg, 'adx_slope_lookback', 5)
adx_slope = np.full(n, np.nan, dtype=np.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)
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
# ── 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)
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,
)
def _assert_host_cache(cache: IndicatorCache) -> None:
"""Every array the forked workers will read must be plain NumPy."""
for name, val in vars(cache).items():
if name in ("df", "index_pos", "rsi_series", "macd_hist_series") or val is None:
continue
if type(val) is not np.ndarray:
raise TypeError(f"IndicatorCache.{name} is {type(val).__name__}, not np.ndarray")
def _precompute_indicators_gpu(df: pd.DataFrame, tech_cfg) -> IndicatorCache:
"""CUDA twin of `_precompute_indicators_cpu`.
Same expressions in the same order (single-op CuPy ufuncs cannot fuse
into an FMA and double add/mul/div/sqrt are correctly rounded on both
sides); the recurrences and the pandas rolling statistics run in the
kernels of backtest/gpu_kernels.py, one thread per series. Everything
comes back to the host once at the end: the replay workers read NumPy.
"""
cp = BACKEND.cp
nan = float("nan")
close_h = df["Close"].to_numpy(dtype=np.float64)
n = len(close_h)
close = cp.asarray(close_h)
high = cp.asarray(df["High"].to_numpy(dtype=np.float64))
low = cp.asarray(df["Low"].to_numpy(dtype=np.float64))
# ── RSI inputs (delta[0] is NaN; NaN > 0 is False so gain[0] = 0.0) ──
delta = cp.empty(n, dtype=cp.float64)
delta[0] = nan
delta[1:] = close[1:] - close[:-1]
gain = cp.where(delta > 0, delta, 0.0)
loss = cp.where(delta < 0, -delta, 0.0)
# ── True range (fmax ignores NaN, like the CPU path) ──
prev_close = cp.empty(n, dtype=cp.float64)
prev_close[0] = nan
prev_close[1:] = close[:-1]
tr1 = high - low
tr2 = cp.abs(high - prev_close)
tr3 = cp.abs(low - prev_close)
tr2[0] = nan
tr3[0] = nan
tr = cp.fmax(cp.fmax(tr1, tr2), tr3)
# ── Directional movement ──
plus_dm_raw = cp.empty(n, dtype=cp.float64)
minus_dm_raw = cp.empty(n, dtype=cp.float64)
plus_dm_raw[0] = nan
minus_dm_raw[0] = nan
plus_dm_raw[1:] = high[1:] - high[:-1]
minus_dm_raw[1:] = -(low[1:] - low[:-1])
cond_plus = (plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0)
plus_dm = cp.where(cond_plus, plus_dm_raw, 0.0)
cond_minus = (minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0)
minus_dm = cp.where(cond_minus, minus_dm_raw, 0.0)
# ── Stage 1: every recurrence that depends only on price, one launch ──
def span_alpha(span: int) -> float:
return 2.0 / (span + 1.0) # identical to _gpu_ewm_span
rsi_p = tech_cfg.rsi_period
atr_p = tech_cfg.atr_period
adx_p = tech_cfg.adx_period
kc_period = getattr(tech_cfg, 'kc_period', 20)
kc_atr_period = getattr(tech_cfg, 'kc_atr_period', 10)
kc_multiplier = getattr(tech_cfg, 'kc_multiplier', 1.5)
have_trend = n >= tech_cfg.ema_trend
jobs = [
(close, span_alpha(tech_cfg.ema_fast), 0),
(close, span_alpha(tech_cfg.ema_medium), 0),
(close, span_alpha(tech_cfg.ema_slow), 0),
(close, span_alpha(tech_cfg.macd_fast), 0),
(close, span_alpha(tech_cfg.macd_slow), 0),
(close, span_alpha(kc_period), 0),
(gain, 1.0 / rsi_p, rsi_p),
(loss, 1.0 / rsi_p, rsi_p),
(tr, 1.0 / atr_p, atr_p),
(tr, 1.0 / adx_p, adx_p),
(plus_dm, 1.0 / adx_p, adx_p),
(minus_dm, 1.0 / adx_p, adx_p),
(tr, 1.0 / kc_atr_period, kc_atr_period),
]
if have_trend:
jobs.append((close, span_alpha(tech_cfg.ema_trend), 0))
out = _gk.ewm_batch_dev([j[0] for j in jobs], [j[1] for j in jobs],
[j[2] for j in jobs])
(ema_fast, ema_medium, ema_slow, ema_macd_fast, ema_macd_slow, kc_mid,
avg_gain, avg_loss, atr, atr_adx, plus_dm_smooth, minus_dm_smooth,
kc_atr) = out[:13]
ema_trend = out[13] if have_trend else None
# ── RSI ──
avg_loss_safe = cp.where(avg_loss == 0, 1e-10, avg_loss)
rs = avg_gain / avg_loss_safe
rsi = 100.0 - (100.0 / (1.0 + rs))
# ── MACD line; DI / DX ──
macd_line = ema_macd_fast - ema_macd_slow
atr_adx_safe = cp.where(atr_adx == 0, nan, atr_adx)
plus_di = 100.0 * plus_dm_smooth / atr_adx_safe
minus_di = 100.0 * minus_dm_smooth / atr_adx_safe
di_sum = plus_di + minus_di
di_sum_safe = cp.where(di_sum == 0, nan, di_sum)
dx = 100.0 * cp.abs(plus_di - minus_di) / di_sum_safe
# ── Stage 2: recurrences on stage-1 outputs, one launch ──
signal_line, adx = _gk.ewm_batch_dev(
[macd_line, dx], [span_alpha(tech_cfg.macd_signal), 1.0 / adx_p], [0, adx_p])
macd_hist = macd_line - signal_line
# ── Bollinger: the pandas rolling replica (mean + var, then zsqrt) ──
(bb_mid,), (bb_var,) = _gk.roll_mean_var_dev([close], tech_cfg.bb_period, ddof=0)
bb_std = _gk.roll_std_from_var_dev(bb_var)
bb_upper = bb_mid + tech_cfg.bb_std * bb_std
bb_lower = bb_mid - tech_cfg.bb_std * bb_std
bb_range = bb_upper - bb_lower
bb_range_safe = cp.where(bb_range == 0, nan, bb_range)
bb_position = (close - bb_lower) / bb_range_safe
bb_position = cp.where(cp.isnan(bb_position), 0.5, bb_position)
bb_mid_safe = cp.where(bb_mid == 0, nan, bb_mid)
bb_band_width = (bb_upper - bb_lower) / bb_mid_safe
# 20-bar mean of the band width: the CPU path uses the literal 20.
(bb_avg_width,), _unused = _gk.roll_mean_var_dev([bb_band_width], 20, ddof=0)
# ── ADX slope ──
adx_slope_lookback = getattr(tech_cfg, 'adx_slope_lookback', 5)
adx_slope = cp.full(n, nan, dtype=cp.float64)
if n > adx_slope_lookback:
adx_slope[adx_slope_lookback:] = adx[adx_slope_lookback:] - adx[:-adx_slope_lookback]
# ── ROC ──
roc_period = getattr(tech_cfg, 'roc_period', 10)
roc = cp.full(n, nan, dtype=cp.float64)
if n > roc_period:
prev_close = close[:-roc_period]
prev_safe = cp.where(prev_close == 0, nan, prev_close)
roc[roc_period:] = ((close[roc_period:] - prev_close) / prev_safe) * 100.0
# ── Keltner ──
kc_upper = kc_mid + kc_multiplier * kc_atr
kc_lower = kc_mid - kc_multiplier * kc_atr
kc_squeeze = ((bb_lower > kc_lower) & (bb_upper < kc_upper)).astype(cp.float64)
# ── One transfer to the host; the workers read NumPy only ──
def h(a):
return a.get()
rsi_h = h(rsi)
macd_hist_h = h(macd_hist)
cache = IndicatorCache(
df=df,
index_pos={ts: i for i, ts in enumerate(df.index)},
ema_fast=h(ema_fast),
ema_medium=h(ema_medium),
ema_slow=h(ema_slow),
ema_trend=h(ema_trend) if ema_trend is not None else None,
rsi=rsi_h,
rsi_series=pd.Series(rsi_h, index=df.index, name="rsi"),
macd_hist=macd_hist_h,
macd_hist_series=pd.Series(macd_hist_h, index=df.index, name="macd_hist"),
bb_mid=h(bb_mid),
bb_upper=h(bb_upper),
bb_lower=h(bb_lower),
bb_position=h(bb_position),
bb_band_width=h(bb_band_width),
bb_avg_width=h(bb_avg_width),
atr=h(atr),
adx=h(adx),
plus_di=h(plus_di),
minus_di=h(minus_di),
adx_slope=h(adx_slope),
roc=h(roc),
kc_mid=h(kc_mid),
kc_upper=h(kc_upper),
kc_lower=h(kc_lower),
kc_squeeze=h(kc_squeeze),
)
_assert_host_cache(cache)
return cache
# ═════════════════════════════════════════════════════════════════════
# GPU-AWARE TECHNICAL ENGINE (subclasses live engine, lookup-only)
# ═════════════════════════════════════════════════════════════════════
class GPUTechnicalEngine(TechnicalEngine):
"""
Drop-in TechnicalEngine that pulls indicator values from a pre-computed
cache instead of recomputing per call. The conditional/scoring/description
logic for each indicator is preserved verbatim by overriding only the
`_compute_*` methods — every downstream consumer (regime classifier,
composite scorer, divergence, structure_sr, candlestick) sees an
IndicatorSignal byte-identical to what the live engine would produce.
"""
def __init__(self, tech_cfg, caches: Dict[str, IndicatorCache]):
super().__init__(tech_cfg)
# cache key = timeframe.value (e.g. "15min", "1h", "4h")
self.caches = caches
self._current_tf: Optional[str] = None
# Cache-miss fallbacks recompute with the live engine's pandas
# math (different EWM seeding than the precomputed cache). That
# is a parity hazard, so it must never happen silently.
self.fallback_count = 0
self._fallback_warned = False
def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis:
# Latch the timeframe so per-indicator overrides know which cache to use.
# Multiple TFs share aligned timestamps (e.g. M15 14:00 == H1 14:00),
# so timestamp-only lookup would match the wrong cache.
self._current_tf = timeframe
try:
return super().analyze(df, symbol, timeframe)
finally:
self._current_tf = None
def _resolve_pos(self, df: pd.DataFrame) -> Tuple[Optional[IndicatorCache], int]:
"""Find this slice's position in its pre-computed cache."""
if df is None or df.empty or self._current_tf is None:
return None, -1
cache = self.caches.get(self._current_tf)
if cache is None:
self._note_fallback(f"no cache for TF {self._current_tf!r}")
return None, -1
pos = cache.index_pos.get(df.index[-1])
if pos is None:
self._note_fallback(
f"{self._current_tf}: slice end {df.index[-1]} not in cache")
return None, -1
return cache, pos
def _note_fallback(self, why: str) -> None:
self.fallback_count += 1
if not self._fallback_warned:
self._fallback_warned = True
logger.warning(
"GPU cache miss → falling back to live-engine recompute "
"(%s). Further misses counted silently; check "
"fallback_count at end of run.", why)
# ── EMA overrides ──────────────────────────────────────────────
def _compute_ema_2(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_ema_2(df, result)
c = self.config
latest = float(df["Close"].iloc[-1])
s = float(cache.ema_slow[pos])
m = float(cache.ema_medium[pos])
t = float(cache.ema_trend[pos]) if cache.ema_trend is not None and not np.isnan(cache.ema_trend[pos]) else None
if t is None or pos < c.ema_trend - 1:
result.indicators["ema_trend"] = IndicatorSignal(
name="EMA Trend", value=0, signal=SignalStrength.NEUTRAL,
confidence=30, description="EMA200 data insufficient", weight=1.0,
)
result.key_levels["ema_medium"] = m
result.key_levels["ema_slow"] = s
return
if latest > s > t:
signal, conf, desc = SignalStrength.STRONG_BUY, 85, "Full bullish: price > EMA50 > EMA200"
elif latest > s and s < t:
signal, conf, desc = SignalStrength.BUY, 60, "Emerging uptrend: price > EMA50, EMA50 < EMA200"
elif latest < s < t:
signal, conf, desc = SignalStrength.STRONG_SELL, 85, "Full bearish: price < EMA50 < EMA200"
elif latest < s and s > t:
signal, conf, desc = SignalStrength.SELL, 60, "Emerging downtrend: price < EMA50, EMA50 > EMA200"
else:
signal, conf, desc = SignalStrength.NEUTRAL, 40, "EMA alignment mixed"
# Crossover detection on prior 10 bars (live: `if len(ema50) >= 10`,
# `for lookback in range(2, min(11, len(ema50)))` — len = pos+1).
if pos >= 9:
for lookback in range(2, min(11, pos + 1)):
prev_s = float(cache.ema_slow[pos - lookback + 1])
prev_t = float(cache.ema_trend[pos - lookback + 1])
if prev_s <= prev_t and s > t:
desc += " | Golden cross (50/200)"
conf = min(conf + 10, 95)
break
if prev_s >= prev_t and s < t:
desc += " | Death cross (50/200)"
conf = min(conf + 10, 95)
break
normalized = {
SignalStrength.STRONG_BUY: 1.0, SignalStrength.BUY: 0.5,
SignalStrength.NEUTRAL: 0.0,
SignalStrength.SELL: -0.5, SignalStrength.STRONG_SELL: -1.0,
}[signal]
result.indicators["ema_trend"] = IndicatorSignal(
name="EMA Trend", value=normalized, signal=signal,
confidence=conf, description=desc, weight=1.0,
)
result.key_levels["ema_trend"] = t
result.key_levels["ema_medium"] = m
result.key_levels["ema_slow"] = s
def _compute_ema_4(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_ema_4(df, result)
c = self.config
latest = float(df["Close"].iloc[-1])
f = float(cache.ema_fast[pos])
m = float(cache.ema_medium[pos])
s = float(cache.ema_slow[pos])
t_arr = cache.ema_trend
t = float(t_arr[pos]) if (t_arr is not None and not np.isnan(t_arr[pos])) else None
# Live engine guards `t` with `len(close) >= ema_trend`; replicate.
if t is not None and pos < c.ema_trend - 1:
t = None
alignment_score = 0
checks = [(latest, f), (f, m), (m, s)]
if t is not None:
checks.append((s, t))
for above, below in checks:
if above > below:
alignment_score += 1
elif above < below:
alignment_score -= 1
max_checks = len(checks)
normalized = alignment_score / max_checks if max_checks > 0 else 0
if normalized > 0.6:
signal = SignalStrength.STRONG_BUY
conf = min(90, 60 + normalized * 30)
elif normalized > 0.2:
signal = SignalStrength.BUY
conf = 55 + normalized * 20
elif normalized < -0.6:
signal = SignalStrength.STRONG_SELL
conf = min(90, 60 + abs(normalized) * 30)
elif normalized < -0.2:
signal = SignalStrength.SELL
conf = 55 + abs(normalized) * 20
else:
signal = SignalStrength.NEUTRAL
conf = 40
# Mirrors the live engine's ema_partial_neutral squash (parity).
if c.ema_partial_neutral and signal in (SignalStrength.BUY, SignalStrength.SELL):
signal = SignalStrength.NEUTRAL
conf = 40
crossover_desc = ""
# Live: `if len(ema_fast) >= 3:` then iloc[-2] — needs len ≥ 2 functionally,
# but live's guard requires len ≥ 3 so we match that to stay byte-identical.
if pos >= 2:
prev_f = float(cache.ema_fast[pos - 1])
prev_m = float(cache.ema_medium[pos - 1])
if prev_f <= prev_m and f > m:
crossover_desc = " | Golden cross (fast/medium)"
conf = min(conf + 10, 95)
elif prev_f >= prev_m and f < m:
crossover_desc = " | Death cross (fast/medium)"
conf = min(conf + 10, 95)
# EMA ribbon compression/expansion detection — mirrors the live
# engine exactly (iloc[-5] ≡ pos-4, `len >= 6` ≡ pos >= 5). ATR is
# already in key_levels because analyze() computes ATR first.
ribbon_desc = ""
atr_val = result.key_levels.get("atr", 0)
ema_values = [f, m, s]
if atr_val > 0:
ribbon_width = (max(ema_values) - min(ema_values)) / atr_val
if ribbon_width < 1.0:
ribbon_desc = " | Ribbon COMPRESSED (trend change imminent)"
if pos >= 5:
prev_vals = [float(cache.ema_fast[pos - 4]),
float(cache.ema_medium[pos - 4]),
float(cache.ema_slow[pos - 4])]
prev_width = (max(prev_vals) - min(prev_vals)) / atr_val if atr_val > 0 else 0
if prev_width < 1.0 and ribbon_width >= 1.0:
ribbon_desc = " | Ribbon EXPANDING (new trend starting)"
conf = min(conf + 10, 95)
desc = f"EMA alignment: {alignment_score}/{max_checks}{crossover_desc}{ribbon_desc}"
result.indicators["ema_trend"] = IndicatorSignal(
name="EMA Trend", value=normalized, signal=signal,
confidence=conf, description=desc, weight=1.0,
)
if t is not None:
result.key_levels["ema_trend"] = t
result.key_levels["ema_fast"] = f
result.key_levels["ema_medium"] = m
result.key_levels["ema_slow"] = s
# ── RSI override ──────────────────────────────────────────────
def _compute_rsi(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_rsi(df, result)
c = self.config
rsi_val = float(cache.rsi[pos])
if np.isnan(rsi_val):
rsi_val = 50.0
if rsi_val >= c.rsi_overbought:
signal = SignalStrength.SELL
conf = min(80, 50 + (rsi_val - c.rsi_overbought))
desc = f"RSI {rsi_val:.1f} — Overbought"
elif rsi_val <= c.rsi_oversold:
signal = SignalStrength.BUY
conf = min(80, 50 + (c.rsi_oversold - rsi_val))
desc = f"RSI {rsi_val:.1f} — Oversold"
elif rsi_val >= c.rsi_neutral_upper:
signal = SignalStrength.BUY
conf = 45
desc = f"RSI {rsi_val:.1f} — Bullish momentum zone"
elif rsi_val <= c.rsi_neutral_lower:
signal = SignalStrength.SELL
conf = 45
desc = f"RSI {rsi_val:.1f} — Bearish momentum zone"
else:
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"RSI {rsi_val:.1f} — Neutral"
result.indicators["rsi"] = IndicatorSignal(
name="RSI", value=rsi_val, signal=signal,
confidence=conf, description=desc, weight=1.5,
)
# Slice the pre-computed series so divergence detector sees only
# bars up to the current decision point (no look-ahead).
result._rsi_series = cache.rsi_series.iloc[: pos + 1] # type: ignore[attr-defined]
# ── MACD override ─────────────────────────────────────────────
def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_macd(df, result)
c = self.config
result._macd_histogram = cache.macd_hist_series.iloc[: pos + 1] # type: ignore[attr-defined]
if not c.macd_as_pillar or pos < 2:
return
hist_now = float(cache.macd_hist[pos])
hist_prev = float(cache.macd_hist[pos - 1])
hist_prev2 = float(cache.macd_hist[pos - 2])
if np.isnan(hist_now) or np.isnan(hist_prev):
return
crossed_up = hist_prev <= 0 < hist_now
crossed_down = hist_prev >= 0 > hist_now
accel_up = hist_now > 0 and hist_now > hist_prev > 0
accel_down = hist_now < 0 and hist_now < hist_prev < 0
fading_up = hist_now > 0 and hist_now < hist_prev
fading_down = hist_now < 0 and hist_now > hist_prev
if pos >= 19:
window = np.abs(cache.macd_hist[pos - 19: pos + 1])
mean_abs = float(np.nanmean(window))
hist_pct = abs(hist_now) / mean_abs if mean_abs > 0 else 1.0
else:
hist_pct = 1.0
if crossed_up or accel_up:
signal = SignalStrength.STRONG_BUY if accel_up else SignalStrength.BUY
conf = min(80, 55 + hist_pct * 10)
desc = "MACD histogram" + (" accelerating ↑" if accel_up else " crossed above zero")
elif crossed_down or accel_down:
signal = SignalStrength.STRONG_SELL if accel_down else SignalStrength.SELL
conf = min(80, 55 + hist_pct * 10)
desc = "MACD histogram" + (" accelerating ↓" if accel_down else " crossed below zero")
elif fading_up:
signal, conf, desc = SignalStrength.NEUTRAL, 35, "MACD histogram positive but fading"
elif fading_down:
signal, conf, desc = SignalStrength.NEUTRAL, 35, "MACD histogram negative but fading"
else:
signal, conf, desc = SignalStrength.NEUTRAL, 30, "MACD histogram flat"
result.indicators["macd"] = IndicatorSignal(
name="MACD", value=hist_now, signal=signal,
confidence=conf, description=desc, weight=1.5,
)
# ── Bollinger override ────────────────────────────────────────
def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_bollinger(df, result)
c = self.config
latest = float(df["Close"].iloc[-1])
upper_val = float(cache.bb_upper[pos])
lower_val = float(cache.bb_lower[pos])
sma_val = float(cache.bb_mid[pos])
if np.isnan(upper_val) or np.isnan(lower_val) or np.isnan(sma_val) or sma_val == 0:
return
bb_range = upper_val - lower_val
bb_position = (latest - lower_val) / bb_range if bb_range != 0 else 0.5
band_width = float(cache.bb_band_width[pos])
is_squeeze = False
if pos >= 19:
avg_width = float(cache.bb_avg_width[pos])
if not np.isnan(avg_width) and avg_width > 0:
is_squeeze = band_width < avg_width * 0.75
squeeze_note = " | SQUEEZE detected (expect breakout)" if is_squeeze else ""
if bb_position > 1.0:
signal, conf = SignalStrength.SELL, 65
desc = f"Price above upper BB ({bb_position:.2f}){squeeze_note}"
elif bb_position < 0.0:
signal, conf = SignalStrength.BUY, 65
desc = f"Price below lower BB ({bb_position:.2f}){squeeze_note}"
elif bb_position > 0.8:
signal, conf = SignalStrength.SELL, 45
desc = f"Price near upper BB ({bb_position:.2f}){squeeze_note}"
elif bb_position < 0.2:
signal, conf = SignalStrength.BUY, 45
desc = f"Price near lower BB ({bb_position:.2f}){squeeze_note}"
else:
signal, conf = SignalStrength.NEUTRAL, 30
desc = f"Price mid-band ({bb_position:.2f}){squeeze_note}"
if is_squeeze:
conf = min(conf + 10, 85)
result.indicators["bollinger"] = IndicatorSignal(
name="Bollinger Bands", value=bb_position, signal=signal,
confidence=conf, description=desc, weight=1.5,
)
result.key_levels["bb_upper"] = upper_val
result.key_levels["bb_lower"] = lower_val
result.key_levels["bb_mid"] = sma_val
# ── ADX override ──────────────────────────────────────────────
def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_adx(df, result)
c = self.config
adx_val = float(cache.adx[pos])
if np.isnan(adx_val):
adx_val = 0.0
plus_di_val = float(cache.plus_di[pos])
minus_di_val = float(cache.minus_di[pos])
if np.isnan(plus_di_val):
plus_di_val = 0.0
if np.isnan(minus_di_val):
minus_di_val = 0.0
if adx_val >= c.adx_very_strong:
trend_strength, conf = "Very strong", 80
elif adx_val >= c.adx_strong_trend:
trend_strength, conf = "Strong", 65
else:
trend_strength, conf = "Weak/Ranging", 40
if plus_di_val > minus_di_val:
signal = SignalStrength.BUY if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL
direction = "bullish"
else:
signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL
direction = "bearish"
adx_slope = 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 > emerging_threshold:
slope_desc = " | TREND EMERGING (ADX rising fast)"
conf = min(conf + 15, 85)
elif adx_val >= c.adx_strong_trend and adx_slope > 0:
slope_desc = f" | Trend strengthening (slope +{adx_slope:.1f})"
conf = min(conf + 5, 85)
elif adx_val >= c.adx_strong_trend and adx_slope < -emerging_threshold:
slope_desc = f" | Trend weakening (slope {adx_slope:.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
# ── ATR override ──────────────────────────────────────────────
def _compute_atr(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_atr(df, result)
latest_close = float(df["Close"].iloc[-1])
atr_val = float(cache.atr[pos])
if np.isnan(atr_val):
atr_val = 0.0
atr_pct = (atr_val / latest_close) * 100 if latest_close != 0 else 0
# Volatility regime: replicate live behavior on the 50-bar window.
if pos >= 49:
window = cache.atr[pos - 49: pos + 1]
atr_percentile = float(np.mean(window < atr_val)) * 100 # bars strictly below current
# Live code: (atr.iloc[-1] > atr.iloc[-50:]).mean() * 100 → fraction of window strictly less than current
else:
atr_percentile = 50.0
if atr_percentile > 80:
vol_regime = "extreme"
elif atr_percentile > 60:
vol_regime = "high"
elif atr_percentile > 30:
vol_regime = "normal"
else:
vol_regime = "low"
result.volatility_regime = vol_regime
result.key_levels["atr"] = atr_val
result.key_levels["atr_pct"] = atr_pct
result.indicators["atr"] = IndicatorSignal(
name="ATR", value=atr_val, signal=SignalStrength.NEUTRAL,
confidence=50,
description=f"ATR {atr_val:.5f} ({atr_pct:.3f}%) | Volatility: {vol_regime} (P{atr_percentile:.0f})",
weight=0.0,
)
def _compute_roc(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_roc(df, result)
c = self.config
n = c.roc_period
# Mirror the live guard (engine._compute_roc: len < n + 5 → skip,
# i.e. emit only from pos >= n + 4). Unreachable in practice —
# analyze() floors at 30 bars and n+4 < 30 for shipped periods —
# but keep the engines textually aligned so a future period/floor
# change can't open a silent divergence window.
if pos < n + 4:
return
roc_val = float(cache.roc[pos])
roc_prev = float(cache.roc[pos - 1]) if pos > 0 else np.nan
if np.isnan(roc_val) or np.isnan(roc_prev):
return
roc_rising = roc_val > roc_prev
roc_abs = abs(roc_val)
threshold = c.roc_threshold
if roc_val > 0 and roc_rising:
if roc_abs > threshold * 2:
signal = SignalStrength.STRONG_BUY
conf = min(80, 60 + roc_abs * 5)
desc = f"ROC +{roc_val:.2f}% — Strong momentum acceleration"
else:
signal = SignalStrength.BUY
conf = min(70, 50 + roc_abs * 10)
desc = f"ROC +{roc_val:.2f}% — Rising momentum"
elif roc_val < 0 and not roc_rising:
if roc_abs > threshold * 2:
signal = SignalStrength.STRONG_SELL
conf = min(80, 60 + roc_abs * 5)
desc = f"ROC {roc_val:.2f}% — Strong bearish momentum"
else:
signal = SignalStrength.SELL
conf = min(70, 50 + roc_abs * 10)
desc = f"ROC {roc_val:.2f}% — Falling momentum"
else:
# Zero-crosses land in the branches above by definition (a
# cross up is positive and rising, a cross down negative and
# falling) — only flat/fading momentum reaches here.
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"ROC {roc_val:+.2f}% — No clear momentum"
result.indicators["roc"] = IndicatorSignal(
name="ROC", value=roc_val, signal=signal,
confidence=conf, description=desc, weight=1.5,
)
def _compute_keltner(self, df: pd.DataFrame, result: TechnicalAnalysis):
cache, pos = self._resolve_pos(df)
if cache is None or pos < 0:
return super()._compute_keltner(df, result)
c = self.config
# Mirror the live guard (engine._compute_keltner: len < max(kc_period,
# kc_atr_period) + 5 → skip). Unreachable below the 30-bar analyze()
# floor with shipped periods; kept aligned for the same reason as ROC.
if pos < max(c.kc_period, c.kc_atr_period) + 4:
return
latest = float(df["Close"].iloc[-1])
kc_upper_val = float(cache.kc_upper[pos])
kc_lower_val = float(cache.kc_lower[pos])
kc_mid_val = float(cache.kc_mid[pos])
if np.isnan(kc_upper_val) or np.isnan(kc_lower_val):
return
is_squeeze = bool(cache.kc_squeeze[pos])
squeeze_released = False
if pos >= 1:
was_squeeze = bool(cache.kc_squeeze[pos - 1])
squeeze_released = was_squeeze and not is_squeeze
macd_direction = 0
if not np.isnan(cache.macd_hist[pos]):
macd_direction = 1 if cache.macd_hist[pos] > 0 else -1
if squeeze_released:
if macd_direction > 0:
signal = SignalStrength.STRONG_BUY
desc = "Keltner SQUEEZE RELEASED — bullish breakout"
elif macd_direction < 0:
signal = SignalStrength.STRONG_SELL
desc = "Keltner SQUEEZE RELEASED — bearish breakout"
else:
signal = SignalStrength.NEUTRAL
desc = "Keltner squeeze released — direction unclear"
conf = 80
elif is_squeeze:
signal = SignalStrength.NEUTRAL
conf = 60
desc = "Keltner SQUEEZE active — breakout imminent"
elif latest > kc_upper_val:
signal = SignalStrength.BUY
conf = 65
desc = "Price above Keltner upper — strong uptrend continuation"
elif latest < kc_lower_val:
signal = SignalStrength.SELL
conf = 65
desc = "Price below Keltner lower — strong downtrend continuation"
else:
signal = SignalStrength.NEUTRAL
conf = 30
desc = "Price within Keltner Channels"
result.indicators["keltner"] = IndicatorSignal(
name="Keltner", value=float(latest - kc_mid_val),
signal=signal, confidence=conf,
description=desc, weight=1.5,
)
result.key_levels["kc_upper"] = kc_upper_val
result.key_levels["kc_lower"] = kc_lower_val
result.key_levels["kc_mid"] = kc_mid_val
# ═════════════════════════════════════════════════════════════════════
# DATA LOADING (API + parquet cache)
# ═════════════════════════════════════════════════════════════════════
CACHE_DIR = Path(__file__).resolve().parent / "cache"
def _history_pad_days(tf: Timeframe, warmup_bars: int) -> int:
"""Calendar days of history fetched BEFORE --start for one timeframe.
Base pad: max(60, warmup_bars // 4) days (62 at the default 250), the
EMA200/ADX warm-up every book since 2026-07 was built on. Slow TFs
need more: 62 days of H4 is 372 bars and of D1 only ~44, while the
live scanner now fetches at least BaseDataProvider.MIN_LIVE_BARS
(1000) bars on every timeframe (2026-09-10 seeding-parity fix — the
engine's EMAs use ewm(adjust=False), so the first bar of the frame
keeps weight (1-2/201)^n in the trend EMA). Target that same depth
here, converting trading to calendar days with 7/5, so the window
START is seeded like the rest of the window and like live. The
intraday pads are unchanged (M5/M15/H1 already hold >= 1488 bars in
62 days), which keeps their cache windows and books byte-identical;
only H4 (234 days) and D1 (1400 days) fetch deeper.
"""
base = max(60, warmup_bars // 4)
bars_per_day = BaseDataProvider.BARS_PER_DAY[tf]
seed_days = int(np.ceil(BaseDataProvider.MIN_LIVE_BARS / bars_per_day * 7 / 5))
return max(base, seed_days)
def _cache_path(
symbol: str, tf: Timeframe, start: datetime, end: datetime,
provider_name: str = "twelve_data",
) -> Path:
safe_sym = symbol.replace("/", "_").replace("=X", "")
# Bars differ between feeds (Oanda vs TD divergence study 2026-07-24),
# so the cache is keyed per provider. twelve_data keeps the legacy
# un-prefixed filename so pinned-book caches stay valid.
prefix = "" if provider_name == "twelve_data" else f"{provider_name}_"
return CACHE_DIR / (
f"{prefix}{safe_sym}_{tf.value}_{start:%Y%m%d}_{end:%Y%m%d}.parquet")
def fetch_history(
provider: BaseDataProvider,
symbol: str,
timeframe: Timeframe,
start: datetime,
end: datetime,
use_cache: bool = True,
clean: bool = True,
) -> pd.DataFrame:
"""Fetch OHLCV between start/end with parquet caching.
Pagination under the provider's per-request bar cap lives in
`BaseDataProvider.fetch_range`, so any registered provider can feed
the backtester.
The parquet cache stores the RAW series; `validate_and_clean` runs on
load. Pass clean=False to get the raw price path — trade resolution must
use it, because cleaning drops >10σ spike bars (news candles) that hit
stops in reality. (Caches written before this change already contain
cleaned data; a refetch is needed to restore dropped spike bars.)
"""
CACHE_DIR.mkdir(parents=True, exist_ok=True)
path = _cache_path(symbol, timeframe, start, end, provider.name)
if use_cache and path.exists():
logger.info(f"Loading cached {symbol} {timeframe.value}: {path.name}")
# Files written before the raw-cache change (2026-07-08, commit
# 6a96ed6) contain CLEANED bars; loading one with clean=False
# would silently miss the spike bars the raw path exists to model.
# The key format predates versioning, so detect by mtime.
_RAW_CACHE_EPOCH = 1783468800.0 # 2026-07-08 00:00 UTC
if path.stat().st_mtime < _RAW_CACHE_EPOCH:
logger.error(
f"{path.name} predates the raw-cache format (2026-07-08) "
f"and likely contains CLEANED bars — delete it and refetch "
f"before trusting trade resolution on this window")
df = pd.read_parquet(path)
if df.index.tz is None:
df.index = df.index.tz_localize("UTC")
return provider.validate_and_clean(df, symbol, timeframe) if clean else df
logger.info(f"Fetching {symbol} {timeframe.value} from {start.date()} to {end.date()}")
df = provider.fetch_range(symbol, timeframe, start, end)
if use_cache:
df.to_parquet(path)
logger.info(f"Cached {len(df)} raw bars to {path.name}")
return provider.validate_and_clean(df, symbol, timeframe) if clean else df
# ═════════════════════════════════════════════════════════════════════
# TRADE SIMULATION (vectorized SL/TP detection)
# ═════════════════════════════════════════════════════════════════════
@dataclass
class Trade:
"""One backtested trade — entry through exit + full reasoning."""
pair: str
entry_time: pd.Timestamp
entry_price: float
sl: float
tp1: float
tp2: float
direction: str # "BUY" or "SELL"
pip_value: float
risk_pips: float
reward_pips: float
rr_planned: float
quality: str # signal grade at entry
final_score: float
final_confidence: float
market_regime: str
volatility_regime: str = ""
strategy_tag: str = "" # "XAU", "XAU_PB", "GENERIC"
# Per-trade trailing overrides (0 = use the run-level params).
trail_arm_r: float = 0.0
trail_dist_r: float = 0.0
# Per-instructions: entry reasoning and indicator breakdown ─────
entry_reasons: List[str] = field(default_factory=list)
entry_warnings: List[str] = field(default_factory=list)
# name -> {value, signal_name, confidence, weight, agrees_with_dir}
indicator_breakdown: Dict[str, Dict] = field(default_factory=dict)
indicators_agreeing: int = 0 # count agreeing with trade direction
indicators_opposing: int = 0 # count opposing
indicators_neutral: int = 0
indicators_total: int = 0 # total scored indicators
# RESEARCH (--export-features only): decision-time feature vector
# (position_sizing.FEATURE_ORDER) and the stake multiplier the setup
# carried. Empty / 1.0 on every normal run, so the default book is
# untouched.
features: Dict[str, float] = field(default_factory=dict)
stake_multiplier: float = 1.0
# RESEARCH (--record only): plain-data decision snapshot built by
# replay_record.snapshot_report — indicator weights/values on every
# timeframe, penalties, setup. Empty on normal runs, so pickling a
# Trade back from a worker costs nothing and the export is untouched.
snapshot: Dict = field(default_factory=dict)
# Filled on exit:
exit_time: Optional[pd.Timestamp] = None
exit_price: Optional[float] = None
exit_reason: str = "" # "TP1", "SL", "TIMEOUT"
bars_held: int = 0
r_multiple: float = 0.0 # realized R (negative if loss)
mae_r: float = 0.0 # max adverse excursion in R
mfe_r: float = 0.0 # max favorable excursion in R
def _build_indicator_breakdown(
report: SignalReport, direction: str
) -> Tuple[Dict[str, Dict], int, int, int, int]:
"""Build per-indicator breakdown and tally direction agreement.
For each scored indicator on the primary timeframe:
- record value/signal/confidence/weight
- flag whether its signal agrees with the trade direction
"""
breakdown: Dict[str, Dict] = {}
agree = oppose = neutral = total = 0
if report.primary_analysis is None:
return breakdown, 0, 0, 0, 0
is_buy = direction == "BUY"
bull = (SignalStrength.BUY, SignalStrength.STRONG_BUY)
bear = (SignalStrength.SELL, SignalStrength.STRONG_SELL)
for name, ind in report.primary_analysis.indicators.items():
if name.startswith("_") or ind.weight == 0:
continue # skip private series + context-only indicators (atr)
agrees = "neutral"
if ind.signal in bull:
agrees = "agree" if is_buy else "oppose"
elif ind.signal in bear:
agrees = "agree" if not is_buy else "oppose"
breakdown[name] = {
"value": float(ind.value) if ind.value is not None else None,
"signal": ind.signal.name,
"confidence": float(ind.confidence),
"weight": float(ind.weight),
"agrees": agrees,
"description": ind.description,
}
total += 1
if agrees == "agree":
agree += 1
elif agrees == "oppose":
oppose += 1
else:
neutral += 1
return breakdown, agree, oppose, neutral, total
def _simulate_trailing(
trades: List[Trade],
opens_arr: np.ndarray,
highs_arr: np.ndarray,
lows_arr: np.ndarray,
closes_arr: np.ndarray,
valid_lens: np.ndarray,
exit_times: List[List[pd.Timestamp]],
spread: float,
trail_arm_r: float,
trail_dist_r: float,
) -> None:
"""Resolve trades with a trailing stop (per-trade CPU loop).
Exit reasons: "SL" (original stop, full loss), "TRAIL" (trailed stop
filled at/above entry region), "TIMEOUT".
"""
for i, t in enumerate(trades):
n = int(valid_lens[i])
if n == 0:
t.exit_reason = "NO_DATA"
continue
o = opens_arr[i, :n]
h = highs_arr[i, :n]
l = lows_arr[i, :n]
c = closes_arr[i, :n]
buy = t.direction == "BUY"
risk = abs(t.entry_price - t.sl)
if risk <= 0:
t.exit_reason = "NO_DATA"
continue
arm = t.trail_arm_r if t.trail_arm_r > 0 else trail_arm_r
dist = t.trail_dist_r if t.trail_dist_r > 0 else trail_dist_r
cur_sl = t.sl
peak = t.entry_price
exit_price = None
exit_idx = n - 1
reason = "TIMEOUT"
for j in range(n):
if np.isnan(h[j]):
# Data ends here: close at the LAST VALID bar, not at
# c[n-1] (which may itself be NaN — a NaN exit price used
# to poison r_multiple and every downstream metric).
if j == 0:
exit_idx = 0
reason = "NO_DATA"
else:
exit_idx = j - 1
exit_price = float(c[exit_idx])
break
hit = (l[j] <= cur_sl) if buy else (h[j] >= cur_sl)
if hit:
exit_price = cur_sl
# Gap through the stop → filled at the open.
if (buy and o[j] < cur_sl) or (not buy and o[j] > cur_sl):
exit_price = o[j]
reason = "SL" if cur_sl == t.sl else "TRAIL"
exit_idx = j
break
# Bar closed: advance peak, then ratchet the stop for the
# NEXT bar.
fav = h[j] if buy else l[j]
peak = max(peak, fav) if buy else min(peak, fav)
exc = ((peak - t.entry_price) if buy
else (t.entry_price - peak)) / risk
if exc >= arm:
lvl = (peak - dist * risk if buy
else peak + dist * risk)
cur_sl = max(cur_sl, lvl) if buy else min(cur_sl, lvl)
if reason == "NO_DATA":
t.exit_reason = reason
continue
if exit_price is None:
exit_price = float(c[exit_idx])
if not np.isfinite(exit_price):
# Last-resort guard: never emit a NaN exit/R into the book.
t.exit_reason = "NO_DATA"
continue
t.exit_reason = reason
t.bars_held = exit_idx + 1
t.exit_price = float(exit_price)
t.exit_time = exit_times[i][exit_idx]
pnl = ((t.exit_price - t.entry_price) if buy
else (t.entry_price - t.exit_price))
t.r_multiple = (pnl - spread) / risk
h_win = h[:t.bars_held]
l_win = l[:t.bars_held]
if buy:
mfe = (np.nanmax(h_win) - t.entry_price) / risk
mae = (t.entry_price - np.nanmin(l_win)) / risk
else:
mfe = (t.entry_price - np.nanmin(l_win)) / risk
mae = (np.nanmax(h_win) - t.entry_price) / risk
t.mfe_r = float(max(0.0, mfe))
t.mae_r = float(max(0.0, mae))
def simulate_trades(
trades: List[Trade],
sim_df: pd.DataFrame,
max_bars: int = 200,
spread: float = 0.0,
trail_arm_r: float = 0.0,
trail_dist_r: float = 0.75,
) -> None:
"""Walk forward bar-by-bar after each entry and resolve the exit.
Two exit modes:
- Static (trail_arm_r <= 0): resolve SL / TP1, vectorized on GPU when
CuPy is available.
- Trailing (trail_arm_r > 0): TP1 is advisory. Once favorable
excursion reaches trail_arm_r × risk, the stop trails
trail_dist_r × risk behind the best price. Stop levels update at
bar CLOSE (intra-bar sequence is unknowable), so a new stop can
only fill from the next bar on — conservative.
SL/TP-same-bar tie: SL assumed first (conservative).
Gap-through fills: a bar *opening* beyond SL/TP fills at the open (worse
than SL on gaps against, better than TP on gaps in favor).
`spread` is a round-trip cost in price units subtracted from every
trade's PnL before computing R.
Times out (no hit) → marked TIMEOUT, exit at the last simulated bar.
"""
if not trades:
return
# Build per-trade post-entry windows (clipped to max_bars).
opens_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
highs_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
lows_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
closes_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
valid_lens = np.zeros(len(trades), dtype=np.int64)
exit_times: List[List[pd.Timestamp]] = []
sim_index = sim_df.index
sim_open = sim_df["Open"].to_numpy(dtype=np.float64)
sim_high = sim_df["High"].to_numpy(dtype=np.float64)
sim_low = sim_df["Low"].to_numpy(dtype=np.float64)
sim_close = sim_df["Close"].to_numpy(dtype=np.float64)
for i, t in enumerate(trades):
# entry_time is an exact bar timestamp from sim_df (we choose the
# first bar AFTER the signal bar as entry — see backtester loop).
try:
entry_pos = sim_index.get_loc(t.entry_time)
except KeyError:
# Snap to next available bar (rare timezone/edge case).
entry_pos = sim_index.searchsorted(t.entry_time)
if entry_pos >= len(sim_index):
exit_times.append([])
continue
end_pos = min(entry_pos + max_bars, len(sim_index))
n = end_pos - entry_pos
valid_lens[i] = n
opens_arr[i, :n] = sim_open[entry_pos:end_pos]
highs_arr[i, :n] = sim_high[entry_pos:end_pos]
lows_arr[i, :n] = sim_low[entry_pos:end_pos]
closes_arr[i, :n] = sim_close[entry_pos:end_pos]
exit_times.append(list(sim_index[entry_pos:end_pos]))
if _gpu_on():
# One CUDA thread per trade, both exit modes (bit-identical twin of
# the two CPU paths below; Module 36 and the reference MD5s).
_resolve_trades_gpu(
trades, opens_arr, highs_arr, lows_arr, closes_arr,
valid_lens, exit_times, spread, trail_arm_r, trail_dist_r,
)
return
if trail_arm_r > 0:
_simulate_trailing(
trades, opens_arr, highs_arr, lows_arr, closes_arr,
valid_lens, exit_times, spread, trail_arm_r, trail_dist_r,
)
return
# CPU static path, vectorized: for each trade, scan its window for the
# first SL/TP hit.
sls = np.array([t.sl for t in trades], dtype=np.float64)[:, None]
tps = np.array([t.tp1 for t in trades], dtype=np.float64)[:, None]
is_buy = np.array([t.direction == "BUY" for t in trades], dtype=bool)[:, None]
# Hit masks. For BUY: SL when low<=sl, TP when high>=tp.
# For SELL: SL when high>=sl, TP when low<=tp.
sl_hit_buy = lows_arr <= sls
tp_hit_buy = highs_arr >= tps
sl_hit_sell = highs_arr >= sls
tp_hit_sell = lows_arr <= tps
sl_hit = np.where(is_buy, sl_hit_buy, sl_hit_sell)
tp_hit = np.where(is_buy, tp_hit_buy, tp_hit_sell)
# Mask out NaN bars (no data past valid_lens).
valid_mask = ~np.isnan(highs_arr)
sl_hit = sl_hit & valid_mask
tp_hit = tp_hit & valid_mask
# First-hit indices (or max_bars if never).
idx_template = np.arange(max_bars)[None, :]
sl_idx = np.where(sl_hit, idx_template, max_bars)
tp_idx = np.where(tp_hit, idx_template, max_bars)
sl_first_arr = np.min(sl_idx, axis=1)
tp_first_arr = np.min(tp_idx, axis=1)
# Resolve each trade and compute realized MFE/MAE *up to exit bar*
# (a trade closed at bar 5 cannot have meaningful excursions at bar 12).
for i, t in enumerate(trades):
n_valid = valid_lens[i]
if n_valid == 0:
t.exit_reason = "NO_DATA"
continue
sl_hit_idx = int(sl_first_arr[i])
tp_hit_idx = int(tp_first_arr[i])
# SL wins ties (conservative same-bar handling).
if sl_hit_idx < n_valid and sl_hit_idx <= tp_hit_idx:
t.exit_reason = "SL"
t.bars_held = sl_hit_idx + 1
exit_price = t.sl
bar_open = opens_arr[i, sl_hit_idx]
# Gap through the stop: the bar opened beyond SL → filled at open.
if not np.isnan(bar_open):
if (t.direction == "BUY" and bar_open < t.sl) or \
(t.direction == "SELL" and bar_open > t.sl):
exit_price = bar_open
t.exit_price = exit_price
t.exit_time = exit_times[i][sl_hit_idx]
elif tp_hit_idx < n_valid:
t.exit_reason = "TP1"
t.bars_held = tp_hit_idx + 1
exit_price = t.tp1
bar_open = opens_arr[i, tp_hit_idx]
# Gap beyond the target in our favor → filled at open.
if not np.isnan(bar_open):
if (t.direction == "BUY" and bar_open > t.tp1) or \
(t.direction == "SELL" and bar_open < t.tp1):
exit_price = bar_open
t.exit_price = exit_price
t.exit_time = exit_times[i][tp_hit_idx]
else:
# Time out at last simulated bar.
t.exit_reason = "TIMEOUT"
t.bars_held = int(n_valid)
last_idx = int(n_valid) - 1
t.exit_price = float(closes_arr[i, last_idx])
t.exit_time = exit_times[i][last_idx]
if not np.isfinite(t.exit_price):
# NaN close on the final window bar (corrupt raw data)
# must not become a NaN R.
t.exit_reason = "NO_DATA"
t.exit_price = None
continue
# Realized R from actual fill prices, net of round-trip spread.
risk = abs(t.entry_price - t.sl)
if risk > 0 and t.exit_price is not None:
pnl = ((t.exit_price - t.entry_price) if t.direction == "BUY"
else (t.entry_price - t.exit_price))
t.r_multiple = (pnl - spread) / risk
# MFE / MAE measured only over bars [0 .. bars_held).
bars = t.bars_held
risk = abs(t.entry_price - t.sl)
if bars > 0 and risk > 0:
h_win = highs_arr[i, :bars]
l_win = lows_arr[i, :bars]
if t.direction == "BUY":
mfe = (np.nanmax(h_win) - t.entry_price) / risk
mae = (t.entry_price - np.nanmin(l_win)) / risk
else:
mfe = (t.entry_price - np.nanmin(l_win)) / risk
mae = (np.nanmax(h_win) - t.entry_price) / risk
t.mfe_r = float(max(0.0, mfe))
t.mae_r = float(max(0.0, mae))
def _resolve_trades_gpu(
trades: List[Trade],
opens_arr: np.ndarray,
highs_arr: np.ndarray,
lows_arr: np.ndarray,
closes_arr: np.ndarray,
valid_lens: np.ndarray,
exit_times: List[List[pd.Timestamp]],
spread: float,
trail_arm_r: float,
trail_dist_r: float,
) -> None:
"""CUDA twin of `_simulate_trailing` (trail_arm_r > 0) and of the static
SL/TP1 branch of `simulate_trades`: the kernel walks every trade's window
in its own thread and returns exit index/price/reason, bars held, R and
the excursions; this host loop only writes them onto the Trade objects,
with the same 'not set' cases as the CPU code."""
trailing = trail_arm_r > 0
entries = np.array([t.entry_price for t in trades], dtype=np.float64)
sls = np.array([t.sl for t in trades], dtype=np.float64)
tps = np.array([t.tp1 for t in trades], dtype=np.float64)
is_buy = np.array([t.direction == "BUY" for t in trades], dtype=np.int8)
arms = np.array([t.trail_arm_r if t.trail_arm_r > 0 else trail_arm_r
for t in trades], dtype=np.float64)
dists = np.array([t.trail_dist_r if t.trail_dist_r > 0 else trail_dist_r
for t in trades], dtype=np.float64)
res = _gk.resolve_trades(
opens_arr, highs_arr, lows_arr, closes_arr, valid_lens,
entries, sls, tps, is_buy, arms, dists, spread, trailing)
for i, t in enumerate(trades):
code = int(res["reason"][i])
if code == _gk.REASON_NO_DATA:
t.exit_reason = "NO_DATA"
continue
bars = int(res["bars_held"][i])
exit_idx = int(res["exit_idx"][i])
if code == _gk.REASON_NO_DATA_AFTER_TIMEOUT:
# Static path: the TIMEOUT bookkeeping ran before the NaN close
# was seen, so bars_held and exit_time are kept and the price
# is None (never a NaN R in the book).
t.exit_reason = "NO_DATA"
t.bars_held = bars
t.exit_price = None
t.exit_time = exit_times[i][exit_idx]
continue
t.exit_reason = _gk.REASON_NAMES[code]
t.bars_held = bars
t.exit_price = float(res["exit_price"][i])
t.exit_time = exit_times[i][exit_idx]
t.r_multiple = float(res["r"][i])
t.mfe_r = float(res["mfe"][i])
t.mae_r = float(res["mae"][i])
# ═════════════════════════════════════════════════════════════════════
# BACKTEST RUNNER
# ═════════════════════════════════════════════════════════════════════
@dataclass
class BacktestConfig:
pair: str
start: datetime
end: datetime
primary_tf: Timeframe = Timeframe.M15
confirmation_tfs: List[Timeframe] = field(
default_factory=lambda: [Timeframe.H1, Timeframe.H4]
)
warmup_bars: int = 250 # min bars before first signal (EMA200 warmup)
cooldown_bars: int = 5 # min bars between trades on same pair
max_hold_bars: int = 300 # cap simulation length per trade
# (300: trailing exits need room to run;
# 200→300 adds ~10R on 12mo M5)
use_cache: bool = True
quality_filter: List[str] = field(
default_factory=lambda: ["A+", "A", "B", "C"]
)
# Restrict to specific strategy tags (XAU / XAU_PB / GENERIC).
# Empty = take all. Use to A/B test whether GENERIC trades add or destroy edge.
strategies_only: List[str] = field(default_factory=list)
# Round-trip transaction cost in price units, charged on every trade
# (XAU/USD retail spread ≈ $0.30). 0 disables.
spread: float = 0.30
# RESEARCH (2026-08): >0 replaces the exit model with a hard TP at
# this multiple of risk from entry (2.0 = fixed 1:2 RR); trailing is
# disabled for ALL trades, including per-trade XAU_PB overrides.
# Entries/stops/gates are untouched, so the signal book is identical
# to the baseline — only the exit policy changes. Never shipped.
fixed_rr_tp: float = 0.0
# 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
# RESEARCH: also compute position_sizing.build_features() at each
# entry and append them to --export-csv. Off by default — the extra
# columns would break comparison against the pinned 63-column books,
# and the feature build costs time per candidate.
export_features: bool = False
# RESEARCH (--record DIR): write manifest.json + trades/indicators/
# events parquet describing every accepted trade to this directory.
# Read-only w.r.t. the signal path — the book is unchanged.
record_dir: Optional[str] = None
# With record_dir set: also store one compact row per ANALYZED
# primary bar (bars.parquet), not just the accepted trades.
record_bars: bool = False
def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
"""Apply XAU weekend + allowed-weekday filter matching live `analyze_pair`.
Market closed Friday 17:00 -> Sunday 18:00 NEW YORK time (= Fri
21:00 UTC in US summer, 22:00 in winter — matches measured Oanda
feed behavior; 2026-08-04 audit I43). Must match `_is_forex_weekend`
in signal_aggregator.py exactly (rule 4; Module 21 agreement test).
Note the weekday gates below stay on UTC weekday — they mirror the
live gates, which evaluate UTC weekdays.
"""
xau = cfg.xau_strategy
if xau.weekend_filter:
ny = ts.astimezone(_NY_TZ)
wd = ny.weekday()
if wd == 4 and ny.hour >= 17:
return "WEEKEND_FILTER"
if wd == 5:
return "WEEKEND_FILTER"
if wd == 6 and ny.hour < 18:
return "WEEKEND_FILTER"
if xau.allowed_weekdays and ts.weekday() not in xau.allowed_weekdays:
return "DAY_FILTER"
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.
Returns (skipped, candidates, bar_rows, cache_fallbacks) — the last
element is this span's GPU-cache-miss count, which the caller sums so
a parity hazard inside a worker process cannot stay invisible.
"""
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"]
# One worker process handles several spans, so the engine's cache-miss
# counter must be per span, not cumulative across them.
engine = aggregator.tech_engine
if hasattr(engine, "fallback_count"):
engine.fallback_count = 0
skipped: Dict[str, int] = {}
candidates: List[Tuple[int, Trade]] = []
is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
record = bool(bt_cfg.record_dir)
record_bars = record and bt_cfg.record_bars
bar_buf = replay_record.new_bar_buffer() if record_bars else None
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,
)
# Live parity: analyze_pair applies the optional higher-TF ADX
# regime filter after delegation, under this exact guard. Calling
# the SAME method on the SAME sliced dfs keeps the two paths
# identical if the (default-off) filter is ever enabled.
if platform_cfg.backtest.regime_filter:
aggregator._apply_regime_filter_live(report, slice_dfs)
# RESEARCH (--record-bars): every ANALYZED bar, setup or not.
if record_bars:
replay_record.add_bar_row(
bar_buf, pos, ts, report,
bt_cfg.primary_tf, bt_cfg.confirmation_tfs)
if report.trade_setup is None:
continue
if report.signal_quality not in bt_cfg.quality_filter:
continue
# Strategy tag: the generator's own stamp (XAU_ATR / XAU_PB /
# XAU_CP / GENERIC), the same string the live log stores in
# setup_json.strategy. Until 2026-09-17 this was inferred from the
# text of the FIRST reason ("XAU STRATEGY" -> "XAU"), which
# labelled every gold ATR setup whose first reason was another
# indicator — and every SELL — as GENERIC (live review §6.3).
# Label-only: entries, exits and R never depended on it, so the
# books are trade-identical; the CSV `strategy` column changed
# and the reference MD5s were re-pinned with that proof.
tag = str(getattr(report.trade_setup, "strategy", "") or "GENERIC")
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
# RESEARCH: decision-time features (same frames the aggregator
# just saw, so live == backtest by construction). Deliberately
# NOT wrapped in try/except — build_features already swallows its
# own failures, so anything escaping here is a real bug.
feats: Dict[str, float] = {}
if bt_cfg.export_features:
feats = position_sizing.build_features(
report, ts_setup, slice_dfs, bt_cfg.primary_tf,
list(bt_cfg.confirmation_tfs),
)
# RESEARCH (--record): plain-data decision snapshot. Read-only —
# it never touches the report or the setup.
snap = (replay_record.snapshot_report(
report, ts_setup, decision_time,
bt_cfg.primary_tf, list(bt_cfg.confirmation_tfs))
if record else {})
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,
features=feats,
snapshot=snap,
stake_multiplier=float(ts_setup.stake_multiplier
if ts_setup.stake_multiplier is not None
else 1.0),
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,
(replay_record.finish_bar_rows(bar_buf) if record_bars else None),
int(getattr(engine, "fallback_count", 0)))
def _reanchor_entries_to_raw(
trades: List[Trade],
signal_ts_list: List[pd.Timestamp],
sim_df: pd.DataFrame,
) -> None:
"""Entry = the first RAW bar strictly after the signal bar.
The replay picks pos+1 in the CLEANED frame. When cleaning dropped a
>10σ spike bar right after the signal bar, that entry skipped the spike
entirely — a fill (and an adverse bar) the live trade would have taken.
SL/TP distances are preserved relative to the corrected entry price,
matching the next-open adjustment in the replay loop.
"""
sim_index = sim_df.index
sim_open = sim_df["Open"].to_numpy(dtype=np.float64)
moved = 0
for t, sig_ts in zip(trades, signal_ts_list):
raw_pos = int(sim_index.searchsorted(sig_ts, side="right"))
if raw_pos >= len(sim_index):
continue # no raw bar after signal — resolution reports NO_DATA
new_time = sim_index[raw_pos]
if new_time == t.entry_time:
continue
new_price = float(sim_open[raw_pos])
if np.isnan(new_price):
continue
sl_dist = abs(t.entry_price - t.sl)
tp1_dist = abs(t.tp1 - t.entry_price)
tp2_dist = abs(t.tp2 - t.entry_price)
if t.direction == "BUY":
t.sl = new_price - sl_dist
t.tp1 = new_price + tp1_dist
t.tp2 = new_price + tp2_dist
else:
t.sl = new_price + sl_dist
t.tp1 = new_price - tp1_dist
t.tp2 = new_price - tp2_dist
t.entry_time = new_time
t.entry_price = new_price
moved += 1
if moved:
logger.info(
f"Re-anchored {moved} entries to the raw price path "
f"(cleaned frame was missing their next bar — usually a >10σ spike)"
)
def _apply_cooldown(
candidates: List[Tuple[int, Trade]], cooldown_bars: int
) -> List[Tuple[int, Trade]]:
"""Ordered cooldown pass over (bar_pos, Trade) candidates: drop any
candidate within `cooldown_bars` bars of the previously ACCEPTED
entry — exactly what a sequential replay loop would have skipped.
Candidates must already be sorted by position (stable merge from the
worker spans)."""
accepted: List[Tuple[int, Trade]] = []
last_entry_pos = -10_000
for pos, trade in candidates:
if pos - last_entry_pos < cooldown_bars:
continue
accepted.append((pos, trade))
last_entry_pos = pos
return accepted
def _apply_fixed_rr_tp(trades: List[Trade], mult: float) -> None:
"""RESEARCH: replace each trade's TP with a hard target at
`mult` × |entry − sl| and zero the per-trade trail params so the
XAU_PB overrides cannot resurrect trailing. Risk is measured off the
final (re-anchored) entry/SL, so call this after
`_reanchor_entries_to_raw`. The caller must also pass
trail_arm_r=0 to `simulate_trades` — dispatch is on the run-level arg."""
for t in trades:
risk = abs(t.entry_price - t.sl)
t.tp1 = (t.entry_price + mult * risk if t.direction == "BUY"
else t.entry_price - mult * risk)
t.tp2 = t.tp1 # static branch never reads tp2; CSV hygiene
t.trail_arm_r = 0.0
t.trail_dist_r = 0.0
t.rr_planned = mult
if t.pip_value:
t.reward_pips = mult * risk / t.pip_value
def run_backtest(
bt_cfg: BacktestConfig,
platform_cfg: Optional[PlatformConfig] = None,
) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]:
"""Replay history and collect trades."""
if platform_cfg is None:
platform_cfg = PlatformConfig(
symbol=bt_cfg.pair,
display_name=PAIR_INFO[bt_cfg.pair]["display"],
primary_timeframe=bt_cfg.primary_tf,
confirmation_timeframes=list(bt_cfg.confirmation_tfs),
)
elif platform_cfg.primary_timeframe != bt_cfg.primary_tf:
# A silent mismatch yields 0 trades: the replayer walks
# bt_cfg.primary_tf bars while the aggregator looks for
# platform_cfg.primary_timeframe in the slices and never finds it.
raise ValueError(
f"primary_tf mismatch: BacktestConfig={bt_cfg.primary_tf.value} "
f"vs PlatformConfig={platform_cfg.primary_timeframe.value}"
)
provider = make_data_provider(platform_cfg)
# ── Load all required timeframes ──
timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs)
# Live parity: analyze_pair also fetches the regime-filter TF when the
# optional H4-ADX gate is enabled. Without it here, M5/M15 primaries
# (whose confirm maps lack H4) would make _apply_regime_filter_live
# silently no-op in the replay while live applies the gate.
flt = platform_cfg.backtest
if flt.regime_filter and flt.regime_filter_timeframe not in timeframes:
timeframes.append(flt.regime_filter_timeframe)
# Same for the daily-trend veto: analyze_pair fetches D1 as a required
# frame when it is on, and the core reads it from the slices (the
# as-of slicing in _replay_span hands it closed daily bars only).
if daily_trend_needed(platform_cfg) and Timeframe.D1 not in timeframes:
timeframes.append(Timeframe.D1)
dfs_full: Dict[Timeframe, pd.DataFrame] = {}
# Which parquet each TF came from (recorded in --record's manifest so
# a study can prove two runs read the same bars).
cache_names: Dict[str, str] = {}
# Per-TF history start; the raw-path fetch below reuses the primary's
# so the cleaned and raw frames come from the SAME cache file.
tf_starts: Dict[Timeframe, datetime] = {}
for tf in timeframes:
tf_start = bt_cfg.start - timedelta(
days=_history_pad_days(tf, bt_cfg.warmup_bars))
tf_starts[tf] = tf_start
cache_names[tf.value] = _cache_path(
bt_cfg.pair, tf, tf_start, bt_cfg.end, provider.name).name
df = fetch_history(
provider, bt_cfg.pair, tf,
start=tf_start, end=bt_cfg.end,
use_cache=bt_cfg.use_cache,
)
if df.empty:
raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}")
dfs_full[tf] = df
# ── Pre-compute indicators on each TF ──
t0 = time.perf_counter()
caches: Dict[str, IndicatorCache] = {
tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True)
for tf in timeframes
}
logger.info(
f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s "
f"({'CuPy GPU' if _gpu_on() else 'NumPy CPU'})")
# The replay workers are forked below and must never touch the device:
# every cache array is already host NumPy; hand pooled memory back too.
BACKEND.release_device_memory()
# ── Inject GPU engine into a fresh aggregator ──
aggregator = SignalAggregator(platform_cfg)
aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches)
# ── Walk primary TF bar-by-bar ──
primary_df = dfs_full[bt_cfg.primary_tf]
# bt_cfg.start/end are already tz-aware UTC datetimes; specifying tz=
# again would raise. Just wrap.
bt_start_ts = pd.Timestamp(bt_cfg.start)
bt_end_ts = pd.Timestamp(bt_cfg.end)
# Start at warmup_bars OR first bar inside the bt window (whichever later).
first_eligible_pos = max(
bt_cfg.warmup_bars,
int(primary_df.index.searchsorted(bt_start_ts)),
)
last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right"))
last_pos = min(last_pos, len(primary_df))
logger.info(
f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: "
f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}"
)
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()
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))]
skipped_reasons: Dict[str, int] = {}
candidates: List[Tuple[int, Trade]] = []
bar_chunks: List = []
cache_fallbacks = 0
for skips, cands, bars, fallbacks in results:
for reason, count in skips.items():
skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
candidates.extend(cands)
if bars is not None:
bar_chunks.append(bars)
cache_fallbacks += fallbacks
candidates.sort(key=lambda pc: pc[0])
# The workers' counters die with their processes, so surface the run
# total here and leave it on the parent engine for callers to read.
aggregator.tech_engine.fallback_count = cache_fallbacks
if cache_fallbacks > 0:
logger.warning(
f"GPU cache fallbacks: {cache_fallbacks} indicator recomputes "
f"(one per _compute_* override per missed bar) fell back to the "
f"live engine's pandas math (different EWM seeding) — parity "
f"hazard; see GPUTechnicalEngine._note_fallback")
else:
logger.info("GPU cache fallbacks: 0")
# ── Phase 2: entry cooldown — the only sequential rule.
accepted = _apply_cooldown(candidates, bt_cfg.cooldown_bars)
trades: List[Trade] = [t for _, t in accepted]
signal_ts_list: List[pd.Timestamp] = [primary_df.index[p] for p, _ in accepted]
replay_seconds = time.perf_counter() - t_loop
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"{workers} worker{'s' if workers != 1 else ''})"
)
if skipped_reasons:
logger.info(f"Skipped bars by reason: {skipped_reasons}")
# ── Resolve trade outcomes ──
# Resolve against the RAW price path: cleaning removes >10σ spike bars
# (news candles) that hit stops in reality. Cache hit → no API call.
sim_df = fetch_history(
provider, bt_cfg.pair, bt_cfg.primary_tf,
start=tf_starts[bt_cfg.primary_tf], end=bt_cfg.end,
use_cache=bt_cfg.use_cache, clean=False,
)
_reanchor_entries_to_raw(trades, signal_ts_list, sim_df)
if bt_cfg.fixed_rr_tp > 0:
_apply_fixed_rr_tp(trades, bt_cfg.fixed_rr_tp)
logger.info(
f"RESEARCH override: fixed-RR TP = {bt_cfg.fixed_rr_tp} x risk; "
f"trailing exit DISABLED for all {len(trades)} trades (incl. XAU_PB)"
)
run_arm = 0.0 if bt_cfg.fixed_rr_tp > 0 else platform_cfg.signals.trail_arm_r
t_sim = time.perf_counter()
simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars,
spread=bt_cfg.spread,
trail_arm_r=run_arm,
trail_dist_r=platform_cfg.signals.trail_dist_r)
logger.info(
f"Trade simulation took {time.perf_counter() - t_sim:.2f}s "
f"({'CuPy GPU' if _gpu_on() else 'NumPy CPU'})"
)
if bt_cfg.record_dir:
replay_record.write_run(
bt_cfg.record_dir, trades=trades, book_df=trades_to_dataframe(trades),
bt_cfg=bt_cfg, platform_cfg=platform_cfg, aggregator=aggregator,
run_info=dict(first_pos=first_eligible_pos, last_pos=last_pos,
n_primary_bars=len(primary_df), n_bars_in_window=n_bars,
skipped=skipped_reasons, n_candidates=len(candidates),
workers=workers,
backend=BACKEND.describe(),
cache_files=cache_names, replay_seconds=replay_seconds,
cache_fallbacks=cache_fallbacks),
sim_df=sim_df, bar_chunks=bar_chunks or None)
logger.info(f"--record: wrote run to {bt_cfg.record_dir}")
return trades, dfs_full
# ═════════════════════════════════════════════════════════════════════
# METRICS & REPORT
# ═════════════════════════════════════════════════════════════════════
@dataclass
class Metrics:
n_trades: int
n_wins: int
n_losses: int
n_timeout: int
win_rate: float
avg_r: float
median_r: float
total_r: float
profit_factor: float
sharpe: float
max_drawdown_r: float
avg_bars_held: float
avg_mfe_r: float
avg_mae_r: float
by_quality: Dict[str, Dict]
by_strategy: Dict[str, Dict]
by_regime: Dict[str, Dict]
def _summarize(rs: List[float]) -> Dict:
if not rs:
return {"n": 0, "win_rate": 0.0, "avg_r": 0.0, "total_r": 0.0}
arr = np.array(rs)
wins = (arr > 0).sum()
return {
"n": len(rs),
"win_rate": float(wins / len(rs)) * 100,
"avg_r": float(arr.mean()),
"total_r": float(arr.sum()),
}
def compute_metrics(trades: List[Trade]) -> Metrics:
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
# Last line of defense: one non-finite R (NaN exit on corrupt data)
# would silently NaN every aggregate below. Exclude loudly.
bad = [t for t in resolved if not np.isfinite(t.r_multiple)]
if bad:
logger.error(
"compute_metrics: excluding %d trade(s) with non-finite R "
"(corrupt exit data) — first at %s", len(bad), bad[0].entry_time)
resolved = [t for t in resolved if np.isfinite(t.r_multiple)]
if not resolved:
return Metrics(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, {}, {}, {})
rs = np.array([t.r_multiple for t in resolved])
wins_mask = rs > 0
losses_mask = rs < 0
gross_profit = float(rs[wins_mask].sum()) if wins_mask.any() else 0.0
gross_loss = float(-rs[losses_mask].sum()) if losses_mask.any() else 0.0
pf = gross_profit / gross_loss if gross_loss > 0 else float("inf")
# Prepend the flat starting equity so an opening losing streak counts
# as drawdown from 0, not from the first trade's already-negative cum.
cum = np.concatenate(([0.0], np.cumsum(rs)))
high_water = np.maximum.accumulate(cum)
drawdown = high_water - cum
max_dd = float(drawdown.max()) if len(drawdown) else 0.0
sharpe = float(rs.mean() / rs.std()) if rs.std() > 0 else 0.0
by_quality = {q: _summarize([t.r_multiple for t in resolved if t.quality == q])
for q in sorted({t.quality for t in resolved})}
by_strategy = {s: _summarize([t.r_multiple for t in resolved if t.strategy_tag == s])
for s in sorted({t.strategy_tag for t in resolved})}
by_regime = {r: _summarize([t.r_multiple for t in resolved if t.market_regime == r])
for r in sorted({t.market_regime for t in resolved})}
return Metrics(
n_trades=len(resolved),
n_wins=int(wins_mask.sum()),
n_losses=int(losses_mask.sum()),
n_timeout=sum(1 for t in resolved if t.exit_reason == "TIMEOUT"),
win_rate=float(wins_mask.mean()) * 100,
avg_r=float(rs.mean()),
median_r=float(np.median(rs)),
total_r=float(rs.sum()),
profit_factor=pf,
sharpe=sharpe,
max_drawdown_r=max_dd,
avg_bars_held=float(np.mean([t.bars_held for t in resolved])),
avg_mfe_r=float(np.mean([t.mfe_r for t in resolved])),
avg_mae_r=float(np.mean([t.mae_r for t in resolved])),
by_quality=by_quality,
by_strategy=by_strategy,
by_regime=by_regime,
)
_DAY_NAMES = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu",
4: "Fri", 5: "Sat", 6: "Sun"}
def _fmt_table(headers: List[str], rows: List[List[str]]) -> str:
"""Render a Markdown-style ASCII table with auto-sized columns."""
if not rows:
return ""
cols = len(headers)
widths = [len(h) for h in headers]
for r in rows:
for i in range(cols):
widths[i] = max(widths[i], len(str(r[i])))
def fmt_row(r):
return "│ " + " │ ".join(str(r[i]).ljust(widths[i]) for i in range(cols)) + " │"
sep = "├" + "┼".join("─" * (w + 2) for w in widths) + "┤"
top = "┌" + "┬".join("─" * (w + 2) for w in widths) + "┐"
bot = "└" + "┴".join("─" * (w + 2) for w in widths) + "┘"
out = [top, fmt_row(headers), sep]
out.extend(fmt_row(r) for r in rows)
out.append(bot)
return "\n".join(out)
def print_report(
pair: str, bt_cfg: BacktestConfig, m: Metrics,
trades: Optional[List[Trade]] = None, detail: bool = False,
) -> None:
bar = "═" * 78
print(f"\n{bar}")
print(f" BACKTEST REPORT — {PAIR_INFO[pair]['display']} "
f"({bt_cfg.start.date()} → {bt_cfg.end.date()})")
print(f" Primary: {bt_cfg.primary_tf.value} "
f"Confirms: {','.join(t.value for t in bt_cfg.confirmation_tfs)}")
print(f"{bar}")
if m.n_trades == 0:
print(" No trades produced — quality filter rejected all signals.")
print(f"{bar}\n")
return
# ── Summary table ──
summary_rows = [
["Trades", f"{m.n_trades}"],
["Wins / Losses / Timeouts", f"{m.n_wins} / {m.n_losses} / {m.n_timeout}"],
["Win rate", f"{m.win_rate:.1f}%"],
["Total R", f"{m.total_r:+.2f}"],
["Avg R / trade", f"{m.avg_r:+.3f}"],
["Median R", f"{m.median_r:+.3f}"],
["Profit factor", f"{m.profit_factor:.2f}"],
["Sharpe (R)", f"{m.sharpe:.2f}"],
["Max DD (R)", f"{m.max_drawdown_r:.2f}"],
["Avg bars held", f"{m.avg_bars_held:.1f}"],
["Avg MFE / MAE", f"{m.avg_mfe_r:.2f}R / {m.avg_mae_r:.2f}R"],
]
print("\n ── Summary ──")
print(_fmt_table(["Metric", "Value"], summary_rows))
def _grouped_table(title: str, groups: Dict[str, Dict]):
if not groups:
return
rows = []
for k, s in groups.items():
rows.append([k, str(s["n"]),
f"{s['win_rate']:5.1f}%",
f"{s['avg_r']:+.3f}",
f"{s['total_r']:+.2f}"])
print(f"\n ── {title} ──")
print(_fmt_table([title.split(" by ")[-1], "n", "WR", "avgR", "totalR"], rows))
_grouped_table("Breakdown by Quality", m.by_quality)
_grouped_table("Breakdown by Strategy", m.by_strategy)
_grouped_table("Breakdown by Regime", m.by_regime)
# ── Per-trade table ──
if trades:
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
if detail or len(resolved) <= 30:
print("\n ── Trade Log ──")
rows = []
for i, t in enumerate(resolved, 1):
rows.append([
str(i),
t.entry_time.strftime("%m-%d %a %H:%M"),
t.direction,
t.strategy_tag,
t.quality,
f"{t.indicators_agreeing}/{t.indicators_total}",
f"{t.final_score:+.1f}",
f"{t.final_confidence:.0f}",
t.market_regime[:8],
t.exit_reason,
str(t.bars_held),
f"{t.r_multiple:+.2f}",
])
print(_fmt_table(
["#", "Entry (UTC)", "Dir", "Strat", "Q", "Agree",
"Score", "Conf", "Regime", "Exit", "Bars", "R"],
rows,
))
else:
print(f"\n Trade log suppressed ({len(resolved)} trades). "
f"Use --detail to print all, or open the CSV.")
print(f"{bar}\n")
def print_trade_reasoning(trades: List[Trade], limit: int = 5) -> None:
"""Per-instructions: dump the *why* for the first/last few trades —
indicators on each TF, agreement count, reasons, warnings."""
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
if not resolved:
return
bar = "─" * 78
print(f"\n{'═' * 78}")
print(f" TRADE REASONING (first {min(limit, len(resolved))} + last "
f"{min(limit, len(resolved))} of {len(resolved)})")
print('═' * 78)
sample = resolved[:limit] + ([] if len(resolved) <= 2 * limit else resolved[-limit:])
for t in sample:
print(f"\n{bar}")
weekday = _DAY_NAMES.get(t.entry_time.weekday(), "?")
print(f" {t.direction} {PAIR_INFO[t.pair]['display']} @ {t.entry_price:.5f} "
f"({weekday} {t.entry_time:%Y-%m-%d %H:%M UTC})")
print(f" Strategy={t.strategy_tag} Quality={t.quality} "
f"Score={t.final_score:+.1f} Confidence={t.final_confidence:.0f} "
f"Regime={t.market_regime}/{t.volatility_regime}")
print(f" SL={t.sl:.5f} TP1={t.tp1:.5f} RR={t.rr_planned:.2f} "
f"({t.risk_pips:.0f}pip risk → {t.reward_pips:.0f}pip target)")
print(f" Indicator agreement: {t.indicators_agreeing}/{t.indicators_total} agree, "
f"{t.indicators_opposing} oppose, {t.indicators_neutral} neutral")
if t.indicator_breakdown:
ind_rows = []
for name, b in t.indicator_breakdown.items():
tag = {"agree": "✓", "oppose": "✗", "neutral": "·"}.get(b["agrees"], "?")
ind_rows.append([
name, b["signal"],
f"{b['value']:.3f}" if b["value"] is not None else "—",
f"{b['confidence']:.0f}",
f"{b['weight']:.1f}",
tag,
b["description"][:40],
])
print(_fmt_table(
["Indicator", "Signal", "Value", "Conf", "Wt", "vs Dir", "Description"],
ind_rows,
))
if t.entry_reasons:
print(f" Entry reasons:")
for r in t.entry_reasons[:5]:
print(f" • {r[:120]}")
if t.entry_warnings:
print(f" Warnings at entry:")
for w in t.entry_warnings[:5]:
print(f" ⚠ {w[:120]}")
weekday_x = _DAY_NAMES.get(t.exit_time.weekday(), "?") if t.exit_time is not None else "?"
print(f" → EXIT: {t.exit_reason} @ {t.exit_price:.5f} after {t.bars_held} bars "
f"({weekday_x} {t.exit_time:%Y-%m-%d %H:%M UTC}) → R={t.r_multiple:+.2f}")
print(f" MFE / MAE: {t.mfe_r:.2f}R / {t.mae_r:.2f}R")
def trades_to_dataframe(trades: List[Trade],
include_features: bool = False) -> pd.DataFrame:
"""Export trades for downstream analysis (e.g. parquet/CSV).
Includes per-instructions: entry reasoning, indicator-by-indicator
signal/value/agreement, and overall agreement count.
`include_features` (research, --export-features) appends
stake_multiplier, r_sized and the position_sizing.FEATURE_ORDER
columns AFTER the existing ones. Default off so the column set stays
byte-identical to the pinned books.
"""
rows = []
for t in trades:
row = {
"pair": t.pair,
"strategy": t.strategy_tag,
"quality": t.quality,
"regime": t.market_regime,
"vol_regime": t.volatility_regime,
"direction": t.direction,
"entry_time": t.entry_time,
"entry_weekday": _DAY_NAMES.get(t.entry_time.weekday(), "?"),
"entry": t.entry_price,
"sl": t.sl,
"tp1": t.tp1,
"rr_planned": t.rr_planned,
"score": t.final_score,
"confidence": t.final_confidence,
"indicators_agree": t.indicators_agreeing,
"indicators_oppose": t.indicators_opposing,
"indicators_neutral": t.indicators_neutral,
"indicators_total": t.indicators_total,
"entry_reasons": " || ".join(t.entry_reasons[:5]),
"warnings": " || ".join(t.entry_warnings[:5]),
"exit_time": t.exit_time,
"exit": t.exit_price,
"exit_reason": t.exit_reason,
"bars_held": t.bars_held,
"r_multiple": t.r_multiple,
"mfe_r": t.mfe_r,
"mae_r": t.mae_r,
}
# Flatten each indicator into columns so spreadsheets can sort/filter.
for name, b in t.indicator_breakdown.items():
row[f"ind_{name}_signal"] = b["signal"]
row[f"ind_{name}_value"] = b["value"]
row[f"ind_{name}_conf"] = b["confidence"]
row[f"ind_{name}_agrees"] = b["agrees"]
if include_features:
row["stake_multiplier"] = t.stake_multiplier
row["r_sized"] = t.r_multiple * t.stake_multiplier
for name in position_sizing.FEATURE_ORDER:
row[name] = t.features.get(name, float("nan"))
rows.append(row)
df = pd.DataFrame(rows)
if include_features and len(df.columns):
# A later trade can introduce an indicator column the first trade
# lacked (pandas orders columns by first appearance), which would
# land AFTER the extras. Force the extras to the tail so the default
# export is always a strict prefix of the feature export.
extras = ["stake_multiplier", "r_sized", *position_sizing.FEATURE_ORDER]
base_cols = [c for c in df.columns if c not in extras]
df = df[base_cols + [c for c in extras if c in df.columns]]
return df
# ═════════════════════════════════════════════════════════════════════
# CLI
# ═════════════════════════════════════════════════════════════════════
def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
p = argparse.ArgumentParser(
description="GPU-accelerated backtester for ForexBot2",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p.add_argument("--pair", required=True, help="XAUUSD | GOLD")
p.add_argument("--days", type=int, default=180, help="Days of history (ignored if --start/--end set)")
p.add_argument("--start", help="ISO date YYYY-MM-DD (overrides --days)")
p.add_argument("--end", help="ISO date YYYY-MM-DD (defaults to today)")
p.add_argument("--primary-tf", default="15min", help="Primary TF (1min/5min/15min/1h/4h)")
p.add_argument("--confirm-tfs", default=None,
help="Comma-separated confirmation TFs. Default: the pair "
"factory's live map (M5→M15,1h | M15→5min,1h | "
"1h→15min,4h | 4h→1h,1day) so backtests replay the "
"exact live config.")
p.add_argument("--warmup-bars", type=int, default=250)
p.add_argument("--cooldown-bars", type=int, default=5)
p.add_argument("--max-hold-bars", type=int, default=300)
p.add_argument("--swing-sr-lookback", type=int, default=0,
help="Override TechnicalConfig.swing_sr_lookback (0 = use config default)")
p.add_argument("--trail-arm", type=float, default=None,
help="Override signals.trail_arm_r for this run (research knob)")
p.add_argument("--trail-dist", type=float, default=None,
help="Override signals.trail_dist_r for this run (research knob)")
p.add_argument("--fixed-rr-tp", type=float, default=None,
help="RESEARCH: exit at a hard TP of MULT x risk (e.g. 2.0 = "
"fixed 1:2 RR) instead of the trailing stop; "
"entries/stops/gates unchanged.")
p.add_argument("--quality", default="A+,A,B,C", help="Comma-separated allowed grades")
p.add_argument("--strategies", default="",
help="Comma-separated strategy tags to include "
"(XAU_ATR,XAU_PB,XAU_CP,GENERIC; the legacy name XAU "
"means XAU_ATR). Empty = all.")
p.add_argument("--cp", action="store_true",
help="Force-enable the chart-pattern entry (XAU_CP) regardless "
"of the factory default — for validating the candidate.")
p.add_argument("--cp-patterns", default=None,
help="Comma-separated pattern subset for --cp (e.g. "
"double_bottom,bull_flag). Default: all detectors.")
p.add_argument("--big-candle", type=float, default=None,
help="Force xau_strategy.big_candle_atr_multiple=K for this "
"run (no entries while the lookback window contains a "
"bar with range >= K x prior ATR) — for validating the "
"candidate.")
p.add_argument("--big-candle-hours", type=float, default=None,
help="Override big_candle_lookback_hours for --big-candle "
"runs (default 24).")
p.add_argument("--set", action="append", default=[], metavar="PATH=JSON",
help="Override any PlatformConfig leaf by dotted path; the "
"value is JSON-parsed for typing (research knob; "
"repeatable). e.g. --set xau_strategy.m5_min_abs_score=25 "
"--set 'xau_strategy.sell_min_quality=[\"A+\",\"A\",\"B\",\"C\"]'")
p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)")
p.add_argument("--spread", type=float, default=None,
help="Round-trip cost in price units per trade (default: pair's "
"default_spread from PAIR_INFO — XAU $0.30; 0 disables)")
p.add_argument("--export-csv", help="Optional path to dump per-trade CSV (incl. indicator breakdown)")
p.add_argument("--export-features", action="store_true",
help="Append the 65 decision-time f_* feature columns "
"(+ stake_multiplier, r_sized) to --export-csv; "
"research flag, default off so pinned 63-column "
"books stay comparable")
p.add_argument("--record", metavar="DIR", default=None,
help="RESEARCH: write manifest.json + trades/indicators/"
"events parquet for every accepted trade to DIR; "
"the book and CSV are unchanged")
p.add_argument("--record-bars", action="store_true",
help="with --record: also write one row per analyzed "
"primary bar (bars.parquet, ~140k rows on 2-yr M5)")
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("--backend", choices=("auto", "gpu", "cpu"), default=None,
help="Compute backend for the indicator pre-compute and the "
"trade resolution: auto = CuPy when CUDA works, else "
"NumPy (the reason is logged); gpu = fail loudly without "
"CUDA; cpu = force the NumPy twins (parity runs). The "
"env var FB2_BACKEND sets the default.")
p.add_argument("--reasoning-samples", type=int, default=3,
help="If --detail not set, print this many sample trades' full reasoning")
p.add_argument("--log-level", default="INFO")
args = p.parse_args(argv)
if args.record_bars and not args.record:
p.error("--record-bars requires --record")
return args
def _resolve_pair(s: str) -> str:
s_up = s.upper().strip()
if s_up in PAIR_ALIASES:
return PAIR_ALIASES[s_up]
if s_up in PAIR_INFO:
return s_up
raise SystemExit(f"Unknown pair '{s}'. Try one of: {', '.join(sorted(PAIR_ALIASES))}")
def _resolve_tf(s: str) -> Timeframe:
for tf in Timeframe:
if tf.value == s:
return tf
raise SystemExit(f"Unknown timeframe '{s}'. Valid: {[t.value for t in Timeframe]}")
def _apply_config_overrides(cfg: PlatformConfig, overrides: List[str]) -> None:
"""Apply repeatable --set PATH=JSON research overrides onto the config.
Unknown paths are a hard error: dataclass instances silently accept
setattr of any name, so a typo'd path would otherwise produce a
no-op run that looks like 'the relaxation had no effect'. Every
applied override is logged — the run log is the config audit trail
required by instructions_for_backtest.txt.
"""
for item in overrides:
path, sep, raw = item.partition("=")
if not sep:
raise SystemExit(f"--set expects PATH=VALUE, got '{item}'")
try:
value = json.loads(raw)
except json.JSONDecodeError:
value = raw # bare-string convenience ("oanda" etc.)
obj = cfg
parts = path.strip().split(".")
for attr in parts[:-1]:
if not hasattr(obj, attr):
raise SystemExit(f"--set: unknown config path '{path}'")
obj = getattr(obj, attr)
leaf = parts[-1]
if not hasattr(obj, leaf):
raise SystemExit(f"--set: unknown config path '{path}'")
old = getattr(obj, leaf)
if hasattr(old, "__dataclass_fields__"):
raise SystemExit(f"--set: '{path}' is a config section, not a leaf")
setattr(obj, leaf, value)
logger.info(f"--set override: {path} = {value!r} (was {old!r})")
def main(argv: Optional[List[str]] = None) -> int:
args = _parse_args(argv)
logging.basicConfig(
level=args.log_level.upper(),
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
if args.backend:
BACKEND.set_mode(args.backend)
pair = _resolve_pair(args.pair)
primary_tf = _resolve_tf(args.primary_tf)
# None = defer to the pair factory's live confirmation map. A hardcoded
# default here would silently backtest a different MTF setup than the
# bot scans (e.g. M15 with 1h+4h instead of the live 5min+1h).
confirm_tfs = (
[_resolve_tf(s.strip()) for s in args.confirm_tfs.split(",") if s.strip()]
if args.confirm_tfs is not None else None
)
# Per-pair tuned configs come from the pair_signals factories (regime
# weights, strategy gates, sessions). Unregistered pairs fall back to
# the inline construction.
from pair_signals import PAIR_PRODUCERS # noqa: E402 (after sys.path fix)
factory = PAIR_PRODUCERS.get(pair)
if factory is not None:
platform_cfg = factory(primary_tf, confirm_tfs)
else:
platform_cfg = PlatformConfig(
symbol=pair,
display_name=PAIR_INFO[pair]["display"],
primary_timeframe=primary_tf,
confirmation_timeframes=confirm_tfs or [Timeframe.H1, Timeframe.H4],
)
# The factory path applies the config.json overlay inside
# make_config; the fallback must match (live == backtest).
from app_config import apply_pair_overrides # noqa: E402
apply_pair_overrides(platform_cfg)
if args.fixed_rr_tp is not None:
if args.fixed_rr_tp <= 0:
raise SystemExit("--fixed-rr-tp must be > 0")
if (args.trail_arm is not None or args.trail_dist is not None or
any(s.split("=", 1)[0].strip().startswith(("signals.trail",
"signals.pb_trail"))
for s in (args.set or []))):
raise SystemExit(
"--fixed-rr-tp conflicts with --trail-arm/--trail-dist/"
"--set signals.*trail* (the flag disables trailing entirely)")
if args.swing_sr_lookback > 0:
platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback
if args.trail_arm is not None:
platform_cfg.signals.trail_arm_r = args.trail_arm
if args.trail_dist is not None:
platform_cfg.signals.trail_dist_r = args.trail_dist
if args.cp:
platform_cfg.xau_strategy.cp_enabled = True
if args.cp_patterns:
platform_cfg.xau_strategy.cp_patterns = [
s.strip() for s in args.cp_patterns.split(",") if s.strip()
]
if args.big_candle is not None:
platform_cfg.xau_strategy.big_candle_atr_multiple = args.big_candle
if args.big_candle_hours is not None:
platform_cfg.xau_strategy.big_candle_lookback_hours = args.big_candle_hours
if args.set:
_apply_config_overrides(platform_cfg, args.set)
confirm_tfs = list(platform_cfg.confirmation_timeframes)
end = (
datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc)
if args.end else datetime.now(timezone.utc)
)
if args.start:
start = datetime.fromisoformat(args.start).replace(tzinfo=timezone.utc)
else:
start = end - timedelta(days=args.days)
spread = (args.spread if args.spread is not None
else PAIR_INFO[pair].get("default_spread", 0.30))
bt_cfg = BacktestConfig(
pair=pair,
start=start,
end=end,
primary_tf=primary_tf,
confirmation_tfs=confirm_tfs,
warmup_bars=args.warmup_bars,
cooldown_bars=args.cooldown_bars,
max_hold_bars=args.max_hold_bars,
use_cache=not args.no_cache,
quality_filter=[q.strip() for q in args.quality.split(",") if q.strip()],
strategies_only=[("XAU_ATR" if s.strip() == "XAU" else s.strip())
for s in args.strategies.split(",") if s.strip()],
spread=spread,
workers=args.workers,
fixed_rr_tp=args.fixed_rr_tp or 0.0,
export_features=args.export_features,
record_dir=args.record,
record_bars=args.record_bars,
)
if args.record:
logger.info(
f"--record ON: decision snapshots → {args.record} "
f"(trades/indicators/events"
f"{'/bars' if args.record_bars else ''} parquet + manifest.json; "
f"the trade book is unchanged)"
)
if args.export_features:
logger.info(
f"--export-features ON: appending stake_multiplier, r_sized and "
f"{len(position_sizing.FEATURE_ORDER)} f_* feature columns to the "
f"CSV export (research flag; book itself is unchanged)"
)
logger.info(
f"Backend: {BACKEND.describe()} | "
f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()} | "
f"Confirm TFs: {','.join(tf.value for tf in confirm_tfs)} | Spread: {spread}"
)
trades, _ = run_backtest(bt_cfg, platform_cfg=platform_cfg)
metrics = compute_metrics(trades)
print_report(pair, bt_cfg, metrics, trades=trades, detail=args.detail)
print_trade_reasoning(trades, limit=args.reasoning_samples)
if args.export_csv:
df = trades_to_dataframe(trades, include_features=args.export_features)
df.to_csv(args.export_csv, index=False)
logger.info(f"Exported {len(df)} trades → {args.export_csv}")
return 0
if __name__ == "__main__":
sys.exit(main())