Audit fixes: cache boundary staleness, 6e session bypass, GPU engine parity
Fixes all high+medium findings from the 2026-07-16 full-codebase audit. High: - data_provider: cache now rejects snapshots missing the latest closed bar (_has_latest_closed_bar, weekend-exempt). The wall-clock TTL served mid-candle snapshots at the next boundary, so live M15/H1 scans ran one full bar behind and the bar-close session check evaluated the wrong hour. - signal_aggregator: gate 6e (SELL quality) only acts on actionable grades. It was re-grading session-blocked F reports to D, which re-opened the M5 pullback path (entry check: quality != "F") on session-blocked bars. - gpu_backtest: port the live EMA-ribbon compression/expansion logic (+10 conf on expansion) into the GPU engine. Its absence diverged composite score/conf on 7-9% of bars and flipped signals on ~0.2%. Medium: - gpu_backtest: Bollinger precompute uses the exact pandas rolling calls (cumsum E[x2]-E[x]2 lost precision at gold price levels; bb_position error up to 7.6e-3 crossed signal thresholds, worst during squeezes). - gpu_backtest: --confirm-tfs defaults to the pair factory's live confirmation map instead of a hardcoded 1h,4h. - gpu_backtest: entries re-anchor to the first RAW bar after the signal bar, so >10-sigma spike bars dropped by cleaning are no longer skipped at entry (favorable bias). - gpu_backtest: per-TF warmup pad — D1 confirms get >=210 bars at window start (62-day pad gave ~61 bars: no EMA200, unlike live). - data_provider: thread-safety locks for cache + throttle (shared provider across the bot's 4-thread pool); outlier filter keeps the snap-back bar after a spike instead of punching a two-bar hole. - telegram_bot: alert fingerprint gains a risk-relative entry bucket (materially-moved setups re-alert) and is committed only after a successful send (failed sends no longer consume the alert for 1h). - engine: disabled_indicators=["macd"] now removes the MACD pillar from scoring (histogram still computed for divergence). Verification: test_all.py 187/187 (new Module 11: 8 audit-fix regressions incl. a live-vs-GPU parity sweep with ribbon coverage); real-data parity sweep over 813 slices of 14mo XAU M15: zero divergence (pre-fix: 7-9% of bars diverged); both original bug repros no longer reproduce; cached end-to-end backtest run passes with the live confirm map. Books shift slightly vs pre-fix runs — comparability note added to CLAUDE.md; pinned tuned numbers need re-validation under the fixed backtester. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Harwinder Sekhon
+127
-89
@@ -159,69 +159,6 @@ def _gpu_ewm_span(x: np.ndarray, span: int) -> np.ndarray:
|
||||
return _gpu_ewm_alpha(x, alpha, min_periods=0)
|
||||
|
||||
|
||||
def _gpu_rolling_mean(x: np.ndarray, window: int) -> np.ndarray:
|
||||
"""Rolling mean via cumsum-diff, NaN-aware (matches pandas
|
||||
`.rolling(window).mean()` default min_periods=window — returns NaN
|
||||
if any value in the window is NaN)."""
|
||||
n = x.shape[0]
|
||||
out = np.full(n, np.nan, dtype=np.float64)
|
||||
if n < window:
|
||||
return out
|
||||
xg = _to_gpu(x.astype(np.float64))
|
||||
nan_mask = cp.isnan(xg)
|
||||
x_filled = cp.where(nan_mask, 0.0, xg)
|
||||
cs = cp.cumsum(x_filled)
|
||||
valid_cnt = cp.cumsum((~nan_mask).astype(cp.int64))
|
||||
|
||||
sum_w = cp.empty(n - window + 1, dtype=cp.float64)
|
||||
cnt_w = cp.empty(n - window + 1, dtype=cp.int64)
|
||||
sum_w[0] = cs[window - 1]
|
||||
sum_w[1:] = cs[window:] - cs[:-window]
|
||||
cnt_w[0] = valid_cnt[window - 1]
|
||||
cnt_w[1:] = valid_cnt[window:] - valid_cnt[:-window]
|
||||
|
||||
# pandas: NaN if any window value is NaN (cnt < window).
|
||||
rolling = cp.where(cnt_w < window, cp.nan, sum_w / window)
|
||||
out[window - 1:] = _to_cpu(rolling)
|
||||
return out
|
||||
|
||||
|
||||
def _gpu_rolling_std(x: np.ndarray, window: int, ddof: int = 0) -> np.ndarray:
|
||||
"""Rolling std via cumsum of x and x^2, NaN-aware. ddof=0 matches
|
||||
pandas std(ddof=0). NaN if any window value is NaN."""
|
||||
n = x.shape[0]
|
||||
out = np.full(n, np.nan, dtype=np.float64)
|
||||
if n < window:
|
||||
return out
|
||||
xg = _to_gpu(x.astype(np.float64))
|
||||
nan_mask = cp.isnan(xg)
|
||||
x_filled = cp.where(nan_mask, 0.0, xg)
|
||||
x2_filled = x_filled * x_filled
|
||||
cs = cp.cumsum(x_filled)
|
||||
cs2 = cp.cumsum(x2_filled)
|
||||
valid_cnt = cp.cumsum((~nan_mask).astype(cp.int64))
|
||||
|
||||
sum_x = cp.empty(n - window + 1, dtype=cp.float64)
|
||||
sum_x2 = cp.empty(n - window + 1, dtype=cp.float64)
|
||||
cnt_w = cp.empty(n - window + 1, dtype=cp.int64)
|
||||
sum_x[0] = cs[window - 1]
|
||||
sum_x[1:] = cs[window:] - cs[:-window]
|
||||
sum_x2[0] = cs2[window - 1]
|
||||
sum_x2[1:] = cs2[window:] - cs2[:-window]
|
||||
cnt_w[0] = valid_cnt[window - 1]
|
||||
cnt_w[1:] = valid_cnt[window:] - valid_cnt[:-window]
|
||||
|
||||
mean = sum_x / window
|
||||
var = sum_x2 / window - mean * mean
|
||||
var = cp.maximum(var, 0.0) # numerical safety against tiny negatives
|
||||
if ddof:
|
||||
var = var * window / max(window - ddof, 1)
|
||||
std = cp.sqrt(var)
|
||||
std = cp.where(cnt_w < window, cp.nan, std)
|
||||
out[window - 1:] = _to_cpu(std)
|
||||
return out
|
||||
|
||||
|
||||
@dataclass
|
||||
class IndicatorCache:
|
||||
"""Pre-computed indicator series for one (symbol, timeframe).
|
||||
@@ -319,8 +256,18 @@ def precompute_indicators(
|
||||
macd_hist_series = pd.Series(macd_hist, index=df.index, name="macd_hist")
|
||||
|
||||
# ── Bollinger ──
|
||||
bb_mid = _gpu_rolling_mean(close, tech_cfg.bb_period)
|
||||
bb_std = _gpu_rolling_std(close, tech_cfg.bb_period, ddof=0)
|
||||
# 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
|
||||
@@ -330,7 +277,10 @@ def precompute_indicators(
|
||||
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 = _gpu_rolling_mean(bb_band_width, 20)
|
||||
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)
|
||||
@@ -595,7 +545,26 @@ class GPUTechnicalEngine(TechnicalEngine):
|
||||
crossover_desc = " | Death cross (fast/medium)"
|
||||
conf = min(conf + 10, 95)
|
||||
|
||||
desc = f"EMA alignment: {alignment_score}/{max_checks}{crossover_desc}"
|
||||
# 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,
|
||||
@@ -1647,6 +1616,53 @@ def _replay_span(span: Tuple[int, int]):
|
||||
return skipped, candidates
|
||||
|
||||
|
||||
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 run_backtest(
|
||||
bt_cfg: BacktestConfig,
|
||||
platform_cfg: Optional[PlatformConfig] = None,
|
||||
@@ -1679,9 +1695,17 @@ def run_backtest(
|
||||
|
||||
dfs_full: Dict[Timeframe, pd.DataFrame] = {}
|
||||
for tf in timeframes:
|
||||
# Slow TFs need a longer pad: 60 days of D1 is only ~44 bars — no
|
||||
# EMA200, truncated swing/divergence windows — while live always has
|
||||
# ≥210 bars (outputsize floor). Target data_provider's min_required
|
||||
# (210 bars); ×7/5 converts trading days to calendar days. Intraday
|
||||
# TFs stay on the base pad, so their books are unchanged.
|
||||
bars_per_day = ForexDataProvider.BARS_PER_DAY[tf]
|
||||
min_pad_days = int(np.ceil(210 / bars_per_day * 7 / 5)) + 5
|
||||
tf_start = bt_cfg.start - timedelta(days=max(pad_days, min_pad_days))
|
||||
df = fetch_history(
|
||||
provider, bt_cfg.pair, tf,
|
||||
start=fetch_start, end=bt_cfg.end,
|
||||
start=tf_start, end=bt_cfg.end,
|
||||
use_cache=bt_cfg.use_cache,
|
||||
)
|
||||
if df.empty:
|
||||
@@ -1763,11 +1787,13 @@ def run_backtest(
|
||||
# inside the cooldown window of the previously ACCEPTED entry is
|
||||
# dropped, exactly as the sequential loop skipped those bars.
|
||||
trades: List[Trade] = []
|
||||
signal_ts_list: List[pd.Timestamp] = []
|
||||
last_entry_pos = -10_000
|
||||
for pos, trade in candidates:
|
||||
if pos - last_entry_pos < bt_cfg.cooldown_bars:
|
||||
continue
|
||||
trades.append(trade)
|
||||
signal_ts_list.append(primary_df.index[pos])
|
||||
last_entry_pos = pos
|
||||
|
||||
logger.info(
|
||||
@@ -1787,6 +1813,7 @@ def run_backtest(
|
||||
start=fetch_start, end=bt_cfg.end,
|
||||
use_cache=bt_cfg.use_cache, clean=False,
|
||||
)
|
||||
_reanchor_entries_to_raw(trades, signal_ts_list, sim_df)
|
||||
t_sim = time.perf_counter()
|
||||
simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars,
|
||||
spread=bt_cfg.spread,
|
||||
@@ -2110,7 +2137,11 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
|
||||
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="1h,4h", help="Comma-separated confirmation TFs")
|
||||
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)
|
||||
@@ -2160,7 +2191,32 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
|
||||
pair = _resolve_pair(args.pair)
|
||||
primary_tf = _resolve_tf(args.primary_tf)
|
||||
confirm_tfs = [_resolve_tf(s.strip()) for s in args.confirm_tfs.split(",") if s.strip()]
|
||||
# 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"],
|
||||
pip_value=PAIR_INFO[pair]["pip_value"],
|
||||
primary_timeframe=primary_tf,
|
||||
confirmation_timeframes=confirm_tfs or [Timeframe.H1, Timeframe.H4],
|
||||
)
|
||||
if args.swing_sr_lookback > 0:
|
||||
platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback
|
||||
confirm_tfs = list(platform_cfg.confirmation_timeframes)
|
||||
|
||||
end = (
|
||||
datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc)
|
||||
@@ -2193,27 +2249,9 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
logger.info(
|
||||
f"Backend: {'CuPy GPU' if _GPU_AVAILABLE else 'NumPy CPU fallback'} | "
|
||||
f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()} | "
|
||||
f"Spread: {spread}"
|
||||
f"Confirm TFs: {','.join(tf.value for tf in confirm_tfs)} | Spread: {spread}"
|
||||
)
|
||||
|
||||
# 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"],
|
||||
pip_value=PAIR_INFO[pair]["pip_value"],
|
||||
primary_timeframe=primary_tf,
|
||||
confirmation_timeframes=confirm_tfs,
|
||||
)
|
||||
if args.swing_sr_lookback > 0:
|
||||
platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback
|
||||
|
||||
trades, _ = run_backtest(bt_cfg, platform_cfg=platform_cfg)
|
||||
metrics = compute_metrics(trades)
|
||||
print_report(pair, bt_cfg, metrics, trades=trades, detail=args.detail)
|
||||
|
||||
Reference in New Issue
Block a user