#!/usr/bin/env python3 """ Forex Signal Intelligence Platform - Telegram Bot =================================================== Exposes all platform functionality through a Telegram bot interface. Usage: export TELEGRAM_BOT_TOKEN="your_bot_token_here" python telegram_bot.py Commands: /scan [TF] — Scan all major pairs /signals [TF] — Only actionable signals (grade B+) /pair EURUSD [TF] — Detailed analysis of one pair /all [TF] — Scan major + cross pairs /tf H1 — Set default timeframe /watch [minutes] — Enable periodic scan alerts /unwatch — Disable alerts /status — Current settings /help — Full command reference """ import asyncio import logging import os from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from typing import Dict, Optional from telegram import BotCommand, Update from telegram.constants import ParseMode from telegram.ext import Application, CallbackContext, CommandHandler, ContextTypes from config import ( CROSS_PAIRS, MAJOR_PAIRS, SYMBOL_DISPLAY, PlatformConfig, SignalStrength, Timeframe, ) from signal_aggregator import SignalAggregator, SignalReport logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────── 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], } SIGNAL_EMOJI = { SignalStrength.STRONG_BUY: "🟢⬆⬆", SignalStrength.BUY: "🟢⬆", SignalStrength.NEUTRAL: "⬜↔", SignalStrength.SELL: "🔴⬇", SignalStrength.STRONG_SELL: "🔴⬇⬇", } SIGNAL_NAME = { SignalStrength.STRONG_BUY: "STRONG BUY", SignalStrength.BUY: "BUY", SignalStrength.NEUTRAL: "NEUTRAL", SignalStrength.SELL: "SELL", SignalStrength.STRONG_SELL: "STRONG SELL", } GRADE_EMOJI = { "A+": "⭐⭐", "A": "⭐", "B": "🔵", "C": "🟡", "D": "🟠", "F": "⚫", } WATCH_MIN_INTERVAL = 5 # minimum watch interval in minutes WATCH_MAX_INTERVAL = 120 # maximum watch interval in minutes WATCH_DEFAULT = 5 # ── Per-chat state ───────────────────────────────────────────────── class ChatState: def __init__(self): self.timeframe: str = "H1" self.pairs: list = list(MAJOR_PAIRS) self.watch_enabled: bool = False # ── HTML helpers ─────────────────────────────────────────────────── def _esc(text: str) -> str: """Escape HTML special characters for Telegram HTML parse mode.""" return text.replace("&", "&").replace("<", "<").replace(">", ">") def _split_messages(lines: list[str], limit: int = 4000) -> list[str]: """Join lines into messages that stay under Telegram's 4096-char limit.""" messages, current, current_len = [], [], 0 for line in lines: line_len = len(line) + 1 # +1 for newline if current and current_len + line_len > limit: messages.append("\n".join(current)) current, current_len = [], 0 current.append(line) current_len += line_len if current: messages.append("\n".join(current)) return messages # ── Message formatters ───────────────────────────────────────────── def _format_scan(reports: list[SignalReport], timeframe: str, title: str = "FOREX SIGNAL SCAN") -> str: now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") lines = [ f"📊 {title} — {timeframe}", "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━", f"🕐 {now}\n", ] for r in reports: sig_em = SIGNAL_EMOJI.get(r.final_signal, "❔") grade_em = GRADE_EMOJI.get(r.signal_quality, "") sig_name = SIGNAL_NAME.get(r.final_signal, "?") event_flag = " ⚠️" if r.event_risk == "danger" else (" ⚡" if r.event_risk == "caution" else "") lines.append( f"{sig_em} {_esc(r.display_name)} " f"{grade_em}{r.signal_quality} " f"{_esc(sig_name):<11} " f"{r.final_confidence:.0f}%{event_flag}" ) lines += ["", "Use /pair EURUSD for detailed analysis"] return "\n".join(lines) def _format_pair_detail(r: SignalReport) -> list[str]: """ Return a list of message strings for a single pair. Split across messages so each stays under Telegram's 4096-char limit. """ sig_em = SIGNAL_EMOJI.get(r.final_signal, "❔") grade_em = GRADE_EMOJI.get(r.signal_quality, "") sig_name = SIGNAL_NAME.get(r.final_signal, "?") # ── Message 1: Header, MTF, Trade Setup, Key Levels ─────────── m1 = [ f"{sig_em} {_esc(r.display_name)} — {_esc(sig_name)}", "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━", f"Price: {r.current_price:.5f}", f"Regime: {_esc(r.market_regime)} Vol: {_esc(r.volatility_regime)}", "", f"Score: {r.final_score:+.1f} " f"Grade: {grade_em}{r.signal_quality} " f"Conf: {r.final_confidence:.0f}%", f"Action: {_esc(r.action_recommendation)}", ] if r.event_risk != "clear": icon = "⚠️" if r.event_risk == "danger" else "⚡" m1.append(f"{icon} Event: {_esc(r.event_description)}") if r.news_description and r.news_description != "Neutral news sentiment": m1.append(f"📰 News: {_esc(r.news_description)}") # Multi-timeframe if r.mtf_signals: agree = "✅ ALIGNED" if r.mtf_agreement else "⚡ DIVERGENT" m1 += ["", f"📊 Multi-Timeframe: {agree}"] for tf, sig in r.mtf_signals.items(): score = r.mtf_scores.get(tf, 0) m1.append( f" {tf:>5} {SIGNAL_EMOJI.get(sig, '?')} " f"{score:+.1f}" ) # Trade setup if r.trade_setup: ts = r.trade_setup dir_em = "🟢" if ts.direction == "BUY" else "🔴" m1 += [ "", "💼 Trade Setup", f" Direction: {dir_em} {ts.direction}", f" Entry: {ts.entry_price:.5f}", f" Stop Loss: {ts.stop_loss:.5f} ({ts.risk_pips:.1f} pips)", f" Take Profit 1: {ts.take_profit_1:.5f} ({ts.reward_pips:.1f} pips)", f" Take Profit 2: {ts.take_profit_2:.5f}", f" Risk:Reward: 1:{ts.risk_reward:.1f}", ] # Key levels if r.primary_analysis and r.primary_analysis.key_levels: lvls = r.primary_analysis.key_levels shown = [(k, lvls[k]) for k in ("r3", "r2", "r1", "pivot", "s1", "s2", "s3") if k in lvls] if shown: m1 += ["", "📍 Key Levels:"] for k, v in shown: arrow = "🔴" if v > r.current_price else "🟢" m1.append(f" {arrow} {k.upper():>6}: {v:.5f}") # ── Message 2: Indicators + Reasons ─────────────────────────── m2 = [] if r.primary_analysis: ind_lines = [] for name, ind in r.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue em = SIGNAL_EMOJI.get(ind.signal, "↔") # Trim long descriptions to keep message readable desc = _esc(ind.description[:90]) ind_lines.append( f" {em} {_esc(ind.name)} {ind.confidence:.0f}%" f"\n {desc}" ) if ind_lines: m2 += ["📉 Indicators:"] + ind_lines if r.bullish_reasons: if m2: m2.append("") m2.append("🟢 Bullish Factors:") for reason in r.bullish_reasons[:5]: m2.append(f" ▲ {_esc(reason)}") if r.bearish_reasons: if m2: m2.append("") m2.append("🔴 Bearish Factors:") for reason in r.bearish_reasons[:5]: m2.append(f" ▼ {_esc(reason)}") if r.warnings: if m2: m2.append("") m2.append("⚠️ Warnings:") for w in r.warnings: m2.append(f" ! {_esc(w)}") result = _split_messages(m1) if m2: result += _split_messages(m2) return result # ══════════════════════════════════════════════════════════════════ # BOT CLASS # ══════════════════════════════════════════════════════════════════ class ForexTelegramBot: def __init__(self): self._chat_states: Dict[int, ChatState] = {} self._executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="forex") def _state(self, chat_id: int) -> ChatState: if chat_id not in self._chat_states: self._chat_states[chat_id] = ChatState() return self._chat_states[chat_id] def _build_config(self, state: ChatState) -> PlatformConfig: config = PlatformConfig() config.primary_timeframe = TF_MAP[state.timeframe] config.confirmation_timeframes = TF_CONFIRMATIONS.get( state.timeframe, [Timeframe.M15, Timeframe.H4] ) config.watch_pairs = list(state.pairs) return config # ── Blocking analysis (run in thread pool) ───────────────────── def _run_scan(self, state: ChatState) -> list[SignalReport]: config = self._build_config(state) return SignalAggregator(config).analyze_all_pairs() def _run_pair(self, symbol: str, state: ChatState) -> Optional[SignalReport]: config = self._build_config(state) config.watch_pairs = [symbol] reports = SignalAggregator(config).analyze_all_pairs() return reports[0] if reports else None async def _analyze(self, func, *args): """Run a blocking analysis function in the thread pool.""" loop = asyncio.get_running_loop() return await loop.run_in_executor(self._executor, func, *args) # ── Parse helpers ────────────────────────────────────────────── @staticmethod def _parse_tf(args: list[str]) -> Optional[str]: """Return the first arg that is a valid timeframe key, or None.""" for a in args: if a.upper() in TF_MAP: return a.upper() return None # ── Commands ─────────────────────────────────────────────────── async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE): text = ( "🌐 Forex Signal Intelligence Platform\n" "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\n" "Multi-layer technical + fundamental signal analysis across major Forex pairs.\n\n" "Commands:\n" " /scan — Scan all major pairs\n" " /signals — Actionable signals only (grade B+)\n" " /pair EURUSD — Detailed single-pair analysis\n" " /all — Include cross pairs in scan\n" " /tf H1 — Set default timeframe\n" " /watch — Periodic scan alerts (every 5 min)\n" " /unwatch — Disable alerts\n" " /status — Current settings\n" " /help — Full command reference\n\n" "⚠️ Educational/research use only. Not financial advice." ) await update.message.reply_text(text, parse_mode=ParseMode.HTML) async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE): text = ( "📖 Command Reference\n\n" "/scan — Scan all major pairs (uses default TF)\n" "/scan H4 — Scan with a specific timeframe\n\n" "/signals — Grade B+ signals only\n" "/signals H4 — With a specific timeframe\n\n" "/pair EURUSD — Detailed analysis (uses default TF)\n" "/pair EURUSD H4 — With a specific timeframe\n\n" "/all — Scan major + cross pairs\n" "/all H4 — With a specific timeframe\n\n" "/tf H1 — Set your default timeframe\n" " Options: M5 M15 M30 H1 H4 D1\n\n" "/watch — Alert every 5 min (actionable signals only)\n" "/watch 15 — Alert every 15 min (5–120 min range)\n" "/unwatch — Stop alerts\n\n" "/status — View current settings\n\n" "Signal key:\n" " 🟢⬆⬆ Strong Buy 🟢⬆ Buy ⬜↔ Neutral\n" " 🔴⬇ Sell 🔴⬇⬇ Strong Sell\n\n" "Grade key:\n" " A+ ⭐⭐ A ⭐ B 🔵 C 🟡 D 🟠 F ⚫" ) await update.message.reply_text(text, parse_mode=ParseMode.HTML) async def cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE): state = self._state(update.effective_chat.id) pairs_display = ", ".join(SYMBOL_DISPLAY.get(p, p) for p in state.pairs) watch_str = "✅ enabled" if state.watch_enabled else "❌ disabled" text = ( "⚙️ Current Settings\n\n" f"Timeframe: {state.timeframe}\n" f"Pairs: {_esc(pairs_display)}\n" f"Watch alerts: {watch_str}" ) 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_scan(self, update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id state = self._state(chat_id) # Allow inline TF override: /scan H4 scan_state = ChatState() scan_state.timeframe = self._parse_tf(context.args or []) or state.timeframe scan_state.pairs = list(state.pairs) msg = await update.message.reply_text("🔄 Scanning markets, please wait…") try: reports = await self._analyze(self._run_scan, scan_state) text = _format_scan(reports, scan_state.timeframe) await msg.edit_text(text, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"Scan error: {e}") await msg.edit_text(f"❌ Scan failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) async def cmd_signals(self, update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id state = self._state(chat_id) scan_state = ChatState() scan_state.timeframe = self._parse_tf(context.args or []) or state.timeframe scan_state.pairs = list(state.pairs) msg = await update.message.reply_text("🔄 Scanning for signals…") try: reports = await self._analyze(self._run_scan, scan_state) actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] if actionable: text = _format_scan(actionable, scan_state.timeframe, title="🎯 ACTIONABLE SIGNALS") else: text = ( f"🎯 ACTIONABLE SIGNALS — {scan_state.timeframe}\n\n" "No grade B+ signals right now.\n" "Use /scan to see all pairs." ) await msg.edit_text(text, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"Signals error: {e}") await msg.edit_text(f"❌ Scan failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) async def cmd_pair(self, update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id state = self._state(chat_id) args = context.args or [] if not args: await update.message.reply_text( "Usage: /pair <symbol> [timeframe]\n" "Example: /pair EURUSD H4", parse_mode=ParseMode.HTML, ) return raw = args[0].upper().replace("/", "").replace("_", "") if not raw.endswith("=X"): raw += "=X" pair_state = ChatState() pair_state.timeframe = self._parse_tf(args[1:]) or state.timeframe display = SYMBOL_DISPLAY.get(raw, raw) msg = await update.message.reply_text(f"🔄 Analyzing {_esc(display)}…") try: report = await self._analyze(self._run_pair, raw, pair_state) if report is None: await msg.edit_text( f"❌ No data returned for {_esc(raw)}.", parse_mode=ParseMode.HTML, ) return messages = _format_pair_detail(report) await msg.edit_text(messages[0], parse_mode=ParseMode.HTML) for m in messages[1:]: await update.message.reply_text(m, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"Pair analysis error for {raw}: {e}") await msg.edit_text(f"❌ Analysis failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) async def cmd_all(self, update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id state = self._state(chat_id) all_state = ChatState() all_state.timeframe = self._parse_tf(context.args or []) or state.timeframe all_state.pairs = list(MAJOR_PAIRS) + list(CROSS_PAIRS) msg = await update.message.reply_text("🔄 Scanning all pairs (major + cross)…") try: reports = await self._analyze(self._run_scan, all_state) text = _format_scan(reports, all_state.timeframe, title="FULL PAIR SCAN") await msg.edit_text(text, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"All-pairs scan error: {e}") await msg.edit_text(f"❌ Scan failed: {_esc(str(e))}", parse_mode=ParseMode.HTML) async def cmd_watch(self, update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id state = self._state(chat_id) interval_minutes = WATCH_DEFAULT if context.args: try: interval_minutes = int(context.args[0]) interval_minutes = max(WATCH_MIN_INTERVAL, min(WATCH_MAX_INTERVAL, interval_minutes)) except ValueError: pass # Remove any existing watch job for this chat for job in context.job_queue.get_jobs_by_name(f"watch_{chat_id}"): job.schedule_removal() state.watch_enabled = True context.job_queue.run_repeating( self._watch_job, interval=interval_minutes * 60, first=5, chat_id=chat_id, name=f"watch_{chat_id}", ) await update.message.reply_text( f"✅ Watch alerts enabled — scanning every {interval_minutes} min.\n" "Alerts fire when grade B+ signals appear. Use /unwatch to stop.", parse_mode=ParseMode.HTML, ) async def cmd_unwatch(self, update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id self._state(chat_id).watch_enabled = False for job in context.job_queue.get_jobs_by_name(f"watch_{chat_id}"): job.schedule_removal() await update.message.reply_text("❌ Watch alerts disabled.") async def cmd_credits(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Show API credit usage against the free-tier daily limit.""" from data_provider import ForexDataProvider config = self._build_config(self._state(update.effective_chat.id)) provider = ForexDataProvider(config) used = provider._daily_credits limit = provider.DAILY_LIMIT remaining = limit - used pct = (used / limit) * 100 if limit else 0 bar_filled = int(pct / 5) bar = "█" * bar_filled + "░" * (20 - bar_filled) status = "🟢" if pct < 60 else ("🟡" if pct < 80 else "🔴") text = ( "📊 API Credit Usage\n\n" f"{bar} {status}\n" f"Used: {used} / {limit} credits\n" f"Remaining: {remaining} credits\n\n" "Free tier: 8 credits/min, 800 credits/day\n" "Each full scan uses ~21 credits (7 pairs × 3 timeframes)\n" "≈ 38 scans available per day" ) await update.message.reply_text(text, parse_mode=ParseMode.HTML) # ── Watch job ────────────────────────────────────────────────── async def _watch_job(self, context: CallbackContext): chat_id = context.job.chat_id state = self._state(chat_id) if not state.watch_enabled: return try: reports = await self._analyze(self._run_scan, state) actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] if actionable: text = ( "🔔 SIGNAL ALERT\n\n" + _format_scan(actionable, state.timeframe, title="ACTIONABLE SIGNALS") ) else: now = datetime.now(timezone.utc).strftime("%H:%M UTC") text = f"📊 Watch scan {now} — No grade B+ signals detected." await context.bot.send_message(chat_id, text, parse_mode=ParseMode.HTML) except Exception as e: logger.error(f"Watch job error for chat {chat_id}: {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" " export TELEGRAM_BOT_TOKEN='your_token_here'" ) async def _register_commands(app: Application) -> None: """Register commands with Telegram so they appear in the / menu.""" await app.bot.set_my_commands([ BotCommand("scan", "Scan all major pairs"), BotCommand("signals", "Actionable signals only (grade B+)"), BotCommand("pair", "Detailed pair analysis — /pair EURUSD"), BotCommand("all", "Scan major + cross pairs"), BotCommand("tf", "Set timeframe — /tf H4"), BotCommand("watch", "Enable periodic alerts"), BotCommand("unwatch", "Disable alerts"), BotCommand("credits", "View API credit usage"), BotCommand("status", "View current settings"), BotCommand("help", "Full command reference"), ]) 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("status", self.cmd_status)) app.add_handler(CommandHandler("tf", self.cmd_tf)) app.add_handler(CommandHandler("scan", self.cmd_scan)) app.add_handler(CommandHandler("signals", self.cmd_signals)) app.add_handler(CommandHandler("pair", self.cmd_pair)) app.add_handler(CommandHandler("all", self.cmd_all)) app.add_handler(CommandHandler("watch", self.cmd_watch)) app.add_handler(CommandHandler("unwatch", self.cmd_unwatch)) app.add_handler(CommandHandler("credits", self.cmd_credits)) 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()