harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Combine all TF outputs into single message, add EST price timestamp

Browse Source 
/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>
This commit is contained in:
Harwinder
2026-05-04 17:39:21 -04:00
parent 0ad741670f
commit f053411343
1 changed files with 165 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_bot.py
+165 -138
View File
@@ -3,7 +3,7 @@
Forex Signal Intelligence Platform - Telegram Bot
==================================================
Trade alert frontend for the signal engine.
Analyzes all timeframes automatically — no TF parameter needed.
All timeframes are analyzed automatically — no TF parameter needed.
Setup:
set TELEGRAM_BOT_TOKEN=your_token_here (Windows CMD)
@@ -11,8 +11,8 @@ Setup:
python telegram_bot.py
Commands:
/signal EURUSD — Signal cards for one pair across all timeframes
/scan — Scan all pairs, show Grade B+ alerts
/signal EURUSD — Combined signal card across all timeframes
/scan — Scan all pairs, compact Grade B+ summary
/help — Command guide
"""
@@ -21,8 +21,9 @@ import html
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
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
@@ -36,8 +37,8 @@ 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.
# ── Config ──────────────────────────────────────────────────────────
SCAN_TIMEFRAMES = ["M15", "H1", "H4", "D1"]
_TF_MAP = {
@@ -54,11 +55,15 @@ _TF_CONFIRMATIONS = {
}
_SUPPORTED_PAIRS = set(MAJOR_PAIRS + CROSS_PAIRS)
_GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5}
_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 "🟢"
@@ -103,85 +108,145 @@ 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
# ── 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}")
# ── 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 (Grade B and above only) ─────────────────────────
if report.trade_setup:
ts = report.trade_setup
lines.append(f"<b>Entry:</b> {ts.entry_price:.5f}")
if synthetic:
lines.append(
f"<b>Take Profit:</b> {ts.take_profit_1:.5f}"
f" <i>(+{ts.reward_pips:.1f} pips)</i>"
"⚠️ <b>Note:</b> Not on Twelve Data free tier — "
"prices are <b>simulated</b> data."
)
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}")
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"<b>Stop Loss:</b> {ts.stop_loss:.5f}"
f" <i>(-{ts.risk_pips:.1f} pips)</i>"
f"{dot} <b>{tf} · {action} · "
f"Grade {_h(r.signal_quality)} · {r.final_confidence:.0f}%</b>"
)
lines.append(f"<b>Risk:Reward:</b> 1:{ts.risk_reward:.1f}")
else:
lines.append(f"<b>Current Price:</b> {report.current_price:.5f}")
lines.append(f"<b>Analysis:</b> {_h(report.action_recommendation)}")
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("")
# ── 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>Confirmation TFs ({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)
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)
@@ -191,7 +256,6 @@ def _build_alert(report: SignalReport, timeframe: str) -> str:
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"
)
@@ -228,7 +292,7 @@ class ForexTelegramBot:
@staticmethod
async def _send_html(update: Update, text: str):
"""Send HTML, splitting into chunks if over 4096 chars."""
"""Send HTML, splitting into chunks at newlines if over 4096 chars."""
LIMIT = 4000
chunks: List[str] = []
while text:
@@ -253,10 +317,10 @@ class ForexTelegramBot:
text = (
"🌐 <b>Forex Signal Intelligence Platform</b>\n\n"
"<b>Commands:</b>\n"
" /signal EURUSD — Signal cards for all timeframes\n"
" /scan — Scan all pairs, Grade B+ alerts\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 automatically:</b> <code>{tfs}</code>\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"
@@ -268,8 +332,8 @@ class ForexTelegramBot:
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.
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:
@@ -283,38 +347,25 @@ class ForexTelegramBot:
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⚠️ <b>Synthetic data warning:</b> "
f"{_h(display)} is not on the Twelve Data free tier. "
"Prices and indicators are <b>simulated</b> — not live market data."
)
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()
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)}"
)
await self._send_html(update, text)
except Exception as e:
logger.error(f"Signal error for {raw}: {e}", exc_info=True)
@@ -324,10 +375,8 @@ class ForexTelegramBot:
"""
/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.
Picks the best-grade result per pair, returns a single compact message.
"""
now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
tfs_str = " · ".join(SCAN_TIMEFRAMES)
msg = await update.message.reply_text(
@@ -335,9 +384,9 @@ class ForexTelegramBot:
)
try:
results_by_tf = await self._analyze_all_tfs(list(MAJOR_PAIRS))
now_str = _now_est()
# For each pair pick the best grade across all TFs
# best_per_pair: {symbol: (tf, report)}
# 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:
@@ -348,7 +397,7 @@ class ForexTelegramBot:
):
best_per_pair[r.symbol] = (tf, r)
# Filter to B+ and sort best-first
# Filter B+ and sort best-first
actionable = [
(tf, r)
for tf, r in best_per_pair.values()
@@ -356,30 +405,8 @@ class ForexTelegramBot:
]
actionable.sort(key=lambda x: _GRADE_RANK.get(x[1].signal_quality, 99))
if not actionable:
await msg.edit_text(
f"🔍 <b>Scan complete</b> — {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"📊 <b>Scan complete</b> — {now_str}\n"
f"Timeframes: <code>{_h(tfs_str)}</code>\n"
f"Grade B+ signals: <b>{len(actionable)}</b> ({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))
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)
@@ -399,7 +426,7 @@ class ForexTelegramBot:
async def _register_commands(app: Application) -> None:
await app.bot.set_my_commands([
BotCommand("signal", "Signal cards for one pair — /signal EURUSD"),
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"),
])