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

Add live dashboard with 60s auto-refresh, disable breakout alerts

Browse Source 
Replace per-scan Telegram messages with a single dashboard message
that edits in-place every 60 seconds. Each pair shows all TFs
horizontally with short signal labels (Buy/Sell/Neutral). WebSocket
live prices feed into the dashboard. Level-breach alerts disabled
(WS still updates cache silently). Removed /stats command.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Harwinder
2026-05-13 18:34:44 -04:00
parent 6b1ffff66a
commit 439e179128
1 changed files with 158 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_bot.py
+155 -33
View File
@@ -95,6 +95,11 @@ _LEVEL_KEYS = (
"ema_trend",
)
# ── Live signal stats cache ──────────────────────────────────────────
# symbol → { tf_label → SignalReport }
_signal_stats: Dict[str, Dict[str, SignalReport]] = {}
_stats_updated: Optional[float] = None
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"},
@@ -302,6 +307,128 @@ def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
return "\n".join(lines)
def _update_signal_stats(reports: List[SignalReport], tf: Timeframe) -> None:
global _stats_updated
for r in reports:
if r.symbol not in _signal_stats:
_signal_stats[r.symbol] = {}
_signal_stats[r.symbol][tf.value] = r
_stats_updated = time.time()
_SHORT_SIGNAL = {
SignalStrength.STRONG_BUY: "Buy",
SignalStrength.BUY: "Buy",
SignalStrength.NEUTRAL: "Neutral",
SignalStrength.SELL: "Sell",
SignalStrength.STRONG_SELL: "Sell",
}
def _format_stats_dashboard() -> str:
if not _signal_stats:
return "📊 *Live Dashboard*\n\n_Waiting for first scan…_"
now = datetime.now(timezone.utc).strftime("%H:%M:%S UTC")
lines = [f"📊 *Live Dashboard* — {now}", ""]
tf_order = ["M15", "1h", "4h"]
for symbol in SCAN_PAIRS:
display = PAIR_INFO.get(symbol, {}).get("display", symbol)
pair_data = _signal_stats.get(symbol, {})
ws_price = _last_prices.get(symbol)
latest = None
for tf in reversed(tf_order):
if tf in pair_data:
latest = pair_data[tf]
break
price = ws_price or (latest.current_price if latest else 0)
if price < 10:
price_str = f"`{price:.5f}`"
else:
price_str = f"`{price:.2f}`"
regime = latest.market_regime.replace("_", " ").title() if latest else ""
tf_parts = []
for tf in tf_order:
r = pair_data.get(tf)
if not r:
tf_parts.append(f"{tf}: ⬜—")
continue
emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
sig = _SHORT_SIGNAL.get(r.final_signal, "Neutral")
tf_parts.append(f"{tf}: {emoji}{sig}")
lines.append(f"*{display}* {price_str} _{regime}_")
lines.append(" | ".join(tf_parts))
if latest and latest.trade_setup and latest.final_signal not in (SignalStrength.NEUTRAL,):
ts = latest.trade_setup
d = "📈" if ts.direction == "BUY" else "📉"
lines.append(
f"{d} {ts.direction} `{ts.entry_price:.3f}` "
f"SL `{ts.stop_loss:.3f}` TP `{ts.take_profit_1:.3f}` "
f"RR 1:{ts.risk_reward:.1f}"
)
lines.append("")
lines.append("_🔄 auto-refresh 60s_")
return "\n".join(lines)
# ── Live dashboard (edit-in-place) ───────────────────────────────────
_dashboard_msg_ids: Dict[int, int] = {}
_dashboard_task: Optional[asyncio.Task] = None
async def _send_dashboard(bot, chat_id: int) -> None:
text = _format_stats_dashboard()
try:
msg = await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
_dashboard_msg_ids[chat_id] = msg.message_id
except Exception:
logger.exception(f"Failed to send dashboard to {chat_id}")
async def _refresh_dashboard(bot) -> None:
text = _format_stats_dashboard()
for chat_id in list(_autoscan_chats):
msg_id = _dashboard_msg_ids.get(chat_id)
if not msg_id:
await _send_dashboard(bot, chat_id)
continue
try:
await bot.edit_message_text(
text, chat_id=chat_id, message_id=msg_id,
parse_mode=ParseMode.MARKDOWN,
)
except Exception as exc:
if "message is not modified" in str(exc).lower():
pass
else:
logger.warning(f"Dashboard edit failed, sending new: {exc}")
await _send_dashboard(bot, chat_id)
async def _dashboard_loop(bot) -> None:
try:
while _autoscan_chats:
await asyncio.sleep(60)
if not _autoscan_chats:
break
await _refresh_dashboard(bot)
except asyncio.CancelledError:
pass
finally:
logger.info("Dashboard refresh loop stopped")
# ── Keyboard helper ───────────────────────────────────────────────────
def scan_keyboard() -> InlineKeyboardMarkup:
@@ -381,23 +508,18 @@ async def _scan_timeframe(
return
_update_level_cache(reports)
_update_signal_stats(reports, tf)
if tf == Timeframe.M5:
logger.info("M5 scan: data analysis only — no Telegram broadcast")
return
await _refresh_dashboard(bot)
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
if not actionable:
logger.info(f"{tf.value} scan: no actionable signals — skipping Telegram message")
return
summary = _format_scan_summary(reports, tf)
for chat_id in list(_autoscan_chats):
try:
await bot.send_message(chat_id, summary, parse_mode=ParseMode.MARKDOWN)
for report in actionable:
try:
detail = format_signal_message(report)
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception:
@@ -451,26 +573,14 @@ async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, le
_scan_cooldowns[pair_symbol] = now
display = PAIR_INFO.get(pair_symbol, {}).get("display", ws_symbol)
label = level_name.replace("_", " ").title()
logger.info(f"Level breach: {display} crossed {label} at {price}")
logger.info(f"Level breach: {display} crossed {label} at {price} (alert disabled, cache updated)")
try:
report = await run_signal(pair_symbol, Timeframe.M15)
_update_level_cache([report])
text = (
f"🔔 *Level Break — {display}*\n"
f"Price crossed *{label}* at `{price}`\n"
f"━━━━━━━━━━━━━━━━━━━━\n"
+ format_signal_message(report)
)
except Exception as exc:
logger.exception(f"Breach scan failed for {display}")
text = f"🔔 *Level Break — {display}*\nPrice crossed {label} at `{price}`\n❌ Scan failed: {exc}"
for chat_id in list(_autoscan_chats):
try:
await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
_update_signal_stats([report], Timeframe.M15)
except Exception:
logger.exception(f"Failed to send breach alert to {chat_id}")
logger.exception(f"Breach scan failed for {display}")
async def _ws_monitor(bot) -> None:
@@ -549,7 +659,7 @@ _ws_task: Optional[asyncio.Task] = None
def _start_scheduler(bot) -> None:
global _ws_task
global _ws_task, _dashboard_task
alive = [t for t in _autoscan_tasks if not t.done()]
if not alive:
_autoscan_tasks.clear()
@@ -563,16 +673,24 @@ def _start_scheduler(bot) -> None:
_ws_task = asyncio.create_task(_ws_monitor(bot))
logger.info("WebSocket level-breach monitor started")
if _dashboard_task is None or _dashboard_task.done():
_dashboard_task = asyncio.create_task(_dashboard_loop(bot))
logger.info("Dashboard 60s refresh loop started")
def _stop_scheduler() -> None:
global _ws_task
global _ws_task, _dashboard_task
for t in _autoscan_tasks:
t.cancel()
_autoscan_tasks.clear()
if _ws_task and not _ws_task.done():
_ws_task.cancel()
_ws_task = None
logger.info("Candle-boundary scheduler and WebSocket monitor stopped")
if _dashboard_task and not _dashboard_task.done():
_dashboard_task.cancel()
_dashboard_task = None
_dashboard_msg_ids.clear()
logger.info("Scheduler, WebSocket, and dashboard loop stopped")
# ── Handlers ──────────────────────────────────────────────────────────
@@ -604,8 +722,8 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"/clearcache — Wipe data cache, force fresh fetch\n"
"/help — This message\n\n"
"*Auto-scan* fires at candle close: M5/M15/H1/H4.\n"
"*WebSocket* monitors live prices and sends Level Break Alerts "
"when price crosses swing S/R, Bollinger bands, or EMA200."
"Live dashboard auto-refreshes every 60s.\n"
"*WebSocket* monitors live prices and updates level cache."
)
await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN)
@@ -717,7 +835,8 @@ async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
await update.message.reply_text(
"✅ *Auto-scan enabled*\n\n"
f"Scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4 at candle close.\n"
f"📡 Live price monitoring active — level-break alerts enabled.\n"
f"📊 Live dashboard refreshes every 60s.\n"
f"📡 Live price monitoring active.\n"
f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
"Use /autoscan again to stop.",
@@ -774,13 +893,14 @@ async def _post_init(app: Application) -> None:
chat_id = int(CHAT_ID)
_autoscan_chats.add(chat_id)
logger.info("Running initial scan to populate level cache for WebSocket monitor...")
logger.info("Running initial scan to populate level cache and signal stats...")
try:
init_reports = []
for symbol in SCAN_PAIRS:
report = await run_signal(symbol, Timeframe.M15)
init_reports.append(report)
_update_level_cache(init_reports)
_update_signal_stats(init_reports, Timeframe.M15)
logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels")
except Exception:
logger.exception("Initial scan failed — WS will wait for first candle scan")
@@ -801,12 +921,14 @@ async def _post_init(app: Application) -> None:
chat_id,
f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n"
f"📡 Live price monitoring active — level-break alerts enabled.\n"
f"📊 Live dashboard refreshes every 60s.\n"
f"📡 Live price monitoring active.\n"
f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
f"/autoscan to stop | Manual buttons via /start",
f"/autoscan to stop",
parse_mode=ParseMode.MARKDOWN,
)
await _send_dashboard(app.bot, chat_id)
except Exception:
logger.exception("Failed to send boot message")