ForexBot2/telegram_bot.py

#!/usr/bin/env python3
"""
ForexBot2 — Telegram Bot
========================
Async Telegram interface with candle-boundary auto-scanner.

Manual scan: 4 pairs × 2 timeframes via inline buttons.
Auto-scan:   Starts automatically on boot if TELEGRAM_CHAT_ID is set.
             /autoscan toggles for additional chats or to stop.
             M15 fires at :00/:15/:30/:45 + 10s buffer.
             H1  fires at :00 + 70s buffer (staggered to respect 8 credits/min).

Credit budget (free tier):
    M15 boundary:  4 pairs × 1 credit  = 4 credits
    H1  boundary:  4 pairs × 1 credit  = 4 credits  (staggered to next minute)
    Top of hour:   4 + 4 = 8 across two minutes → ≤ 8 / min
    Daily total:   384 (M15) + 96 (H1) = 480 / day (limit 800)

Usage:
    python telegram_bot.py
    TELEGRAM_TOKEN=<token> TELEGRAM_CHAT_ID=<id> python telegram_bot.py
"""

import asyncio
import logging
import os
import sys
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import List, Optional

from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ParseMode
from telegram.ext import (
    Application,
    CallbackQueryHandler,
    CommandHandler,
    ContextTypes,
)

from config import (
    PlatformConfig, SignalStrength, Timeframe,
    EURUSD, USDJPY, GBPUSD, XAUUSD, PAIR_INFO,
)
from data_provider import ForexDataProvider
from signal_aggregator import SignalAggregator, SignalReport

# ── Force UTF-8 on Windows ────────────────────────────────────────────
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    level=logging.INFO,
    datefmt="%H:%M:%S",
)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logger = logging.getLogger("telegram_bot")

TOKEN = os.environ.get("TELEGRAM_TOKEN", "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE")
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022")

# ── Pairs & buttons ───────────────────────────────────────────────────

SCAN_PAIRS = [EURUSD, USDJPY, GBPUSD, XAUUSD]

SCAN_BUTTONS = [
    {"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"},
    {"cb": "scan_eurusd_h1",  "symbol": EURUSD, "tf": Timeframe.H1,  "label": "EUR/USD 1H"},
    {"cb": "scan_usdjpy_m15", "symbol": USDJPY, "tf": Timeframe.M15, "label": "USD/JPY 15M"},
    {"cb": "scan_usdjpy_h1",  "symbol": USDJPY, "tf": Timeframe.H1,  "label": "USD/JPY 1H"},
    {"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"},
    {"cb": "scan_gbpusd_h1",  "symbol": GBPUSD, "tf": Timeframe.H1,  "label": "GBP/USD 1H"},
    {"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"},
    {"cb": "scan_xauusd_h1",  "symbol": XAUUSD, "tf": Timeframe.H1,  "label": "XAU/USD 1H"},
]

SCAN_LOOKUP = {b["cb"]: b for b in SCAN_BUTTONS}

# Manual scans use richer confirmations (may spend extra cached credits)
TF_CONFIRMATIONS = {
    Timeframe.M15: [Timeframe.M5, Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15, Timeframe.H4],
}

# Auto-scan uses only cross-TF (already cached, zero extra credits)
AUTOSCAN_CONFIRMATIONS = {
    Timeframe.M15: [Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15],
}

_executor = ThreadPoolExecutor(max_workers=2)


# ── Shared data provider (single cache across all scans) ─────────────

_shared_dp: Optional[ForexDataProvider] = None


def _get_shared_dp() -> ForexDataProvider:
    global _shared_dp
    if _shared_dp is None:
        _shared_dp = ForexDataProvider(PlatformConfig())
    return _shared_dp


# ── Signal formatting ─────────────────────────────────────────────────

_SIGNAL_EMOJI = {
    SignalStrength.STRONG_BUY:  "🟢",
    SignalStrength.BUY:         "🟩",
    SignalStrength.NEUTRAL:     "⬜",
    SignalStrength.SELL:        "🟥",
    SignalStrength.STRONG_SELL: "🔴",
}

_QUALITY_EMOJI = {
    "A+": "🏆", "A": "🥇", "B": "🥈", "C": "🥉", "D": "⚠️", "F": "❌",
}

_ARROW = {
    SignalStrength.STRONG_BUY:  "⬆️",
    SignalStrength.BUY:         "↗️",
    SignalStrength.NEUTRAL:     "➡️",
    SignalStrength.SELL:        "↘️",
    SignalStrength.STRONG_SELL: "⬇️",
}


def _signal_label(sig: SignalStrength) -> str:
    return sig.name.replace("_", " ")


def _mtf_line(report: SignalReport) -> str:
    if not report.mtf_signals:
        return ""
    parts = []
    for tf, sig in report.mtf_signals.items():
        score = report.mtf_scores.get(tf, 0)
        parts.append(f"{tf}: {_signal_label(sig)} {score:+.1f}")
    return " | ".join(parts)


def _indicator_line(name: str, ind) -> str:
    arrow = _ARROW.get(ind.signal, "➡️")
    short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description
    return f"{arrow} *{ind.name}*: {short}"


def format_signal_message(report: SignalReport) -> str:
    sig = report.final_signal
    emoji = _SIGNAL_EMOJI.get(sig, "⬜")
    qe = _QUALITY_EMOJI.get(report.signal_quality, "")
    label = _signal_label(sig)
    mtf_tag = "✅ MTF aligned" if report.mtf_agreement else "❌ MTF diverged"

    lines: list[str] = []

    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    lines += [
        f"🔍 *{report.display_name} Signal Analysis*  ({report.primary_analysis.timeframe if report.primary_analysis else ''})",
        f"🕐 {now}",
        "",
        f"💰 Price: *{report.current_price:.5f}*" if report.current_price < 10 else f"💰 Price: *{report.current_price:.3f}*",
        f"📊 Regime: {report.market_regime.replace('_', ' ').title()} | Vol: {report.volatility_regime}",
        "",
        "━━━━━━━━━━━━━━━━━━━━",
        f"{emoji} *{label}*  {qe}{report.signal_quality}  |  Score: {report.final_score:+.1f}  |  {report.final_confidence:.0f}% conf",
        f"_{report.action_recommendation}_",
        f"{mtf_tag}",
    ]

    mtf = _mtf_line(report)
    if mtf:
        lines += ["", f"📈 *MTF*: {mtf}"]

    if report.primary_analysis:
        lines += ["", "━━━━━━━━━━━━━━━━━━━━", "📋 *Indicators*"]
        for name, ind in report.primary_analysis.indicators.items():
            if name.startswith("_") or ind.weight == 0:
                continue
            lines.append(_indicator_line(name, ind))

    if report.warnings:
        lines.append("")
        for w in report.warnings:
            lines.append(f"⚠️ {w}")

    if report.trade_setup and sig in (SignalStrength.STRONG_BUY, SignalStrength.STRONG_SELL):
        ts = report.trade_setup
        dir_emoji = "📈" if ts.direction == "BUY" else "📉"
        lines += [
            "",
            "━━━━━━━━━━━━━━━━━━━━",
            f"🚀 *TRADE SETUP*",
            f"{dir_emoji} Direction: *{ts.direction}*",
            f"▶ Entry:  `{ts.entry_price:.3f}`",
            f"🛑 SL:     `{ts.stop_loss:.3f}`  ({ts.risk_pips:.1f} pips)",
            f"🎯 TP1:    `{ts.take_profit_1:.3f}`  ({ts.reward_pips:.1f} pips)",
        ]
        if ts.take_profit_2:
            lines.append(f"🎯 TP2:    `{ts.take_profit_2:.3f}`")
        lines.append(f"⚖️  RR:     1:{ts.risk_reward:.1f}")

    lines += ["", "━━━━━━━━━━━━━━━━━━━━",
              "_Educational / research use only._"]

    return "\n".join(lines)


# ── Auto-scan summary format ─────────────────────────────────────────

def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
    now = datetime.now(timezone.utc).strftime("%H:%M UTC")
    lines = [
        f"📊 *{tf.value} Auto-Scan* | {now}",
        "━━━━━━━━━━━━━━━━━━━━",
    ]
    for r in reports:
        emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
        sig = _signal_label(r.final_signal)
        mtf = "✅" if r.mtf_agreement else "❌"
        lines.append(
            f"{emoji} *{r.display_name}*  {sig}  "
            f"{_QUALITY_EMOJI.get(r.signal_quality, '')}{r.signal_quality}  "
            f"{r.final_score:+.1f}  {r.final_confidence:.0f}%  {mtf}MTF"
        )
    lines.append("━━━━━━━━━━━━━━━━━━━━")

    actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")]
    if actionable:
        names = ", ".join(r.display_name for r in actionable)
        lines.append(f"🔔 Actionable: {names}")
    else:
        lines.append("No actionable signals this cycle.")

    return "\n".join(lines)


# ── Keyboard helper ───────────────────────────────────────────────────

def scan_keyboard() -> InlineKeyboardMarkup:
    rows = []
    for i in range(0, len(SCAN_BUTTONS), 2):
        rows.append([
            InlineKeyboardButton(f"🔍 {b['label']}", callback_data=b["cb"])
            for b in SCAN_BUTTONS[i:i + 2]
        ])
    return InlineKeyboardMarkup(rows)


# ── Signal runner (shared data provider for cache reuse) ──────────────

def _run_signal_sync(
    symbol: str,
    timeframe: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
) -> SignalReport:
    if confirmation_tfs is None:
        confirmation_tfs = TF_CONFIRMATIONS.get(
            timeframe, [Timeframe.M15, Timeframe.H4]
        )
    config = PlatformConfig()
    config.primary_timeframe = timeframe
    config.confirmation_timeframes = confirmation_tfs
    aggregator = SignalAggregator(config)
    aggregator.data_provider = _get_shared_dp()
    return aggregator.analyze_pair(symbol=symbol)


async def run_signal(
    symbol: str,
    timeframe: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
) -> SignalReport:
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        _executor, _run_signal_sync, symbol, timeframe, confirmation_tfs
    )


# ── Candle-boundary scheduler ────────────────────────────────────────

_autoscan_chats: set[int] = set()
_autoscan_tasks: list[asyncio.Task] = []


def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
    """Seconds until the next candle close + buffer."""
    now = datetime.now(timezone.utc)
    elapsed_minutes = now.minute % interval_minutes
    elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
    interval_seconds = interval_minutes * 60
    remaining = interval_seconds - elapsed_seconds + buffer
    if remaining <= buffer:
        remaining += interval_seconds
    return remaining


async def _scan_timeframe(
    bot,
    tf: Timeframe,
    confirmation_tfs: List[Timeframe],
) -> None:
    """Scan all pairs for one timeframe and broadcast results."""
    reports: List[SignalReport] = []
    for symbol in SCAN_PAIRS:
        try:
            report = await run_signal(symbol, tf, confirmation_tfs)
            reports.append(report)
        except Exception:
            logger.exception(f"Auto-scan failed for {symbol} {tf.value}")

    if not reports:
        return

    summary = _format_scan_summary(reports, tf)
    actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")]

    for chat_id in list(_autoscan_chats):
        try:
            await bot.send_message(chat_id, summary, parse_mode=ParseMode.MARKDOWN)
            for report in actionable:
                detail = format_signal_message(report)
                await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
        except Exception:
            logger.exception(f"Failed to send to chat {chat_id}")


async def _m15_loop(bot) -> None:
    """Sleep until each M15 candle close, then scan all pairs."""
    try:
        while _autoscan_chats:
            wait = _seconds_until_boundary(15, buffer=10)
            logger.info(f"M15 auto-scan: next in {wait:.0f}s")
            await asyncio.sleep(wait)
            if not _autoscan_chats:
                break
            logger.info("M15 auto-scan: firing")
            await _scan_timeframe(
                bot, Timeframe.M15, AUTOSCAN_CONFIRMATIONS[Timeframe.M15]
            )
    except asyncio.CancelledError:
        pass
    finally:
        logger.info("M15 auto-scan loop stopped")


async def _h1_loop(bot) -> None:
    """Sleep until each H1 candle close + 60s stagger, then scan all pairs."""
    try:
        while _autoscan_chats:
            wait = _seconds_until_boundary(60, buffer=70)
            logger.info(f"H1 auto-scan: next in {wait:.0f}s")
            await asyncio.sleep(wait)
            if not _autoscan_chats:
                break
            logger.info("H1 auto-scan: firing")
            await _scan_timeframe(
                bot, Timeframe.H1, AUTOSCAN_CONFIRMATIONS[Timeframe.H1]
            )
    except asyncio.CancelledError:
        pass
    finally:
        logger.info("H1 auto-scan loop stopped")


def _start_scheduler(bot) -> None:
    """Start the M15 and H1 scanning loops if not already running."""
    alive = [t for t in _autoscan_tasks if not t.done()]
    if not alive:
        _autoscan_tasks.clear()
        _autoscan_tasks.append(asyncio.create_task(_m15_loop(bot)))
        _autoscan_tasks.append(asyncio.create_task(_h1_loop(bot)))
        logger.info("Candle-boundary scheduler started")


def _stop_scheduler() -> None:
    """Cancel all scanning loops."""
    for t in _autoscan_tasks:
        t.cancel()
    _autoscan_tasks.clear()
    logger.info("Candle-boundary scheduler stopped")


# ── Handlers ──────────────────────────────────────────────────────────

async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    scanning = "running" if _autoscan_chats else "off"
    text = (
        "👋 *ForexBot2 — Signal Intelligence*\n\n"
        "Pick a pair and timeframe to scan.\n"
        "Strong signals include a full trade setup with entry, SL and TP.\n\n"
        f"Auto-scan: *{scanning}* — /autoscan to toggle"
    )
    await update.message.reply_text(
        text,
        parse_mode=ParseMode.MARKDOWN,
        reply_markup=scan_keyboard(),
    )


async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    chat_id = update.effective_chat.id

    if chat_id in _autoscan_chats:
        _autoscan_chats.discard(chat_id)
        if not _autoscan_chats:
            _stop_scheduler()
        await update.message.reply_text(
            "⏹ Auto-scan stopped.",
            parse_mode=ParseMode.MARKDOWN,
        )
        return

    _autoscan_chats.add(chat_id)
    _start_scheduler(context.bot)

    next_m15 = _seconds_until_boundary(15, buffer=10)
    next_h1 = _seconds_until_boundary(60, buffer=70)
    await update.message.reply_text(
        "✅ *Auto-scan enabled*\n\n"
        f"Scanning {len(SCAN_PAIRS)} pairs on M15 + H1 at candle close.\n"
        f"Next M15 scan in {next_m15 / 60:.0f} min\n"
        f"Next H1 scan in {next_h1 / 60:.0f} min\n\n"
        "Use /autoscan again to stop.",
        parse_mode=ParseMode.MARKDOWN,
    )


async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.callback_query
    await query.answer()

    btn = SCAN_LOOKUP.get(query.data)
    if not btn:
        return

    pair_display = btn["label"]
    await query.edit_message_text(
        f"⏳ Scanning {pair_display}…",
        parse_mode=ParseMode.MARKDOWN,
    )

    try:
        report = await run_signal(btn["symbol"], btn["tf"])
        text = format_signal_message(report)
    except Exception as exc:
        logger.exception("Signal scan failed")
        text = f"❌ Scan failed: {exc}"

    await query.edit_message_text(
        text,
        parse_mode=ParseMode.MARKDOWN,
        reply_markup=scan_keyboard(),
    )


# ── Boot-time auto-start ──────────────────────────────────────────────

async def _post_init(app: Application) -> None:
    """Start the candle-boundary scheduler on boot if CHAT_ID is set."""
    if not CHAT_ID:
        logger.info("TELEGRAM_CHAT_ID not set — auto-scan disabled on boot. "
                     "Use /autoscan in chat to enable.")
        return

    chat_id = int(CHAT_ID)
    _autoscan_chats.add(chat_id)
    _start_scheduler(app.bot)

    next_m15 = _seconds_until_boundary(15, buffer=10)
    next_h1 = _seconds_until_boundary(60, buffer=70)
    logger.info(
        f"Auto-scan started for chat {chat_id} — "
        f"M15 in {next_m15:.0f}s, H1 in {next_h1:.0f}s"
    )

    try:
        await app.bot.send_message(
            chat_id,
            f"🤖 *ForexBot2 started*\n\n"
            f"Auto-scanning {len(SCAN_PAIRS)} pairs on M15 + H1.\n"
            f"Next M15 in {next_m15 / 60:.0f} min | "
            f"Next H1 in {next_h1 / 60:.0f} min\n\n"
            f"/autoscan to stop | Manual buttons via /start",
            parse_mode=ParseMode.MARKDOWN,
        )
    except Exception:
        logger.exception("Failed to send boot message")


# ── Entry point ───────────────────────────────────────────────────────

def main() -> None:
    if not TOKEN:
        sys.exit("TELEGRAM_TOKEN not set")

    logger.info("Starting ForexBot2 Telegram bot…")

    app = (
        Application.builder()
        .token(TOKEN)
        .post_init(_post_init)
        .build()
    )

    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("autoscan", cmd_autoscan))
    app.add_handler(CallbackQueryHandler(cb_scan, pattern="^scan_"))

    logger.info("Bot running — press Ctrl+C to stop")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()