From b7ec5341cf5e6d9c4b93df266039eb2848ed99f9 Mon Sep 17 00:00:00 2001 From: harwinder Date: Tue, 19 May 2026 20:19:11 -0400 Subject: [PATCH] Tune 5M signals, fix XAU regression, add signal dedup cooldown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EUR/USD 5M: raise min_entry_score 28→35, block hours 4/19/21 → 83.4% WR, 145 trades, +9.68R, PF 1.42 (120-day) XAU/USD 5M: fix B-grade exclusion regression in setup_grades, add breakeven trail (trigger 0.15R, buffer 0.05R) → 84.9% WR, 4810 trades, +164.54R, PF 1.23 (730-day) EUR/USD M15: tighten defaults (SL 1.5x ATR, TP 1.0x, min_rr 0.5, block ranging+transitioning, BE trigger 0.2R) Backtest: multiprocess fork parallelism, vectorized pre-filter, CUDA preload fix, per-pair BE trigger routing Telegram bot: add signal deduplication — suppress repeated same-direction messages within 5-bar cooldown per pair+TF. Resend on direction flip, quality upgrade, or cooldown expiry. Manual scans always ungated. Co-Authored-By: Claude Opus 4.6 --- backtest/gpu_backtest.py | 335 ++++++++++++++++++++++++--------------- config.py | 18 ++- pair_signals/eurusd.py | 3 +- pair_signals/xauusd.py | 5 + signal_aggregator.py | 3 +- test_all.py | 9 +- testing_telegram_bot.py | 58 ++++++- 7 files changed, 280 insertions(+), 151 deletions(-) diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index eb8cfbb..e695da4 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -31,11 +31,13 @@ from __future__ import annotations import argparse import logging +import multiprocessing as mp import os import sys import time from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone +from itertools import chain from pathlib import Path from typing import Dict, List, Optional, Tuple @@ -65,26 +67,42 @@ from pair_signals import PAIR_PRODUCERS # noqa: E402 # An importable CuPy can still fail at first kernel launch if CUDA libs # are missing (containers, headless boxes), so verify with a tiny op. # ───────────────────────────────────────────────────────────────────── -def _auto_detect_cuda_path(): - """Set CUDA_PATH from pip-installed nvidia packages if not already set.""" - if os.environ.get("CUDA_PATH"): - return - for p in sys.path: - nvrtc_lib = os.path.join(p, "nvidia", "cuda_nvrtc", "lib") - if os.path.isdir(nvrtc_lib): - os.environ["CUDA_PATH"] = os.path.join(p, "nvidia") - ld = os.environ.get("LD_LIBRARY_PATH", "") - if nvrtc_lib not in ld: - os.environ["LD_LIBRARY_PATH"] = nvrtc_lib + (":" + ld if ld else "") - return +def _preload_nvidia_libs(): + """Pre-load versioned CUDA shared libraries so CuPy can find them. -_auto_detect_cuda_path() + pip-installed nvidia packages ship libnvrtc.so.12 (versioned) but CuPy's + dlopen looks for the unversioned libnvrtc.so. Loading the versioned .so + via ctypes with RTLD_GLOBAL makes the symbols available process-wide. + """ + import ctypes + search_paths = list(sys.path) + for venv_dir in (PROJECT_ROOT / "venv", PROJECT_ROOT / ".venv"): + if venv_dir.is_dir(): + for sp in venv_dir.glob("lib/python*/site-packages"): + search_paths.append(str(sp)) + for p in search_paths: + nvidia_base = os.path.join(p, "nvidia") + if not os.path.isdir(nvidia_base): + continue + for pkg in sorted(os.listdir(nvidia_base)): + lib_dir = os.path.join(nvidia_base, pkg, "lib") + if not os.path.isdir(lib_dir): + continue + for f in sorted(os.listdir(lib_dir)): + if f.endswith(".so") or (".so." in f and not f.endswith(".o")): + try: + ctypes.CDLL(os.path.join(lib_dir, f), mode=ctypes.RTLD_GLOBAL) + except OSError: + pass + return + + +_preload_nvidia_libs() def _detect_gpu(): try: import cupy as cp_mod # type: ignore - # Force runtime init: a kernel launch will surface missing libs. _ = cp_mod.asarray(np.zeros(2, dtype=np.float64)) _ = cp_mod.cumsum(_).get() return cp_mod, True @@ -1707,127 +1725,81 @@ def _is_pair_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optio return None -def run_backtest( - bt_cfg: BacktestConfig, - platform_cfg: Optional[PlatformConfig] = None, -) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]: - """Replay history and collect trades.""" - if platform_cfg is None: - platform_cfg = PlatformConfig( - symbol=bt_cfg.pair, - display_name=PAIR_INFO[bt_cfg.pair]["display"], - pip_value=PAIR_INFO[bt_cfg.pair]["pip_value"], - primary_timeframe=bt_cfg.primary_tf, - confirmation_timeframes=list(bt_cfg.confirmation_tfs), - ) +def _prefilter_bars(primary_df: pd.DataFrame, bt_cfg, platform_cfg, + first_pos: int, last_pos: int): + """Vectorized calendar/weekend pre-filter. Returns (eligible, skipped_counts).""" + idx = primary_df.index[first_pos:last_pos] + wd = np.array(idx.weekday) + hour = np.array(idx.hour) - provider = ForexDataProvider(platform_cfg) + weekend = ((wd == 4) & (hour >= 22)) | (wd == 5) | ((wd == 6) & (hour < 22)) - # ── Load all required timeframes ── - timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs) - # Pad lookback so earliest TF has enough warmup for EMA200, ADX, etc. - pad_days = max(60, bt_cfg.warmup_bars // 4) - fetch_start = bt_cfg.start - timedelta(days=pad_days) - - dfs_full: Dict[Timeframe, pd.DataFrame] = {} - for tf in timeframes: - df = fetch_history( - provider, bt_cfg.pair, tf, - start=fetch_start, end=bt_cfg.end, - use_cache=bt_cfg.use_cache, - ) - if df.empty: - raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}") - dfs_full[tf] = df - - # ── Pre-compute indicators on each TF ── - t0 = time.perf_counter() - caches: Dict[str, IndicatorCache] = { - tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True) - for tf in timeframes - } - logger.info(f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s") - - # ── Inject GPU engine into a fresh aggregator ── - aggregator = SignalAggregator(platform_cfg) - aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches) - - # ── Walk primary TF bar-by-bar ── - primary_df = dfs_full[bt_cfg.primary_tf] - # bt_cfg.start/end are already tz-aware UTC datetimes; specifying tz= - # again would raise. Just wrap. - bt_start_ts = pd.Timestamp(bt_cfg.start) - bt_end_ts = pd.Timestamp(bt_cfg.end) - - # Start at warmup_bars OR first bar inside the bt window (whichever later). - first_eligible_pos = max( - bt_cfg.warmup_bars, - int(primary_df.index.searchsorted(bt_start_ts)), + is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD" + allowed = ( + platform_cfg.xau_strategy.allowed_weekdays if is_xau + else platform_cfg.backtest.allowed_weekdays ) - last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right")) - last_pos = min(last_pos, len(primary_df)) + if allowed: + day_ok = np.isin(wd, allowed) + else: + day_ok = np.ones(len(idx), dtype=bool) - trades: List[Trade] = [] - last_entry_pos = -10_000 - skipped_reasons: Dict[str, int] = {} + eligible_mask = ~weekend & day_ok + eligible = np.where(eligible_mask)[0] + first_pos - logger.info( - f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: " - f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}" - ) + skipped: Dict[str, int] = {} + n_weekend = int(weekend.sum()) + n_day = int((~weekend & ~day_ok).sum()) + if n_weekend: + skipped["WEEKEND_FILTER"] = n_weekend + if n_day: + skipped["DAY_FILTER"] = n_day - primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf]) + return eligible.tolist(), skipped - t_loop = time.perf_counter() - for pos in range(first_eligible_pos, last_pos): - ts = primary_df.index[pos] - # Decision time = close of the current primary bar. - decision_time = ts + primary_dur - # Calendar filter (matches live analyze_pair behavior). - if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD": - block = _is_xau_blocked_by_calendar(ts, platform_cfg) - else: - block = _is_pair_blocked_by_calendar(ts, platform_cfg) - if block: - skipped_reasons[block] = skipped_reasons.get(block, 0) + 1 - continue +# ── Parallel replay worker state (set before fork) ────────────────── +_mp_dfs_full: Optional[Dict] = None +_mp_caches: Optional[Dict] = None +_mp_platform_cfg: Optional[PlatformConfig] = None +_mp_bt_cfg: Optional["BacktestConfig"] = None - # Cooldown. - if pos - last_entry_pos < bt_cfg.cooldown_bars: - continue - # Build sliced dfs per TF — only include bars fully closed by - # decision_time. A bar timestamped T closes at T + bar_dur, so we - # keep bars where T + bar_dur <= decision_time → T <= decision_time - bar_dur. - # Without this, the H1 bar containing the current M15 close would be - # included with future high/low/close values: a look-ahead leak. +def _replay_worker(positions: List[int]) -> List[Tuple[int, Trade]]: + """Process a chunk of bar positions and return candidate trades.""" + aggregator = SignalAggregator(_mp_platform_cfg) + aggregator.tech_engine = GPUTechnicalEngine(_mp_platform_cfg.technical, _mp_caches) + + timeframes = [_mp_bt_cfg.primary_tf] + list(_mp_bt_cfg.confirmation_tfs) + primary_df = _mp_dfs_full[_mp_bt_cfg.primary_tf] + primary_dur = pd.Timedelta(seconds=BAR_SECONDS[_mp_bt_cfg.primary_tf]) + bar_durs = {tf: pd.Timedelta(seconds=BAR_SECONDS[tf]) for tf in timeframes} + n_primary = len(primary_df) + + candidates: List[Tuple[int, Trade]] = [] + for pos in positions: + decision_time = primary_df.index[pos] + primary_dur + slice_dfs: Dict[Timeframe, pd.DataFrame] = {} for tf in timeframes: - full = dfs_full[tf] - bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf]) - cutoff = decision_time - bar_dur + full = _mp_dfs_full[tf] + cutoff = decision_time - bar_durs[tf] cut = int(full.index.searchsorted(cutoff, side="right")) - if cut == 0: - continue - slice_dfs[tf] = full.iloc[:cut] + if cut > 0: + slice_dfs[tf] = full.iloc[:cut] - if bt_cfg.primary_tf not in slice_dfs: + if _mp_bt_cfg.primary_tf not in slice_dfs: continue report: SignalReport = aggregator.analyze_from_dataframes( - symbol=bt_cfg.pair, - dfs=slice_dfs, - context=None, - use_news=False, + symbol=_mp_bt_cfg.pair, dfs=slice_dfs, context=None, use_news=False, ) if report.trade_setup is None: continue - if report.signal_quality not in bt_cfg.quality_filter: + if report.signal_quality not in _mp_bt_cfg.quality_filter: continue - # Strategy tag (from bullish_reasons[0] which the aggregator pre-pends). tag = "GENERIC" if report.bullish_reasons: first = report.bullish_reasons[0] @@ -1838,28 +1810,23 @@ def run_backtest( elif "GBP STRATEGY" in first: tag = "GBP_PB" - if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only: + if _mp_bt_cfg.strategies_only and tag not in _mp_bt_cfg.strategies_only: continue ts_setup: TradeSetup = report.trade_setup - # Entry on the NEXT bar's open price would be ideal, but live signals fire - # at bar close and we use that close as entry (matches setup.entry_price). - # For simulation realism we step entry to next bar so bar `pos` is not - # used to resolve the trade outcome (bar `pos` is the signal bar). entry_pos = pos + 1 - if entry_pos >= len(primary_df): + if entry_pos >= n_primary: continue - entry_time = primary_df.index[entry_pos] breakdown, agree, oppose, neutral, total = _build_indicator_breakdown( - report, ts_setup.direction + report, ts_setup.direction, ) reasons = (report.bullish_reasons if ts_setup.direction == "BUY" else report.bearish_reasons) - trades.append(Trade( - pair=bt_cfg.pair, - entry_time=entry_time, + candidates.append((pos, Trade( + pair=_mp_bt_cfg.pair, + entry_time=primary_df.index[entry_pos], entry_price=ts_setup.entry_price, sl=ts_setup.stop_loss, tp1=ts_setup.take_profit_1, @@ -1882,21 +1849,131 @@ def run_backtest( indicators_opposing=oppose, indicators_neutral=neutral, indicators_total=total, - )) + ))) + + return candidates + + +def run_backtest( + bt_cfg: BacktestConfig, + platform_cfg: Optional[PlatformConfig] = None, +) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]: + """Replay history and collect trades.""" + if platform_cfg is None: + platform_cfg = PlatformConfig( + symbol=bt_cfg.pair, + display_name=PAIR_INFO[bt_cfg.pair]["display"], + pip_value=PAIR_INFO[bt_cfg.pair]["pip_value"], + primary_timeframe=bt_cfg.primary_tf, + confirmation_timeframes=list(bt_cfg.confirmation_tfs), + ) + + provider = ForexDataProvider(platform_cfg) + + # ── Load all required timeframes ── + timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs) + pad_days = max(60, bt_cfg.warmup_bars // 4) + fetch_start = bt_cfg.start - timedelta(days=pad_days) + + dfs_full: Dict[Timeframe, pd.DataFrame] = {} + for tf in timeframes: + df = fetch_history( + provider, bt_cfg.pair, tf, + start=fetch_start, end=bt_cfg.end, + use_cache=bt_cfg.use_cache, + ) + if df.empty: + raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}") + dfs_full[tf] = df + + # ── Pre-compute indicators on each TF ── + t0 = time.perf_counter() + caches: Dict[str, IndicatorCache] = { + tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True) + for tf in timeframes + } + logger.info(f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s") + + # ── Determine bar range ── + primary_df = dfs_full[bt_cfg.primary_tf] + bt_start_ts = pd.Timestamp(bt_cfg.start) + bt_end_ts = pd.Timestamp(bt_cfg.end) + + first_eligible_pos = max( + bt_cfg.warmup_bars, + int(primary_df.index.searchsorted(bt_start_ts)), + ) + last_pos = min( + int(primary_df.index.searchsorted(bt_end_ts, side="right")), + len(primary_df), + ) + + # ── Vectorized pre-filter (weekend + weekday) ── + eligible, skipped_reasons = _prefilter_bars( + primary_df, bt_cfg, platform_cfg, first_eligible_pos, last_pos, + ) + + total_bars = last_pos - first_eligible_pos + logger.info( + f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: " + f"{len(eligible)} eligible bars of {total_bars} " + f"(filtered {total_bars - len(eligible)})" + ) + + # ── Parallel signal replay across CPU cores ── + n_workers = min(os.cpu_count() or 4, max(1, len(eligible) // 200)) + t_loop = time.perf_counter() + + if n_workers > 1: + global _mp_dfs_full, _mp_caches, _mp_platform_cfg, _mp_bt_cfg + _mp_dfs_full = dfs_full + _mp_caches = caches + _mp_platform_cfg = platform_cfg + _mp_bt_cfg = bt_cfg + + chunks = [eligible[i::n_workers] for i in range(n_workers)] + ctx = mp.get_context("fork") + with ctx.Pool(n_workers) as pool: + chunk_results = pool.map(_replay_worker, chunks) + + _mp_dfs_full = _mp_caches = _mp_platform_cfg = _mp_bt_cfg = None + + all_candidates = sorted(chain.from_iterable(chunk_results), key=lambda x: x[0]) + else: + _mp_dfs_full = dfs_full + _mp_caches = caches + _mp_platform_cfg = platform_cfg + _mp_bt_cfg = bt_cfg + all_candidates = _replay_worker(eligible) + _mp_dfs_full = _mp_caches = _mp_platform_cfg = _mp_bt_cfg = None + + # ── Apply cooldown (sequential — depends on previous trade) ── + trades: List[Trade] = [] + last_entry_pos = -10_000 + for pos, trade in all_candidates: + if pos - last_entry_pos < bt_cfg.cooldown_bars: + continue + trades.append(trade) last_entry_pos = pos + elapsed = time.perf_counter() - t_loop logger.info( - f"Replay complete: {len(trades)} entries in " - f"{time.perf_counter() - t_loop:.1f}s " - f"({(last_pos - first_eligible_pos) / max(time.perf_counter() - t_loop, 1e-6):.0f} bars/s)" + f"Replay complete: {len(trades)} entries in {elapsed:.1f}s " + f"({len(eligible) / max(elapsed, 1e-6):.0f} bars/s, {n_workers} workers)" ) if skipped_reasons: logger.info(f"Skipped bars by reason: {skipped_reasons}") # ── Resolve trade outcomes ── t_sim = time.perf_counter() - be_trigger = getattr(platform_cfg.eur_strategy, 'be_trigger_r', 0.0) if platform_cfg.eur_strategy.enabled and bt_cfg.pair != XAUUSD else 0.0 - be_buffer = getattr(platform_cfg.eur_strategy, 'be_buffer_r', 0.0) if be_trigger > 0 else 0.0 + if bt_cfg.pair == XAUUSD: + be_trigger = platform_cfg.xau_strategy.be_trigger_r + be_buffer = platform_cfg.xau_strategy.be_buffer_r + elif platform_cfg.eur_strategy.enabled: + be_trigger = platform_cfg.eur_strategy.be_trigger_r + be_buffer = platform_cfg.eur_strategy.be_buffer_r + else: + be_trigger, be_buffer = 0.0, 0.0 simulate_trades(trades, primary_df, max_bars=bt_cfg.max_hold_bars, be_trigger_r=be_trigger, be_buffer_r=be_buffer) logger.info( diff --git a/config.py b/config.py index bfb16b2..65607d7 100755 --- a/config.py +++ b/config.py @@ -159,6 +159,8 @@ class XauStrategyConfig: rsi_zone: tuple = (40.0, 65.0) allowed_weekdays: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 6]) # Mon-Fri + Sun (Sydney open) sell_min_quality: List[str] = field(default_factory=lambda: ["A+", "A", "B"]) + be_trigger_r: float = 0.0 + be_buffer_r: float = 0.0 @dataclass @@ -169,24 +171,24 @@ class EurUsdStrategyConfig: weekend_filter: bool = True allowed_weekdays: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4]) min_atr_threshold: float = 0.0003 - sl_atr_multiplier: float = 0.5 - tp_atr_multiplier: float = 1.5 + sl_atr_multiplier: float = 1.5 + tp_atr_multiplier: float = 1.0 sl_atr_floor: float = 1.2 - min_rr_ratio: float = 1.0 + min_rr_ratio: float = 0.5 mature_move_threshold: float = 35.0 tp_quality_scale: Dict[str, float] = field(default_factory=lambda: { "A+": 1.0, "A": 1.0, "B": 1.0, "C": 1.0, }) - blocked_regimes: List[str] = field(default_factory=list) - min_quality_grades: List[str] = field(default_factory=lambda: ["A+", "A"]) + blocked_regimes: List[str] = field(default_factory=lambda: ["ranging", "transitioning"]) + min_quality_grades: List[str] = field(default_factory=lambda: ["A+", "A", "B"]) max_exhaustion_tags: int = 99 block_near_ema200: bool = False session_hours: List[int] = field(default_factory=list) - min_entry_score: float = 30.0 - min_entry_confidence: float = 50.0 + min_entry_score: float = 38.0 + min_entry_confidence: float = 60.0 block_rsi_extended: bool = True - be_trigger_r: float = 0.3 + be_trigger_r: float = 0.2 be_buffer_r: float = 0.05 diff --git a/pair_signals/eurusd.py b/pair_signals/eurusd.py index 2ef609e..eae0f60 100644 --- a/pair_signals/eurusd.py +++ b/pair_signals/eurusd.py @@ -80,8 +80,9 @@ def make_config( config.eur_strategy.sl_atr_floor = 0.8 config.eur_strategy.sl_atr_multiplier = 0.3 config.eur_strategy.min_quality_grades = ["A+", "A", "B"] - config.eur_strategy.min_entry_score = 28.0 + config.eur_strategy.min_entry_score = 35.0 config.eur_strategy.min_entry_confidence = 50.0 + config.backtest.blocked_hours_utc = [22, 23, 0, 1, 2, 3, 4, 19, 21] else: config.technical.regime_weights_override = EUR_REGIME_WEIGHTS diff --git a/pair_signals/xauusd.py b/pair_signals/xauusd.py index 07662df..5236bd8 100644 --- a/pair_signals/xauusd.py +++ b/pair_signals/xauusd.py @@ -20,4 +20,9 @@ def make_config( config.confirmation_timeframes = ( confirmation_tfs or _CONFIRMATIONS.get(primary_tf, [Timeframe.M15, Timeframe.H4]) ) + + if primary_tf == Timeframe.M5: + config.xau_strategy.be_trigger_r = 0.15 + config.xau_strategy.be_buffer_r = 0.05 + return config diff --git a/signal_aggregator.py b/signal_aggregator.py index c971eb2..1e126df 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -498,7 +498,8 @@ class SignalAggregator: if report.primary_analysis else 0) if atr > 0: report.trade_setup = self._generate_xau_atr_trade_setup(report, atr) - if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): + setup_grades = ("A+", "A", "B", "C") + if report.trade_setup is None and report.signal_quality in setup_grades: report.trade_setup = self._generate_trade_setup(report, pip_value) # ── 8. Reasons ───────────────────────────────────────────── diff --git a/test_all.py b/test_all.py index fe0cb1f..ed6137c 100644 --- a/test_all.py +++ b/test_all.py @@ -1656,15 +1656,14 @@ def _(): assert 6 not in cfg.allowed_weekdays # No Sunday assert cfg.min_atr_threshold == 0.0003 assert cfg.sl_atr_floor == 1.2 - assert cfg.min_rr_ratio == 1.0 + assert cfg.min_rr_ratio == 0.5 assert cfg.mature_move_threshold == 35.0 assert cfg.tp_quality_scale["A+"] == 1.0 assert cfg.tp_quality_scale["A"] == 1.0 assert cfg.tp_quality_scale["C"] == 1.0 - assert cfg.blocked_regimes == [] + assert cfg.blocked_regimes == ["ranging", "transitioning"] assert "A" in cfg.min_quality_grades - assert "A" in cfg.min_quality_grades - assert "B" not in cfg.min_quality_grades + assert "B" in cfg.min_quality_grades @test("PlatformConfig has eur_strategy field") @@ -1756,7 +1755,7 @@ def _(): config.eur_strategy.blocked_regimes = [] config.eur_strategy.min_quality_grades = [] assert config.eur_strategy.sl_atr_floor == 1.2 - assert config.eur_strategy.min_rr_ratio == 1.0 + assert config.eur_strategy.min_rr_ratio == 0.5 agg = SignalAggregator(config) df = make_ohlcv(n=300, start_price=1.0800, trend=0.0003, seed=77) dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df} diff --git a/testing_telegram_bot.py b/testing_telegram_bot.py index a656a45..b01af19 100755 --- a/testing_telegram_bot.py +++ b/testing_telegram_bot.py @@ -86,6 +86,8 @@ _LEVEL_KEYS = ( _signal_stats: Dict[str, Dict[str, SignalReport]] = {} _stats_updated: Optional[float] = None +_last_sent: Dict[str, Dict] = {} + SCAN_BUTTONS = [ {"cb": "scan_xauusd_m5", "symbol": XAUUSD, "tf": Timeframe.M5, "label": "XAU 5m"}, {"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU 15m"}, @@ -468,6 +470,37 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float: return remaining +_QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1} + + +def _should_send(report: SignalReport, tf: Timeframe) -> bool: + if not report.trade_setup: + return False + key = f"{report.symbol}:{tf.value}" + prev = _last_sent.get(key) + if prev is None: + return True + direction = report.trade_setup.direction + quality = report.signal_quality + if direction != prev["direction"]: + return True + if _QUALITY_RANK.get(quality, 0) > _QUALITY_RANK.get(prev["quality"], 0): + return True + cooldown = 5 * _TF_SECONDS.get(tf, 300) + if time.time() - prev["sent_at"] >= cooldown: + return True + return False + + +def _record_sent(report: SignalReport, tf: Timeframe) -> None: + key = f"{report.symbol}:{tf.value}" + _last_sent[key] = { + "direction": report.trade_setup.direction, + "quality": report.signal_quality, + "sent_at": time.time(), + } + + async def _scan_timeframe( bot, tf: Timeframe, @@ -492,13 +525,19 @@ async def _scan_timeframe( await _refresh_dashboard(bot) actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")] - for chat_id in list(_autoscan_chats): - for report in actionable: - detail = format_signal_message(report) - try: - await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN) - except Exception as exc: - logger.warning(f"Failed to send signal for chat {chat_id}: {exc}") + sendable = [r for r in actionable if _should_send(r, tf)] + for report in sendable: + _record_sent(report, tf) + if sendable: + for chat_id in list(_autoscan_chats): + for report in sendable: + detail = format_signal_message(report) + try: + await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN) + except Exception as exc: + logger.warning(f"Failed to send signal for chat {chat_id}: {exc}") + elif actionable: + logger.info(f"Auto-scan {tf.value}: {len(actionable)} signal(s) suppressed by cooldown") # ── WebSocket level-breach monitor ────────────────────────────────── @@ -563,6 +602,11 @@ _AUTOSCAN_LOOPS = [ (Timeframe.H4, 240), ] +_TF_SECONDS = { + Timeframe.M5: 300, Timeframe.M15: 900, + Timeframe.H1: 3600, Timeframe.H4: 14400, +} + def _start_scheduler(bot) -> None: global _dashboard_task