From 6e2fdbd684b45b71aa5ab91db24c07fb6e500c4f Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Tue, 21 Jul 2026 13:18:05 -0400 Subject: [PATCH] telegram_bot: credentials/quality/scan-loops from config.json, startup guard - Credentials, alert quality, scan pairs/timeframes, log level and candle-loop intervals now resolve via app_config (env -> config.json -> code default) instead of hardcoded fallbacks. - _get_shared_dp() uses make_data_provider() for provider-registry dispatch. - main() refuses to start without a token/chat_id, raising SystemExit with an actionable message instead of a bare sys.exit string. - test_all.py Module 13: two new tests covering credential/quality/ loop wiring via a reload, and the startup guard. Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b --- telegram_bot.py | 43 +++++++++++++++++++++++++++---------------- test_all.py | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 16 deletions(-) diff --git a/telegram_bot.py b/telegram_bot.py index 182a570..1b23633 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -65,11 +65,16 @@ from config import ( PlatformConfig, SignalStrength, Timeframe, XAUUSD, PAIR_INFO, ) -from app_config import provider_api_key -from data_provider import ForexDataProvider +from app_config import ( + get_app_config, provider_api_key, telegram_token, telegram_chat_id, + telegram_channel_id, +) +from data_provider import ForexDataProvider, make_data_provider from pair_signals import PAIR_PRODUCERS from signal_aggregator import SignalAggregator, SignalReport +_APP = get_app_config() + # ── Force UTF-8 on Windows ──────────────────────────────────────────── if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") @@ -86,13 +91,14 @@ _file_handler = logging.handlers.TimedRotatingFileHandler( when="midnight", backupCount=30, encoding="utf-8", ) _file_handler.setFormatter(_log_formatter) -logging.basicConfig(level=logging.INFO, handlers=[_console_handler, _file_handler]) +logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO), + handlers=[_console_handler, _file_handler]) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) logger = logging.getLogger("telegram_bot") -TOKEN = os.environ.get("TELEGRAM_TOKEN", "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE") -CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022") +TOKEN = telegram_token() +CHAT_ID = telegram_chat_id() # Owner lock: only this user/chat may drive the bot (commands + buttons). # The bot is the product's signal source — without the lock, anyone who @@ -103,7 +109,7 @@ OWNER_ID: Optional[int] = int(CHAT_ID) if CHAT_ID else None # Telegram). When set, auto-scan signal alerts and level-breach alerts # are also posted there. Manual scans, status, cache and boot messages # stay owner-DM only. Set TELEGRAM_CHANNEL_ID="" to disable the feed. -_CHANNEL_ID_RAW = os.environ.get("TELEGRAM_CHANNEL_ID", "-1003942265371") +_CHANNEL_ID_RAW = telegram_channel_id() CHANNEL_ID: Optional[int] = int(_CHANNEL_ID_RAW) if _CHANNEL_ID_RAW else None # ── Pairs & buttons ─────────────────────────────────────────────────── @@ -111,7 +117,7 @@ CHANNEL_ID: Optional[int] = int(_CHANNEL_ID_RAW) if _CHANNEL_ID_RAW else None # the single switch for single- vs multi-pair operation. Buttons, WS # subscriptions, and scan loops all derive from it. -SCAN_PAIRS = list(PlatformConfig().scan_pairs) +SCAN_PAIRS = list(_APP.scan_pairs) _PAIRS_LABEL = " + ".join(PAIR_INFO[p]["display"] for p in SCAN_PAIRS) @@ -193,7 +199,7 @@ _shared_dp: Optional[ForexDataProvider] = None def _get_shared_dp() -> ForexDataProvider: global _shared_dp if _shared_dp is None: - _shared_dp = ForexDataProvider(PlatformConfig()) + _shared_dp = make_data_provider(PlatformConfig()) return _shared_dp @@ -476,7 +482,7 @@ async def run_signal( _QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1} _ALERT_QUALITY_CYCLE = ["C", "B", "A"] # C+ → B+ → A-only -_alert_min_quality = "C" +_alert_min_quality = _APP.alert_min_quality # Fingerprint of the last alert per (symbol, timeframe) so a persisting # signal doesn't re-send an identical message every candle. @@ -795,11 +801,12 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None: logger.info(f"{label} auto-scan loop stopped") +_TF_INTERVAL_MINUTES = { + Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30, + Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440, +} _AUTOSCAN_LOOPS = [ - (Timeframe.M5, 5), - (Timeframe.M15, 15), - (Timeframe.H1, 60), - (Timeframe.H4, 240), + (tf, _TF_INTERVAL_MINUTES[tf]) for tf in _APP.scan_timeframes ] @@ -815,7 +822,8 @@ def _start_scheduler(bot) -> None: _autoscan_tasks.append( asyncio.create_task(_candle_loop(bot, tf, interval)) ) - logger.info("Candle-boundary scheduler started (M5/M15/H1/H4)") + logger.info("Candle-boundary scheduler started (%s)", + "/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS)) if _ws_task is None or _ws_task.done(): _ws_task = asyncio.create_task(_ws_monitor(bot)) @@ -1260,8 +1268,11 @@ async def _post_shutdown(app: Application) -> None: # ── Entry point ─────────────────────────────────────────────────────── def main() -> None: - if not TOKEN: - sys.exit("TELEGRAM_TOKEN not set") + if not TOKEN or not CHAT_ID: + raise SystemExit( + "telegram_bot: missing Telegram credentials. Set telegram.token " + "and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and " + "TELEGRAM_CHAT_ID environment variables.") logger.info("Starting ForexBot2 Telegram bot…") diff --git a/test_all.py b/test_all.py index c2b5e4c..331ce7d 100644 --- a/test_all.py +++ b/test_all.py @@ -2766,6 +2766,43 @@ def _(): "credentials must not live in committed source" +@test("telegram_bot: credentials, quality, pairs, loops from app_config") +def _(): + import importlib + import telegram_bot + app = AppConfig(telegram_token="json-tok", telegram_chat_id="42", + telegram_channel_id="", alert_min_quality="B", + scan_timeframes=[Timeframe.M15, Timeframe.H1], + provider_settings={"twelve_data": {"api_key": "k"}}) + set_app_config(app) + try: + with patch.dict(os.environ, {}, clear=True): + tb = importlib.reload(telegram_bot) + assert tb.TOKEN == "json-tok" + assert tb.OWNER_ID == 42 + assert tb.CHANNEL_ID is None, '"" channel id must disable the feed' + assert tb._alert_min_quality == "B" + assert [tf for tf, _ in tb._AUTOSCAN_LOOPS] == \ + [Timeframe.M15, Timeframe.H1] + assert tb._AUTOSCAN_LOOPS == [(Timeframe.M15, 15), + (Timeframe.H1, 60)] + finally: + set_app_config(_TEST_APP) + importlib.reload(telegram_bot) + + +@test("telegram_bot: main() refuses to start without credentials") +def _(): + import telegram_bot + if telegram_bot.TOKEN: # only meaningful in the hermetic no-cred env + return + try: + telegram_bot.main() + assert False, "main() must refuse to start without token/chat_id" + except SystemExit as e: + assert "credentials" in str(e) + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════