From fd1e136e24ae256a9817e0ea6aad8712460d96bd Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Sun, 6 Sep 2026 16:50:21 -0400 Subject: [PATCH] chore: remove test-only helpers and the duplicate confirmation map (no behaviour change) - telegram_bot._format_scan_summary: multi-pair "auto-scan summary" leftover; production sends per-report format_signal_message. Its only caller was a test. - telegram_bot.scan_keyboard: thin wrapper only a test called; main_menu_keyboard already builds from _scan_button_rows directly. - telegram_bot.AUTOSCAN_CONFIRMATIONS was byte-identical to TF_CONFIRMATIONS; the candle loop now reads the one map (app_config comment updated). - SignalLogger.news_for: one-line wrapper over the module-level read_news that every real reader uses; the test now calls read_news like its neighbours. - pair_signals.get_supported_pairs: PAIR_PRODUCERS is the registry; the test lists it directly. Suite: 461 passed, 0 failed (two helper-only tests removed). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_0156YhogLytzj9P2iH1skGbL --- app_config.py | 2 +- pair_signals/__init__.py | 4 ---- signal_log.py | 5 ----- telegram_bot.py | 44 +--------------------------------------- test_all.py | 29 ++++---------------------- 5 files changed, 6 insertions(+), 78 deletions(-) diff --git a/app_config.py b/app_config.py index 1d191d8..e9e16cc 100644 --- a/app_config.py +++ b/app_config.py @@ -82,7 +82,7 @@ load_dotenv() _VALID_QUALITIES = ("A+", "A", "B", "C") _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") # Timeframes the bot's candle-boundary scheduler supports (must have -# AUTOSCAN_CONFIRMATIONS entries in telegram_bot.py). +# TF_CONFIRMATIONS entries in telegram_bot.py). _SCANNABLE_TFS = ("M5", "M15", "H1", "H4") _KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level", diff --git a/pair_signals/__init__.py b/pair_signals/__init__.py index 1abc310..882ce0f 100644 --- a/pair_signals/__init__.py +++ b/pair_signals/__init__.py @@ -21,10 +21,6 @@ def register_pair(symbol: str, config_factory: Callable): PAIR_PRODUCERS[symbol] = config_factory -def get_supported_pairs() -> List[str]: - return list(PAIR_PRODUCERS.keys()) - - def produce_signal( symbol: str, primary_tf: Timeframe, diff --git a/signal_log.py b/signal_log.py index c8f4ade..171e2a9 100644 --- a/signal_log.py +++ b/signal_log.py @@ -978,11 +978,6 @@ class SignalLogger: (now_iso, digest)) return digest - def news_for(self, rowid: int) -> Tuple[Optional[str], Optional[str]]: - """(events_json, headlines_json) for a row, legacy or interned.""" - with self._lock: - return read_news(self._conn, rowid) - # ── advisory-event seeding / dedup ──────────────────────────── def _seed_outcome(self, rowid: int, row: Dict[str, Any]) -> None: diff --git a/telegram_bot.py b/telegram_bot.py index ccd0aab..b1104c6 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -188,13 +188,6 @@ TF_CONFIRMATIONS = { Timeframe.H4: [Timeframe.H1, Timeframe.D1], } -AUTOSCAN_CONFIRMATIONS = { - Timeframe.M5: [Timeframe.M15, Timeframe.H1], - Timeframe.M15: [Timeframe.M5, Timeframe.H1], - Timeframe.H1: [Timeframe.M15, Timeframe.H4], - Timeframe.H4: [Timeframe.H1, Timeframe.D1], -} - _executor = ThreadPoolExecutor(max_workers=4) # ── Markdown escaping ───────────────────────────────────────────────── @@ -441,37 +434,6 @@ def format_signal_message(report: SignalReport) -> str: return "\n".join(lines) -# ── Auto-scan summary format ───────────────────────────────────────── - -def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str: - now = datetime.now(timezone.utc).strftime("%H:%M UTC") - lines = [ - f"📊 *{tf.value} Auto-Scan* | {now}", - "━━━━━━━━━━━━━━━━━━━━", - ] - for r in reports: - emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜") - sig = _signal_label(r.final_signal) - mtf = "✅" if r.mtf_agreement else "❌" - lines.append( - f"{emoji} *{r.display_name}* {sig} " - f"{_QUALITY_EMOJI.get(r.signal_quality, '')}{r.signal_quality} " - f"{r.final_score:+.1f} {r.final_confidence:.0f}% {mtf}MTF" - ) - lines.append("━━━━━━━━━━━━━━━━━━━━") - - actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")] - if actionable: - names = ", ".join(r.display_name for r in actionable) - lines.append(f"🔔 Actionable: {names}") - else: - lines.append("No actionable signals this cycle.") - - lines += ["", _DISCLAIMER] - - return "\n".join(lines) - - # ── Keyboard helper ─────────────────────────────────────────────────── def _scan_button_rows() -> List[List[InlineKeyboardButton]]: @@ -490,10 +452,6 @@ def _scan_button_rows() -> List[List[InlineKeyboardButton]]: return rows -def scan_keyboard() -> InlineKeyboardMarkup: - return InlineKeyboardMarkup(_scan_button_rows()) - - def main_menu_keyboard() -> InlineKeyboardMarkup: """Full interactive menu: scans + settings + status in one keyboard.""" autoscan_label = "🔔 Auto-scan: ON" if _autoscan_chats else "🔕 Auto-scan: OFF" @@ -1147,7 +1105,7 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None: logger.info(f"{label} auto-scan: firing") try: await _scan_timeframe( - bot, tf, AUTOSCAN_CONFIRMATIONS[tf] + bot, tf, TF_CONFIRMATIONS[tf] ) except Exception: # One failed cycle must not kill the loop for good — diff --git a/test_all.py b/test_all.py index cded7e5..f946668 100644 --- a/test_all.py +++ b/test_all.py @@ -1612,9 +1612,9 @@ print("\n═══ Module 7: telegram_bot.py ═══") from telegram_bot import ( _escape_md, _signal_label, _mtf_line, _indicator_summary, - format_signal_message, _format_scan_summary, _news_text, + format_signal_message, _news_text, _seconds_until_boundary, _check_level_breach, - scan_keyboard, SCAN_BUTTONS, SCAN_LOOKUP, + SCAN_BUTTONS, SCAN_LOOKUP, _update_level_cache, _last_prices, _level_cache, _level_cache_ts, ) @@ -1757,21 +1757,6 @@ def _(): assert "[Bullish]" not in msg and "[Bearish]" not in msg and "[Neutral]" not in msg -@test("_format_scan_summary produces valid text") -def _(): - config = PlatformConfig() - agg = SignalAggregator(config) - dfs = { - Timeframe.H1: make_ohlcv(300, seed=42), - Timeframe.M15: make_ohlcv(300, seed=43), - Timeframe.H4: make_ohlcv(300, seed=44), - } - reports = [agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)] - summary = _format_scan_summary(reports, Timeframe.H1) - assert "Auto-Scan" in summary - assert "XAU/USD" in summary - - @test("_seconds_until_boundary returns positive value") def _(): result = _seconds_until_boundary(15, buffer=10) @@ -1878,12 +1863,6 @@ def _(): assert "label" in btn -@test("scan_keyboard returns InlineKeyboardMarkup") -def _(): - kb = scan_keyboard() - assert kb is not None - - @test("_update_level_cache populates cache from reports") def _(): config = PlatformConfig() @@ -2070,7 +2049,7 @@ print("\n═══ Module 9: pair_signals registry ═══") @test("pair_signals registry includes XAU/USD only") def _(): import pair_signals - pairs = pair_signals.get_supported_pairs() + pairs = list(pair_signals.PAIR_PRODUCERS) assert XAUUSD in pairs, f"XAU missing from registry: {pairs}" assert pairs == [XAUUSD], f"registry must be XAU-only: {pairs}" @@ -8900,7 +8879,7 @@ def _(): legacy_id = lg._conn.execute("SELECT MAX(id) FROM signals").fetchone()[0] ev, hl = slog.read_news(lg._conn, legacy_id) assert json.loads(hl)[0]["headline"] == "legacy" - ev, hl = lg.news_for(r1) + ev, hl = slog.read_news(lg._conn, r1) assert json.loads(hl)[0]["headline"] == "Fed signals rate hike" view = dict(lg._conn.execute( "SELECT id, headlines_resolved FROM signals_with_news").fetchall())