Harwinder
4aba243472
- Replace /scan+/pair commands with /signal (single pair) and /scan (all pairs, B+ only) - Implement exact visual template: grade dot (🟢/🟡/🔴), TRADE ALERT header, Entry/TP/SL/R:R block, full indicator list with value + condition - Grade dot enforced: A/A+=🟢, B=🟡, C/D/F=🔴 - /scan filters to minimum Grade B confidence before sending alerts - HTML parse mode with html.escape() on all dynamic content - Per-chat default timeframe state preserved Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
468 lines
18 KiB
Python
468 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Forex Signal Intelligence Platform - Telegram Bot
|
|
==================================================
|
|
Trade alert frontend for the signal engine.
|
|
|
|
Setup:
|
|
set TELEGRAM_BOT_TOKEN=your_token_here
|
|
python telegram_bot.py
|
|
|
|
Commands:
|
|
/signal EURUSD [TF] — Signal card for one pair
|
|
/scan [TF] — Scan all pairs, show Grade B+ alerts
|
|
/tf H1 — Set default timeframe
|
|
/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
|
|
|
|
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 maps ──────────────────────────────────────────────────
|
|
|
|
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 helpers ──────────────────────────────────────────────
|
|
|
|
def _grade_dot(quality: str) -> str:
|
|
"""Colored dot matching confidence grade per spec."""
|
|
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:
|
|
"""
|
|
Extract a short readable condition from the indicator description.
|
|
Prefer text after ' — ' (e.g. 'RSI 68.2 — Overbought' → 'Overbought').
|
|
Fall back to the signal enum name.
|
|
"""
|
|
desc = ind.description
|
|
if " — " in desc:
|
|
return desc.split(" — ", 1)[1].strip()
|
|
if " | " in desc:
|
|
# e.g. 'MACD bullish crossover | Hist: 0.00008' — take first clause
|
|
return desc.split(" | ", 1)[0].strip()
|
|
return ind.signal.name.replace("_", " ").title()
|
|
|
|
|
|
def _fmt_float(val: float) -> str:
|
|
"""Human-readable float with appropriate precision."""
|
|
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:
|
|
"""HTML-escape a string for safe use inside HTML parse mode."""
|
|
return html.escape(str(text))
|
|
|
|
|
|
def _build_alert(report: SignalReport, timeframe: str) -> str:
|
|
"""
|
|
Build a single trade alert card using the required visual template.
|
|
Returns an HTML-formatted string ready for Telegram.
|
|
|
|
Template:
|
|
{Dot} **TRADE ALERT: {ACTION} {PAIR}**
|
|
Grade / confidence / timeframe subtitle
|
|
|
|
**Entry:** {price}
|
|
**Take Profit:** {price} (+pips)
|
|
**Stop Loss:** {price} (-pips)
|
|
**Risk:Reward:** 1:2.0
|
|
|
|
**Technical Indicators:**
|
|
• {Name}: {Value} ({Condition})
|
|
...
|
|
|
|
Multi-Timeframe block
|
|
Warnings / events / news
|
|
Market context line
|
|
Disclaimer
|
|
"""
|
|
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} <b>TRADE ALERT: {action} {pair}</b>")
|
|
lines.append(f"<i>Grade: {grade} | Confidence: {conf:.0f}% | {timeframe}</i>")
|
|
lines.append("")
|
|
|
|
# ── Trade setup (only when grade B or above) ─────────────────────
|
|
if report.trade_setup:
|
|
ts = report.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}")
|
|
else:
|
|
# No actionable setup — show price and recommendation
|
|
lines.append(f"<b>Current Price:</b> {report.current_price:.5f}")
|
|
lines.append(f"<b>Analysis:</b> {_h(report.action_recommendation)}")
|
|
|
|
lines.append("")
|
|
|
|
# ── Technical Indicators ─────────────────────────────────────────
|
|
if report.primary_analysis and report.primary_analysis.indicators:
|
|
lines.append("<b>Technical Indicators:</b>")
|
|
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"<b>Multi-Timeframe ({agree_str}):</b>")
|
|
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"📅 <b>Event Risk:</b> {_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"<i>Regime: {_h(report.market_regime)} | "
|
|
f"Vol: {_h(report.volatility_regime)} | "
|
|
f"Score: {report.final_score:+.1f}</i>"
|
|
)
|
|
lines.append("<i>⚠️ Educational use only. Not financial advice.</i>")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ── 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")
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────
|
|
|
|
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: Optional[List[str]] = None) -> 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 or MAJOR_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(self, func, *args):
|
|
loop = asyncio.get_running_loop()
|
|
return await loop.run_in_executor(self._executor, func, *args)
|
|
|
|
@staticmethod
|
|
def _parse_tf(args: List[str]) -> Optional[str]:
|
|
for a in args:
|
|
if a.upper() in TF_MAP:
|
|
return a.upper()
|
|
return None
|
|
|
|
@staticmethod
|
|
async def _send_html(update: Update, text: str):
|
|
"""Send HTML message, splitting at newlines if over Telegram's 4096-char limit."""
|
|
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):
|
|
state = self._state(update.effective_chat.id)
|
|
text = (
|
|
"🌐 <b>Forex Signal Intelligence Platform</b>\n\n"
|
|
"<b>Commands:</b>\n"
|
|
" /signal EURUSD — Signal card for a pair\n"
|
|
" /signal EURUSD H4 — With specific timeframe\n"
|
|
" /scan — Scan all pairs (Grade B+ alerts only)\n"
|
|
" /scan H4 — Scan at a specific timeframe\n"
|
|
" /tf H1 — Set your 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 (default, intraday)\n"
|
|
" <code>H4 </code> — 4 hours (swing trading)\n"
|
|
" <code>D1 </code> — Daily (position trading)\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"
|
|
f"Your default timeframe: <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 <timeframe></b>\nOptions: <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_signal(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""
|
|
/signal EURUSD [TF]
|
|
Analyze a single pair and return the full trade alert card.
|
|
"""
|
|
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>/signal <pair> [timeframe]</b>\n"
|
|
"Example: <code>/signal 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.replace("=X", ""))
|
|
|
|
msg = await update.message.reply_text(f"🔄 Analyzing {_h(display)} ({tf})…")
|
|
try:
|
|
reports = await self._analyze(self._run_scan, tf, [raw])
|
|
if not reports:
|
|
await msg.edit_text(f"❌ No data returned for {_h(display)}")
|
|
return
|
|
|
|
r = reports[0]
|
|
alert = _build_alert(r, tf)
|
|
await msg.delete()
|
|
await self._send_html(update, alert)
|
|
|
|
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 [TF]
|
|
Analyze all major pairs and send a trade alert card for every
|
|
signal rated Grade B or above.
|
|
"""
|
|
chat_id = update.effective_chat.id
|
|
state = self._state(chat_id)
|
|
tf = self._parse_tf(context.args or []) or state.timeframe
|
|
|
|
now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
|
msg = await update.message.reply_text(f"🔄 Scanning all pairs ({tf})…")
|
|
|
|
try:
|
|
reports = await self._analyze(self._run_scan, tf, list(MAJOR_PAIRS))
|
|
|
|
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")]
|
|
|
|
if not actionable:
|
|
await msg.edit_text(
|
|
f"🔍 <b>Scan complete</b> — {now_str}\n\n"
|
|
"No Grade B+ signals found at this time.\n"
|
|
"Try a different timeframe with <code>/scan H4</code>.",
|
|
parse_mode=ParseMode.HTML,
|
|
)
|
|
return
|
|
|
|
# Summary header
|
|
grades = {"A+": 0, "A": 0, "B": 0}
|
|
for r in actionable:
|
|
grades[r.signal_quality] = grades.get(r.signal_quality, 0) + 1
|
|
|
|
grade_str = " ".join(
|
|
f"{g}: {count}" for g, count in grades.items() if count
|
|
)
|
|
summary = (
|
|
f"📊 <b>Scan complete</b> — {now_str}\n"
|
|
f"Timeframe: <b>{tf}</b>\n"
|
|
f"Grade B+ signals: <b>{len(actionable)}</b> ({grade_str})\n\n"
|
|
"Trade alerts follow ↓"
|
|
)
|
|
await msg.edit_text(summary, parse_mode=ParseMode.HTML)
|
|
|
|
# One card per actionable signal
|
|
for r in actionable:
|
|
alert = _build_alert(r, tf)
|
|
await self._send_html(update, alert)
|
|
|
|
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: 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", "Trade alert for one pair — /signal EURUSD"),
|
|
BotCommand("scan", "Scan all pairs, Grade B+ alerts — /scan H4"),
|
|
BotCommand("tf", "Set default timeframe — /tf H4"),
|
|
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("tf", self.cmd_tf))
|
|
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()
|