ForexBot2/telegram_bot.py

#!/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

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
# plus gold to keep credits low and signal quality high.
SCAN_PAIRS = ["EURUSD=X", "USDJPY=X", "GBPUSD=X", "XAUUSD=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"🌐 🧠 <b>{_h(display)} — Signal Analysis</b>")
    lines.append(f"🕐 {now_str}")
    lines.append(f"<i>{_GRADE_LEGEND}</i>")

    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"<b>{tf}</b> — ❌ Data unavailable")
            if r.warnings:
                lines.append(f"<i>{_h(r.warnings[0])}</i>")
            continue

        action = _signal_action(r.final_signal)
        sym    = _grade_symbol(r.signal_quality)

        lines.append("")
        lines.append("────────────────────────")
        lines.append(f"{sym} <b>{tf} · {action} · Grade {_h(r.signal_quality)}</b>")
        lines.append("")

        # ── Trade setup (always kept intact) ─────────────────────────
        if r.trade_setup:
            ts = r.trade_setup
            lines.append(f"<b>Entry:</b> {ts.entry_price:.5f}")
            lines.append(
                f"<b>Take Profit:</b> {ts.take_profit_1:.5f}"
                f"  <i>(+{ts.reward_pips:.1f} pips)</i>"
            )
            if abs(ts.take_profit_2 - ts.take_profit_1) > 1e-8:
                lines.append(f"<b>Take Profit 2:</b> {ts.take_profit_2:.5f}")
            lines.append(
                f"<b>Stop Loss:</b> {ts.stop_loss:.5f}"
                f"  <i>(-{ts.risk_pips:.1f} pips)</i>"
            )
            lines.append(f"<b>Risk:Reward:</b> 1:{ts.risk_reward:.1f}")
            lines.append("")
        else:
            lines.append(f"<b>Current Price:</b> {r.current_price:.5f}")
            lines.append(f"<b>Analysis:</b> {_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"<i>MTF: {agree_str} | Score: {r.final_score:+.1f}"
            f" | {_h(r.market_regime)} | Vol: {_h(r.volatility_regime)}</i>"
        )

        # ── 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"📅 <b>Event Risk:</b> {_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("<i>⚠️ Educational use only. Not financial advice.</i>")
    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"📊 🧠 <b>Scan Complete</b> · {now_str}")
    lines.append(f"Timeframes: <code>{_h(tfs_str)}</code> | Best signal per pair")
    lines.append(f"<i>{_GRADE_LEGEND}</i>")
    lines.append("")

    for tf, r in actionable:
        action = _signal_action(r.final_signal)
        sym    = _grade_symbol(r.signal_quality)

        lines.append(
            f"{sym} <b>{_h(r.display_name)} · {action}</b>"
            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("<i>⚠️ Educational use only. Not financial advice.</i>")
    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 = (
            "🌐 <b>Forex Signal Intelligence Platform</b>\n\n"
            "<b>Commands:</b>\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"<b>Timeframes analyzed:</b> <code>{tfs}</code>\n\n"
            "<b>Signal Grades:</b>\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"
            "<i>⚠️ Educational use only. Not financial advice.</i>"
        )
        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: <b>/signal &lt;pair&gt;</b>\n"
                "Example: <code>/signal EURUSD</code>",
                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()