From fe7d4c5d4a2e6496ed8fda23ea6643fa80c846ea Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Thu, 16 Jul 2026 10:24:39 -0400 Subject: [PATCH] Audit fixes: cache boundary staleness, 6e session bypass, GPU engine parity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- CLAUDE.md | 22 ++-- backtest/gpu_backtest.py | 216 +++++++++++++++++++++--------------- data_provider.py | 87 +++++++++++---- engine.py | 4 + signal_aggregator.py | 4 + telegram_bot.py | 53 +++++++-- test_all.py | 231 +++++++++++++++++++++++++++++++++++++++ 7 files changed, 483 insertions(+), 134 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 66636b9..a85ac10 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,7 +2,7 @@ Signal-intelligence platform for **XAU/USD (gold — live, validated)** and **EUR/USD (functional but OFF by default — no validated edge, see below)**. It fetches OHLCV from Twelve Data, scores 9 technical indicators per timeframe with regime-adaptive weights, checks multi-timeframe agreement and news/calendar risk, grades the signal (A+…F), and generates a full trade setup (entry, SL, TP1/TP2, trailing plan). Delivery: Telegram bot (auto-scanner + subscriber channel) and a CLI. A realism-focused backtester replays history through the *exact same* signal code. -This file is the authoritative documentation. `DOCUMENTATION.md` is stale (May 2026) — do not trust it. `backtest/instructions_for_backtest.txt` lists what every backtest report must record. +This file is the authoritative developer/agent reference. The human-facing docs live in `docs/` (`user_guide.html` + `beginner_guide.html`, and their PDF builds) — keep them in sync when signal behavior changes. `backtest/instructions_for_backtest.txt` lists what every backtest report must record. **Read this whole file before changing signal logic.** Every numeric threshold below was either backtested or is load-bearing for a backtested result. Rules for changes are at the bottom ("Rules for making changes"). @@ -22,7 +22,7 @@ This file is the authoritative documentation. `DOCUMENTATION.md` is stale (May 2 | `pair_signals/xauusd.py` | 71 | XAU config factory + per-timeframe overrides (docstring = validation record) | | `pair_signals/eurusd.py` | 158 | EUR config factory + EUR regime weights (docstring = no-edge verdict + tuning history) | | `main.py` | ~348 | CLI: `python main.py --pair XAUUSD --timeframe H1 --detail/--json` | -| `test_all.py` | ~2361 | QA suite: **179 tests in 10 modules**, no live API needed. Must stay green. | +| `test_all.py` | ~2570 | QA suite: **187 tests in 11 modules**, no live API needed. Must stay green. | | `backtest/gpu_backtest.py` | ~2231 | Bar-replay backtester (CuPy→NumPy fallback) using the real `SignalAggregator` | | `backtest/diagnose_filters.py` | 609 | Filter-diagnosis tooling | | `backtest/research/` | — | Calendar research infra (`ff_calendar_scraper.py`, `calendar_store.py`) + result CSVs | @@ -67,7 +67,7 @@ Signal currency everywhere is the `SignalStrength` enum: `STRONG_BUY=2, BUY=1, N | `default_spread` | 0.30 | 0.0001 | round-trip cost charged by the backtester when `--spread` not given | ### TechnicalConfig (indicator parameters) -EMA 9/21/50/200 (`ema_mode`: `"4ema"` default or `"2ema"` = 50/200 only); RSI 14 (OB 70 / OS 30 / neutral zone 40–60; `rsi_mode`: `"regime_aware"` default or `"classic"`); MACD 12/26/9 (`macd_as_pillar=True` — MACD is a scored indicator, not just divergence input); BB 20/2.0σ; ATR 14; ADX 14 (strong ≥25, very strong ≥50); divergence lookback 40; swing-S/R lookback 50; ROC period 10, threshold 1.0%; Keltner: EMA 20 mid ± 1.5 × ATR(10); ADX slope lookback 5, "emerging" threshold 3.0; `disabled_indicators` list; `regime_weights_override` dict (used by EUR). +EMA 9/21/50/200 (`ema_mode`: `"4ema"` default or `"2ema"` = 50/200 only); RSI 14 (OB 70 / OS 30 / neutral zone 40–60; `rsi_mode`: `"regime_aware"` default or `"classic"`); MACD 12/26/9 (`macd_as_pillar=True` — MACD is a scored indicator, not just divergence input); BB 20/2.0σ; ATR 14; ADX 14 (strong ≥25, very strong ≥50); divergence lookback 40; swing-S/R lookback 50; ROC period 10, threshold 1.0%; Keltner: EMA 20 mid ± 1.5 × ATR(10); ADX slope lookback 5, "emerging" threshold 3.0; `disabled_indicators` list (removes the pillar from scoring; for MACD the histogram is still computed — divergence needs it); `regime_weights_override` dict (used by EUR). ### SignalConfig (scoring & trade geometry — XAU-calibrated; EUR uses its own copies) | Field | Default | Meaning | @@ -193,7 +193,7 @@ Two entry points: - **6b session filter** (bar-label hour + weekday; XAU reads `xau_strategy.blocked_hours_utc`/`allowed_weekdays`, others read `config.backtest.*`) → NEUTRAL/F. - **6c transitioning boost**: regime `transitioning` and grade B/C/D → A if |score|≥30 & conf≥50 & MTF; else B if |score|≥25 & conf≥45 (early trend entries). - **6d exhausted-trend cap**: trending regime + exhaustion_count ≥2 + grade C/D → D, setup killed. - - **6e XAU SELL quality gate**: sell-direction & grade not in `sell_min_quality` → D. + - **6e XAU SELL quality gate**: sell-direction & grade actionable (A+..C) & not in `sell_min_quality` → D. The actionable guard is load-bearing: without it a session-blocked F report was *upgraded* to D, re-opening the M5 pullback path (its entry check is `!= "F"`). - **6f MTF opposition**: ≥2 TFs pointing against the trade direction → D (never trade against consensus). - **6g XAU SELL volatility gate**: sell-direction & vol regime not in {high, extreme} → D. - **6g2 XAU regime gates**: regime in `blocked_regimes` (M15: mixed `trending`) → D; sell-direction & regime in `sell_blocked_regimes` → D. @@ -232,10 +232,10 @@ Two entry points: - Twelve Data SDK (`TDClient.time_series(...).as_pandas()`), symbols normalized to `"XAU/USD"` form; renames lowercase columns → capitalized; sorts oldest-first; localizes UTC. - **Rate limit**: 55 credits/min sliding window (`_throttle`), blocks with sleep when exhausted. -- **In-memory cache**: key `{symbol}_{tf}`, TTL = candle duration (M5 300 s … H4 14400 s); stale cached data (freshness check) is evicted and refetched. +- **In-memory cache**: key `{symbol}_{tf}`, TTL = candle duration (M5 300 s … H4 14400 s); a cached frame is evicted and refetched if it fails the freshness check **or the boundary check** (`_has_latest_closed_bar`: the frame must contain the most recent fully closed bar, weekend-exempt — a wall-clock TTL alone would serve mid-candle snapshots at the next boundary and scans would run one bar behind). Cache and throttle are lock-protected (one provider is shared across the bot's thread pool). - **`_drop_forming_candle`**: Twelve Data includes the current partial candle; the scanner fires seconds after boundary, so the last bar is dropped if its close time is in the future. Without this every indicator would read a seconds-old stub and signals would repaint. - **Freshness** (`_is_data_fresh`): last bar older than 3× candle duration → stale (warns; live path suppresses signal if the PRIMARY TF is stale). Weekends exempt (Fri 22:00→Sun 22:00 UTC). -- **`_validate_and_clean`**: drop all-NaN rows → ffill (limit 3) → drop NaN closes → sort + dedupe (keep last) → drop future timestamps (>now+5 min) → repair OHLC inconsistencies (High=max, Low=min of O/C/existing) → remove >10σ return outliers → warn if < ~210 bars. +- **`_validate_and_clean`**: drop all-NaN rows → ffill (limit 3) → drop NaN closes → sort + dedupe (keep last) → drop future timestamps (>now+5 min) → repair OHLC inconsistencies (High=max, Low=min of O/C/existing) → remove >10σ return outliers (spike bar only — the opposite-direction snap-back bar right after it is kept, else every spike punched a two-bar hole) → warn if < ~210 bars. - `outputsize = max(300, min(days×bars_per_day+20, 5000))`; TD caps at 5000/call (the backtester paginates by date). - `get_latest_price` / `get_spread_estimate` (median M5 High−Low) helpers. @@ -265,7 +265,7 @@ Run: `python telegram_bot.py` (env `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGR - **Owner lock**: every command/button is wrapped in `_owner_only` — only `TELEGRAM_CHAT_ID` gets responses; others get a refusal. The bot is the owner's control panel; the product is the channel. - **Subscriber channel** (`TELEGRAM_CHANNEL_ID`, `-100…`; `""` disables): auto-scan signal alerts + breach alerts also post there (bot needs channel-admin). Monetization = Telegram Stars subscription invite links; Telegram handles membership, the bot has zero payment code. The channel feed keeps running even when the owner toggles DM auto-scan off (`_feed_active()` = DM chats OR channel). - **Auto-scan**: boot-starts if `TELEGRAM_CHAT_ID` set (initial M15 scan populates the level cache first). Candle-boundary loops for M5/M15/H1/H4, firing at boundary + 10 s. Each scan runs all `scan_pairs` concurrently, updates the level cache, then broadcasts. -- **Alert gating**: quality threshold (`_alert_min_quality`, default "C", menu cycles C→B→A) + fingerprint dedup per (symbol, TF) on (signal, quality, direction) with 1-hour re-arm — a persisting identical signal doesn't re-send every candle. +- **Alert gating**: quality threshold (`_alert_min_quality`, default "C", menu cycles C→B→A) + fingerprint dedup per (symbol, TF) on (signal, quality, direction, entry bucket) with 1-hour re-arm — a persisting identical signal doesn't re-send every candle, but a setup that moved ≥ half its risk distance (new pullback/swing anchor) re-alerts. The fingerprint is committed only **after** a message reaches ≥1 target (`_commit_alert`), so a failed Telegram send doesn't consume the alert. - **WebSocket monitor**: subscribes every `scan_pairs` symbol to TD quotes WS; 10 s heartbeat; exponential reconnect (≤60 s). On tick crossing any cached key level (`swing_high/low_1..3`, `ema_trend`, `bb_upper/lower`, `kc_upper/lower`; cache TTL 1 h) → immediate M15 rescan → "LEVEL BREAK" alert if actionable+changed (same gates as auto-scan), 5-min per-pair cooldown. - **Commands**: `/start` (menu), `/scan` (buttons: per pair — M5/M15/H1/H4 + Scan All TFs), `/autoscan` (toggle DM copy), `/status` (loops, WS, live prices, level cache, cooldowns), `/cache` (age per pair/TF), `/clearcache`, `/help`. - Signal messages include the full indicator breakdown, warnings, trade setup, **trailing-stop management plan** (arm price + trail distance, from Signal/pb params by `strategy`), and a risk disclaimer on every outbound message. @@ -284,7 +284,7 @@ Run: `python telegram_bot.py` (env `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGR Replays history bar-by-bar through the **real `SignalAggregator.analyze_from_dataframes`** — same gates, same setup math. News is disabled (`use_news=False`): backtests are pure technical. **Realism model (all validated fixes — do not regress):** -- Entry at the **next bar's open** after the signal bar; SL/TP distances preserved relative to actual entry. +- Entry at the **next bar's open** after the signal bar; SL/TP distances preserved relative to actual entry. Entries are **re-anchored to the raw price path** (`_reanchor_entries_to_raw`): if cleaning dropped a >10σ spike bar right after the signal bar, the trade enters at the spike bar's open, not the bar after it — skipping it was a favorable bias. - Round-trip `--spread` subtracted from every trade's PnL before computing R (default = pair's `default_spread`: XAU $0.30, EUR 1 pip). Results before 2026-07-08 (+59.5R baseline era) charged no spread — not comparable. - Trade resolution runs against the **RAW price path** (`fetch_history(clean=False)`): the parquet cache stores raw bars; cleaning (which drops >10σ news spikes that hit stops in reality) is applied only to the signal-side data. Caches written before this fix contain cleaned data — refetch to restore spike bars. - Same-bar SL+TP tie → SL first (conservative). Gap through SL/TP → filled at the bar open. @@ -306,13 +306,15 @@ python backtest/gpu_backtest.py --pair EURUSD --primary-tf 5min --confirm-tfs 15 ``` Report: summary metrics (total R, PF, Sharpe on R, max DD, MFE/MAE), breakdowns by quality/strategy/regime, trade log, and per-trade reasoning dumps (indicator-by-indicator agreement at entry — required by `instructions_for_backtest.txt`). `--export-csv` flattens every indicator's signal/value/agreement into columns. -Data: paginated TD fetches (5000-bar chunks) cached as parquet in `backtest/cache/` keyed by symbol/TF/window; warmup pad ≥60 days before the window for EMA200. +**Comparability note (2026-07-16):** the parity/realism fixes of that date (EMA-ribbon port to the GPU engine, exact-pandas Bollinger precompute, raw-path entry re-anchoring, per-TF D1 warmup, live confirm-map CLI default, outlier snap-back retention) all shift books slightly vs earlier runs. Books produced before these fixes are not directly comparable — re-run the baseline before any before/after tuning comparison. + +Data: paginated TD fetches (5000-bar chunks) cached as parquet in `backtest/cache/` keyed by symbol/TF/window; warmup pad ≥60 days before the window for EMA200, extended per-TF so slow confirmation TFs get ≥210 bars at window start (D1 gets ~299 days — 62 days gave only ~61 bars: no EMA200, unlike live's 300-bar outputsize floor). `--confirm-tfs` defaults to the pair factory's live confirmation map (pass it explicitly to override); a hardcoded default here previously made the standard invocation backtest a different MTF setup than the bot scans. --- ## test_all.py — QA suite -`python test_all.py` — **179 tests, 10 modules** (config, data_provider, engine, news_engine, signal_aggregator, main, telegram_bot, integration, pair_signals registry + engine-bug regressions, EUR/USD support). No network needed (mocked). Uses a custom `@test("name")` decorator that runs immediately (NOT pytest — don't grep for `def test_`). Exit code non-zero on failure. Includes: pair-isolation tests (no gold logic on EUR configs and vice versa), the xauusd-factory ≡ inline-config equivalence test, and engine bug-fix regression tests (Module 9b). **Any signal-logic change must keep this green and should add tests.** +`python test_all.py` — **187 tests, 11 modules** (config, data_provider, engine, news_engine, signal_aggregator, main, telegram_bot, integration, pair_signals registry + engine-bug regressions, EUR/USD support, 2026-07-16 audit-fix regressions). No network needed (mocked). Uses a custom `@test("name")` decorator that runs immediately (NOT pytest — don't grep for `def test_`). Exit code non-zero on failure. Includes: pair-isolation tests (no gold logic on EUR configs and vice versa), the xauusd-factory ≡ inline-config equivalence test, engine bug-fix regression tests (Module 9b), and a live-vs-GPU engine parity sweep incl. the EMA-ribbon path (Module 11). **Any signal-logic change must keep this green and should add tests.** --- diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 1388698..ac4be6d 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -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) diff --git a/data_provider.py b/data_provider.py index 3cdfeaa..69e5256 100755 --- a/data_provider.py +++ b/data_provider.py @@ -6,6 +6,7 @@ Fetches and validates OHLCV via the official `twelvedata` Python SDK. import logging import os +import threading import time from collections import deque from typing import Dict, Optional @@ -75,6 +76,10 @@ class ForexDataProvider: ) self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} + # The bot shares one provider across a thread pool; cache eviction + # and the throttle deque are check-then-act and need locking. + self._cache_lock = threading.Lock() + self._throttle_lock = threading.Lock() self._request_times: deque = deque() self._td: Optional[TDClient] = ( TDClient(apikey=self._api_key) if self._api_key else None @@ -103,18 +108,22 @@ class ForexDataProvider: now = time.time() cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval) - if ( - cache_key in self._cache_timestamps - and (now - self._cache_timestamps[cache_key]) < cache_ttl - and cache_key in self._cache - ): + with self._cache_lock: cached = self._cache.get(cache_key) - if cached is not None and not cached.empty: - if self._is_data_fresh(cached, symbol, timeframe): - return cached - logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching") - del self._cache[cache_key] - del self._cache_timestamps[cache_key] + cached_at = self._cache_timestamps.get(cache_key) + if ( + cached is not None + and cached_at is not None + and (now - cached_at) < cache_ttl + and not cached.empty + ): + if (self._is_data_fresh(cached, symbol, timeframe) + and self._has_latest_closed_bar(cached, timeframe)): + return cached + logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching") + with self._cache_lock: + self._cache.pop(cache_key, None) + self._cache_timestamps.pop(cache_key, None) outputsize = max( 300, @@ -130,10 +139,27 @@ class ForexDataProvider: f"last bar {df.index[-1]} is too old. Signal quality may be degraded." ) - self._cache[cache_key] = df - self._cache_timestamps[cache_key] = now + with self._cache_lock: + self._cache[cache_key] = df + self._cache_timestamps[cache_key] = now return df + def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool: + """Cached data must include the most recent fully closed bar. + + The TTL alone is wall-clock based: a snapshot fetched mid-candle is + still within TTL at the next candle boundary but is missing the bar + that just closed, so every boundary scan would analyze — and run the + bar-close session check against — the previous bar. + """ + now = pd.Timestamp.now(tz="UTC") + if _is_forex_weekend(now): + return True + candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) + dur = pd.Timedelta(seconds=candle_seconds) + expected_label = now.floor(dur) - dur + return df.index[-1] >= expected_label + def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame: """Drop the still-forming last candle so analysis only sees closed bars. @@ -209,19 +235,21 @@ class ForexDataProvider: def _throttle(self, credits: int = 1) -> None: """Pro tier: 55 credits / minute. Block until budget allows.""" while True: - now = time.time() - while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: - self._request_times.popleft() + with self._throttle_lock: + now = time.time() + while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: + self._request_times.popleft() - available = self.RATE_LIMIT - len(self._request_times) - if available >= credits: - for _ in range(credits): - self._request_times.append(time.time()) - return + available = self.RATE_LIMIT - len(self._request_times) + if available >= credits: + for _ in range(credits): + self._request_times.append(time.time()) + return - wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 + used = len(self._request_times) + wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 logger.info( - f"Rate limit: {len(self._request_times)}/{self.RATE_LIMIT} credits used " + f"Rate limit: {used}/{self.RATE_LIMIT} credits used " f"— waiting {wait:.1f}s..." ) time.sleep(wait) @@ -324,7 +352,18 @@ class ForexDataProvider: returns = df["Close"].pct_change() std_ret = returns.std() if std_ret > 0: - outlier_mask = (returns - returns.mean()).abs() > 10 * std_ret + dev = returns - returns.mean() + outlier_mask = dev.abs() > 10 * std_ret + # A spike bar makes the NEXT bar's return look extreme too (it is + # measured against the spike). If the previous bar is flagged and + # this bar reverses direction, it is the legitimate snap-back — + # keep it, or every spike would punch a two-bar hole in the data. + recovery = ( + outlier_mask + & outlier_mask.shift(1, fill_value=False) + & (dev * dev.shift(1) < 0) + ) + outlier_mask = outlier_mask & ~recovery if outlier_mask.any(): logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars") df = df[~outlier_mask] diff --git a/engine.py b/engine.py index 095971e..1287658 100755 --- a/engine.py +++ b/engine.py @@ -92,7 +92,11 @@ class TechnicalEngine: if "bollinger" not in disabled: self._compute_bollinger(df, result) if "adx" not in disabled: self._compute_adx(df, result) + # MACD always runs — divergence needs the histogram — but the + # scored pillar must still honour disabled_indicators. self._compute_macd(df, result) + if "macd" in disabled: + result.indicators.pop("macd", None) if "roc" not in disabled: self._compute_roc(df, result) if "keltner" not in disabled: self._compute_keltner(df, result) diff --git a/signal_aggregator.py b/signal_aggregator.py index ff97f2d..7a234c7 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -739,9 +739,13 @@ class SignalAggregator: report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL) or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0) ) + # Actionable grades only: without this guard the gate would "upgrade" + # a session-blocked F report to D, which re-opens the M5 pullback + # path (its entry check is quality != "F"). if (self.config.xau_strategy.enabled and report.display_name == "XAU/USD" and xau_is_sell_dir + and report.signal_quality in ("A+", "A", "B", "C") and report.signal_quality not in self.config.xau_strategy.sell_min_quality): report.warnings.append( f"SELL FILTER: quality {report.signal_quality} below minimum for SELL" diff --git a/telegram_bot.py b/telegram_bot.py index 40c528c..067724c 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -488,21 +488,41 @@ def _passes_quality(quality: str) -> bool: return _QUALITY_RANK.get(quality, 0) >= _QUALITY_RANK.get(_alert_min_quality, 1) -def _should_alert(key: str, report: SignalReport) -> bool: - """True when this signal is new/changed for the key, or re-armed.""" +def _alert_fingerprint(report: SignalReport) -> tuple: + """Alert identity for dedup. Includes a risk-relative entry bucket: + a persisting signal whose entry drifts a few ticks between candles + keeps the same fingerprint, while a setup that moved by half its risk + distance or more (new pullback, new swing anchor) re-alerts instead + of being swallowed for the full re-arm hour.""" setup = report.trade_setup - fp = ( + entry_bucket = None + if setup is not None: + sl_dist = abs(setup.entry_price - setup.stop_loss) + entry_bucket = round(setup.entry_price / max(sl_dist * 0.5, 1e-9)) + return ( report.final_signal.name, report.signal_quality, setup.direction if setup else None, + entry_bucket, ) - now = time.time() - if (_last_alert_fp.get(key) == fp - and now - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS): - return False - _last_alert_fp[key] = fp - _last_alert_ts[key] = now - return True + + +def _should_alert(key: str, report: SignalReport) -> bool: + """True when this signal is new/changed for the key, or re-armed. + + Pure check — call _commit_alert only after the message actually went + out, so a failed Telegram send doesn't consume the alert for an hour. + """ + fp = _alert_fingerprint(report) + return not ( + _last_alert_fp.get(key) == fp + and time.time() - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS + ) + + +def _commit_alert(key: str, report: SignalReport) -> None: + _last_alert_fp[key] = _alert_fingerprint(report) + _last_alert_ts[key] = time.time() # ── Candle-boundary scheduler ──────────────────────────────────────── @@ -563,13 +583,20 @@ async def _scan_timeframe( if _passes_quality(r.signal_quality) and _should_alert(f"{r.symbol}_{tf.value}", r) ] + sent_ok: Set[int] = set() for chat_id in _signal_targets(): - for report in actionable: + for i, report in enumerate(actionable): detail = format_signal_message(report) try: await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN) + sent_ok.add(i) except Exception as exc: logger.warning(f"Failed to send signal for chat {chat_id}: {exc}") + # Commit fingerprints only for alerts that reached at least one target; + # a fully failed send stays eligible for the next scan. + for i, report in enumerate(actionable): + if i in sent_ok: + _commit_alert(f"{report.symbol}_{tf.value}", report) # ── WebSocket level-breach monitor ────────────────────────────────── @@ -658,11 +685,15 @@ async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, le f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n" ) detail = header + format_signal_message(report) + sent = False for chat_id in _signal_targets(): try: await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN) + sent = True except Exception as exc: logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}") + if sent: + _commit_alert(f"{pair_symbol}_{Timeframe.M15.value}", report) async def _ws_heartbeat(ws) -> None: diff --git a/test_all.py b/test_all.py index 40ce9f9..0db3c2d 100644 --- a/test_all.py +++ b/test_all.py @@ -2341,6 +2341,237 @@ def _(): assert eur_report.display_name == "EUR/USD" +# ══════════════════════════════════════════════════════════════════════ +# MODULE 11: 2026-07-16 audit-fix regressions +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 11: 2026-07-16 audit fixes ═══") + +_BT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backtest") +if _BT_DIR not in sys.path: + sys.path.insert(0, _BT_DIR) + + +def make_xau_bars(n=400, end="2026-07-08 16:55", freq="5min", drift=-0.8, seed=7): + """Seeded XAU-priced bars ending at a FIXED timestamp (deterministic).""" + rng = np.random.default_rng(seed) + idx = pd.date_range(end=pd.Timestamp(end, tz="UTC"), periods=n, freq=freq) + close = 3400 + np.cumsum(drift + rng.normal(0, 0.5, n)) + open_ = np.roll(close, 1) + open_[0] = close[0] + high = np.maximum(open_, close) + rng.uniform(0.5, 2.0, n) + low = np.minimum(open_, close) - rng.uniform(0.5, 2.0, n) + df = pd.DataFrame({ + "Open": open_, "High": high, "Low": low, + "Close": close, "Volume": rng.uniform(1e3, 5e3, n), + }, index=idx) + df.index.name = "Datetime" + return df + + +@test("Gate 6e: session-blocked sell-direction report stays F (no F→D upgrade)") +def _(): + # Bar LABELED in a blocked hour (16:xx UTC, Wednesday) with a bearish + # score: gate 6b sets F; before the fix, 6e re-graded it D, which + # re-opened the M5 pullback path (entry check: quality != "F"). + from pair_signals.xauusd import make_config as xau_make_config + primary = make_xau_bars(end="2026-07-08 16:55", freq="5min") + m15 = make_xau_bars(end="2026-07-08 16:45", freq="15min") + h1 = make_xau_bars(end="2026-07-08 16:00", freq="60min") + agg = SignalAggregator(xau_make_config(Timeframe.M5, [Timeframe.M15, Timeframe.H1])) + report = agg.analyze_from_dataframes( + XAUUSD, {Timeframe.M5: primary, Timeframe.M15: m15, Timeframe.H1: h1}, + use_news=False, + ) + assert any("SESSION FILTER" in w for w in report.warnings), \ + f"session filter did not fire: {report.warnings}" + assert report.primary_score < 0, "test setup must produce a sell-direction score" + assert report.signal_quality == "F", \ + f"blocked bar must stay F, got {report.signal_quality}" + assert report.trade_setup is None + assert not any("SELL FILTER: quality" in w for w in report.warnings), \ + "6e must not fire on non-actionable grades" + + +@test("GPU engine parity: composite + ema_trend identical to live (incl. ribbon)") +def _(): + from gpu_backtest import precompute_indicators, GPUTechnicalEngine + from engine import TechnicalEngine + cfg = PlatformConfig() + # Two-phase series: flat chop (EMA ribbon compresses) then a strong + # trend (ribbon expands) — guarantees the sweep crosses both ribbon + # states, the exact logic the GPU engine was missing. + rng = np.random.default_rng(11) + n = 650 + drift = np.where(np.arange(n) < 450, 0.0, 1.2) + idx = pd.date_range(end=pd.Timestamp("2026-06-30 12:00", tz="UTC"), + periods=n, freq="60min") + close = 3400 + np.cumsum(drift + rng.normal(0, 0.5, n)) + open_ = np.roll(close, 1) + open_[0] = close[0] + high = np.maximum(open_, close) + rng.uniform(0.5, 2.0, n) + low = np.minimum(open_, close) - rng.uniform(0.5, 2.0, n) + df = pd.DataFrame({"Open": open_, "High": high, "Low": low, "Close": close, + "Volume": rng.uniform(1e3, 5e3, n)}, index=idx) + df.index.name = "Datetime" + cache = precompute_indicators(df, cfg.technical) + gpu = GPUTechnicalEngine(cfg.technical, {"1h": cache}) + live = TechnicalEngine(cfg.technical) + seen_compressed = seen_expanding = False + for i in range(250, n + 1, 3): + s = df.iloc[:i] + r_live = live.analyze(s, "XAU/USD", "1h") + r_gpu = gpu.analyze(s, "XAU/USD", "1h") + assert r_live.composite_signal == r_gpu.composite_signal, f"signal flip @ {i}" + assert abs(r_live.composite_score - r_gpu.composite_score) < 1e-9, f"score @ {i}" + assert abs(r_live.confidence - r_gpu.confidence) < 1e-9, f"confidence @ {i}" + el = r_live.indicators["ema_trend"] + eg = r_gpu.indicators["ema_trend"] + assert el.confidence == eg.confidence, f"ema_trend conf @ {i}" + assert el.description == eg.description, f"ema_trend desc @ {i}" + bl = r_live.indicators["bollinger"] + bg = r_gpu.indicators["bollinger"] + assert bl.signal == bg.signal and bl.description == bg.description, f"bollinger @ {i}" + if "Ribbon COMPRESSED" in el.description: + seen_compressed = True + if "Ribbon EXPANDING" in el.description: + seen_expanding = True + assert seen_compressed, "sweep never saw a compressed ribbon — no teeth" + assert seen_expanding, "sweep never saw a ribbon expansion (+10 conf path) — no teeth" + + +@test("disabled_indicators=['macd'] removes the pillar but keeps the histogram") +def _(): + from engine import TechnicalEngine + from config import TechnicalConfig + df = make_xau_ohlcv(n=300, seed=5) + result = TechnicalEngine(TechnicalConfig(disabled_indicators=["macd"])).analyze( + df, "XAU/USD", "1h") + assert "macd" not in result.indicators, "disabled MACD still scored" + assert getattr(result, "_macd_histogram", None) is not None, \ + "histogram must survive for divergence" + baseline = TechnicalEngine(TechnicalConfig()).analyze(df, "XAU/USD", "1h") + assert "macd" in baseline.indicators + + +@test("Outlier filter drops the spike bar but keeps the snap-back bar") +def _(): + provider = ForexDataProvider(PlatformConfig()) + df = make_ohlcv(n=300, seed=3) + k = 150 + spike_ts, after_ts = df.index[k], df.index[k + 1] + spike_val = df["Close"].iloc[k - 1] * 1.05 # ~+5% on a ~0.1%-σ series + df.loc[spike_ts, ["Open", "High", "Low", "Close"]] = spike_val + cleaned = provider._validate_and_clean(df, "EUR/USD", Timeframe.H1) + assert spike_ts not in cleaned.index, "spike bar must be removed" + assert after_ts in cleaned.index, \ + "snap-back bar after the spike must be kept (two-bar hole regression)" + + +@test("Cache: pre-boundary snapshot refetched; boundary-complete snapshot served") +def _(): + def bars_ending(last): + idx = pd.date_range(end=last, periods=300, freq="15min", tz="UTC") + v = np.linspace(3000.0, 3100.0, 300) + return pd.DataFrame({"Open": v, "High": v + 1, "Low": v - 1, + "Close": v, "Volume": 1000.0}, index=idx) + + for _attempt in range(2): + provider = ForexDataProvider(PlatformConfig()) + started = pd.Timestamp.now(tz="UTC").floor("15min") + last_closed = started - pd.Timedelta(minutes=15) + stale = bars_ending(last_closed - pd.Timedelta(minutes=15)) + fresh = bars_ending(last_closed) + key = "XAU/USD_15min" + with patch("data_provider._is_forex_weekend", return_value=False): + # (a) snapshot cached mid-candle 5 min ago, missing the + # just-closed bar: TTL says fresh, boundary check must refetch. + provider._cache[key] = stale + provider._cache_timestamps[key] = time.time() - 300 + with patch.object(provider, "_fetch_twelve_data", + return_value=fresh.copy()) as mock_fetch: + served = provider.fetch_ohlcv("XAU/USD", Timeframe.M15) + # (b) cache now boundary-complete: served without any fetch. + with patch.object(provider, "_fetch_twelve_data", + side_effect=AssertionError("must not refetch")): + served2 = provider.fetch_ohlcv("XAU/USD", Timeframe.M15) + if pd.Timestamp.now(tz="UTC").floor("15min") != started: + continue # boundary crossed mid-test — retry once + assert mock_fetch.called, "stale pre-boundary snapshot was served from cache" + assert served.index[-1] == last_closed + assert served2.index[-1] == last_closed + break + + +@test("Alert dedup: drift dedups, materially-moved setup re-alerts, no early commit") +def _(): + import telegram_bot as tb + from signal_aggregator import SignalReport, TradeSetup + + def rpt(entry, sl): + r = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="t") + r.final_signal = SignalStrength.BUY + r.signal_quality = "B" + r.trade_setup = TradeSetup( + pair="XAU/USD", display_name="XAU/USD", direction="BUY", + entry_price=entry, stop_loss=sl, take_profit_1=entry + 12, + take_profit_2=entry + 20, risk_reward=1.5, + risk_pips=80.0, reward_pips=120.0, pip_value=0.10, + ) + return r + + key = "TEST_ALERT_FP" + tb._last_alert_fp.pop(key, None) + tb._last_alert_ts.pop(key, None) + r1 = rpt(3400.0, 3392.0) # risk $8 → $4 buckets + assert tb._should_alert(key, r1) + assert tb._should_alert(key, r1), "_should_alert must not commit state itself" + tb._commit_alert(key, r1) + assert not tb._should_alert(key, r1), "identical signal must dedup" + assert not tb._should_alert(key, rpt(3400.5, 3392.5)), \ + "sub-bucket drift must dedup" + assert tb._should_alert(key, rpt(3412.0, 3404.0)), \ + "setup moved 1.5x risk — must re-alert" + tb._last_alert_fp.pop(key, None) + tb._last_alert_ts.pop(key, None) + + +@test("Backtester --confirm-tfs defaults to None (factory's live map)") +def _(): + from gpu_backtest import _parse_args + args = _parse_args(["--pair", "XAUUSD"]) + assert args.confirm_tfs is None + + +@test("Backtest entries re-anchor to the first RAW bar after the signal bar") +def _(): + from gpu_backtest import Trade, _reanchor_entries_to_raw + idx = pd.date_range("2026-01-05 10:00", periods=6, freq="5min", tz="UTC") + opens = np.array([100.0, 101.0, 110.0, 102.0, 103.0, 104.0]) + sim = pd.DataFrame({"Open": opens, "High": opens + 1, + "Low": opens - 1, "Close": opens}, index=idx) + + def mk_trade(entry_time, entry_price): + return Trade(pair="XAU/USD", entry_time=entry_time, entry_price=entry_price, + sl=entry_price - 8, tp1=entry_price + 12, tp2=entry_price + 20, + direction="BUY", pip_value=0.10, risk_pips=80.0, + reward_pips=120.0, rr_planned=1.5, quality="B", + final_score=30.0, final_confidence=60.0, + market_regime="trending_up") + + # Cleaned frame lacked the spike at idx[2], so the replay entered at idx[3]. + t = mk_trade(idx[3], 102.0) + _reanchor_entries_to_raw([t], [idx[1]], sim) + assert t.entry_time == idx[2] and t.entry_price == 110.0 + assert t.sl == 102.0 and t.tp1 == 122.0 and t.tp2 == 130.0, \ + "SL/TP distances must be preserved relative to the corrected entry" + + # Entry already the first raw bar after the signal → untouched. + t2 = mk_trade(idx[2], 110.0) + _reanchor_entries_to_raw([t2], [idx[1]], sim) + assert t2.entry_time == idx[2] and t2.entry_price == 110.0 and t2.sl == 102.0 + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════