#!/usr/bin/env python3
"""
Forex Signal Intelligence Platform - Telegram Bot
==================================================
Trade alert frontend for the signal engine.
Analyzes all timeframes 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 — Signal cards for one pair across all timeframes
/scan — Scan all pairs, show Grade B+ alerts
/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, Tuple
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 signal_aggregator import SignalAggregator, SignalReport
logger = logging.getLogger(__name__)
# ── Timeframes run on every analysis ───────────────────────────────
# Each TF is run as the primary; its confirmation TFs are auto-selected.
SCAN_TIMEFRAMES = ["M15", "H1", "H4", "D1"]
_TF_MAP = {
"M5": Timeframe.M5, "M15": Timeframe.M15,
"M30": Timeframe.M30, "H1": Timeframe.H1,
"H4": Timeframe.H4, "D1": Timeframe.D1,
}
_TF_CONFIRMATIONS = {
"M15": [Timeframe.M5, Timeframe.H1],
"H1": [Timeframe.M15, Timeframe.H4],
"H4": [Timeframe.H1, Timeframe.D1],
"D1": [Timeframe.H4, Timeframe.W1],
}
_SUPPORTED_PAIRS = set(MAJOR_PAIRS + CROSS_PAIRS)
_GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5}
# ── Formatting helpers ──────────────────────────────────────────────
def _grade_dot(quality: str) -> str:
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:
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))
def _build_alert(report: SignalReport, timeframe: str) -> str:
"""Build one trade alert card (HTML) for a given report + timeframe."""
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 (Grade B and above only) ─────────────────────────
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:
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"Confirmation TFs ({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)
# ══════════════════════════════════════════════════════════════════════
# Bot
# ══════════════════════════════════════════════════════════════════════
class ForexTelegramBot:
def __init__(self):
# One worker per timeframe so all TFs run concurrently
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.H4])
cfg.watch_pairs = list(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_all_tfs(
self, pairs: List[str]
) -> Dict[str, List[SignalReport]]:
"""Run all SCAN_TIMEFRAMES concurrently, return {tf: [reports]}."""
loop = asyncio.get_running_loop()
tasks = {
tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs))
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 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 = (
"🌐 Forex Signal Intelligence Platform\n\n"
"Commands:\n"
" /signal EURUSD — Signal cards for all timeframes\n"
" /scan — Scan all pairs, Grade B+ alerts\n"
" /help — This message\n\n"
f"Timeframes analyzed automatically: {tfs}\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"
"⚠️ Educational use only. Not financial advice."
)
await update.message.reply_text(text, parse_mode=ParseMode.HTML)
async def cmd_signal(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
/signal EURUSD
Runs the pair through all timeframes concurrently and sends
one trade alert card per timeframe.
"""
args = context.args or []
if not args:
await update.message.reply_text(
"Usage: /signal <pair>\n"
"Example: /signal EURUSD",
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)
synthetic_warning = ""
if raw not in _SUPPORTED_PAIRS:
synthetic_warning = (
"\n\n⚠️ Synthetic data warning: "
f"{_h(display)} is not on the Twelve Data free tier. "
"Prices and indicators are simulated — not live market data."
)
msg = await update.message.reply_text(
f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…"
)
try:
results_by_tf = await self._analyze_all_tfs([raw])
await msg.delete()
sent = 0
for tf in SCAN_TIMEFRAMES:
reports = results_by_tf.get(tf, [])
if not reports:
continue
alert = _build_alert(reports[0], tf) + synthetic_warning
await self._send_html(update, alert)
sent += 1
if sent == 0:
await update.message.reply_text(
f"❌ No data returned for {_h(display)}"
)
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.
For each pair, picks the highest-grade result across all TFs
and sends a card for Grade B and above.
"""
now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
tfs_str = " · ".join(SCAN_TIMEFRAMES)
msg = await update.message.reply_text(
f"🔄 Scanning all pairs across {_h(tfs_str)}…"
)
try:
results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS))
# For each pair pick the best grade across all TFs
# best_per_pair: {symbol: (tf, report)}
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 to 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))
if not actionable:
await msg.edit_text(
f"🔍 Scan complete — {now_str}\n\n"
"No Grade B+ signals found across any timeframe right now.",
parse_mode=ParseMode.HTML,
)
return
# Tally grades
grade_counts: Dict[str, int] = {}
for _, r in actionable:
grade_counts[r.signal_quality] = grade_counts.get(r.signal_quality, 0) + 1
grade_str = " ".join(f"{g}: {n}" for g, n in grade_counts.items())
summary = (
f"📊 Scan complete — {now_str}\n"
f"Timeframes: {_h(tfs_str)}\n"
f"Grade B+ signals: {len(actionable)} ({grade_str})\n\n"
"Best signal per pair ↓"
)
await msg.edit_text(summary, parse_mode=ParseMode.HTML)
for tf, r in actionable:
await self._send_html(update, _build_alert(r, tf))
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 cards 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()