#!/usr/bin/env python3 """ Forex Signal Intelligence Platform - Telegram Bot ================================================== Trade alert frontend for the signal engine. Setup: set TELEGRAM_BOT_TOKEN=your_token_here 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 """ import asyncio import html import logging import os from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from typing import Dict, List, Optional from telegram import BotCommand, Update from telegram.constants import ParseMode from telegram.ext import Application, CommandHandler, ContextTypes from config import ( MAJOR_PAIRS, SYMBOL_DISPLAY, PlatformConfig, SignalStrength, Timeframe, ) from signal_aggregator import SignalAggregator, SignalReport logger = logging.getLogger(__name__) # ── Timeframe maps ────────────────────────────────────────────────── 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], "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], } # ── Formatting helpers ────────────────────────────────────────────── def _grade_dot(quality: str) -> str: """Colored dot matching confidence grade per spec.""" if quality in ("A+", "A"): return "🟢" if quality == "B": return "🟡" return "🔴" 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 _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) 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: """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 """ dot = _grade_dot(report.signal_quality) action = _signal_action(report.final_signal) pair = _h(report.display_name) grade = _h(report.signal_quality) conf = report.final_confidence lines: List[str] = [] # ── Header ─────────────────────────────────────────────────────── lines.append(f"{dot} TRADE ALERT: {action} {pair}") lines.append(f"Grade: {grade} | Confidence: {conf:.0f}% | {timeframe}") lines.append("") # ── Trade setup (only when grade B or above) ───────────────────── if report.trade_setup: ts = report.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}") 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)}") lines.append("") # ── Technical Indicators ───────────────────────────────────────── if report.primary_analysis and report.primary_analysis.indicators: lines.append("Technical Indicators:") for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue val_str = _fmt_float(ind.value) condition = _h(_indicator_condition(ind)) ind_name = _h(ind.name) lines.append(f"• {ind_name}: {val_str} ({condition})") # ── Multi-Timeframe ─────────────────────────────────────────────── if report.mtf_signals: lines.append("") agree_str = "✅ Aligned" if report.mtf_agreement else "⚠️ Divergent" lines.append(f"Multi-Timeframe ({agree_str}):") for tf, sig in report.mtf_signals.items(): score = report.mtf_scores.get(tf, 0) sig_name = _h(sig.name.replace("_", " ")) lines.append(f" {_h(tf)}: {sig_name} ({score:+.0f})") # ── Warnings ───────────────────────────────────────────────────── if report.warnings: lines.append("") for w in report.warnings: lines.append(f"⚠️ {_h(w)}") # ── Calendar / news ─────────────────────────────────────────────── if report.event_risk != "clear" and report.event_description: lines.append(f"📅 Event Risk: {_h(report.event_description)}") if report.news_description and report.news_description != "Neutral news sentiment": lines.append(f"📰 {_h(report.news_description)}") # ── Context footer ──────────────────────────────────────────────── lines.append("") lines.append( f"Regime: {_h(report.market_regime)} | " f"Vol: {_h(report.volatility_regime)} | " f"Score: {report.final_score:+.1f}" ) lines.append("⚠️ Educational use only. Not financial advice.") 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") # ── 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: 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) 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): 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 @staticmethod async def _send_html(update: Update, text: str): """Send HTML message, splitting at newlines if over Telegram's 4096-char limit.""" 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): state = self._state(update.effective_chat.id) 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 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. """ chat_id = update.effective_chat.id state = self._state(chat_id) args = context.args or [] if not args: await update.message.reply_text( "Usage: /signal <pair> [timeframe]\n" "Example: /signal EURUSD H4", parse_mode=ParseMode.HTML, ) return 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", "")) msg = await update.message.reply_text(f"🔄 Analyzing {_h(display)} ({tf})…") 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 r = reports[0] alert = _build_alert(r, tf) await msg.delete() await self._send_html(update, alert) 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 [TF] Analyze all major pairs and send a trade alert card for every signal rated Grade B or 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})…") try: reports = await self._analyze(self._run_scan, tf, list(MAJOR_PAIRS)) actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] 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.", 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 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"Grade B+ signals: {len(actionable)} ({grade_str})\n\n" "Trade alerts follow ↓" ) 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) 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: 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("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("tf", self.cmd_tf)) 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()