From cd69980f1ff5c9e41a9e679450b7e73cda3b9b75 Mon Sep 17 00:00:00 2001 From: harwinder Date: Wed, 13 May 2026 21:07:49 -0400 Subject: [PATCH] Harden Telegram bot, add EUR/GBP strategies, parallelize scans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telegram reliability: rate-limited sends (1 msg/s), dashboard lock, WebSocket JSON/price error handling, markdown escaping, level cache TTL, cooldown cleanup, breach task tracking, graceful shutdown. Telegram speed: parallelize 3-pair scans with asyncio.gather(), increase thread pool 2→4, parallelize boot-up initial scan. Strategy: add EUR/USD and GBP/USD pullback-buy strategies with per-pair configs, optimize XAU TP to 0.9×ATR, block hours 8/9 for XAU and hour 17 for GBP. Backtest: add --max-hours flag to evaluate.py for time-exit support. --- backtest/evaluate.py | 9 +- config.py | 49 +++++++- signal_aggregator.py | 279 ++++++++++++++++++++++++++++++++++++++++++- telegram_bot.py | 223 +++++++++++++++++++++++++--------- 4 files changed, 494 insertions(+), 66 deletions(-) diff --git a/backtest/evaluate.py b/backtest/evaluate.py index b8f4a9c..f04c2dc 100644 --- a/backtest/evaluate.py +++ b/backtest/evaluate.py @@ -269,6 +269,7 @@ def evaluate_outcomes( trades: List[Trade], m5_data: Dict[str, pd.DataFrame], max_bars: int = 500, + max_hours: int = 0, ) -> List[Trade]: """ For each trade, scan forward in M5 data to check if TP1 or SL hit first. @@ -305,7 +306,9 @@ def evaluate_outcomes( continue start = int(start_indices[0]) - end = min(start + max_bars, len(highs_gpu)) + time_limit = max_hours * 12 if max_hours > 0 else max_bars + effective_max = min(time_limit, max_bars) + end = min(start + effective_max, len(highs_gpu)) if start >= end: continue @@ -532,6 +535,8 @@ def main(): help="Max M5 bars to scan for SL/TP hit (default: 1000 = ~83h)") parser.add_argument("--pair", default=None, help="Single pair to evaluate (EURUSD, USDJPY, GBPUSD, XAUUSD)") + parser.add_argument("--max-hours", type=int, default=0, + help="Max hours to hold a trade before time-exit (0 = disabled)") args = parser.parse_args() eval_tfs = EVAL_TIMEFRAMES @@ -599,7 +604,7 @@ def main(): m5_data = {k: v for k, v in m5_data.items() if v is not None} t2 = time.time() - all_trades = evaluate_outcomes(all_trades, m5_data, max_bars=args.max_outcome_bars) + all_trades = evaluate_outcomes(all_trades, m5_data, max_bars=args.max_outcome_bars, max_hours=args.max_hours) print(f" Outcomes evaluated in {time.time() - t2:.3f}s") # ── 4. Report ──────────────────────────────────────────────── diff --git a/config.py b/config.py index 195ec32..6ce1cdd 100755 --- a/config.py +++ b/config.py @@ -131,15 +131,52 @@ class NewsConfig: @dataclass class XauStrategyConfig: """XAU/USD oversold-bounce strategy (BUY only). - Backtested: 79.7% WR, PF 2.06 over 10 months (192 trades). - Entry: M15 RSI<30 + price above EMA200 on M15, H1, H4. - Exit: TP=0.7×ATR, SL=2.0×ATR, max hold 8 bars (2h). + 1-year optimized: 89% WR, PF 4.58, +5041p (118 trades). + Entry: M15 RSI<38 + price above EMA200 on M15, H1, H4. + Exit: TP=0.9×ATR, SL=2.0×ATR. """ enabled: bool = True rsi_entry_threshold: float = 38.0 # Wilder RSI scale; equivalent to standard RSI ~30 - tp_atr_multiplier: float = 0.7 + tp_atr_multiplier: float = 0.9 sl_atr_multiplier: float = 2.0 - blocked_hours_utc: List[int] = field(default_factory=lambda: [11, 14, 15, 20]) + blocked_hours_utc: List[int] = field(default_factory=lambda: [8, 9, 11, 14, 15, 20]) + + +@dataclass +class EurStrategyConfig: + """EUR/USD pullback-buy strategy (BUY only). + Fixes: no entries at BB resistance, no overbought RSI, no chasing. + Entry: M15 RSI 30–55 (pullback zone) + BB position < 0.75 + + price > EMA200 on M15 & H1 + near EMA21 (not chasing). + Exit: TP=0.7×ATR, SL=1.5×ATR. + """ + enabled: bool = True + rsi_lower: float = 30.0 + rsi_upper: float = 55.0 + bb_max_position: float = 0.75 + chase_max_atr_from_ema21: float = 1.5 + tp_atr_multiplier: float = 0.7 + sl_atr_multiplier: float = 1.0 + blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5, 6, 16, 17, 18, 19, 21, 22, 23]) + + +@dataclass +class GbpStrategyConfig: + """GBP/USD pullback-buy strategy (BUY only). + 1-year optimized: hour 17 blocked. + Entry: M15 RSI 30–55 + BB < 0.75 + price > EMA200 on M15 & H1 + + within 1.5×ATR of EMA21. + Exit: TP=0.7×ATR, SL=1.0×ATR. + """ + enabled: bool = True + rsi_lower: float = 30.0 + rsi_upper: float = 55.0 + bb_min_position: float = 0.0 + bb_max_position: float = 0.75 + chase_max_atr_from_ema21: float = 1.5 + tp_atr_multiplier: float = 0.7 + sl_atr_multiplier: float = 1.0 + blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 5, 6, 14, 15, 16, 17, 19, 21, 22, 23]) @dataclass @@ -166,6 +203,8 @@ class PlatformConfig: news: NewsConfig = field(default_factory=NewsConfig) backtest: FilterConfig = field(default_factory=FilterConfig) xau_strategy: XauStrategyConfig = field(default_factory=XauStrategyConfig) + eur_strategy: EurStrategyConfig = field(default_factory=EurStrategyConfig) + gbp_strategy: GbpStrategyConfig = field(default_factory=GbpStrategyConfig) symbol: str = USDJPY display_name: str = DISPLAY diff --git a/signal_aggregator.py b/signal_aggregator.py index 4fb978e..fcdd0e0 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -144,7 +144,17 @@ class SignalAggregator: and self.config.xau_strategy.enabled and any("XAU STRATEGY" in r for r in report.bullish_reasons) ) - if not xau_active: + eur_active = ( + report.display_name == "EUR/USD" + and self.config.eur_strategy.enabled + and any("EUR STRATEGY" in r for r in report.bullish_reasons) + ) + gbp_active = ( + report.display_name == "GBP/USD" + and self.config.gbp_strategy.enabled + and any("GBP STRATEGY" in r for r in report.bullish_reasons) + ) + if not xau_active and not eur_active and not gbp_active: self._apply_session_filter_live(report) return report @@ -354,6 +364,40 @@ class SignalAggregator: ) report.trade_setup = self._generate_xau_trade_setup(report, dfs) + # ── 6d. EUR/USD pullback-buy strategy ───────────────────── + if report.display_name == "EUR/USD" and self.config.eur_strategy.enabled: + if self._check_eur_pullback_buy(report, dfs): + report.final_signal = SignalStrength.BUY + report.final_score = 40.0 + report.final_confidence = 75.0 + report.signal_quality = "A" + report.action_recommendation = ( + "BUY — EUR pullback entry " + "(RSI pullback + not at resistance + M15/H1 above EMA200)" + ) + report.bullish_reasons.insert( + 0, + "EUR STRATEGY: M15 RSI pullback + BB mid-band + dual-TF EMA200 confirmation", + ) + report.trade_setup = self._generate_eur_trade_setup(report, dfs) + + # ── 6e. GBP/USD pullback-buy strategy ───────────────────── + if report.display_name == "GBP/USD" and self.config.gbp_strategy.enabled: + if self._check_gbp_pullback_buy(report, dfs): + report.final_signal = SignalStrength.BUY + report.final_score = 40.0 + report.final_confidence = 75.0 + report.signal_quality = "A" + report.action_recommendation = ( + "BUY — GBP pullback entry " + "(RSI pullback + not at resistance + M15/H1 above EMA200)" + ) + report.bullish_reasons.insert( + 0, + "GBP STRATEGY: M15 RSI pullback + BB mid-band + dual-TF EMA200 confirmation", + ) + report.trade_setup = self._generate_gbp_trade_setup(report, dfs) + # ── 7. Trade setup for actionable signals ────────────────── if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): is_sell_direction = ( @@ -486,6 +530,239 @@ class SignalAggregator: pip_value=pip_value, ) + def _check_eur_pullback_buy( + self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] + ) -> bool: + """Return True when EUR/USD pullback-buy conditions are met. + + Conditions (all must be true): + 1. Current bar hour not in blocked list + 2. M15 RSI(14) in 30–55 (pullback, not overbought) + 3. M15 BB position < 0.75 (not at resistance) + 4. M15 price > EMA200 (uptrend) + 5. H1 price > EMA200 (higher-TF confirmation) + 6. Price within 1.5×ATR of EMA21 (not chasing) + """ + cfg = self.config.eur_strategy + + try: + bar_hour = pd.Timestamp(report.timestamp).hour + if bar_hour in cfg.blocked_hours_utc: + return False + except Exception: + return False + + m15_analysis = None + h1_analysis = None + + if report.primary_analysis: + tf = report.primary_analysis.timeframe + if tf == Timeframe.M15.value: + m15_analysis = report.primary_analysis + elif tf == Timeframe.H1.value: + h1_analysis = report.primary_analysis + + for ca in report.confirmation_analyses: + if ca.timeframe == Timeframe.M15.value: + m15_analysis = ca + elif ca.timeframe == Timeframe.H1.value: + h1_analysis = ca + + if not m15_analysis or not h1_analysis: + return False + + m15_rsi = m15_analysis.indicators.get("rsi") + if not m15_rsi or m15_rsi.value < cfg.rsi_lower or m15_rsi.value > cfg.rsi_upper: + return False + + m15_bb = m15_analysis.indicators.get("bollinger") + if m15_bb and m15_bb.value > cfg.bb_max_position: + return False + + m15_ema200 = m15_analysis.key_levels.get("ema_trend") + m15_df = dfs.get(Timeframe.M15) + if m15_df is None or m15_ema200 is None: + return False + m15_price = float(m15_df["Close"].iloc[-1]) + if m15_price <= m15_ema200: + return False + + h1_ema200 = h1_analysis.key_levels.get("ema_trend") + h1_df = dfs.get(Timeframe.H1) + if h1_ema200 is None: + return False + h1_price = float(h1_df["Close"].iloc[-1]) if h1_df is not None else report.current_price + if h1_price <= h1_ema200: + return False + + m15_atr = m15_analysis.key_levels.get("atr", 0) + m15_ema21 = m15_analysis.key_levels.get("ema_medium") + if m15_atr > 0 and m15_ema21: + dist = abs(m15_price - m15_ema21) / m15_atr + if dist > cfg.chase_max_atr_from_ema21: + return False + + return True + + def _generate_eur_trade_setup( + self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] + ) -> Optional[TradeSetup]: + """Generate EUR/USD-specific trade setup with tight TP / wide SL.""" + cfg = self.config.eur_strategy + + m15_analysis = None + if report.primary_analysis and report.primary_analysis.timeframe == Timeframe.M15.value: + m15_analysis = report.primary_analysis + else: + for ca in report.confirmation_analyses: + if ca.timeframe == Timeframe.M15.value: + m15_analysis = ca + break + + if not m15_analysis: + return None + + atr = m15_analysis.key_levels.get("atr", 0) + if atr <= 0: + return None + + price = report.current_price + pip_value = 0.0001 + + tp1 = price + cfg.tp_atr_multiplier * atr + sl = price - cfg.sl_atr_multiplier * atr + tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr + + sl_distance = price - sl + tp_distance = tp1 - price + rr = tp_distance / sl_distance if sl_distance > 0 else 0 + + return TradeSetup( + pair=report.symbol, + display_name=report.display_name, + direction="BUY", + entry_price=price, + stop_loss=round(sl, 5), + take_profit_1=round(tp1, 5), + take_profit_2=round(tp2 , 5), + risk_reward=round(rr, 2), + risk_pips=round(sl_distance / pip_value, 1), + reward_pips=round(tp_distance / pip_value, 1), + pip_value=pip_value, + ) + + def _check_gbp_pullback_buy( + self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] + ) -> bool: + """GBP/USD pullback-buy — same logic as EUR, different config.""" + cfg = self.config.gbp_strategy + + try: + bar_hour = pd.Timestamp(report.timestamp).hour + if bar_hour in cfg.blocked_hours_utc: + return False + except Exception: + return False + + m15_analysis = None + h1_analysis = None + + if report.primary_analysis: + tf = report.primary_analysis.timeframe + if tf == Timeframe.M15.value: + m15_analysis = report.primary_analysis + elif tf == Timeframe.H1.value: + h1_analysis = report.primary_analysis + + for ca in report.confirmation_analyses: + if ca.timeframe == Timeframe.M15.value: + m15_analysis = ca + elif ca.timeframe == Timeframe.H1.value: + h1_analysis = ca + + if not m15_analysis or not h1_analysis: + return False + + m15_rsi = m15_analysis.indicators.get("rsi") + if not m15_rsi or m15_rsi.value < cfg.rsi_lower or m15_rsi.value > cfg.rsi_upper: + return False + + m15_bb = m15_analysis.indicators.get("bollinger") + if m15_bb and (m15_bb.value < cfg.bb_min_position or m15_bb.value > cfg.bb_max_position): + return False + + m15_ema200 = m15_analysis.key_levels.get("ema_trend") + m15_df = dfs.get(Timeframe.M15) + if m15_df is None or m15_ema200 is None: + return False + m15_price = float(m15_df["Close"].iloc[-1]) + if m15_price <= m15_ema200: + return False + + h1_ema200 = h1_analysis.key_levels.get("ema_trend") + h1_df = dfs.get(Timeframe.H1) + if h1_ema200 is None: + return False + h1_price = float(h1_df["Close"].iloc[-1]) if h1_df is not None else report.current_price + if h1_price <= h1_ema200: + return False + + m15_atr = m15_analysis.key_levels.get("atr", 0) + m15_ema21 = m15_analysis.key_levels.get("ema_medium") + if m15_atr > 0 and m15_ema21: + dist = abs(m15_price - m15_ema21) / m15_atr + if dist > cfg.chase_max_atr_from_ema21: + return False + + return True + + def _generate_gbp_trade_setup( + self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] + ) -> Optional[TradeSetup]: + """Generate GBP/USD trade setup with tight TP / balanced SL.""" + cfg = self.config.gbp_strategy + + m15_analysis = None + if report.primary_analysis and report.primary_analysis.timeframe == Timeframe.M15.value: + m15_analysis = report.primary_analysis + else: + for ca in report.confirmation_analyses: + if ca.timeframe == Timeframe.M15.value: + m15_analysis = ca + break + + if not m15_analysis: + return None + + atr = m15_analysis.key_levels.get("atr", 0) + if atr <= 0: + return None + + price = report.current_price + pip_value = 0.0001 + + tp1 = price + cfg.tp_atr_multiplier * atr + sl = price - cfg.sl_atr_multiplier * atr + tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr + + sl_distance = price - sl + tp_distance = tp1 - price + rr = tp_distance / sl_distance if sl_distance > 0 else 0 + + return TradeSetup( + pair=report.symbol, + display_name=report.display_name, + direction="BUY", + entry_price=price, + stop_loss=round(sl, 5), + take_profit_1=round(tp1, 5), + take_profit_2=round(tp2, 5), + risk_reward=round(rr, 2), + risk_pips=round(sl_distance / pip_value, 1), + reward_pips=round(tp_distance / pip_value, 1), + pip_value=pip_value, + ) + def _check_mtf_agreement(self, report: SignalReport) -> bool: signals = list(report.mtf_signals.values()) if len(signals) < 2: diff --git a/telegram_bot.py b/telegram_bot.py index 1b1cbc1..e777b4a 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -35,11 +35,12 @@ import asyncio import json import logging import os +import signal import sys import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Set import websockets @@ -104,7 +105,7 @@ SCAN_BUTTONS = [ {"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"}, {"cb": "scan_eurusd_h1", "symbol": EURUSD, "tf": Timeframe.H1, "label": "EUR/USD 1H"}, {"cb": "scan_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"}, -{"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"}, + {"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"}, {"cb": "scan_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"}, {"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"}, {"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"}, @@ -128,7 +129,60 @@ AUTOSCAN_CONFIRMATIONS = { Timeframe.H4: [Timeframe.H1, Timeframe.D1], } -_executor = ThreadPoolExecutor(max_workers=2) +_executor = ThreadPoolExecutor(max_workers=4) + +# ── Telegram rate limiter (1 msg/sec per chat) ──────────────────────── + +_send_lock = asyncio.Lock() +_last_send_per_chat: Dict[int, float] = {} +_TELEGRAM_MIN_INTERVAL = 1.0 +_TELEGRAM_SEND_TIMEOUT = 30 + + +async def _rate_limited_send(bot, chat_id: int, text: str, **kwargs) -> Optional[object]: + async with _send_lock: + now = time.time() + last = _last_send_per_chat.get(chat_id, 0) + wait = _TELEGRAM_MIN_INTERVAL - (now - last) + if wait > 0: + await asyncio.sleep(wait) + try: + result = await asyncio.wait_for( + bot.send_message(chat_id, text, **kwargs), + timeout=_TELEGRAM_SEND_TIMEOUT, + ) + _last_send_per_chat[chat_id] = time.time() + return result + except asyncio.TimeoutError: + logger.error(f"Telegram send_message timed out for chat {chat_id}") + return None + except Exception: + logger.exception(f"Telegram send_message failed for chat {chat_id}") + return None + + +async def _rate_limited_edit(bot, text: str, chat_id: int, message_id: int, **kwargs) -> bool: + try: + await asyncio.wait_for( + bot.edit_message_text(text, chat_id=chat_id, message_id=message_id, **kwargs), + timeout=_TELEGRAM_SEND_TIMEOUT, + ) + return True + except asyncio.TimeoutError: + logger.error(f"Telegram edit_message timed out for chat {chat_id}") + return False + except Exception as exc: + if "message is not modified" not in str(exc).lower(): + logger.warning(f"Telegram edit_message failed for chat {chat_id}: {exc}") + return False + + +# ── Markdown escaping ───────────────────────────────────────────────── + +def _escape_md(text: str) -> str: + for ch in ("*", "_", "`", "[", "]"): + text = text.replace(ch, "\\" + ch) + return text # ── Shared data provider (single cache across all scans) ───────────── @@ -183,7 +237,7 @@ def _mtf_line(report: SignalReport) -> str: def _indicator_line(name: str, ind) -> str: tag = _SIGNAL_TAG.get(ind.signal, "Neutral") short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description - return f"[{tag}] *{ind.name}*: {short}" + return f"[{tag}] *{_escape_md(ind.name)}*: {_escape_md(short)}" def format_signal_message(report: SignalReport) -> str: @@ -196,16 +250,18 @@ def format_signal_message(report: SignalReport) -> str: lines: list[str] = [] now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + regime_str = _escape_md(report.market_regime.replace("_", " ").title()) + vol_str = _escape_md(str(report.volatility_regime)) lines += [ f"🔍 *{report.display_name} Signal Analysis* ({report.primary_analysis.timeframe if report.primary_analysis else ''})", f"🕐 {now}", "", f"💰 Price: *{report.current_price:.5f}*" if report.current_price < 10 else f"💰 Price: *{report.current_price:.3f}*", - f"📊 Regime: {report.market_regime.replace('_', ' ').title()} | Vol: {report.volatility_regime}", + f"📊 Regime: {regime_str} | Vol: {vol_str}", "", "━━━━━━━━━━━━━━━━━━━━", f"{emoji} *{label}* {qe}{report.signal_quality} | Score: {report.final_score:+.1f} | {report.final_confidence:.0f}% conf", - f"_{report.action_recommendation}_", + f"_{_escape_md(report.action_recommendation)}_", f"{mtf_tag}", ] @@ -223,7 +279,7 @@ def format_signal_message(report: SignalReport) -> str: if report.warnings: lines.append("") for w in report.warnings: - lines.append(f"⚠️ {w}") + lines.append(f"⚠️ {_escape_md(w)}") if report.trade_setup and sig not in (SignalStrength.NEUTRAL,): ts = report.trade_setup @@ -366,7 +422,7 @@ def _format_stats_dashboard() -> str: sig = _SHORT_SIGNAL.get(r.final_signal, "Neutral") tf_parts.append(f"{label}: {emoji}{sig}") - lines.append(f"*{display}* {price_str} _{regime}_") + lines.append(f"*{display}* {price_str} _{_escape_md(regime)}_") lines.append(" | ".join(tf_parts)) if latest and latest.trade_setup and latest.final_signal not in (SignalStrength.NEUTRAL,): @@ -390,40 +446,35 @@ _dashboard_msg_ids: Dict[int, int] = {} _dashboard_task: Optional[asyncio.Task] = None _last_dashboard_refresh: float = 0 _DASHBOARD_THROTTLE = 5 +_dashboard_lock = asyncio.Lock() async def _send_dashboard(bot, chat_id: int) -> None: text = _format_stats_dashboard() - try: - msg = await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN) + msg = await _rate_limited_send(bot, chat_id, text, parse_mode=ParseMode.MARKDOWN) + if msg: _dashboard_msg_ids[chat_id] = msg.message_id - except Exception: - logger.exception(f"Failed to send dashboard to {chat_id}") async def _refresh_dashboard(bot, force: bool = False) -> None: global _last_dashboard_refresh - now = time.time() - if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE: - return - _last_dashboard_refresh = now + async with _dashboard_lock: + now = time.time() + if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE: + return + _last_dashboard_refresh = now - text = _format_stats_dashboard() - for chat_id in list(_autoscan_chats): - msg_id = _dashboard_msg_ids.get(chat_id) - if not msg_id: - await _send_dashboard(bot, chat_id) - continue - try: - await bot.edit_message_text( - text, chat_id=chat_id, message_id=msg_id, + text = _format_stats_dashboard() + for chat_id in list(_autoscan_chats): + msg_id = _dashboard_msg_ids.get(chat_id) + if not msg_id: + await _send_dashboard(bot, chat_id) + continue + ok = await _rate_limited_edit( + bot, text, chat_id=chat_id, message_id=msg_id, parse_mode=ParseMode.MARKDOWN, ) - except Exception as exc: - if "message is not modified" in str(exc).lower(): - pass - else: - logger.warning(f"Dashboard edit failed, sending new: {exc}") + if not ok: await _send_dashboard(bot, chat_id) @@ -484,7 +535,7 @@ async def run_signal( # ── Candle-boundary scheduler ──────────────────────────────────────── -_autoscan_chats: set[int] = set() +_autoscan_chats: Set[int] = set() _autoscan_tasks: list[asyncio.Task] = [] @@ -507,13 +558,14 @@ async def _scan_timeframe( confirmation_tfs: List[Timeframe], ) -> None: """Scan all pairs for one timeframe and broadcast results.""" + tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS] + results = await asyncio.gather(*tasks, return_exceptions=True) reports: List[SignalReport] = [] - for symbol in SCAN_PAIRS: - try: - report = await run_signal(symbol, tf, confirmation_tfs) - reports.append(report) - except Exception: - logger.exception(f"Auto-scan failed for {symbol} {tf.value}") + for i, result in enumerate(results): + if isinstance(result, Exception): + logger.exception(f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}") + else: + reports.append(result) if not reports: return @@ -530,17 +582,17 @@ async def _scan_timeframe( 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: - try: - detail = format_signal_message(report) - await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN) - except Exception: - logger.exception(f"Failed to send to chat {chat_id}") + detail = format_signal_message(report) + await _rate_limited_send(bot, chat_id, detail, parse_mode=ParseMode.MARKDOWN) # ── WebSocket level-breach monitor ────────────────────────────────── _level_cache: Dict[str, Dict[str, float]] = {} +_level_cache_ts: Dict[str, float] = {} +_LEVEL_CACHE_TTL = 3600 _last_prices: Dict[str, float] = {} +_last_price_ts: Dict[str, float] = {} _scan_cooldowns: Dict[str, float] = {} _cfg = PlatformConfig() @@ -548,8 +600,11 @@ _WS_API_KEY = os.environ.get("TWELVE_DATA_API_KEY") or _cfg.twelve_data_api_key _WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}" _WS_COOLDOWN = _cfg.ws_cooldown_seconds +_breach_tasks: Set[asyncio.Task] = set() + def _update_level_cache(reports: List[SignalReport]) -> None: + now = time.time() for r in reports: if r.primary_analysis and r.primary_analysis.key_levels: levels = { @@ -558,6 +613,7 @@ def _update_level_cache(reports: List[SignalReport]) -> None: } if levels: _level_cache[r.symbol] = levels + _level_cache_ts[r.symbol] = now def _check_level_breach(symbol: str, price: float) -> Optional[str]: @@ -565,6 +621,10 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]: if not levels: return None + cache_age = time.time() - _level_cache_ts.get(symbol, 0) + if cache_age > _LEVEL_CACHE_TTL: + return None + last = _last_prices.get(symbol) if last is None: return None @@ -576,6 +636,13 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]: return None +def _cleanup_cooldowns() -> None: + now = time.time() + expired = [k for k, v in _scan_cooldowns.items() if now - v > _WS_COOLDOWN * 2] + for k in expired: + del _scan_cooldowns[k] + + async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None: now = time.time() if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN: @@ -611,7 +678,12 @@ async def _ws_monitor(bot) -> None: if not _autoscan_chats: break - msg = json.loads(raw) + try: + msg = json.loads(raw) + except json.JSONDecodeError: + logger.warning(f"WebSocket: malformed JSON, skipping") + continue + if msg.get("event") != "price": continue @@ -620,14 +692,24 @@ async def _ws_monitor(bot) -> None: if not pair_symbol: continue - price = float(msg["price"]) + try: + price = float(msg["price"]) + except (KeyError, ValueError, TypeError) as e: + logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}") + continue + breached = _check_level_breach(pair_symbol, price) _last_prices[pair_symbol] = price + _last_price_ts[pair_symbol] = time.time() if breached: - asyncio.create_task( + task = asyncio.create_task( _handle_breach(bot, ws_symbol, pair_symbol, price, breached) ) + _breach_tasks.add(task) + task.add_done_callback(_breach_tasks.discard) + + _cleanup_cooldowns() except asyncio.CancelledError: break @@ -701,6 +783,9 @@ def _stop_scheduler() -> None: _dashboard_task.cancel() _dashboard_task = None _dashboard_msg_ids.clear() + for t in list(_breach_tasks): + t.cancel() + _breach_tasks.clear() logger.info("Scheduler, WebSocket, and dashboard loop stopped") @@ -729,7 +814,7 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: "/start — Show scan buttons\n" "/autoscan — Toggle auto-scan on/off\n" "/status — Bot status, WS connection, live prices\n" - "/cache — Show when each pair/TF was last fetched\n" + "/cache — Show data cache age per pair/TF\n" "/clearcache — Wipe data cache, force fresh fetch\n" "/help — This message\n\n" "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n" @@ -759,19 +844,21 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None for ws_sym, pair in _WS_TO_PAIR.items(): price = _last_prices.get(pair) display = PAIR_INFO.get(pair, {}).get("display", ws_sym) + age = time.time() - _last_price_ts.get(pair, 0) if price is None: lines.append(f" {display}: waiting...") elif price < 10: - lines.append(f" {display}: `{price:.5f}`") + lines.append(f" {display}: `{price:.5f}` ({age:.0f}s ago)") else: - lines.append(f" {display}: `{price:.3f}`") + lines.append(f" {display}: `{price:.3f}` ({age:.0f}s ago)") else: 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()) if cached_levels: - lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs") + oldest_age = time.time() - min(_level_cache_ts.values()) if _level_cache_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") @@ -782,6 +869,10 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None if active_cooldowns: lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown") + active_breach = sum(1 for t in _breach_tasks if not t.done()) + if active_breach: + lines.append(f"*Breach tasks:* {active_breach} running") + await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN) @@ -906,13 +997,19 @@ async def _post_init(app: Application) -> None: logger.info("Running initial scan to populate level cache and signal stats...") try: - init_reports = [] - for symbol in SCAN_PAIRS: - report = await run_signal(symbol, Timeframe.M15) - init_reports.append(report) - _update_level_cache(init_reports) - _update_signal_stats(init_reports, Timeframe.M15) - logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels") + tasks = [ + asyncio.wait_for(run_signal(sym, Timeframe.M15), timeout=30) + for sym in SCAN_PAIRS + ] + results = await asyncio.gather(*tasks, return_exceptions=True) + init_reports = [r for r in results if isinstance(r, SignalReport)] + for i, r in enumerate(results): + if isinstance(r, Exception): + logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}") + if init_reports: + _update_level_cache(init_reports) + _update_signal_stats(init_reports, Timeframe.M15) + logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels") except Exception: logger.exception("Initial scan failed — WS will wait for first candle scan") @@ -928,8 +1025,8 @@ async def _post_init(app: Application) -> None: ) try: - await app.bot.send_message( - chat_id, + await _rate_limited_send( + app.bot, chat_id, f"🤖 *ForexBot2 started*\n\n" f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n" f"📊 Live dashboard refreshes every 60s.\n" @@ -944,6 +1041,15 @@ async def _post_init(app: Application) -> None: logger.exception("Failed to send boot message") +# ── Graceful shutdown ───────────────────────────────────────────────── + +async def _post_shutdown(app: Application) -> None: + logger.info("Shutting down ForexBot2...") + _stop_scheduler() + _executor.shutdown(wait=False) + logger.info("Shutdown complete") + + # ── Entry point ─────────────────────────────────────────────────────── def main() -> None: @@ -956,6 +1062,7 @@ def main() -> None: Application.builder() .token(TOKEN) .post_init(_post_init) + .post_shutdown(_post_shutdown) .build() )