ForexBot2/telegram_bot.py

#!/usr/bin/env python3
"""
Forex Signal Intelligence Platform - Telegram Bot
===================================================
Mirrors main.py output exactly via Telegram.

Setup:
    export TELEGRAM_BOT_TOKEN="your_token_here"
    python telegram_bot.py

Commands:
    /scan [TF]        — Full signal scan (all major pairs)
    /pair EURUSD [TF] — Detailed analysis for one pair
    /tf H1            — Set default timeframe
    /help             — Commands + timeframe guide
"""

import asyncio
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 config ───────────────────────────────────────────────

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 (mirrors main.py, no ANSI colours) ──────────────────

WIDTH = 80


def _score_bar(score: float, width: int = 30) -> str:
    normalized = (score + 100) / 200
    filled = int(normalized * width)
    center = width // 2
    bar = ""
    for i in range(width):
        if i == center:
            bar += "│"
        elif (score >= 0 and center < i <= filled) or \
             (score < 0 and filled <= i < center):
            bar += "█"
        else:
            bar += "░"
    return bar


def _format_header(timeframe: str) -> str:
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    lines = [
        "",
        "═" * WIDTH,
        "  ╔═══════════════════════════════════════════════════════╗",
        "  ║   FOREX SIGNAL INTELLIGENCE PLATFORM                 ║",
        "  ║   Multi-Layer Signal Analysis Engine                  ║",
        "  ╚═══════════════════════════════════════════════════════╝",
        f"  Scan Time: {now}   TF: {timeframe}",
        "═" * WIDTH,
        "",
    ]
    return "\n".join(lines)


def _format_table(reports: List[SignalReport]) -> str:
    lines = [
        "  ┌─────────────┬───────────┬──────────────────────────────────┬────────┬────────────┐",
        "  │ Pair        │ Signal    │ Score                            │ Grade  │ Confidence │",
        "  ├─────────────┼───────────┼──────────────────────────────────┼────────┼────────────┤",
    ]
    for r in reports:
        sig = r.final_signal.name.replace("_", " ")
        bar = _score_bar(r.final_score, 30)
        pair = f"{r.display_name:<11}"
        signal_str = f"{sig:<9}"
        grade_str = f"{r.signal_quality:<6}"
        conf_str = f"{r.final_confidence:5.1f}%"
        warn = " ⚠" if r.event_risk == "danger" else (" △" if r.event_risk == "caution" else "  ")
        lines.append(f"  │ {pair} │ {signal_str} │ {bar} │ {grade_str} │ {conf_str:>8}{warn} │")
    lines.append(
        "  └─────────────┴───────────┴──────────────────────────────────┴────────┴────────────┘"
    )
    lines += [
        "",
        "  Legend: ██ Bullish | ██ Bearish | │ Center (0) | ⚠ Event Risk | △ Caution",
    ]
    return "\n".join(lines)


def _format_calendar(reports: List[SignalReport]) -> str:
    events_shown = set()
    lines = []
    for r in reports:
        if r.event_risk != "clear" and r.event_description not in events_shown:
            events_shown.add(r.event_description)
            lines.append(f"    ● {r.event_description} → affects {r.display_name}")
    if lines:
        return "\n  📅 ECONOMIC CALENDAR ALERTS\n" + "\n".join(lines)
    return "\n  📅 No high-impact events in the near term"


def _format_detail(r: SignalReport) -> List[str]:
    """
    Returns a list of message strings for one pair's detail report.
    Split into two messages so each stays under Telegram's 4096-char limit.
    """
    sig_name = r.final_signal.name.replace("_", " ")

    # ── Part 1: header, signal, MTF, indicators ────────────────────
    p1 = ["─" * WIDTH]
    p1.append(
        f"  {r.display_name}  │  Price: {r.current_price:.5f}"
        f"  │  Regime: {r.market_regime}  │  Vol: {r.volatility_regime}"
    )
    p1.append(
        f"\n  SIGNAL: {sig_name}  │  Score: {r.final_score:+.1f}"
        f"  │  Grade: {r.signal_quality}  │  Confidence: {r.final_confidence:.0f}%"
    )
    p1.append(f"  Action: {r.action_recommendation}")

    if r.mtf_signals:
        agree = "✓ ALIGNED" if r.mtf_agreement else "✗ DIVERGENT"
        p1.append(f"\n  Multi-Timeframe: {agree}")
        for tf, sig in r.mtf_signals.items():
            score = r.mtf_scores.get(tf, 0)
            p1.append(f"    {tf:>5}: {sig.name.replace('_', ' '):<12} ({score:+.1f})")

    if r.primary_analysis:
        p1.append(f"\n  Indicators ({r.primary_analysis.timeframe}):")
        for name, ind in r.primary_analysis.indicators.items():
            if name.startswith("_") or ind.weight == 0:
                continue
            ind_name = f"{ind.name:<18}"
            sig_str = f"{ind.signal.name.replace('_', ' '):<12}"
            desc = ind.description[:60]
            p1.append(f"    {ind_name} {sig_str} ({ind.confidence:.0f}%) │ {desc}")

    # ── Part 2: reasons, warnings, trade setup, key levels ─────────
    p2 = []

    if r.bullish_reasons:
        p2.append("  ▲ Bullish Factors:")
        for reason in r.bullish_reasons[:5]:
            p2.append(f"    + {reason}")

    if r.bearish_reasons:
        p2.append("\n  ▼ Bearish Factors:")
        for reason in r.bearish_reasons[:5]:
            p2.append(f"    - {reason}")

    if r.warnings:
        p2.append("\n  ⚠ Warnings:")
        for w in r.warnings:
            p2.append(f"    ! {w}")

    if r.news_description and r.news_description != "Neutral news sentiment":
        p2.append(f"\n  📰 News: {r.news_description}")
    if r.event_description and r.event_risk != "clear":
        p2.append(f"  📅 Event: {r.event_description}")

    if r.trade_setup:
        ts = r.trade_setup
        p2 += [
            "\n  Trade Setup:",
            f"    Direction:     {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}",
        ]

    if r.primary_analysis and r.primary_analysis.key_levels:
        lvls = r.primary_analysis.key_levels
        p2.append("\n  Key Levels:")
        for k in ("r3", "r2", "r1", "pivot", "s1", "s2", "s3"):
            if k in lvls:
                p2.append(f"    {k.upper():>6}: {lvls[k]:.5f}")

    result = ["\n".join(p1)]
    if p2:
        result.append("\n".join(p2))
    return result


# ── 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")

    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: list = None) -> PlatformConfig:
        config = PlatformConfig()
        config.primary_timeframe = TF_MAP[tf]
        config.confirmation_timeframes = TF_CONFIRMATIONS.get(tf, [Timeframe.M15, Timeframe.H4])
        config.watch_pairs = list(pairs or MAJOR_PAIRS)
        return config

    def _run_scan(self, tf: str, pairs: list) -> 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) -> Optional[str]:
        for a in args:
            if a.upper() in TF_MAP:
                return a.upper()
        return None

    @staticmethod
    async def _send_pre(update: Update, text: str):
        """Send text as a monospace pre block, splitting if over 4096 chars."""
        LIMIT = 3800  # leave room for <pre> tags
        chunks = []
        while text:
            if len(text) <= LIMIT:
                chunks.append(text)
                break
            # Split at last newline before limit
            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(
                f"<pre>{chunk}</pre>", 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 = (
            "<b>🌐 Forex Signal Intelligence Platform</b>\n\n"
            "<b>Commands:</b>\n"
            "  /scan — Scan all major pairs\n"
            "  /scan H4 — Scan with specific timeframe\n"
            "  /pair EURUSD — Detailed analysis\n"
            "  /pair EURUSD H4 — With specific timeframe\n"
            "  /tf H1 — Set default timeframe\n"
            "  /help — This message\n\n"
            "<b>Timeframes:</b>\n"
            "  <code>M5 </code> — 5 minutes\n"
            "  <code>M15</code> — 15 minutes\n"
            "  <code>M30</code> — 30 minutes\n"
            "  <code>H1 </code> — 1 hour  ← recommended for intraday\n"
            "  <code>H4 </code> — 4 hours ← recommended for swing trading\n"
            "  <code>D1 </code> — Daily   ← recommended for position trading\n\n"
            f"Current default: <b>{state.timeframe}</b>\n\n"
            "<i>⚠️ Educational use only. Not financial advice.</i>"
        )
        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: <b>/tf &lt;timeframe&gt;</b>\n"
                f"Options: <code>{opts}</code>",
                parse_mode=ParseMode.HTML,
            )
            return
        self._state(update.effective_chat.id).timeframe = tf
        await update.message.reply_text(
            f"✅ Default timeframe set to <b>{tf}</b>", 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)
        tf = self._parse_tf(context.args or []) or state.timeframe

        msg = await update.message.reply_text(f"🔄 Scanning markets ({tf})…")
        try:
            reports = await self._analyze(self._run_scan, tf, list(MAJOR_PAIRS))

            # ── summary block (header + table + calendar) ──────────
            summary = (
                _format_header(tf)
                + _format_table(reports)
                + _format_calendar(reports)
            )
            await msg.edit_text(f"<pre>{summary}</pre>", parse_mode=ParseMode.HTML)

            # ── detailed block for each actionable signal ──────────
            actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")]
            if actionable:
                header_line = "\n  ═══ ACTIONABLE SIGNALS (Grade B or above) ═══\n"
                first = True
                for r in actionable:
                    parts = _format_detail(r)
                    if first:
                        parts[0] = header_line + parts[0]
                        first = False
                    for part in parts:
                        await self._send_pre(update, part)
            else:
                await self._send_pre(
                    update,
                    "\n  No high-grade signals detected.\n"
                    "  Use /scan with a different timeframe to see more.\n",
                )

        except Exception as e:
            logger.error(f"Scan error: {e}")
            await msg.edit_text(f"❌ Scan failed: {e}")

    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: <b>/pair &lt;symbol&gt; [timeframe]</b>\n"
                "Example: <code>/pair EURUSD H4</code>",
                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)

        msg = await update.message.reply_text(f"🔄 Analyzing {display} ({tf})…")
        try:
            reports = await self._analyze(self._run_scan, tf, [raw])
            if not reports:
                await msg.edit_text(f"❌ No data returned for {raw}")
                return

            r = reports[0]
            summary = _format_header(tf) + _format_table([r])
            await msg.edit_text(f"<pre>{summary}</pre>", parse_mode=ParseMode.HTML)

            for part in _format_detail(r):
                await self._send_pre(update, part)

        except Exception as e:
            logger.error(f"Pair error for {raw}: {e}")
            await msg.edit_text(f"❌ Analysis failed: {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:
            await app.bot.set_my_commands([
                BotCommand("scan", "Scan all major pairs"),
                BotCommand("pair", "Detailed pair analysis — /pair EURUSD"),
                BotCommand("tf",   "Set default timeframe — /tf H4"),
                BotCommand("help", "Commands and timeframe 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("scan",  self.cmd_scan))
        app.add_handler(CommandHandler("pair",  self.cmd_pair))

        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()