#!/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"] # Auto-scan job: fires every AUTO_SCAN_INTERVAL seconds and pushes any A/A+ # signal to ALERT_CHAT_ID (env var). 10 min cadence ≈ 595 credits/day on # cold-cache average — fits inside the 800/day Twelve Data free-tier budget. AUTO_SCAN_INTERVAL = 600 ALERT_GRADES = {"A+", "A"} _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) def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str: """ Auto-scan alert message — focused on actionable trade info plus a prominent news/event block (the user wants news context surfaced on every A-grade alert). """ lines: List[str] = [] sym = _grade_symbol(r.signal_quality) action = _signal_action(r.final_signal) lines.append( f"{sym} {_h(r.signal_quality)} ALERT · {_h(r.display_name)}" f" · {action} · {tf}" ) lines.append(f"🕐 {now_str}") lines.append("") # ── Trade setup ────────────────────────────────────────────── 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("") # ── Top confluence reasons (top 3, direction-matched) ──────── if r.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): reasons = r.bullish_reasons[:3] elif r.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): reasons = r.bearish_reasons[:3] else: reasons = [] if reasons: lines.append("📊 Confluence:") for reason in reasons: lines.append(f" • {_h(reason)}") lines.append("") # ── News & Events (always shown on alerts) ─────────────────── lines.append("📰 News & Events:") sentiment = r.news_sentiment if sentiment > 0.2: sent_label = f"📈 Bullish news bias (score {sentiment:+.2f})" elif sentiment < -0.2: sent_label = f"📉 Bearish news bias (score {sentiment:+.2f})" else: sent_label = f"📊 Neutral news (score {sentiment:+.2f})" lines.append(f" {sent_label}") if r.news_description and r.news_description.strip().lower() not in ( "", "neutral news sentiment", "no recent news", ): lines.append(f" {_h(r.news_description)}") risk_label = { "clear": "✅ Clear (no major events nearby)", "caution": "⚠️ Caution — upcoming event", "danger": "🚨 Danger — high-impact event imminent", }.get(r.event_risk, _h(r.event_risk)) lines.append(f" Event Risk: {risk_label}") if r.event_description: lines.append(f" {_h(r.event_description)}") # Sentiment vs technical alignment hint — useful one-liner for the trader if abs(sentiment) > 0.2: score_dir = 1 if r.final_score > 0 else -1 if r.final_score < 0 else 0 sent_dir = 1 if sentiment > 0 else -1 if score_dir != 0 and score_dir == sent_dir: lines.append(" ✅ News aligns with technical signal") elif score_dir != 0: lines.append(" ⚠️ News conflicts with technical signal") lines.append("") # ── MTF / regime / vol summary ──────────────────────────────── agree_str = "✅ Aligned" if r.mtf_agreement else "⚠️ Divergent" lines.append( f"MTF: {agree_str} | Score: {r.final_score:+.1f}" f" | Confidence: {r.final_confidence:.0f}%" f" | {_h(r.market_regime)} | Vol: {_h(r.volatility_regime)}" ) # ── Critical warnings ──────────────────────────────────────── for w in r.warnings: if any(k in w for k in ("EVENT RISK", "CONFLICT", "EXTREME")): lines.append(f"⚠️ {_h(w)}") 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" ) # Persistent ForexDataProvider so the OHLCV cache survives across # /scan and /signal calls. Without this each call burns API credits # re-fetching the same bars and waits up to 60s on the rate limit. self._provider = ForexDataProvider(PlatformConfig()) # Dedup keys for the auto-scan alert job: (symbol, tf, direction, # candle_open_timestamp). New bar → new key → re-alert allowed if the # setup persists. Bounded to avoid unbounded growth on long uptime. self._alerted_keys: set = set() # ── 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 = self._provider # reuse across calls — see __init__ 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_myid(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """ Returns the current chat's ID so the operator can set ALERT_CHAT_ID in .env and receive auto-scan alerts here. """ chat = update.effective_chat text = ( f"Chat ID: {chat.id}\n" f"Type: {chat.type}\n\n" "Set ALERT_CHAT_ID=<id> in .env " "and restart the bot to receive A-grade alerts here." ) await update.message.reply_text(text, parse_mode=ParseMode.HTML) # ── Auto-scan job (every AUTO_SCAN_INTERVAL seconds) ───────────── async def _auto_scan_job(self, context: ContextTypes.DEFAULT_TYPE): """ Periodic scan that pushes A/A+ alerts to the configured chat. Dedup: one alert per (symbol, tf, direction, candle_open) — so a persistent setup re-alerts once per new bar, not every 10 min. """ chat_id = context.job.data["chat_id"] try: results_by_tf = await self._analyze_all_tfs(list(SCAN_PAIRS)) except Exception as e: logger.error(f"Auto-scan failed: {e}", exc_info=True) return now_str = _now_est() sent = 0 for tf, reports in results_by_tf.items(): for r in reports: if r.signal_quality not in ALERT_GRADES: continue # Skip if engine blocked it via event danger if r.event_risk == "danger": continue ts = r.primary_analysis.timestamp if r.primary_analysis else "" key = (r.symbol, tf, r.final_signal.name, ts) if key in self._alerted_keys: continue self._alerted_keys.add(key) text = _build_alert_card(tf, r, now_str) try: await context.bot.send_message( chat_id=chat_id, text=text, parse_mode=ParseMode.HTML, ) sent += 1 logger.info( f"Alert sent: {r.symbol} {tf} {r.final_signal.name} " f"({r.signal_quality})" ) except Exception as e: logger.error(f"Failed to send alert for {r.symbol} {tf}: {e}") # Bound dedup memory: keep most recent 256 keys if len(self._alerted_keys) > 256: self._alerted_keys = set(list(self._alerted_keys)[-256:]) if sent == 0: logger.debug("Auto-scan complete: no new A-grade signals") 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("myid", "Show this chat's ID (for ALERT_CHAT_ID setup)"), 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)) app.add_handler(CommandHandler("myid", self.cmd_myid)) # ── Auto-scan job: only enabled when ALERT_CHAT_ID is set ───── alert_chat_raw = os.environ.get("ALERT_CHAT_ID") if alert_chat_raw: if app.job_queue is None: logger.warning( "ALERT_CHAT_ID set but JobQueue is unavailable. " "Install with: pip install 'python-telegram-bot[job-queue]'" ) else: try: chat_id = int(alert_chat_raw) except ValueError: chat_id = alert_chat_raw # channel @username form app.job_queue.run_repeating( self._auto_scan_job, interval=AUTO_SCAN_INTERVAL, first=15, # first run 15s after startup name="auto_scan", data={"chat_id": chat_id}, ) logger.info( f"Auto-scan enabled: every {AUTO_SCAN_INTERVAL}s → " f"chat {chat_id} (grades: {sorted(ALERT_GRADES)})" ) else: logger.info( "Auto-scan disabled — set ALERT_CHAT_ID in .env to enable. " "Use /myid in your target chat to find the ID." ) 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()