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

Harden Telegram bot, add EUR/GBP strategies, parallelize scans

Browse Source 
Telegram reliability: rate-limited sends (1 msg/s), dashboard lock,
WebSocket JSON/price error handling, markdown escaping, level cache
TTL, cooldown cleanup, breach task tracking, graceful shutdown.

Telegram speed: parallelize 3-pair scans with asyncio.gather(),
increase thread pool 2→4, parallelize boot-up initial scan.

Strategy: add EUR/USD and GBP/USD pullback-buy strategies with
per-pair configs, optimize XAU TP to 0.9×ATR, block hours 8/9
for XAU and hour 17 for GBP.

Backtest: add --max-hours flag to evaluate.py for time-exit support.
This commit is contained in:
harwinder
2026-05-13 21:07:49 -04:00
parent 67a9169eb3
commit cd69980f1f
4 changed files with 494 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_bot.py
+165 -58
View File
@@ -35,11 +35,12 @@ import asyncio
import json
import logging
import os
import signal
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import Dict, List, Optional
from typing import Dict, List, Optional, Set
import websockets
@@ -104,7 +105,7 @@ 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"},
{"cb": "scan_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"},
{"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"},
{"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"},
{"cb": "scan_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"},
{"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"},
{"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"},
@@ -128,7 +129,60 @@ AUTOSCAN_CONFIRMATIONS = {
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
}
_executor = ThreadPoolExecutor(max_workers=2)
_executor = ThreadPoolExecutor(max_workers=4)
# ── Telegram rate limiter (1 msg/sec per chat) ────────────────────────
_send_lock = asyncio.Lock()
_last_send_per_chat: Dict[int, float] = {}
_TELEGRAM_MIN_INTERVAL = 1.0
_TELEGRAM_SEND_TIMEOUT = 30
async def _rate_limited_send(bot, chat_id: int, text: str, **kwargs) -> Optional[object]:
async with _send_lock:
now = time.time()
last = _last_send_per_chat.get(chat_id, 0)
wait = _TELEGRAM_MIN_INTERVAL - (now - last)
if wait > 0:
await asyncio.sleep(wait)
try:
result = await asyncio.wait_for(
bot.send_message(chat_id, text, **kwargs),
timeout=_TELEGRAM_SEND_TIMEOUT,
)
_last_send_per_chat[chat_id] = time.time()
return result
except asyncio.TimeoutError:
logger.error(f"Telegram send_message timed out for chat {chat_id}")
return None
except Exception:
logger.exception(f"Telegram send_message failed for chat {chat_id}")
return None
async def _rate_limited_edit(bot, text: str, chat_id: int, message_id: int, **kwargs) -> bool:
try:
await asyncio.wait_for(
bot.edit_message_text(text, chat_id=chat_id, message_id=message_id, **kwargs),
timeout=_TELEGRAM_SEND_TIMEOUT,
)
return True
except asyncio.TimeoutError:
logger.error(f"Telegram edit_message timed out for chat {chat_id}")
return False
except Exception as exc:
if "message is not modified" not in str(exc).lower():
logger.warning(f"Telegram edit_message failed for chat {chat_id}: {exc}")
return False
# ── Markdown escaping ─────────────────────────────────────────────────
def _escape_md(text: str) -> str:
for ch in ("*", "_", "`", "[", "]"):
text = text.replace(ch, "\\" + ch)
return text
# ── Shared data provider (single cache across all scans) ─────────────
@@ -183,7 +237,7 @@ def _mtf_line(report: SignalReport) -> str:
def _indicator_line(name: str, ind) -> str:
tag = _SIGNAL_TAG.get(ind.signal, "Neutral")
short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description
return f"[{tag}] *{ind.name}*: {short}"
return f"[{tag}] *{_escape_md(ind.name)}*: {_escape_md(short)}"
def format_signal_message(report: SignalReport) -> str:
@@ -196,16 +250,18 @@ def format_signal_message(report: SignalReport) -> str:
lines: list[str] = []
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
regime_str = _escape_md(report.market_regime.replace("_", " ").title())
vol_str = _escape_md(str(report.volatility_regime))
lines += [
f"🔍 *{report.display_name} Signal Analysis* ({report.primary_analysis.timeframe if report.primary_analysis else ''})",
f"🕐 {now}",
"",
f"💰 Price: *{report.current_price:.5f}*" if report.current_price < 10 else f"💰 Price: *{report.current_price:.3f}*",
f"📊 Regime: {report.market_regime.replace('_', ' ').title()} | Vol: {report.volatility_regime}",
f"📊 Regime: {regime_str} | Vol: {vol_str}",
"",
"━━━━━━━━━━━━━━━━━━━━",
f"{emoji} *{label}* {qe}{report.signal_quality} | Score: {report.final_score:+.1f} | {report.final_confidence:.0f}% conf",
f"_{report.action_recommendation}_",
f"_{_escape_md(report.action_recommendation)}_",
f"{mtf_tag}",
]
@@ -223,7 +279,7 @@ def format_signal_message(report: SignalReport) -> str:
if report.warnings:
lines.append("")
for w in report.warnings:
lines.append(f"⚠️ {w}")
lines.append(f"⚠️ {_escape_md(w)}")
if report.trade_setup and sig not in (SignalStrength.NEUTRAL,):
ts = report.trade_setup
@@ -366,7 +422,7 @@ def _format_stats_dashboard() -> str:
sig = _SHORT_SIGNAL.get(r.final_signal, "Neutral")
tf_parts.append(f"{label}: {emoji}{sig}")
lines.append(f"*{display}* {price_str} _{regime}_")
lines.append(f"*{display}* {price_str} _{_escape_md(regime)}_")
lines.append(" | ".join(tf_parts))
if latest and latest.trade_setup and latest.final_signal not in (SignalStrength.NEUTRAL,):
@@ -390,40 +446,35 @@ _dashboard_msg_ids: Dict[int, int] = {}
_dashboard_task: Optional[asyncio.Task] = None
_last_dashboard_refresh: float = 0
_DASHBOARD_THROTTLE = 5
_dashboard_lock = asyncio.Lock()
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)
msg = await _rate_limited_send(bot, chat_id, text, parse_mode=ParseMode.MARKDOWN)
if msg:
_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, force: bool = False) -> None:
global _last_dashboard_refresh
now = time.time()
if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE:
return
_last_dashboard_refresh = now
async with _dashboard_lock:
now = time.time()
if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE:
return
_last_dashboard_refresh = now
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,
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
ok = await _rate_limited_edit(
bot, 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}")
if not ok:
await _send_dashboard(bot, chat_id)
@@ -484,7 +535,7 @@ async def run_signal(
# ── Candle-boundary scheduler ────────────────────────────────────────
_autoscan_chats: set[int] = set()
_autoscan_chats: Set[int] = set()
_autoscan_tasks: list[asyncio.Task] = []
@@ -507,13 +558,14 @@ async def _scan_timeframe(
confirmation_tfs: List[Timeframe],
) -> None:
"""Scan all pairs for one timeframe and broadcast results."""
tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
results = await asyncio.gather(*tasks, return_exceptions=True)
reports: List[SignalReport] = []
for symbol in SCAN_PAIRS:
try:
report = await run_signal(symbol, tf, confirmation_tfs)
reports.append(report)
except Exception:
logger.exception(f"Auto-scan failed for {symbol} {tf.value}")
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.exception(f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}")
else:
reports.append(result)
if not reports:
return
@@ -530,17 +582,17 @@ async def _scan_timeframe(
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
for chat_id in list(_autoscan_chats):
for report in actionable:
try:
detail = format_signal_message(report)
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception:
logger.exception(f"Failed to send to chat {chat_id}")
detail = format_signal_message(report)
await _rate_limited_send(bot, chat_id, detail, parse_mode=ParseMode.MARKDOWN)
# ── WebSocket level-breach monitor ──────────────────────────────────
_level_cache: Dict[str, Dict[str, float]] = {}
_level_cache_ts: Dict[str, float] = {}
_LEVEL_CACHE_TTL = 3600
_last_prices: Dict[str, float] = {}
_last_price_ts: Dict[str, float] = {}
_scan_cooldowns: Dict[str, float] = {}
_cfg = PlatformConfig()
@@ -548,8 +600,11 @@ _WS_API_KEY = os.environ.get("TWELVE_DATA_API_KEY") or _cfg.twelve_data_api_key
_WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}"
_WS_COOLDOWN = _cfg.ws_cooldown_seconds
_breach_tasks: Set[asyncio.Task] = set()
def _update_level_cache(reports: List[SignalReport]) -> None:
now = time.time()
for r in reports:
if r.primary_analysis and r.primary_analysis.key_levels:
levels = {
@@ -558,6 +613,7 @@ def _update_level_cache(reports: List[SignalReport]) -> None:
}
if levels:
_level_cache[r.symbol] = levels
_level_cache_ts[r.symbol] = now
def _check_level_breach(symbol: str, price: float) -> Optional[str]:
@@ -565,6 +621,10 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]:
if not levels:
return None
cache_age = time.time() - _level_cache_ts.get(symbol, 0)
if cache_age > _LEVEL_CACHE_TTL:
return None
last = _last_prices.get(symbol)
if last is None:
return None
@@ -576,6 +636,13 @@ def _check_level_breach(symbol: str, price: float) -> Optional[str]:
return None
def _cleanup_cooldowns() -> None:
now = time.time()
expired = [k for k, v in _scan_cooldowns.items() if now - v > _WS_COOLDOWN * 2]
for k in expired:
del _scan_cooldowns[k]
async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None:
now = time.time()
if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN:
@@ -611,7 +678,12 @@ async def _ws_monitor(bot) -> None:
if not _autoscan_chats:
break
msg = json.loads(raw)
try:
msg = json.loads(raw)
except json.JSONDecodeError:
logger.warning(f"WebSocket: malformed JSON, skipping")
continue
if msg.get("event") != "price":
continue
@@ -620,14 +692,24 @@ async def _ws_monitor(bot) -> None:
if not pair_symbol:
continue
price = float(msg["price"])
try:
price = float(msg["price"])
except (KeyError, ValueError, TypeError) as e:
logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}")
continue
breached = _check_level_breach(pair_symbol, price)
_last_prices[pair_symbol] = price
_last_price_ts[pair_symbol] = time.time()
if breached:
asyncio.create_task(
task = asyncio.create_task(
_handle_breach(bot, ws_symbol, pair_symbol, price, breached)
)
_breach_tasks.add(task)
task.add_done_callback(_breach_tasks.discard)
_cleanup_cooldowns()
except asyncio.CancelledError:
break
@@ -701,6 +783,9 @@ def _stop_scheduler() -> None:
_dashboard_task.cancel()
_dashboard_task = None
_dashboard_msg_ids.clear()
for t in list(_breach_tasks):
t.cancel()
_breach_tasks.clear()
logger.info("Scheduler, WebSocket, and dashboard loop stopped")
@@ -729,7 +814,7 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"/start — Show scan buttons\n"
"/autoscan — Toggle auto-scan on/off\n"
"/status — Bot status, WS connection, live prices\n"
"/cache — Show when each pair/TF was last fetched\n"
"/cache — Show data cache age per pair/TF\n"
"/clearcache — Wipe data cache, force fresh fetch\n"
"/help — This message\n\n"
"*Auto-scan* fires at candle close: M5/M15/H1/H4.\n"
@@ -759,19 +844,21 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
for ws_sym, pair in _WS_TO_PAIR.items():
price = _last_prices.get(pair)
display = PAIR_INFO.get(pair, {}).get("display", ws_sym)
age = time.time() - _last_price_ts.get(pair, 0)
if price is None:
lines.append(f" {display}: waiting...")
elif price < 10:
lines.append(f" {display}: `{price:.5f}`")
lines.append(f" {display}: `{price:.5f}` ({age:.0f}s ago)")
else:
lines.append(f" {display}: `{price:.3f}`")
lines.append(f" {display}: `{price:.3f}` ({age:.0f}s ago)")
else:
lines.append("\n_No live prices yet (WS not streaming)_")
cached_pairs = len(_level_cache)
cached_levels = sum(len(v) for v in _level_cache.values())
if cached_levels:
lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs")
oldest_age = time.time() - min(_level_cache_ts.values()) if _level_cache_ts else 0
lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs (oldest {oldest_age:.0f}s)")
else:
lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan")
@@ -782,6 +869,10 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
if active_cooldowns:
lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown")
active_breach = sum(1 for t in _breach_tasks if not t.done())
if active_breach:
lines.append(f"*Breach tasks:* {active_breach} running")
await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
@@ -906,13 +997,19 @@ async def _post_init(app: Application) -> None:
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")
tasks = [
asyncio.wait_for(run_signal(sym, Timeframe.M15), timeout=30)
for sym in SCAN_PAIRS
]
results = await asyncio.gather(*tasks, return_exceptions=True)
init_reports = [r for r in results if isinstance(r, SignalReport)]
for i, r in enumerate(results):
if isinstance(r, Exception):
logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}")
if init_reports:
_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")
@@ -928,8 +1025,8 @@ async def _post_init(app: Application) -> None:
)
try:
await app.bot.send_message(
chat_id,
await _rate_limited_send(
app.bot, chat_id,
f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n"
f"📊 Live dashboard refreshes every 60s.\n"
@@ -944,6 +1041,15 @@ async def _post_init(app: Application) -> None:
logger.exception("Failed to send boot message")
# ── Graceful shutdown ─────────────────────────────────────────────────
async def _post_shutdown(app: Application) -> None:
logger.info("Shutting down ForexBot2...")
_stop_scheduler()
_executor.shutdown(wait=False)
logger.info("Shutdown complete")
# ── Entry point ───────────────────────────────────────────────────────
def main() -> None:
@@ -956,6 +1062,7 @@ def main() -> None:
Application.builder()
.token(TOKEN)
.post_init(_post_init)
.post_shutdown(_post_shutdown)
.build()
)