diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index fc5e55b..cabd4c7 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -1008,6 +1008,16 @@ def fetch_history( 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") @@ -1159,6 +1169,15 @@ def _simulate_trailing( 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: @@ -1180,8 +1199,15 @@ def _simulate_trailing( 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) @@ -1354,6 +1380,12 @@ def simulate_trades( 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) @@ -1857,6 +1889,14 @@ def _summarize(rs: List[float]) -> Dict: 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, {}, {}, {}) diff --git a/providers/base.py b/providers/base.py index aac87a6..183ee5e 100644 --- a/providers/base.py +++ b/providers/base.py @@ -86,9 +86,15 @@ class BaseDataProvider(ABC): self.config = config self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} + # Requested fetch depth per key: a cached short frame must not + # satisfy a caller asking for a longer lookback. + self._cache_outputsize: Dict[str, int] = {} # The bot shares one provider across a thread pool; cache eviction # is check-then-act and needs locking. self._cache_lock = threading.Lock() + # Single-flight: one API fetch per (symbol, TF) at a time; other + # threads wait and then read the cache. + self._inflight: Dict[str, threading.Event] = {} # ── Abstract: the only provider-specific piece ───────────────── @@ -119,68 +125,97 @@ class BaseDataProvider(ABC): lookback_days = self.config.data_lookback_days cache_key = f"{symbol}_{timeframe.value}" - now = time.time() - - cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval) - with self._cache_lock: - cached = self._cache.get(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, min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, self.MAX_BARS_PER_REQUEST), ) - df = self._fetch_raw(symbol, timeframe, outputsize) - df = self._drop_forming_candle(df, timeframe) - df = self._validate_and_clean(df, symbol, timeframe) - # The closed-bar guarantee must hold for FRESH responses too, not - # just cache hits: the scanner fires seconds after the boundary, - # and a lagging feed can answer without the just-closed bar — the - # scan would silently analyze (and session-check) the previous - # bar. Give the feed one short grace retry, then proceed loudly. - if not df.empty and not self._has_latest_closed_bar(df, timeframe): - time.sleep(self.FRESH_FETCH_RETRY_SECONDS) - try: - retry = self._fetch_raw(symbol, timeframe, outputsize) - retry = self._drop_forming_candle(retry, timeframe) - retry = self._validate_and_clean(retry, symbol, timeframe) - if retry is not None and not retry.empty: - df = retry - except DataIntegrityError as e: + while True: + now = time.time() + cache_ttl = self.CANDLE_SECONDS.get( + timeframe, self.config.refresh_interval) + with self._cache_lock: + cached = self._cache.get(cache_key) + cached_at = self._cache_timestamps.get(cache_key) + cached_size = self._cache_outputsize.get(cache_key, 0) + if ( + cached is not None + and cached_at is not None + and (now - cached_at) < cache_ttl + and not cached.empty + # A caller asking for MORE history than the cached fetch + # requested must not be served the short frame silently. + and cached_size >= outputsize + ): + 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) + self._cache_outputsize.pop(cache_key, None) + + # Single-flight: concurrent scans for the same (symbol, TF) + # must not each hit the API (the four TF loops fire together + # at aligned boundaries and Oanda has no client throttle). + # First caller claims the key and fetches; the rest wait, + # then re-check the cache. + with self._cache_lock: + inflight = self._inflight.get(cache_key) + if inflight is None: + self._inflight[cache_key] = threading.Event() + break + inflight.wait(timeout=90) + + try: + df = self._fetch_raw(symbol, timeframe, outputsize) + df = self._drop_forming_candle(df, timeframe) + df = self._validate_and_clean(df, symbol, timeframe) + + # The closed-bar guarantee must hold for FRESH responses too, + # not just cache hits: the scanner fires seconds after the + # boundary, and a lagging feed can answer without the + # just-closed bar — the scan would silently analyze (and + # session-check) the previous bar. Give the feed one short + # grace retry, then proceed loudly. + if not df.empty and not self._has_latest_closed_bar(df, timeframe): + time.sleep(self.FRESH_FETCH_RETRY_SECONDS) + try: + retry = self._fetch_raw(symbol, timeframe, outputsize) + retry = self._drop_forming_candle(retry, timeframe) + retry = self._validate_and_clean(retry, symbol, timeframe) + if retry is not None and not retry.empty: + df = retry + except DataIntegrityError as e: + logger.warning( + f"{symbol} {timeframe.value}: boundary retry refetch " + f"failed ({e}) — keeping original data") + if not self._has_latest_closed_bar(df, timeframe): + logger.warning( + f"{symbol} {timeframe.value}: feed lagging — just-closed bar " + f"still missing after retry; this scan may run one bar behind" + ) + + if not self._is_data_fresh(df, symbol, timeframe): logger.warning( - f"{symbol} {timeframe.value}: boundary retry refetch " - f"failed ({e}) — keeping original data") - if not self._has_latest_closed_bar(df, timeframe): - logger.warning( - f"{symbol} {timeframe.value}: feed lagging — just-closed bar " - f"still missing after retry; this scan may run one bar behind" + f"STALE DATA from API: {symbol} {timeframe.value} — " + f"last bar {df.index[-1]} is too old. Signal quality may be degraded." ) - if not self._is_data_fresh(df, symbol, timeframe): - logger.warning( - f"STALE DATA from API: {symbol} {timeframe.value} — " - f"last bar {df.index[-1]} is too old. Signal quality may be degraded." - ) - - with self._cache_lock: - self._cache[cache_key] = df - self._cache_timestamps[cache_key] = now - return df + with self._cache_lock: + self._cache[cache_key] = df + self._cache_timestamps[cache_key] = now + self._cache_outputsize[cache_key] = outputsize + return df + finally: + # Release waiters even when the fetch raised — they re-check + # the cache and (on a miss) claim the key themselves. + with self._cache_lock: + ev = self._inflight.pop(cache_key, None) + if ev is not None: + ev.set() def fetch_range( self, @@ -199,6 +234,7 @@ class BaseDataProvider(ABC): chunks = [] cursor_end = end empty_streak = 0 + error_streak = 0 while cursor_end > start: cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk)) try: @@ -207,19 +243,27 @@ class BaseDataProvider(ABC): start=cursor_start, end=cursor_end, ) except Exception as e: - # Providers raise for windows with no data (weekends, - # holidays, asset gaps). Skip the window, step back, retry - # — but bail after a few consecutive empties. + # A raising chunk is a PROVIDER ERROR (auth, 429, 5xx, + # malformed payload) — not "no data". Three in a row means + # the rest of the history is unreachable: raise instead of + # returning a silently truncated frame (the backtester + # would treat it as complete). logger.warning( f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}" ) - empty_streak += 1 - if empty_streak >= 3: - break + error_streak += 1 + if error_streak >= 3: + raise self.DATA_ERROR( + f"{symbol} {timeframe.value}: {error_streak} " + f"consecutive chunk failures walking back from " + f"{cursor_end.date()} — refusing to return " + f"truncated history (last error: {e})") from e cursor_end = cursor_start - timedelta(seconds=1) continue if chunk is None or chunk.empty: + # Genuinely empty windows (weekends/holidays/asset gaps) + # are normal; several in a row = start of history. empty_streak += 1 if empty_streak >= 3: break @@ -227,9 +271,18 @@ class BaseDataProvider(ABC): continue empty_streak = 0 + error_streak = 0 chunks.append(chunk) # Step back; dedup on concat. - cursor_end = chunk.index[0] - timedelta(seconds=1) + new_cursor = chunk.index[0] - timedelta(seconds=1) + if new_cursor >= cursor_end: + # A provider ignoring start/end would pin the cursor and + # loop forever — refuse to spin. + logger.warning( + f"{symbol} {timeframe.value}: pagination made no " + f"progress at {cursor_end} — stopping") + break + cursor_end = new_cursor if len(chunk) < 2: break @@ -243,7 +296,17 @@ class BaseDataProvider(ABC): df = self._drop_forming_candle(df, timeframe) start_ts = pd.Timestamp(start) end_ts = pd.Timestamp(end) - return df[(df.index >= start_ts) & (df.index <= end_ts)] + df = df[(df.index >= start_ts) & (df.index <= end_ts)] + # Coverage sanity: a head materially later than the requested + # start means history was truncated (empty-streak bail or provider + # depth limit). Loud, not fatal — some assets genuinely start late. + if not df.empty and (df.index[0] - start_ts) > pd.Timedelta(days=7): + logger.error( + f"{symbol} {timeframe.value}: fetched history starts " + f"{df.index[0]} but {start_ts} was requested — " + f"~{(df.index[0] - start_ts).days} days missing from the " + f"head; treat downstream results accordingly") + return df def is_data_fresh( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe @@ -323,8 +386,19 @@ class BaseDataProvider(ABC): return df candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) now = pd.Timestamp.now(tz="UTC") - if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now: + # Loop, not a single slice: a misbehaving feed can return MORE + # than one not-yet-closed bar, and a single iloc[:-1] would let + # the residual forming bar through cleaning (future-tolerance is + # +5 min) straight into the signal engine. + dropped = 0 + while (not df.empty + and df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now): df = df.iloc[:-1] + dropped += 1 + if dropped > 1: + logger.warning( + "%s: feed returned %d forming/future bars — all dropped", + timeframe.value, dropped) return df def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool: @@ -372,6 +446,20 @@ class BaseDataProvider(ABC): logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps") df = df[~future_mask] + # Non-positive prices are corrupt, never real gold quotes. Reject + # the row BEFORE the consistency repair below — repairing High/Low + # from a corrupt 0.0 Open/Close would propagate the zero into + # previously-valid fields (2026-08-04 audit, I45). + nonpositive = (df[["Open", "High", "Low", "Close"]] <= 0).any(axis=1) + if nonpositive.any(): + logger.warning( + f"{symbol}: Dropping {nonpositive.sum()} bars with " + f"non-positive prices (corrupt feed data)") + df = df[~nonpositive] + if df.empty: + raise self.DATA_ERROR( + f"{symbol}: all bars had non-positive prices") + invalid_ohlc = ( (df["High"] < df["Low"]) | (df["High"] < df["Open"]) | @@ -386,7 +474,10 @@ class BaseDataProvider(ABC): df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) + # inf returns (division by a zero/na Close) would NaN the std and + # silently disarm the whole 10σ pass — neutralize them first. returns = df["Close"].pct_change() + returns = returns.replace([float("inf"), float("-inf")], float("nan")) std_ret = returns.std() if std_ret > 0: dev = returns - returns.mean() diff --git a/providers/oanda.py b/providers/oanda.py index 1805f03..2ea5435 100644 --- a/providers/oanda.py +++ b/providers/oanda.py @@ -149,23 +149,37 @@ class OandaDataProvider(BaseDataProvider): rows = [] last_bid_close = None last_ask_close = None + skipped_malformed = 0 for c in candles: if not c.get("complete", False): continue mid = c.get("mid", {}) bid = c.get("bid", {}) ask = c.get("ask", {}) + # A candle missing any OHLC field must be SKIPPED, never + # defaulted to 0.0: a zero price flows through cleaning as a + # "legit" bar and poisons ATR/Keltner/SL math — and a + # mid-frame zero Close makes pct_change() emit inf, which + # NaNs the std and disarms the 10σ outlier filter for the + # whole frame (2026-08-04 audit, I45). + if any(mid.get(k) is None for k in ("o", "h", "l", "c")): + skipped_malformed += 1 + continue rows.append({ "Datetime": pd.Timestamp(c["time"]).tz_convert("UTC"), - "Open": float(mid.get("o", 0)), - "High": float(mid.get("h", 0)), - "Low": float(mid.get("l", 0)), - "Close": float(mid.get("c", 0)), + "Open": float(mid["o"]), + "High": float(mid["h"]), + "Low": float(mid["l"]), + "Close": float(mid["c"]), "Volume": int(c.get("volume", 0)), }) if bid and ask: last_bid_close = float(bid.get("c", 0)) last_ask_close = float(ask.get("c", 0)) + if skipped_malformed: + logger.warning( + "Oanda returned %d malformed candle(s) for %s (missing " + "OHLC fields) — skipped", skipped_malformed, instrument) if not rows: raise OandaDataError( diff --git a/providers/twelve_data.py b/providers/twelve_data.py index 288ff7a..8bf94cc 100644 --- a/providers/twelve_data.py +++ b/providers/twelve_data.py @@ -210,6 +210,12 @@ class TwelveDataProvider(BaseDataProvider): The TD WS speaks display symbols ("XAU/USD"); ticks are mapped back to the canonical pair symbols passed in. The caller owns the reconnect/backoff loop around this generator. + + `on_connect` fires on the FIRST RECEIVED frame, not after the + subscribe send (2026-08-04 audit H38b, mirroring the Tiingo + provider) — firing pre-receive let a rejected API key reset the + caller's reconnect backoff on every attempt, pinning it at the + minimum forever. """ import websockets @@ -222,11 +228,14 @@ class TwelveDataProvider(BaseDataProvider): } await ws.send(json.dumps(subscribe)) logger.info(f"WebSocket connected, subscribed to {len(ws_to_pair)} symbols") - if on_connect is not None: - on_connect() hb_task = asyncio.create_task(self._ws_heartbeat(ws)) + connected = False try: async for raw in ws: + if not connected: + connected = True + if on_connect is not None: + on_connect() parsed = self._parse_ws_price(raw) if parsed is None: continue diff --git a/telegram_bot.py b/telegram_bot.py index d32a281..23ec767 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -600,6 +600,45 @@ def _commit_alert(key: str, report: SignalReport) -> None: _last_alert_setup.pop(key, None) +# Per-key locks serializing the _should_alert → send → _commit_alert +# window. The auto-scan loop and a concurrent breach handler share the +# "XAU/USD_15min" dedup key; without the lock both pass _should_alert +# while the other's send is still awaiting Telegram, and the same signal +# goes out twice (2026-08-04 audit H36). Per-key rather than one global +# lock so a slow send on one timeframe cannot head-of-line-block another +# timeframe's alert. Keys are symbol × TF, so the dict stays tiny. +_alert_locks: Dict[str, asyncio.Lock] = {} + + +def _alert_lock(key: str) -> asyncio.Lock: + lock = _alert_locks.get(key) + if lock is None: + lock = _alert_locks[key] = asyncio.Lock() + return lock + + +async def _send_alert(bot, chat_id: int, text: str) -> bool: + """Markdown alert send with a plain-text fallback on parse failures, + mirroring _safe_edit (2026-08-04 audit H35 — alert paths previously + dropped the message on a BadRequest with only a warning logged). + Returns True when a message reached the chat by either attempt.""" + try: + await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN) + return True + except BadRequest as exc: + logger.warning(f"Markdown alert failed for chat {chat_id} ({exc}); " + f"retrying as plain text") + try: + await bot.send_message(chat_id, text) + return True + except Exception as exc2: + logger.warning(f"Failed to send alert to chat {chat_id}: {exc2}") + return False + except Exception as exc: + logger.warning(f"Failed to send alert to chat {chat_id}: {exc}") + return False + + # ── Candle-boundary scheduler ──────────────────────────────────────── _autoscan_chats: Set[int] = set() @@ -644,7 +683,13 @@ async def _scan_timeframe( reports: List[SignalReport] = [] for i, result in enumerate(results): if isinstance(result, Exception): - logger.exception(f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}") + # gather(return_exceptions=True) means there is no active + # exception here — logger.exception would log "NoneType: None" + # instead of the real stack; pass the instance explicitly. + logger.error( + f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}", + exc_info=result, + ) else: reports.append(result) @@ -653,36 +698,51 @@ async def _scan_timeframe( _update_level_cache(reports) - actionable = [ + candidates = [ r for r in reports if r.final_signal != SignalStrength.NEUTRAL # same gate as breach path and _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 i, report in enumerate(actionable): + for report in candidates: + key = f"{report.symbol}_{tf.value}" + # The lock spans check → send → commit so a concurrent breach + # handler on the same key cannot double-alert (H36). + async with _alert_lock(key): + if not _should_alert(key, report): + continue 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) + delivered = False + for chat_id in _signal_targets(): + if await _send_alert(bot, chat_id, detail): + delivered = True + # Commit only when the alert reached at least one target; + # a fully failed send stays eligible for the next scan. + if delivered: + _commit_alert(key, report) # ── WebSocket level-breach monitor ────────────────────────────────── -_level_cache: Dict[str, Dict[str, float]] = {} -_level_cache_ts: Dict[str, float] = {} +# Level cache keyed symbol → timeframe → {level_name: price}, with a +# matching per-TF write timestamp. Keyed per TF (2026-08-04 audit H34) +# because every scan path writes here — the M5/M15/H1/H4 boundary loops, +# the breach rescan, the manual Scan-All button — and a single flat dict +# meant the last writer wiped the other timeframes' swing/EMA levels +# within minutes. Breach checks scan the union of all fresh TF entries. +_level_cache: Dict[str, Dict[str, Dict[str, float]]] = {} +_level_cache_ts: Dict[str, Dict[str, float]] = {} _LEVEL_CACHE_TTL = 3600 _last_prices: Dict[str, float] = {} _last_price_ts: Dict[str, float] = {} _scan_cooldowns: Dict[str, float] = {} +# Breaches are edge-triggered (only the crossing tick fires), so one that +# lands while the symbol is inside the rescan cooldown would otherwise be +# dropped forever (H33b). Park it here; the tick loop fires it once the +# cooldown expires. symbol → (price, level_name, recorded_at). +_pending_breach: Dict[str, tuple] = {} +# A failed breach rescan shrinks the just-set cooldown to this retry +# window instead of silencing the symbol for the full cooldown (H33a). +_BREACH_RETRY_SECONDS = 30 _cfg = PlatformConfig() _WS_COOLDOWN = _cfg.ws_cooldown_seconds @@ -691,6 +751,8 @@ _breach_tasks: Set[asyncio.Task] = set() def _update_level_cache(reports: List[SignalReport]) -> None: + """Each report updates only its own timeframe's slot — an M5 scan can + no longer overwrite the H1/H4 swing levels (H34).""" now = time.time() for r in reports: if r.primary_analysis and r.primary_analysis.key_levels: @@ -699,26 +761,31 @@ def _update_level_cache(reports: List[SignalReport]) -> None: if k in _LEVEL_KEYS } if levels: - _level_cache[r.symbol] = levels - _level_cache_ts[r.symbol] = now + tf = r.primary_analysis.timeframe + _level_cache.setdefault(r.symbol, {})[tf] = levels + _level_cache_ts.setdefault(r.symbol, {})[tf] = now def _check_level_breach(symbol: str, price: float) -> Optional[str]: - levels = _level_cache.get(symbol) - if not levels: - return None - - cache_age = time.time() - _level_cache_ts.get(symbol, 0) - if cache_age > _LEVEL_CACHE_TTL: + """Scan the union of every fresh timeframe's levels; returns + ' ' (e.g. '15min swing_high_1') on a crossing so the + alert label says which timeframe's level broke.""" + per_tf = _level_cache.get(symbol) + if not per_tf: return None last = _last_prices.get(symbol) if last is None: return None - for name, level in levels.items(): - if (last < level <= price) or (last > level >= price): - return name + now = time.time() + ts_map = _level_cache_ts.get(symbol, {}) + for tf, levels in per_tf.items(): + if now - ts_map.get(tf, 0) > _LEVEL_CACHE_TTL: + continue # stale TF entry — skip, fresher TFs still count + for name, level in levels.items(): + if (last < level <= price) or (last > level >= price): + return f"{tf} {name}" return None @@ -733,9 +800,17 @@ def _cleanup_cooldowns() -> None: async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -> None: now = time.time() if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN: + # Park the crossing instead of dropping it — price holding beyond + # the level never re-fires the edge trigger, so without this the + # breach would be lost forever (H33b). The tick loop fires it + # once the cooldown expires; a newer crossing overwrites it. + _pending_breach[pair_symbol] = (price, level_name, now) return + # Early-set on purpose: this is what stops concurrent breach tasks + # from stacking rescans. The failure path below rolls it back. _scan_cooldowns[pair_symbol] = now + _pending_breach.pop(pair_symbol, None) # this crossing is being handled display = PAIR_INFO.get(pair_symbol, {}).get("display", pair_symbol) label = level_name.replace("_", " ").title() logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15") @@ -745,29 +820,42 @@ async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) - _update_level_cache([report]) except Exception: logger.exception(f"Breach scan failed for {display}") + # Shrink the just-set cooldown to a short retry window so a + # transient fetch error doesn't silence the symbol for the full + # cooldown (H33a); the floor bounds the retry rate while the + # provider stays down. Re-park the breach so the tick loop + # retries it (setdefault: a newer crossing recorded meanwhile + # wins over this stale one). + retry = min(_BREACH_RETRY_SECONDS, _WS_COOLDOWN) + _scan_cooldowns[pair_symbol] = now - _WS_COOLDOWN + retry + _pending_breach.setdefault(pair_symbol, (price, level_name, now)) return # Alert only when the rescan produced an actionable, changed signal — # same gates as the auto-scanner, so a breach can't spam the channel. if (report.final_signal == SignalStrength.NEUTRAL - or not _passes_quality(report.signal_quality) - or not _should_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)): + or not _passes_quality(report.signal_quality)): return - header = ( - f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed " - f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n" - ) - sent = False - for chat_id in _signal_targets(): + key = f"{pair_symbol}_{Timeframe.M15.value}" + # Lock spans check → send → commit; without it this handler and an + # in-flight M15 auto-scan both pass _should_alert and double-alert + # the same signal (H36). + async with _alert_lock(key): + if not _should_alert(key, report): + return + + header = ( + f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed " + f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n" + ) detail = header + format_signal_message(report) - 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) + sent = False + for chat_id in _signal_targets(): + if await _send_alert(bot, chat_id, detail): + sent = True + if sent: + _commit_alert(key, report) def _breach_task_done(task: asyncio.Task) -> None: @@ -799,13 +887,16 @@ async def _ws_monitor(bot) -> None: backoff = 1 while _feed_active(): + got_tick = False try: - def _on_connect() -> None: - nonlocal backoff - backoff = 1 - - async for pair_symbol, price in dp.iter_ticks( - SCAN_PAIRS, on_connect=_on_connect): + async for pair_symbol, price in dp.iter_ticks(SCAN_PAIRS): + if not got_tick: + # Reset backoff only once the connection actually + # produced data — resetting on connect let a + # rejected key or accept-then-close server pin the + # backoff at minimum forever (H38a/H38b). + got_tick = True + backoff = 1 if not _feed_active(): break @@ -814,11 +905,32 @@ async def _ws_monitor(bot) -> None: _last_price_ts[pair_symbol] = time.time() if breached: + # A fresh crossing supersedes any parked one. + _pending_breach.pop(pair_symbol, None) task = asyncio.create_task( _handle_breach(bot, pair_symbol, price, breached) ) _breach_tasks.add(task) task.add_done_callback(_breach_task_done) + elif pair_symbol in _pending_breach: + pend = _pending_breach[pair_symbol] + if time.time() - pend[2] > _LEVEL_CACHE_TTL: + # Ancient parked breach (feed gap/weekend) — an + # alert about an hours-old crossing would only + # mislead; the level cache it came from is stale + # anyway. + _pending_breach.pop(pair_symbol) + elif (time.time() - _scan_cooldowns.get(pair_symbol, 0) + >= _WS_COOLDOWN): + # Crossing parked during the cooldown: fire it now + # that the window expired — the edge trigger alone + # would have dropped it forever (H33b). + _pending_breach.pop(pair_symbol) + task = asyncio.create_task( + _handle_breach(bot, pair_symbol, pend[0], pend[1]) + ) + _breach_tasks.add(task) + task.add_done_callback(_breach_task_done) _cleanup_cooldowns() @@ -828,6 +940,16 @@ async def _ws_monitor(bot) -> None: logger.exception(f"WebSocket error, reconnecting in {backoff}s") await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) + else: + # Clean server close: the async-for ended without raising. + # This must back off and log like the error path — silently + # reconnecting with zero delay hot-loops against a server + # that accepts and immediately closes (H38a). + if _feed_active(): + logger.warning( + f"WebSocket closed cleanly, reconnecting in {backoff}s") + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 60) logger.info("WebSocket monitor stopped") @@ -870,15 +992,25 @@ _ws_task: Optional[asyncio.Task] = None def _start_scheduler(bot) -> None: global _ws_task - alive = [t for t in _autoscan_tasks if not t.done()] - if not alive: - _autoscan_tasks.clear() - for tf, interval in _AUTOSCAN_LOOPS: - _autoscan_tasks.append( - asyncio.create_task(_candle_loop(bot, tf, interval)) - ) + # Per-TF respawn (2026-08-04 audit H37): drop finished loop tasks and + # recreate only the missing ones. The old all-or-nothing alive check + # left a single dead TF loop unmonitored forever while its siblings + # lived — and with the channel feed active, /autoscan never reaches + # _stop_scheduler, so nothing short of a process restart recovered it. + _autoscan_tasks[:] = [t for t in _autoscan_tasks if not t.done()] + alive_names = {t.get_name() for t in _autoscan_tasks} + started = [] + for tf, interval in _AUTOSCAN_LOOPS: + name = f"candle_loop_{tf.name}" + if name in alive_names: + continue + _autoscan_tasks.append( + asyncio.create_task(_candle_loop(bot, tf, interval), name=name) + ) + started.append(tf.name) + if started: logger.info("Candle-boundary scheduler started (%s)", - "/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS)) + "/".join(started)) if _ws_task is None or _ws_task.done(): _ws_task = asyncio.create_task(_ws_monitor(bot)) @@ -1006,9 +1138,16 @@ def _status_text() -> str: lines.append("\n_No live prices yet (WS not streaming)_") cached_pairs = len(_level_cache) - cached_levels = sum(len(v) for v in _level_cache.values()) + cached_levels = sum( + len(levels) for per_tf in _level_cache.values() + for levels in per_tf.values() + ) if cached_levels: - oldest_age = time.time() - min(_level_cache_ts.values()) if _level_cache_ts else 0 + all_ts = [ + ts for per_tf in _level_cache_ts.values() + for ts in per_tf.values() + ] + oldest_age = time.time() - min(all_ts) if all_ts else 0 lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs (oldest {oldest_age:.0f}s)") else: lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan") @@ -1311,7 +1450,11 @@ async def _post_init(app: Application) -> None: logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}") if init_reports: _update_level_cache(init_reports) - logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels") + n_levels = sum( + len(levels) for per_tf in _level_cache.values() + for levels in per_tf.values() + ) + logger.info(f"Level cache populated: {n_levels} levels") except Exception: logger.exception("Initial scan failed — WS will wait for first candle scan") diff --git a/test_all.py b/test_all.py index 41adc32..942f1da 100644 --- a/test_all.py +++ b/test_all.py @@ -1800,13 +1800,17 @@ def _(): assert result <= 240 * 60 + 10 + 1 +# NOTE 2026-08-04 (audit H34): the level cache is keyed symbol → TF → +# levels and _check_level_breach returns " " — these tests +# were updated to the new shape in the same change. + @test("_check_level_breach detects upward cross") def _(): - _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} - _level_cache_ts["TEST_PAIR"] = time.time() + _level_cache["TEST_PAIR"] = {"15min": {"swing_high_1": 1.1000}} + _level_cache_ts["TEST_PAIR"] = {"15min": time.time()} _last_prices["TEST_PAIR"] = 1.0999 result = _check_level_breach("TEST_PAIR", 1.1001) - assert result == "swing_high_1" + assert result == "15min swing_high_1" # cleanup del _level_cache["TEST_PAIR"] del _level_cache_ts["TEST_PAIR"] @@ -1815,11 +1819,11 @@ def _(): @test("_check_level_breach detects downward cross") def _(): - _level_cache["TEST_PAIR"] = {"swing_low_1": 1.0900} - _level_cache_ts["TEST_PAIR"] = time.time() + _level_cache["TEST_PAIR"] = {"15min": {"swing_low_1": 1.0900}} + _level_cache_ts["TEST_PAIR"] = {"15min": time.time()} _last_prices["TEST_PAIR"] = 1.0901 result = _check_level_breach("TEST_PAIR", 1.0899) - assert result == "swing_low_1" + assert result == "15min swing_low_1" del _level_cache["TEST_PAIR"] del _level_cache_ts["TEST_PAIR"] del _last_prices["TEST_PAIR"] @@ -1827,8 +1831,8 @@ def _(): @test("_check_level_breach no breach when price stays same side") def _(): - _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} - _level_cache_ts["TEST_PAIR"] = time.time() + _level_cache["TEST_PAIR"] = {"15min": {"swing_high_1": 1.1000}} + _level_cache_ts["TEST_PAIR"] = {"15min": time.time()} _last_prices["TEST_PAIR"] = 1.0990 result = _check_level_breach("TEST_PAIR", 1.0995) assert result is None @@ -1839,8 +1843,8 @@ def _(): @test("_check_level_breach returns None for expired cache") def _(): - _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} - _level_cache_ts["TEST_PAIR"] = time.time() - 7200 # 2h old + _level_cache["TEST_PAIR"] = {"15min": {"swing_high_1": 1.1000}} + _level_cache_ts["TEST_PAIR"] = {"15min": time.time() - 7200} # 2h old _last_prices["TEST_PAIR"] = 1.0999 result = _check_level_breach("TEST_PAIR", 1.1001) assert result is None @@ -1851,8 +1855,8 @@ def _(): @test("_check_level_breach returns None when no previous price") def _(): - _level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000} - _level_cache_ts["TEST_PAIR"] = time.time() + _level_cache["TEST_PAIR"] = {"15min": {"swing_high_1": 1.1000}} + _level_cache_ts["TEST_PAIR"] = {"15min": time.time()} if "TEST_PAIR" in _last_prices: del _last_prices["TEST_PAIR"] result = _check_level_breach("TEST_PAIR", 1.1001) @@ -3424,7 +3428,9 @@ def _(): provider2._base_url = "https://test" provider2._cache = {} provider2._cache_timestamps = {} + provider2._cache_outputsize = {} provider2._cache_lock = threading.Lock() + provider2._inflight = {} provider2._spread_cache = {} provider2._session = type("S", (), { "get": lambda self, *a, **k: mock_resp, @@ -4081,7 +4087,9 @@ def _(): provider.config = PlatformConfig() provider._cache = {} provider._cache_timestamps = {} + provider._cache_outputsize = {} provider._cache_lock = threading.Lock() + provider._inflight = {} provider._spread_cache = {} provider.FRESH_FETCH_RETRY_SECONDS = 0 # instance override — no sleep @@ -5637,6 +5645,677 @@ def _(): "first cache miss must emit a WARNING" +def _m21_stub_provider(bars=320, tf=Timeframe.H1): + """Stub provider whose _fetch_raw serves a deterministic frame and + counts invocations (2026-08-04 audit, Batch E).""" + import providers.base as pbase + + class _Stub(pbase.BaseDataProvider): + PROVIDER = "stub21" + calls = [] + + def _fetch_raw(self, symbol, timeframe, outputsize, + start=None, end=None): + self.calls.append(outputsize) + time_mod.sleep(0.05) + end_ts = pd.Timestamp.now(tz="UTC").floor("1h") - pd.Timedelta(hours=1) + idx = pd.date_range(end=end_ts, periods=min(outputsize, bars), + freq="1h", tz="UTC") + base = 2300 + np.cumsum(np.ones(len(idx)) * 0.1) + return pd.DataFrame( + {"Open": base, "High": base + 1, "Low": base - 1, + "Close": base, "Volume": 1.0}, + index=pd.DatetimeIndex(idx, name="Datetime")) + + def get_pair_info(self, symbol): + return PAIR_INFO[XAUUSD] + + return _Stub(PlatformConfig()) + + +import time as time_mod + + +@test("provider cache: longer-lookback request is NOT served the short frame") +def _(): + from unittest.mock import patch + with patch("providers.base._is_forex_weekend", lambda now: False): + p = _m21_stub_provider() + p.fetch_ohlcv(XAUUSD, Timeframe.H1, lookback_days=1) + assert len(p.calls) == 1 and p.calls[0] == 300 + p.fetch_ohlcv(XAUUSD, Timeframe.H1, lookback_days=60) + assert len(p.calls) == 2, \ + "a longer-lookback call within TTL must refetch, not serve the " \ + "short cached frame" + assert p.calls[1] == 1460 + + +@test("provider cache: concurrent same-key fetches are single-flight") +def _(): + from unittest.mock import patch + import threading as _t + with patch("providers.base._is_forex_weekend", lambda now: False): + p = _m21_stub_provider() + barrier = _t.Barrier(4) + results = [] + + def worker(): + barrier.wait() + results.append(p.fetch_ohlcv(XAUUSD, Timeframe.H1)) + + threads = [_t.Thread(target=worker) for _ in range(4)] + for th in threads: th.start() + for th in threads: th.join() + assert len(results) == 4 + assert len(p.calls) == 1, \ + f"4 concurrent fetches must produce 1 API call, got {len(p.calls)}" + + +@test("provider: multi-bar forming/future tail is fully dropped") +def _(): + p = _m21_stub_provider() + now = pd.Timestamp.now(tz="UTC") + idx = pd.DatetimeIndex( + [now - pd.Timedelta(hours=3), now - pd.Timedelta(hours=2), + now - pd.Timedelta(minutes=30), now + pd.Timedelta(minutes=30)], + name="Datetime") + df = pd.DataFrame({"Open": 1.0, "High": 2.0, "Low": 0.5, "Close": 1.5, + "Volume": 1.0}, index=idx) + out = p._drop_forming_candle(df, Timeframe.H1) + assert len(out) == 2, \ + f"both not-yet-closed bars must be dropped, kept {len(out)}" + + +@test("provider fetch_range: 3 consecutive chunk ERRORS raise, no silent truncation") +def _(): + import providers.base as pbase + + class _Err(pbase.BaseDataProvider): + PROVIDER = "err21" + + def _fetch_raw(self, symbol, timeframe, outputsize, + start=None, end=None): + raise self.DATA_ERROR("HTTP 429 simulated") + + def get_pair_info(self, symbol): + return PAIR_INFO[XAUUSD] + + p = _Err(PlatformConfig()) + from datetime import datetime as _dt, timezone as _tz, timedelta as _td + end = _dt.now(_tz.utc) + try: + # M5 → ~15-day chunks, so a 90-day window would walk ≥3 chunks. + p.fetch_range(XAUUSD, Timeframe.M5, end - _td(days=90), end) + assert False, "must raise after repeated chunk failures" + except pbase.DataIntegrityError as e: + assert "truncated" in str(e) or "consecutive" in str(e) + + +@test("provider cleaning: non-positive bar dropped, 10-sigma filter stays armed") +def _(): + p = _m21_stub_provider() + n = 260 + idx = pd.date_range("2026-01-05", periods=n, freq="1h", tz="UTC") + close = 2300 + np.cumsum(np.random.default_rng(5).normal(0, 1.0, n)) + df = pd.DataFrame({"Open": close, "High": close + 1, "Low": close - 1, + "Close": close, "Volume": 1.0}, index=idx) + df.iloc[100, df.columns.get_loc("Close")] = 0.0 # corrupt mid-frame bar + df.iloc[200, df.columns.get_loc("Close")] = close[200] * 1.5 # true spike + out = p._validate_and_clean(df, XAUUSD, Timeframe.H1) + assert (out[["Open", "High", "Low", "Close"]] > 0).all().all(), \ + "non-positive bar must be dropped" + assert idx[200] not in out.index, \ + "the 10-sigma filter must survive the corrupt bar and still " \ + "remove the genuine spike (inf-return kill-switch fixed)" + assert out["Low"].min() > 1000, "no zero may leak into Low via repair" + + +@test("oanda: malformed candle (missing OHLC field) is skipped, not zeroed") +def _(): + from providers.oanda import OandaDataProvider + prov = OandaDataProvider.__new__(OandaDataProvider) + prov._spread_cache = {} + + good = {"complete": True, "time": "2026-08-04T10:00:00Z", "volume": 10, + "mid": {"o": "3000.0", "h": "3001.0", "l": "2999.0", "c": "3000.5"}} + bad = {"complete": True, "time": "2026-08-04T11:00:00Z", "volume": 10, + "mid": {"o": "3000.5", "h": "3001.5", "c": "3000.9"}} # no "l" + good2 = {"complete": True, "time": "2026-08-04T12:00:00Z", "volume": 10, + "mid": {"o": "3000.9", "h": "3002.0", "l": "3000.0", "c": "3001.4"}} + + class _Resp: + status_code = 200 + text = "" + def json(self): + return {"candles": [good, bad, good2]} + + class _Sess: + def get(self, url, params=None, timeout=None): + return _Resp() + + prov._session = _Sess() + prov._base_url = "http://x" + df = prov._fetch_oanda_candles(XAUUSD, Timeframe.H1, 10) + assert len(df) == 2, "malformed candle must be skipped" + assert (df[["Open", "High", "Low", "Close"]] > 0).all().all(), \ + "no 0.0-defaulted prices may appear" + + +@test("backtester: NaN bar inside a trailing window cannot poison R/metrics") +def _(): + import gpu_backtest as _gb + t = _gb.Trade(pair="XAU/USD", + entry_time=pd.Timestamp("2026-01-05 10:00", tz="UTC"), + entry_price=2400.0, sl=2384.0, tp1=2416.0, tp2=2432.0, + direction="BUY", pip_value=0.1, risk_pips=160.0, + reward_pips=160.0, rr_planned=1.0, quality="B", + final_score=30.0, final_confidence=60.0, + market_regime="trending_up") + idx = pd.date_range("2026-01-05 10:00", periods=10, freq="15min", tz="UTC") + o = np.full(10, 2401.0); h = np.full(10, 2402.0) + l = np.full(10, 2399.0); c = np.full(10, 2401.5) + for arr in (o, h, l, c): + arr[3] = np.nan # data gap mid-window + c[9] = np.nan + sim = pd.DataFrame({"Open": o, "High": h, "Low": l, "Close": c, + "Volume": 1.0}, index=idx) + _gb.simulate_trades([t], sim, max_bars=10, spread=0.30, + trail_arm_r=0.7, trail_dist_r=0.75) + assert t.exit_reason == "TIMEOUT" and t.bars_held == 3, \ + f"must exit at last valid bar pre-gap, got {t.exit_reason}/{t.bars_held}" + assert np.isfinite(t.r_multiple), "NaN R must never be emitted" + m = _gb.compute_metrics([t]) + assert np.isfinite(m.total_r) and np.isfinite(m.avg_r) + + +@test("backtester: compute_metrics excludes a non-finite R loudly, not NaN-ly") +def _(): + import gpu_backtest as _gb + def mk(r): + t = _gb.Trade(pair="XAU/USD", + entry_time=pd.Timestamp("2026-01-05 10:00", tz="UTC"), + entry_price=2400.0, sl=2384.0, tp1=2416.0, tp2=2432.0, + direction="BUY", pip_value=0.1, risk_pips=160.0, + reward_pips=160.0, rr_planned=1.0, quality="B", + final_score=30.0, final_confidence=60.0, + market_regime="trending_up") + t.exit_reason = "SL"; t.bars_held = 5; t.r_multiple = r + return t + m = _gb.compute_metrics([mk(1.0), mk(float("nan")), mk(-1.0)]) + assert m.n_trades == 2, "non-finite R trade must be excluded" + assert np.isfinite(m.total_r) and abs(m.total_r) < 1e-9 + + +# ── Batch C: telegram_bot delivery-path fixes (2026-08-04 audit) ───── +# Regression kernels from the H33–H38 verification harness. All hermetic: +# fake bots, fake streaming providers, patched run_signal — no network. + +import asyncio as _aio21 + +import telegram_bot as _tb21 +from telegram.error import BadRequest as _BadRequest21 + + +def _m21c_report(tf="15min", quality="A", levels=None, + entry=3300.0, sl=3290.0): + from engine import TechnicalAnalysis + r = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="t") + r.final_signal = SignalStrength.BUY + r.signal_quality = quality + r.final_score = 30.0 + r.final_confidence = 60.0 + r.market_regime = "trending_up" + r.volatility_regime = "normal" + r.current_price = entry + r.primary_analysis = TechnicalAnalysis( + symbol="XAU/USD", timeframe=tf, timestamp="t", + key_levels=dict(levels) if levels else {}, + ) + r.trade_setup = TradeSetup( + pair="XAU/USD", display_name="XAU/USD", direction="BUY", + entry_price=entry, stop_loss=sl, take_profit_1=entry + 15, + take_profit_2=entry + 25, risk_reward=1.5, risk_pips=100.0, + reward_pips=150.0, pip_value=0.10, + trail_arm_r=0.7, trail_dist_r=0.85, + ) + return r + + +class _M21cBot: + """send_message stub: optionally raise (exc / md_exc for Markdown + attempts only) or block on an asyncio.Event before returning.""" + + def __init__(self, exc=None, md_exc=None, gate=None): + self.exc = exc + self.md_exc = md_exc + self.gate = gate + self.attempts = [] + self.sent = [] + + async def send_message(self, chat_id, text, parse_mode=None, **kw): + self.attempts.append(parse_mode) + if self.gate is not None: + await self.gate.wait() + if self.exc is not None: + raise self.exc + if self.md_exc is not None and parse_mode is not None: + raise self.md_exc + self.sent.append(text) + + +def _m21c_reset(): + _tb21._scan_cooldowns.clear() + _tb21._pending_breach.clear() + _tb21._last_alert_fp.clear() + _tb21._last_alert_ts.clear() + _tb21._last_alert_setup.clear() + _tb21._level_cache.clear() + _tb21._level_cache_ts.clear() + _tb21._last_prices.clear() + _tb21._last_price_ts.clear() + _tb21._autoscan_chats.discard(911) + + +@test("H33a: failed breach rescan shrinks cooldown to the retry window") +def _(): + async def main(): + _m21c_reset() + _tb21._autoscan_chats.add(911) + calls = {"n": 0} + + async def failing(*a, **k): + calls["n"] += 1 + raise RuntimeError("simulated rescan failure") + + good = _m21c_report() + + async def working(*a, **k): + calls["n"] += 1 + return good + + bot = _M21cBot() + with patch.object(_tb21, "_WS_COOLDOWN", 0.4), \ + patch.object(_tb21, "_BREACH_RETRY_SECONDS", 0.05): + with patch.object(_tb21, "run_signal", failing): + await _tb21._handle_breach(bot, "XAU/USD", 3301.0, "swing_high_1") + assert calls["n"] == 1 + remaining = _tb21._WS_COOLDOWN - ( + time.time() - _tb21._scan_cooldowns["XAU/USD"]) + assert remaining <= _tb21._BREACH_RETRY_SECONDS + 0.02, \ + f"cooldown must be rolled back to the retry window, " \ + f"got {remaining:.2f}s remaining" + assert "XAU/USD" in _tb21._pending_breach, \ + "failed rescan must re-park the breach for retry" + await _aio21.sleep(0.07) # retry window elapses + with patch.object(_tb21, "run_signal", working): + await _tb21._handle_breach(bot, "XAU/USD", 3301.0, "swing_high_1") + assert calls["n"] == 2, "post-retry-window breach must rescan" + assert len(bot.sent) == 1, "and the alert must go out" + _m21c_reset() + + _aio21.run(main()) + + +@test("H33b: breach during cooldown is parked and fired after expiry") +def _(): + async def main(): + _m21c_reset() + _tb21._autoscan_chats.add(911) + _tb21._update_level_cache( + [_m21c_report(levels={"swing_high_1": 3300.0})]) + _tb21._scan_cooldowns["XAU/USD"] = time.time() # cooldown active + rescans = {"n": 0} + + async def working(*a, **k): + rescans["n"] += 1 + return _m21c_report() + + bot = _M21cBot() + mid_state = {} + connects = {"n": 0} + + class FakeDP: + name = "fake" + supports_streaming = True + _api_key = "k" + + async def iter_ticks(self, symbols, on_connect=None): + # No asserts in here: an in-generator failure would be + # swallowed by the monitor's reconnect loop. Record state, + # assert after; the connect counter is a safety valve. + connects["n"] += 1 + if connects["n"] > 1: + raise _aio21.CancelledError + yield "XAU/USD", 3299.0 # below level + yield "XAU/USD", 3301.0 # CROSS — inside cooldown → parked + await _aio21.sleep(0) # let the breach task park it + mid_state["rescans"] = rescans["n"] + mid_state["parked"] = "XAU/USD" in _tb21._pending_breach + await _aio21.sleep(0.1) # cooldown (0.08s) expires + yield "XAU/USD", 3302.0 # no crossing — pending must fire + await _aio21.sleep(0.05) # let the fired task run + raise _aio21.CancelledError + + with patch.object(_tb21, "_WS_COOLDOWN", 0.08), \ + patch.object(_tb21, "run_signal", working), \ + patch.object(_tb21, "_get_streaming_dp", lambda: FakeDP()): + await _tb21._ws_monitor(bot) + if _tb21._breach_tasks: + await _aio21.gather(*_tb21._breach_tasks, + return_exceptions=True) + assert connects["n"] == 1, "monitor must stop on the first pass" + assert mid_state == {"rescans": 0, "parked": True}, \ + f"cross inside cooldown must park, not rescan: {mid_state}" + assert rescans["n"] == 1, "parked breach must fire after expiry" + assert len(bot.sent) == 1 + assert "3301" in bot.sent[0], \ + "alert must carry the original crossing price" + assert "XAU/USD" not in _tb21._pending_breach, "pending consumed" + _m21c_reset() + + _aio21.run(main()) + + +@test("H34: level cache is per-timeframe — M5 scan no longer wipes H1 levels") +def _(): + _m21c_reset() + _tb21._update_level_cache([_m21c_report( + tf="1h", levels={"swing_high_1": 3350.0, "ema_trend": 3310.0})]) + _tb21._update_level_cache([_m21c_report( + tf="5min", levels={"swing_high_1": 3311.0})]) + cache = _tb21._level_cache["XAU/USD"] + assert set(cache) == {"1h", "5min"}, f"both TF slots must exist: {cache}" + assert cache["1h"]["swing_high_1"] == 3350.0, \ + "M5 write must not touch the H1 slot" + # Breach check scans the union: crossing the H1-only level fires + # with a TF-qualified name. + _tb21._last_prices["XAU/USD"] = 3349.0 + assert _tb21._check_level_breach("XAU/USD", 3351.0) == "1h swing_high_1" + # A second M5 scan replaces only the M5 slot. + _tb21._update_level_cache([_m21c_report( + tf="5min", levels={"swing_low_1": 3290.0})]) + assert _tb21._level_cache["XAU/USD"]["5min"] == {"swing_low_1": 3290.0} + assert _tb21._level_cache["XAU/USD"]["1h"]["swing_high_1"] == 3350.0 + _m21c_reset() + + +@test("H35: Markdown parse failure falls back to plain text, then commits") +def _(): + async def main(): + _m21c_reset() + _tb21._autoscan_chats.add(911) + rep = _m21c_report() + + async def rs(*a, **k): + return rep + + key = f"XAU/USD_{Timeframe.M15.value}" + with patch.object(_tb21, "run_signal", rs): + # Markdown attempt fails → plain-text retry succeeds → commit. + bot = _M21cBot(md_exc=_BadRequest21("Can't parse entities")) + await _tb21._scan_timeframe( + bot, Timeframe.M15, [Timeframe.M5, Timeframe.H1]) + assert bot.attempts == [_tb21.ParseMode.MARKDOWN, None], \ + f"expected md attempt then plain retry, got {bot.attempts}" + assert len(bot.sent) == 1 + assert key in _tb21._last_alert_fp, \ + "plain-text delivery must commit the fingerprint" + + # Both attempts fail → NOT committed (commit-after-send holds). + _m21c_reset() + _tb21._autoscan_chats.add(911) + bot2 = _M21cBot(exc=_BadRequest21("Can't parse entities")) + await _tb21._scan_timeframe( + bot2, Timeframe.M15, [Timeframe.M5, Timeframe.H1]) + assert key not in _tb21._last_alert_fp, \ + "a fully failed send must stay eligible for the next scan" + + # Breach path uses the same fallback helper. + _m21c_reset() + _tb21._autoscan_chats.add(911) + bot3 = _M21cBot(md_exc=_BadRequest21("Can't parse entities")) + await _tb21._handle_breach(bot3, "XAU/USD", 3301.0, "swing_high_1") + assert len(bot3.sent) == 1 and key in _tb21._last_alert_fp + _m21c_reset() + + _aio21.run(main()) + + +@test("H36: concurrent scan + breach on one key produce a single alert") +def _(): + async def main(): + _m21c_reset() + _tb21._autoscan_chats.add(911) + rep = _m21c_report() + + async def rs(*a, **k): + return rep + + gate = _aio21.Event() + bot = _M21cBot(gate=gate) + with patch.object(_tb21, "run_signal", rs): + t1 = _aio21.create_task(_tb21._scan_timeframe( + bot, Timeframe.M15, [Timeframe.M5, Timeframe.H1])) + await _aio21.sleep(0.02) # scan holds the key lock, blocked in send + t2 = _aio21.create_task(_tb21._handle_breach( + bot, "XAU/USD", 3301.0, "swing_high_1")) + await _aio21.sleep(0.02) # breach is queued on the lock + assert len(bot.attempts) == 1, \ + "breach must wait on the alert lock, not race into send" + gate.set() + await _aio21.gather(t1, t2) + assert len(bot.sent) == 1, \ + f"same fingerprint must alert exactly once, got {len(bot.sent)}" + _m21c_reset() + + _aio21.run(main()) + + +@test("H37: _start_scheduler respawns only the dead TF loop") +def _(): + async def main(): + saved_tasks = list(_tb21._autoscan_tasks) + saved_ws = _tb21._ws_task + _tb21._autoscan_tasks.clear() + _tb21._ws_task = None + + async def fake_loop(bot, tf, interval): + await _aio21.sleep(30) + + class NoStream: + name = "stub" + supports_streaming = False + + try: + with patch.object(_tb21, "_AUTOSCAN_LOOPS", + [(Timeframe.M5, 5), (Timeframe.M15, 15)]), \ + patch.object(_tb21, "_candle_loop", fake_loop), \ + patch.object(_tb21, "_get_streaming_dp", lambda: NoStream()): + _tb21._start_scheduler(None) + assert len(_tb21._autoscan_tasks) == 2 + m5 = next(t for t in _tb21._autoscan_tasks + if t.get_name() == "candle_loop_M5") + m15 = next(t for t in _tb21._autoscan_tasks + if t.get_name() == "candle_loop_M15") + m5.cancel() + await _aio21.sleep(0.01) # let the cancellation land + assert m5.done() and not m15.done() + + _tb21._start_scheduler(None) + names = sorted(t.get_name() for t in _tb21._autoscan_tasks) + assert names == ["candle_loop_M15", "candle_loop_M5"], names + assert m15 in _tb21._autoscan_tasks, \ + "the live loop must be kept, not restarted" + assert m5 not in _tb21._autoscan_tasks, \ + "the dead loop must be replaced by a fresh task" + finally: + for t in _tb21._autoscan_tasks: + t.cancel() + await _aio21.gather(*_tb21._autoscan_tasks, + return_exceptions=True) + if _tb21._ws_task is not None: + _tb21._ws_task.cancel() + await _aio21.gather(_tb21._ws_task, return_exceptions=True) + _tb21._autoscan_tasks[:] = saved_tasks + _tb21._ws_task = saved_ws + + _aio21.run(main()) + + +@test("H38a: clean WS close backs off; backoff resets only after ticks") +def _(): + async def main(): + _m21c_reset() + _tb21._autoscan_chats.add(911) + sleeps = [] + real_asyncio = _aio21 + + class AsyncioProxy: + def __getattr__(self, name): + return getattr(real_asyncio, name) + + @staticmethod + async def sleep(t): + sleeps.append(t) + await real_asyncio.sleep(0) + + def make_dp(tick_per_conn): + connects = {"n": 0} + + class FakeDP: + name = "fake" + supports_streaming = True + _api_key = "k" + + async def iter_ticks(self, symbols, on_connect=None): + connects["n"] += 1 + if connects["n"] >= 3: + raise real_asyncio.CancelledError + if tick_per_conn: + yield "XAU/USD", 3300.0 + return # clean close, no exception + + return FakeDP() + + try: + # No ticks: backoff must GROW across clean closes (1, then 2). + with patch.object(_tb21, "_get_streaming_dp", + lambda: make_dp(False)): + _tb21.asyncio = AsyncioProxy() + await _tb21._ws_monitor(_M21cBot()) + assert sleeps == [1, 2], \ + f"clean closes without data must back off, got {sleeps}" + # One tick per connection: backoff resets, sleeps stay at 1. + sleeps.clear() + with patch.object(_tb21, "_get_streaming_dp", + lambda: make_dp(True)): + await _tb21._ws_monitor(_M21cBot()) + assert sleeps == [1, 1], \ + f"a connection that produced ticks must reset backoff, " \ + f"got {sleeps}" + finally: + _tb21.asyncio = real_asyncio + _m21c_reset() + + _aio21.run(main()) + + +@test("H38b: TD iter_ticks fires on_connect on first RECEIVED frame only") +def _(): + import types as _types + + class FakeWS: + def __init__(self, messages): + self.messages = list(messages) + self.sent = [] + + async def send(self, m): + self.sent.append(m) + + def __aiter__(self): + return self + + async def __anext__(self): + if self.messages: + return self.messages.pop(0) + raise StopAsyncIteration + + class CM: + def __init__(self, ws): + self.ws = ws + + async def __aenter__(self): + return self.ws + + async def __aexit__(self, *a): + return False + + async def main(): + from providers.twelve_data import TwelveDataProvider + fake_mod = _types.ModuleType("websockets") + saved = sys.modules.get("websockets") + sys.modules["websockets"] = fake_mod + try: + td = TwelveDataProvider(PlatformConfig()) + # Server sends NOTHING (rejected key / silent close): the + # subscribe frame goes out but on_connect must NOT fire. + ws1 = FakeWS([]) + fake_mod.connect = lambda url, **kw: CM(ws1) + fired = [] + async for _ in td.iter_ticks(["XAU/USD"], + on_connect=lambda: fired.append(1)): + pass + await _aio21.sleep(0) # let the heartbeat task cancel cleanly + assert len(ws1.sent) == 1, "subscribe frame must have been sent" + assert not fired, \ + "on_connect must not fire before any server frame (H38b)" + # First received frame (even a non-price one) → fires. + ws2 = FakeWS([json.dumps({"event": "subscribe-status", + "status": "ok"})]) + fake_mod.connect = lambda url, **kw: CM(ws2) + async for _ in td.iter_ticks(["XAU/USD"], + on_connect=lambda: fired.append(1)): + pass + await _aio21.sleep(0) + assert fired == [1], "on_connect must fire on first received frame" + finally: + if saved is not None: + sys.modules["websockets"] = saved + else: + sys.modules.pop("websockets", None) + + _aio21.run(main()) + + +@test("scan-failure log carries the real traceback (not 'NoneType: None')") +def _(): + async def main(): + _m21c_reset() + + async def rs(*a, **k): + raise RuntimeError("m21-traceback-marker") + + import logging as _lg + records = [] + h = _lg.Handler() + h.emit = lambda rec: records.append(rec) + _tb21.logger.addHandler(h) + try: + with patch.object(_tb21, "run_signal", rs): + await _tb21._scan_timeframe(_M21cBot(), Timeframe.M15, []) + finally: + _tb21.logger.removeHandler(h) + fails = [r for r in records if "Auto-scan failed" in r.getMessage()] + assert fails, "the per-pair failure must be logged" + text = _lg.Formatter("%(message)s").format(fails[0]) + assert "NoneType: None" not in text + assert "m21-traceback-marker" in text and "RuntimeError" in text, \ + f"log must include the real traceback, got: {text[-160:]!r}" + _m21c_reset() + + _aio21.run(main()) + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════