diff --git a/telegram_bot.py b/telegram_bot.py index 9b32d25..1a784d8 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -3,16 +3,17 @@ Forex Signal Intelligence Platform - Telegram Bot ================================================== Trade alert frontend for the signal engine. +Analyzes all timeframes automatically — no TF parameter needed. Setup: - set TELEGRAM_BOT_TOKEN=your_token_here + set TELEGRAM_BOT_TOKEN=your_token_here (Windows CMD) + export TELEGRAM_BOT_TOKEN=your_token_here (Linux/Mac) python telegram_bot.py Commands: - /signal EURUSD [TF] — Signal card for one pair - /scan [TF] — Scan all pairs, show Grade B+ alerts - /tf H1 — Set default timeframe - /help — Command guide + /signal EURUSD — Signal cards for one pair across all timeframes + /scan — Scan all pairs, show Grade B+ alerts + /help — Command guide """ import asyncio @@ -21,44 +22,44 @@ import logging import os from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Tuple from telegram import BotCommand, Update from telegram.constants import ParseMode from telegram.ext import Application, CommandHandler, ContextTypes from config import ( - MAJOR_PAIRS, SYMBOL_DISPLAY, + MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY, PlatformConfig, SignalStrength, Timeframe, ) from signal_aggregator import SignalAggregator, SignalReport logger = logging.getLogger(__name__) -# ── Timeframe maps ────────────────────────────────────────────────── +# ── Timeframes run on every analysis ─────────────────────────────── +# Each TF is run as the primary; its confirmation TFs are auto-selected. +SCAN_TIMEFRAMES = ["M15", "H1", "H4", "D1"] -TF_MAP = { - "M5": Timeframe.M5, - "M15": Timeframe.M15, - "M30": Timeframe.M30, - "H1": Timeframe.H1, - "H4": Timeframe.H4, - "D1": Timeframe.D1, +_TF_MAP = { + "M5": Timeframe.M5, "M15": Timeframe.M15, + "M30": Timeframe.M30, "H1": Timeframe.H1, + "H4": Timeframe.H4, "D1": Timeframe.D1, } -TF_CONFIRMATIONS = { - "M5": [Timeframe.M1, Timeframe.M15], +_TF_CONFIRMATIONS = { "M15": [Timeframe.M5, Timeframe.H1], - "M30": [Timeframe.M15, Timeframe.H1], "H1": [Timeframe.M15, Timeframe.H4], "H4": [Timeframe.H1, Timeframe.D1], "D1": [Timeframe.H4, Timeframe.W1], } +_SUPPORTED_PAIRS = set(MAJOR_PAIRS + CROSS_PAIRS) + +_GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5} + # ── Formatting helpers ────────────────────────────────────────────── def _grade_dot(quality: str) -> str: - """Colored dot matching confidence grade per spec.""" if quality in ("A+", "A"): return "🟢" if quality == "B": @@ -77,22 +78,15 @@ def _signal_action(signal: SignalStrength) -> str: def _indicator_condition(ind) -> str: - """ - Extract a short readable condition from the indicator description. - Prefer text after ' — ' (e.g. 'RSI 68.2 — Overbought' → 'Overbought'). - Fall back to the signal enum name. - """ desc = ind.description if " — " in desc: return desc.split(" — ", 1)[1].strip() if " | " in desc: - # e.g. 'MACD bullish crossover | Hist: 0.00008' — take first clause return desc.split(" | ", 1)[0].strip() return ind.signal.name.replace("_", " ").title() def _fmt_float(val: float) -> str: - """Human-readable float with appropriate precision.""" if val == 0.0: return "0" av = abs(val) @@ -106,33 +100,11 @@ def _fmt_float(val: float) -> str: def _h(text: str) -> str: - """HTML-escape a string for safe use inside HTML parse mode.""" return html.escape(str(text)) def _build_alert(report: SignalReport, timeframe: str) -> str: - """ - Build a single trade alert card using the required visual template. - Returns an HTML-formatted string ready for Telegram. - - Template: - {Dot} **TRADE ALERT: {ACTION} {PAIR}** - Grade / confidence / timeframe subtitle - - **Entry:** {price} - **Take Profit:** {price} (+pips) - **Stop Loss:** {price} (-pips) - **Risk:Reward:** 1:2.0 - - **Technical Indicators:** - • {Name}: {Value} ({Condition}) - ... - - Multi-Timeframe block - Warnings / events / news - Market context line - Disclaimer - """ + """Build one trade alert card (HTML) for a given report + timeframe.""" dot = _grade_dot(report.signal_quality) action = _signal_action(report.final_signal) pair = _h(report.display_name) @@ -146,7 +118,7 @@ def _build_alert(report: SignalReport, timeframe: str) -> str: lines.append(f"Grade: {grade} | Confidence: {conf:.0f}% | {timeframe}") lines.append("") - # ── Trade setup (only when grade B or above) ───────────────────── + # ── Trade setup (Grade B and above only) ───────────────────────── if report.trade_setup: ts = report.trade_setup lines.append(f"Entry: {ts.entry_price:.5f}") @@ -162,7 +134,6 @@ def _build_alert(report: SignalReport, timeframe: str) -> str: ) lines.append(f"Risk:Reward: 1:{ts.risk_reward:.1f}") else: - # No actionable setup — show price and recommendation lines.append(f"Current Price: {report.current_price:.5f}") lines.append(f"Analysis: {_h(report.action_recommendation)}") @@ -183,7 +154,7 @@ def _build_alert(report: SignalReport, timeframe: str) -> str: if report.mtf_signals: lines.append("") agree_str = "✅ Aligned" if report.mtf_agreement else "⚠️ Divergent" - lines.append(f"Multi-Timeframe ({agree_str}):") + lines.append(f"Confirmation TFs ({agree_str}):") for tf, sig in report.mtf_signals.items(): score = report.mtf_scores.get(tf, 0) sig_name = _h(sig.name.replace("_", " ")) @@ -214,53 +185,50 @@ def _build_alert(report: SignalReport, timeframe: str) -> str: return "\n".join(lines) -# ── Per-chat state ────────────────────────────────────────────────── - -class ChatState: - def __init__(self): - self.timeframe: str = "H1" - - # ══════════════════════════════════════════════════════════════════════ # Bot # ══════════════════════════════════════════════════════════════════════ class ForexTelegramBot: def __init__(self): - self._states: Dict[int, ChatState] = {} - self._executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="forex") + # One worker per timeframe so all TFs run concurrently + self._executor = ThreadPoolExecutor( + max_workers=len(SCAN_TIMEFRAMES), thread_name_prefix="forex" + ) - # ── Helpers ─────────────────────────────────────────────────────── + # ── Engine helpers ──────────────────────────────────────────────── - def _state(self, chat_id: int) -> ChatState: - if chat_id not in self._states: - self._states[chat_id] = ChatState() - return self._states[chat_id] - - def _build_config(self, tf: str, pairs: Optional[List[str]] = None) -> PlatformConfig: + def _build_config(self, tf: str, pairs: List[str]) -> PlatformConfig: cfg = PlatformConfig() - cfg.primary_timeframe = TF_MAP[tf] - cfg.confirmation_timeframes = TF_CONFIRMATIONS.get(tf, [Timeframe.M15, Timeframe.H4]) - cfg.watch_pairs = list(pairs or MAJOR_PAIRS) + cfg.primary_timeframe = _TF_MAP[tf] + cfg.confirmation_timeframes = _TF_CONFIRMATIONS.get(tf, [Timeframe.M15, Timeframe.H4]) + cfg.watch_pairs = list(pairs) return cfg def _run_scan(self, tf: str, pairs: List[str]) -> List[SignalReport]: return SignalAggregator(self._build_config(tf, pairs)).analyze_all_pairs() - async def _analyze(self, func, *args): + async def _analyze_all_tfs( + self, pairs: List[str] + ) -> Dict[str, List[SignalReport]]: + """Run all SCAN_TIMEFRAMES concurrently, return {tf: [reports]}.""" loop = asyncio.get_running_loop() - return await loop.run_in_executor(self._executor, func, *args) - - @staticmethod - def _parse_tf(args: List[str]) -> Optional[str]: - for a in args: - if a.upper() in TF_MAP: - return a.upper() - return None + tasks = { + tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs)) + for tf in SCAN_TIMEFRAMES + } + results: Dict[str, List[SignalReport]] = {} + for tf, task in tasks.items(): + try: + results[tf] = await task + except Exception as e: + logger.error(f"TF {tf} scan failed: {e}", exc_info=True) + results[tf] = [] + return results @staticmethod async def _send_html(update: Update, text: str): - """Send HTML message, splitting at newlines if over Telegram's 4096-char limit.""" + """Send HTML, splitting into chunks if over 4096 chars.""" LIMIT = 4000 chunks: List[str] = [] while text: @@ -281,60 +249,33 @@ class ForexTelegramBot: await self.cmd_help(update, context) async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE): - state = self._state(update.effective_chat.id) + tfs = " · ".join(SCAN_TIMEFRAMES) text = ( "🌐 Forex Signal Intelligence Platform\n\n" "Commands:\n" - " /signal EURUSD — Signal card for a pair\n" - " /signal EURUSD H4 — With specific timeframe\n" - " /scan — Scan all pairs (Grade B+ alerts only)\n" - " /scan H4 — Scan at a specific timeframe\n" - " /tf H1 — Set your default timeframe\n" - " /help — This message\n\n" - "Timeframes:\n" - " M5 — 5 minutes\n" - " M15 — 15 minutes\n" - " M30 — 30 minutes\n" - " H1 — 1 hour (default, intraday)\n" - " H4 — 4 hours (swing trading)\n" - " D1 — Daily (position trading)\n\n" + " /signal EURUSD — Signal cards for all timeframes\n" + " /scan — Scan all pairs, Grade B+ alerts\n" + " /help — This message\n\n" + f"Timeframes analyzed automatically: {tfs}\n\n" "Signal Grades:\n" " 🟢 A / A+ — High confidence, consider entry\n" " 🟡 B — Decent setup, wait for confirmation\n" " 🔴 C / D — Low confidence, watch only\n\n" - f"Your default timeframe: {state.timeframe}\n\n" "⚠️ Educational use only. Not financial advice." ) await update.message.reply_text(text, parse_mode=ParseMode.HTML) - async def cmd_tf(self, update: Update, context: ContextTypes.DEFAULT_TYPE): - tf = self._parse_tf(context.args or []) - if not tf: - opts = " ".join(TF_MAP.keys()) - await update.message.reply_text( - f"Usage: /tf <timeframe>\nOptions: {opts}", - parse_mode=ParseMode.HTML, - ) - return - self._state(update.effective_chat.id).timeframe = tf - await update.message.reply_text( - f"✅ Default timeframe set to {tf}", - parse_mode=ParseMode.HTML, - ) - async def cmd_signal(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """ - /signal EURUSD [TF] - Analyze a single pair and return the full trade alert card. + /signal EURUSD + Runs the pair through all timeframes concurrently and sends + one trade alert card per timeframe. """ - chat_id = update.effective_chat.id - state = self._state(chat_id) - args = context.args or [] - + args = context.args or [] if not args: await update.message.reply_text( - "Usage: /signal <pair> [timeframe]\n" - "Example: /signal EURUSD H4", + "Usage: /signal <pair>\n" + "Example: /signal EURUSD", parse_mode=ParseMode.HTML, ) return @@ -342,20 +283,38 @@ class ForexTelegramBot: raw = args[0].upper().replace("/", "").replace("_", "") if not raw.endswith("=X"): raw += "=X" - tf = self._parse_tf(args[1:]) or state.timeframe display = SYMBOL_DISPLAY.get(raw, raw.replace("=X", "")) + tfs_str = " · ".join(SCAN_TIMEFRAMES) - msg = await update.message.reply_text(f"🔄 Analyzing {_h(display)} ({tf})…") + synthetic_warning = "" + if raw not in _SUPPORTED_PAIRS: + synthetic_warning = ( + "\n\n⚠️ Synthetic data warning: " + f"{_h(display)} is not on the Twelve Data free tier. " + "Prices and indicators are simulated — not live market data." + ) + + msg = await update.message.reply_text( + f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…" + ) try: - reports = await self._analyze(self._run_scan, tf, [raw]) - if not reports: - await msg.edit_text(f"❌ No data returned for {_h(display)}") - return + results_by_tf = await self._analyze_all_tfs([raw]) - r = reports[0] - alert = _build_alert(r, tf) await msg.delete() - await self._send_html(update, alert) + + sent = 0 + for tf in SCAN_TIMEFRAMES: + reports = results_by_tf.get(tf, []) + if not reports: + continue + alert = _build_alert(reports[0], tf) + synthetic_warning + await self._send_html(update, alert) + sent += 1 + + if sent == 0: + await update.message.reply_text( + f"❌ No data returned for {_h(display)}" + ) except Exception as e: logger.error(f"Signal error for {raw}: {e}", exc_info=True) @@ -363,51 +322,64 @@ class ForexTelegramBot: async def cmd_scan(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """ - /scan [TF] - Analyze all major pairs and send a trade alert card for every - signal rated Grade B or above. + /scan + Analyzes all major pairs across all timeframes concurrently. + For each pair, picks the highest-grade result across all TFs + and sends a card for Grade B and above. """ - chat_id = update.effective_chat.id - state = self._state(chat_id) - tf = self._parse_tf(context.args or []) or state.timeframe - now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") - msg = await update.message.reply_text(f"🔄 Scanning all pairs ({tf})…") + tfs_str = " · ".join(SCAN_TIMEFRAMES) + msg = await update.message.reply_text( + f"🔄 Scanning all pairs across {_h(tfs_str)}…" + ) try: - reports = await self._analyze(self._run_scan, tf, list(MAJOR_PAIRS)) + results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS)) - actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] + # For each pair pick the best grade across all TFs + # best_per_pair: {symbol: (tf, report)} + best_per_pair: Dict[str, Tuple[str, SignalReport]] = {} + for tf, reports in results_by_tf.items(): + for r in reports: + current = best_per_pair.get(r.symbol) + if current is None or ( + _GRADE_RANK.get(r.signal_quality, 99) + < _GRADE_RANK.get(current[1].signal_quality, 99) + ): + best_per_pair[r.symbol] = (tf, r) + + # Filter to B+ and sort best-first + actionable = [ + (tf, r) + for tf, r in best_per_pair.values() + if r.signal_quality in ("A+", "A", "B") + ] + actionable.sort(key=lambda x: _GRADE_RANK.get(x[1].signal_quality, 99)) if not actionable: await msg.edit_text( f"🔍 Scan complete — {now_str}\n\n" - "No Grade B+ signals found at this time.\n" - "Try a different timeframe with /scan H4.", + "No Grade B+ signals found across any timeframe right now.", parse_mode=ParseMode.HTML, ) return - # Summary header - grades = {"A+": 0, "A": 0, "B": 0} - for r in actionable: - grades[r.signal_quality] = grades.get(r.signal_quality, 0) + 1 + # Tally grades + grade_counts: Dict[str, int] = {} + for _, r in actionable: + grade_counts[r.signal_quality] = grade_counts.get(r.signal_quality, 0) + 1 + grade_str = " ".join(f"{g}: {n}" for g, n in grade_counts.items()) - grade_str = " ".join( - f"{g}: {count}" for g, count in grades.items() if count - ) summary = ( f"📊 Scan complete — {now_str}\n" - f"Timeframe: {tf}\n" + f"Timeframes: {_h(tfs_str)}\n" f"Grade B+ signals: {len(actionable)} ({grade_str})\n\n" - "Trade alerts follow ↓" + "Best signal per pair ↓" ) await msg.edit_text(summary, parse_mode=ParseMode.HTML) - # One card per actionable signal - for r in actionable: - alert = _build_alert(r, tf) - await self._send_html(update, alert) + for tf, r in actionable: + await self._send_html(update, _build_alert(r, tf)) except Exception as e: logger.error(f"Scan error: {e}", exc_info=True) @@ -421,15 +393,14 @@ class ForexTelegramBot: raise ValueError( "TELEGRAM_BOT_TOKEN environment variable is not set.\n" "Get a token from @BotFather on Telegram, then:\n" - " Windows: set TELEGRAM_BOT_TOKEN=your_token\n" - " Linux/Mac: export TELEGRAM_BOT_TOKEN=your_token" + " Windows CMD: set TELEGRAM_BOT_TOKEN=your_token\n" + " Linux/Mac: export TELEGRAM_BOT_TOKEN=your_token" ) async def _register_commands(app: Application) -> None: await app.bot.set_my_commands([ - BotCommand("signal", "Trade alert for one pair — /signal EURUSD"), - BotCommand("scan", "Scan all pairs, Grade B+ alerts — /scan H4"), - BotCommand("tf", "Set default timeframe — /tf H4"), + BotCommand("signal", "Signal cards for one pair — /signal EURUSD"), + BotCommand("scan", "Scan all pairs, best Grade B+ per pair"), BotCommand("help", "Commands and grade guide"), ]) @@ -442,7 +413,6 @@ class ForexTelegramBot: app.add_handler(CommandHandler("start", self.cmd_start)) app.add_handler(CommandHandler("help", self.cmd_help)) - app.add_handler(CommandHandler("tf", self.cmd_tf)) app.add_handler(CommandHandler("signal", self.cmd_signal)) app.add_handler(CommandHandler("scan", self.cmd_scan))