#!/usr/bin/env python3 """ Forex Signal Intelligence Platform - Telegram Bot ================================================== Trade alert frontend for the signal engine. All timeframes are analyzed automatically — no TF parameter needed. Setup: 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 — Combined signal card across all timeframes /scan — Scan all pairs, compact Grade B+ summary /help — Command guide """ import asyncio import html import logging import os from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Dict, List, Optional, Tuple from zoneinfo import ZoneInfo # Load .env so TELEGRAM_BOT_TOKEN / GEMINI_API_KEY / TWELVE_DATA_API_KEY # don't need to be exported manually. try: from dotenv import load_dotenv load_dotenv() except ImportError: pass from telegram import BotCommand, Update from telegram.constants import ParseMode from telegram.ext import Application, CommandHandler, ContextTypes from config import ( MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY, PlatformConfig, SignalStrength, Timeframe, ) from data_provider import ForexDataProvider from signal_aggregator import SignalAggregator, SignalReport logger = logging.getLogger(__name__) # ── Config ────────────────────────────────────────────────────────── SCAN_TIMEFRAMES = ["M15", "H1"] # two cards (M15 and H1) per /scan; cache shared # Pairs analyzed by the /scan command. Limited to the most liquid majors # to keep credits low and signal quality high. SCAN_PAIRS = ["EURUSD=X", "USDJPY=X", "GBPUSD=X", "AUDUSD=X"] _TF_MAP = { "M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30, "H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1, } # D1 kept as confirmation for higher-timeframe trend context. # W1 dropped — only the D1 scan used it, and we no longer scan D1. _TF_CONFIRMATIONS = { "M15": [Timeframe.H1, Timeframe.D1], "H1": [Timeframe.M15, Timeframe.D1], } _GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5} _GRADE_LEGEND = "Grade scale: 🔥 A+ · ✅ A · 👍 B · ⚠️ C · 📉 D · ❌ F" _EST = ZoneInfo("America/New_York") # ── Formatting helpers ────────────────────────────────────────────── def _now_est() -> str: return datetime.now(_EST).strftime("%I:%M %p EST · %a %d %b %Y") def _signal_action(signal: SignalStrength) -> str: return { SignalStrength.STRONG_BUY: "STRONG BUY", SignalStrength.BUY: "BUY", SignalStrength.STRONG_SELL: "STRONG SELL", SignalStrength.SELL: "SELL", SignalStrength.NEUTRAL: "NEUTRAL", }.get(signal, "NEUTRAL") def _grade_symbol(quality: str) -> str: return {"A+": "🔥", "A": "✅", "B": "👍", "C": "⚠️", "D": "📉"}.get(quality, "❌") def _indicator_condition(ind) -> str: desc = ind.description if " — " in desc: return desc.split(" — ", 1)[1].strip() if " | " in desc: return desc.split(" | ", 1)[0].strip() return ind.signal.name.replace("_", " ").title() def _fmt_float(val: float) -> str: if val == 0.0: return "0" av = abs(val) if av >= 1000: return f"{val:.1f}" if av >= 10: return f"{val:.2f}" if av >= 0.01: return f"{val:.4f}" return f"{val:.6f}" def _h(text: str) -> str: return html.escape(str(text)) # ── Message builders ──────────────────────────────────────────────── def _build_combined_signal( results_by_tf: Dict[str, List[SignalReport]], display: str, now_str: str, ) -> str: """ Single message showing all timeframes stacked. Entry/TP/SL always preserved; trimmed: per-card disclaimer, MTF detail block (folded to one line), news unless non-neutral. """ lines: List[str] = [] lines.append(f"🌐 🧠 {_h(display)} — Signal Analysis") lines.append(f"🕐 {now_str}") lines.append(f"{_GRADE_LEGEND}") for tf in SCAN_TIMEFRAMES: reports = results_by_tf.get(tf, []) if not reports: continue r = reports[0] # If primary TF data was unavailable the engine returns grade F with a warning. # Show a clean "data unavailable" line instead of a garbled card. if r.signal_quality == "F" and any( "unavailable" in w.lower() or "api" in w.lower() or "credits" in w.lower() for w in r.warnings ): lines.append("") lines.append("────────────────────────") lines.append(f"{tf} — ❌ Data unavailable") if r.warnings: lines.append(f"{_h(r.warnings[0])}") continue action = _signal_action(r.final_signal) sym = _grade_symbol(r.signal_quality) lines.append("") lines.append("────────────────────────") lines.append(f"{sym} {tf} · {action} · Grade {_h(r.signal_quality)}") lines.append("") # ── Trade setup (always kept intact) ───────────────────────── if r.trade_setup: ts = r.trade_setup lines.append(f"Entry: {ts.entry_price:.5f}") lines.append( f"Take Profit: {ts.take_profit_1:.5f}" f" (+{ts.reward_pips:.1f} pips)" ) if abs(ts.take_profit_2 - ts.take_profit_1) > 1e-8: lines.append(f"Take Profit 2: {ts.take_profit_2:.5f}") lines.append( f"Stop Loss: {ts.stop_loss:.5f}" f" (-{ts.risk_pips:.1f} pips)" ) lines.append(f"Risk:Reward: 1:{ts.risk_reward:.1f}") lines.append("") else: lines.append(f"Current Price: {r.current_price:.5f}") lines.append(f"Analysis: {_h(r.action_recommendation)}") lines.append("") # ── Indicators ──────────────────────────────────────────────── if r.primary_analysis and r.primary_analysis.indicators: for name, ind in r.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue lines.append( f"• {_h(ind.name)}: {_fmt_float(ind.value)}" f" ({_h(_indicator_condition(ind))})" ) # ── MTF summary (one line) ──────────────────────────────────── agree_str = "✅ Aligned" if r.mtf_agreement else "⚠️ Divergent" lines.append( f"MTF: {agree_str} | Score: {r.final_score:+.1f}" f" | {_h(r.market_regime)} | Vol: {_h(r.volatility_regime)}" ) # ── Critical warnings only ──────────────────────────────────── for w in r.warnings: if any(k in w for k in ("EVENT RISK", "CONFLICT", "EXTREME")): lines.append(f"⚠️ {_h(w)}") # ── Event / news only when relevant ────────────────────────── if r.event_risk == "danger" and r.event_description: lines.append(f"📅 Event Risk: {_h(r.event_description)}") if r.news_description and r.news_description not in ("", "Neutral news sentiment"): lines.append(f"📰 {_h(r.news_description)}") lines.append("") lines.append("⚠️ Educational use only. Not financial advice.") return "\n".join(lines) def _build_scan_message( actionable: List[Tuple[str, SignalReport]], now_str: str, tfs_str: str, ) -> str: """ Compact single message: one line header + one trade line per pair. Full entry/TP/SL preserved; no indicator list (use /signal to drill in). """ lines: List[str] = [] lines.append(f"📊 🧠 Scan Complete · {now_str}") lines.append(f"Timeframes: {_h(tfs_str)} | Best signal per pair") lines.append(f"{_GRADE_LEGEND}") lines.append("") for tf, r in actionable: action = _signal_action(r.final_signal) sym = _grade_symbol(r.signal_quality) lines.append( f"{sym} {_h(r.display_name)} · {action}" f" · {_h(r.signal_quality)} · {tf}" ) if r.trade_setup: ts = r.trade_setup lines.append( f" Entry: {ts.entry_price:.5f} | " f"TP: {ts.take_profit_1:.5f} (+{ts.reward_pips:.0f}p) | " f"SL: {ts.stop_loss:.5f} (-{ts.risk_pips:.0f}p) | " f"R:R 1:{ts.risk_reward:.1f}" ) else: lines.append(" Watch only") # Show danger-level events inline if r.event_risk == "danger" and r.event_description: lines.append(f" 📅 {_h(r.event_description)}") lines.append("") if not actionable: lines.append("No Grade B+ signals found across any timeframe.") lines.append("") lines.append("⚠️ Educational use only. Not financial advice.") return "\n".join(lines) # ══════════════════════════════════════════════════════════════════════ # Bot # ══════════════════════════════════════════════════════════════════════ class ForexTelegramBot: def __init__(self): self._executor = ThreadPoolExecutor( max_workers=len(SCAN_TIMEFRAMES), thread_name_prefix="forex" ) # ── Engine helpers ──────────────────────────────────────────────── 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.D1]) cfg.watch_pairs = list(pairs) return cfg def _run_scan(self, tf: str, pairs: List[str], provider: ForexDataProvider) -> List[SignalReport]: return SignalAggregator(self._build_config(tf, pairs), data_provider=provider).analyze_all_pairs() async def _analyze_all_tfs( self, pairs: List[str], progress=None, ) -> Dict[str, List[SignalReport]]: """Single combined prefetch, then run all SCAN_TIMEFRAMES concurrently off the warm cache.""" # Collect all unique timeframes needed across every TF scan all_tfs = list({ tf for tf_name in SCAN_TIMEFRAMES for tf in [_TF_MAP[tf_name]] + _TF_CONFIRMATIONS.get(tf_name, []) }) provider = ForexDataProvider(PlatformConfig()) loop = asyncio.get_running_loop() if progress: await progress(f"📡 Fetching market data ({len(all_tfs)} timeframes × {len(pairs)} pairs)…") # Run blocking prefetch off the event loop so we don't freeze the bot await loop.run_in_executor(None, provider.prefetch_batch, list(pairs), all_tfs) if progress: await progress("🧠 Analyzing pairs across timeframes…") tasks = { tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs), provider) 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, splitting into chunks at newlines if over 4096 chars.""" LIMIT = 4000 chunks: List[str] = [] while text: if len(text) <= LIMIT: chunks.append(text) break split_at = text.rfind("\n", 0, LIMIT) if split_at == -1: split_at = LIMIT chunks.append(text[:split_at]) text = text[split_at:].lstrip("\n") for chunk in chunks: await update.message.reply_text(chunk, parse_mode=ParseMode.HTML) # ── Commands ────────────────────────────────────────────────────── async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE): await self.cmd_help(update, context) async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE): tfs = " · ".join(SCAN_TIMEFRAMES) text = ( "🌐 Forex Signal Intelligence Platform\n\n" "Commands:\n" " /signal EURUSD — Full signal analysis across all timeframes\n" " /scan — Scan all pairs, compact Grade B+ summary\n" " /help — This message\n\n" f"Timeframes analyzed: {tfs}\n\n" "Signal Grades:\n" " 🔥 A+ — Strongest signal, MTF aligned, high confidence\n" " ✅ A — Strong signal, MTF aligned\n" " 👍 B — Decent setup, wait for confirmation\n" " ⚠️ C — Weak signal, watch only\n" " 📉 D — Very weak, no action\n" " ❌ F — No clear signal\n\n" "⚠️ Educational use only. Not financial advice." ) await update.message.reply_text(text, parse_mode=ParseMode.HTML) async def cmd_signal(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """ /signal EURUSD Runs all timeframes concurrently, returns a single combined message with one stacked section per TF. Entry/TP/SL always preserved. """ args = context.args or [] if not args: await update.message.reply_text( "Usage: /signal <pair>\n" "Example: /signal EURUSD", parse_mode=ParseMode.HTML, ) return raw = args[0].upper().replace("/", "").replace("_", "") if not raw.endswith("=X"): raw += "=X" display = SYMBOL_DISPLAY.get(raw, raw.replace("=X", "")) tfs_str = " · ".join(SCAN_TIMEFRAMES) msg = await update.message.reply_text( f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…" ) async def progress(text: str): try: await msg.edit_text(text) except Exception: pass # ignore edit-rate errors / no-change errors try: results_by_tf = await self._analyze_all_tfs([raw], progress=progress) now_str = _now_est() # Check we got at least one result if not any(results_by_tf.values()): await msg.edit_text(f"❌ No data returned for {_h(display)}") return text = _build_combined_signal(results_by_tf, display, now_str) await msg.delete() await self._send_html(update, text) except Exception as e: logger.error(f"Signal error for {raw}: {e}", exc_info=True) await msg.edit_text(f"❌ Analysis failed: {_h(str(e))}") async def cmd_scan(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """ /scan Analyzes all major pairs across all timeframes concurrently. Picks the best-grade result per pair, returns a single compact message. """ tfs_str = " · ".join(SCAN_TIMEFRAMES) msg = await update.message.reply_text( f"🔄 Scanning all pairs across {_h(tfs_str)}…" ) async def progress(text: str): try: await msg.edit_text(text) except Exception: pass try: results_by_tf = await self._analyze_all_tfs(list(SCAN_PAIRS), progress=progress) now_str = _now_est() # Best grade per pair across all TFs 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 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)) text = _build_scan_message(actionable, now_str, tfs_str) await msg.edit_text(text, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"Scan error: {e}", exc_info=True) await msg.edit_text(f"❌ Scan failed: {_h(str(e))}") # ── Application builder ─────────────────────────────────────────── def build_application(self) -> Application: token = os.environ.get("TELEGRAM_BOT_TOKEN") if not token: raise ValueError( "TELEGRAM_BOT_TOKEN environment variable is not set.\n" "Get a token from @BotFather on Telegram, then:\n" " 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", "Signal analysis for one pair — /signal EURUSD"), BotCommand("scan", "Scan all pairs, best Grade B+ per pair"), BotCommand("help", "Commands and grade guide"), ]) app = ( Application.builder() .token(token) .post_init(_register_commands) .build() ) app.add_handler(CommandHandler("start", self.cmd_start)) app.add_handler(CommandHandler("help", self.cmd_help)) app.add_handler(CommandHandler("signal", self.cmd_signal)) app.add_handler(CommandHandler("scan", self.cmd_scan)) return app # ══════════════════════════════════════════════════════════════════════ def main(): logging.basicConfig( level=logging.WARNING, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%H:%M:%S", ) bot = ForexTelegramBot() app = bot.build_application() print("🤖 Forex Signal Bot started. Press Ctrl+C to stop.") app.run_polling(drop_pending_updates=True) if __name__ == "__main__": main()