Harwinder
f053411343
/signal: one combined message with all 4 TF sections stacked (────── dividers), entry/TP/SL always preserved, indicators per TF, MTF folded to one footer line, disclaimer once at bottom, news/warnings only when relevant /scan: single compact message — one header+trade line per pair, full entry/TP/SL kept, no indicator list (use /signal to drill in), danger events shown inline Both commands show price pull time as EST (e.g. 09:42 AM EST · Mon 04 May 2026) Scanning progress message no longer shows a single TF in parentheses Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
465 lines
18 KiB
Python
465 lines
18 KiB
Python
#!/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 signal_aggregator import SignalAggregator, SignalReport
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Config ──────────────────────────────────────────────────────────
|
|
|
|
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}
|
|
_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 _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))
|
|
|
|
|
|
# ── Message builders ────────────────────────────────────────────────
|
|
|
|
def _build_combined_signal(
|
|
results_by_tf: Dict[str, List[SignalReport]],
|
|
display: str,
|
|
now_str: str,
|
|
synthetic: bool = False,
|
|
) -> 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}")
|
|
|
|
if synthetic:
|
|
lines.append(
|
|
"⚠️ <b>Note:</b> Not on Twelve Data free tier — "
|
|
"prices are <b>simulated</b> data."
|
|
)
|
|
|
|
for tf in SCAN_TIMEFRAMES:
|
|
reports = results_by_tf.get(tf, [])
|
|
if not reports:
|
|
continue
|
|
r = reports[0]
|
|
|
|
dot = _grade_dot(r.signal_quality)
|
|
action = _signal_action(r.final_signal)
|
|
|
|
lines.append("")
|
|
lines.append("────────────────────────")
|
|
lines.append(
|
|
f"{dot} <b>{tf} · {action} · "
|
|
f"Grade {_h(r.signal_quality)} · {r.final_confidence:.0f}%</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("")
|
|
|
|
for tf, r in actionable:
|
|
dot = _grade_dot(r.signal_quality)
|
|
action = _signal_action(r.final_signal)
|
|
|
|
lines.append(
|
|
f"{dot} <b>{_h(r.display_name)}</b> · {action}"
|
|
f" · {_h(r.signal_quality)} · {tf} · {r.final_confidence:.0f}%"
|
|
)
|
|
|
|
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.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 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 / A+ — High confidence, consider entry\n"
|
|
" 🟡 B — Decent setup, wait for confirmation\n"
|
|
" 🔴 C / D — Low confidence, watch only\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 <pair></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)
|
|
synthetic = raw not in _SUPPORTED_PAIRS
|
|
|
|
msg = await update.message.reply_text(
|
|
f"🔄 Analyzing {_h(display)} across {_h(tfs_str)}…"
|
|
)
|
|
try:
|
|
results_by_tf = await self._analyze_all_tfs([raw])
|
|
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, synthetic)
|
|
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)}…"
|
|
)
|
|
try:
|
|
results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS))
|
|
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()
|