Harwinder
67a9169eb3
Timeframe keys stored as enum values (15min/1h/4h) but dashboard looked for M15/1h/4h — GBP/USD showed no price or signal text. Now uses Timeframe enum directly with display labels. Reduced dashboard refresh from 60s to 5s with throttle guard.
976 lines
35 KiB
Python
Executable File
976 lines
35 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
ForexBot2 — Telegram Bot
|
||
========================
|
||
Async Telegram interface with candle-boundary auto-scanner and
|
||
real-time WebSocket level-breach monitor.
|
||
|
||
Manual scan: 4 pairs × 3 timeframes (M15/H1/H4) via inline buttons.
|
||
Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set.
|
||
/autoscan toggles for additional chats or to stop.
|
||
M5 fires at :00/:05/:10/… + 10s buffer (data only, no signals).
|
||
M15 fires at :00/:15/:30/:45 + 10s buffer.
|
||
H1 fires at :00 + 10s buffer.
|
||
H4 fires at :00 every 4 hours + 10s buffer.
|
||
WebSocket: Streams live ticks from TwelveData for EUR/USD & XAU/USD
|
||
(Grow plan supports only these 2 WS symbols).
|
||
When price crosses a key structure level (swing S/R, BB,
|
||
EMA200) cached from the last boundary scan, triggers an
|
||
immediate M15 scan and sends a Level Break Alert.
|
||
5-minute cooldown per pair prevents spam.
|
||
GBP/USD & USD/JPY are monitored via REST scans only.
|
||
|
||
Plan: TwelveData Grow (55 credits/min, 8 trial WS credits).
|
||
Credit budget (worst case, 4-hour boundary):
|
||
Unique fetches: 4 pairs × ~6 TFs = ~24 calls (rest cached) → fits in 55/min.
|
||
WebSocket: 2 symbols = 2 credits (one-time on connect).
|
||
Level breach scans: ~3 REST calls each, ~10/hour worst case.
|
||
|
||
Usage:
|
||
python telegram_bot.py
|
||
TELEGRAM_TOKEN=<token> TELEGRAM_CHAT_ID=<id> python telegram_bot.py
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Optional
|
||
|
||
import websockets
|
||
|
||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||
from telegram.constants import ParseMode
|
||
from telegram.ext import (
|
||
Application,
|
||
CallbackQueryHandler,
|
||
CommandHandler,
|
||
ContextTypes,
|
||
)
|
||
|
||
from config import (
|
||
PlatformConfig, SignalStrength, Timeframe,
|
||
EURUSD, GBPUSD, XAUUSD, PAIR_INFO,
|
||
)
|
||
from data_provider import ForexDataProvider
|
||
from signal_aggregator import SignalAggregator, SignalReport
|
||
|
||
# ── Force UTF-8 on Windows ────────────────────────────────────────────
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
logging.basicConfig(
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
level=logging.INFO,
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||
logger = logging.getLogger("telegram_bot")
|
||
|
||
TOKEN = os.environ.get("TELEGRAM_TOKEN", "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE")
|
||
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022")
|
||
|
||
# ── Pairs & buttons ───────────────────────────────────────────────────
|
||
|
||
SCAN_PAIRS = [EURUSD, GBPUSD, XAUUSD]
|
||
|
||
CORRELATION_GROUPS = {
|
||
"usd_negative": {"pairs": {EURUSD, GBPUSD}, "label": "USD-weakness"},
|
||
}
|
||
|
||
_WS_TO_PAIR = {
|
||
"EUR/USD": EURUSD,
|
||
"XAU/USD": XAUUSD,
|
||
}
|
||
_WS_SYMBOLS = list(_WS_TO_PAIR.keys())
|
||
|
||
_LEVEL_KEYS = (
|
||
"swing_high_1", "swing_high_2", "swing_high_3",
|
||
"swing_low_1", "swing_low_2", "swing_low_3",
|
||
"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"},
|
||
{"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_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"},
|
||
{"cb": "scan_xauusd_h1", "symbol": XAUUSD, "tf": Timeframe.H1, "label": "XAU/USD 1H"},
|
||
{"cb": "scan_xauusd_h4", "symbol": XAUUSD, "tf": Timeframe.H4, "label": "XAU/USD 4H"},
|
||
]
|
||
|
||
SCAN_LOOKUP = {b["cb"]: b for b in SCAN_BUTTONS}
|
||
|
||
TF_CONFIRMATIONS = {
|
||
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
|
||
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
|
||
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
|
||
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
|
||
}
|
||
|
||
AUTOSCAN_CONFIRMATIONS = {
|
||
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
|
||
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
|
||
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
|
||
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
|
||
}
|
||
|
||
_executor = ThreadPoolExecutor(max_workers=2)
|
||
|
||
|
||
# ── Shared data provider (single cache across all scans) ─────────────
|
||
|
||
_shared_dp: Optional[ForexDataProvider] = None
|
||
|
||
|
||
def _get_shared_dp() -> ForexDataProvider:
|
||
global _shared_dp
|
||
if _shared_dp is None:
|
||
_shared_dp = ForexDataProvider(PlatformConfig())
|
||
return _shared_dp
|
||
|
||
|
||
# ── Signal formatting ─────────────────────────────────────────────────
|
||
|
||
_SIGNAL_EMOJI = {
|
||
SignalStrength.STRONG_BUY: "🟢",
|
||
SignalStrength.BUY: "🟩",
|
||
SignalStrength.NEUTRAL: "⬜",
|
||
SignalStrength.SELL: "🟥",
|
||
SignalStrength.STRONG_SELL: "🔴",
|
||
}
|
||
|
||
_QUALITY_EMOJI = {
|
||
"A+": "🏆", "A": "🥇", "B": "🥈", "C": "🥉", "D": "⚠️", "F": "❌",
|
||
}
|
||
|
||
_SIGNAL_TAG = {
|
||
SignalStrength.STRONG_BUY: "Strong Buy",
|
||
SignalStrength.BUY: "Bullish",
|
||
SignalStrength.NEUTRAL: "Neutral",
|
||
SignalStrength.SELL: "Bearish",
|
||
SignalStrength.STRONG_SELL: "Strong Sell",
|
||
}
|
||
|
||
|
||
def _signal_label(sig: SignalStrength) -> str:
|
||
return sig.name.replace("_", " ")
|
||
|
||
|
||
def _mtf_line(report: SignalReport) -> str:
|
||
if not report.mtf_signals:
|
||
return ""
|
||
parts = []
|
||
for tf, sig in report.mtf_signals.items():
|
||
score = report.mtf_scores.get(tf, 0)
|
||
parts.append(f"{tf}: {_signal_label(sig)} {score:+.1f}")
|
||
return " | ".join(parts)
|
||
|
||
|
||
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}"
|
||
|
||
|
||
def format_signal_message(report: SignalReport) -> str:
|
||
sig = report.final_signal
|
||
emoji = _SIGNAL_EMOJI.get(sig, "⬜")
|
||
qe = _QUALITY_EMOJI.get(report.signal_quality, "")
|
||
label = _signal_label(sig)
|
||
mtf_tag = "✅ MTF aligned" if report.mtf_agreement else "❌ MTF diverged"
|
||
|
||
lines: list[str] = []
|
||
|
||
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
||
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"{emoji} *{label}* {qe}{report.signal_quality} | Score: {report.final_score:+.1f} | {report.final_confidence:.0f}% conf",
|
||
f"_{report.action_recommendation}_",
|
||
f"{mtf_tag}",
|
||
]
|
||
|
||
mtf = _mtf_line(report)
|
||
if mtf:
|
||
lines += ["", f"📈 *MTF*: {mtf}"]
|
||
|
||
if report.primary_analysis:
|
||
lines += ["", "━━━━━━━━━━━━━━━━━━━━", "📋 *Indicators*"]
|
||
for name, ind in report.primary_analysis.indicators.items():
|
||
if name.startswith("_") or ind.weight == 0:
|
||
continue
|
||
lines.append(_indicator_line(name, ind))
|
||
|
||
if report.warnings:
|
||
lines.append("")
|
||
for w in report.warnings:
|
||
lines.append(f"⚠️ {w}")
|
||
|
||
if report.trade_setup and sig not in (SignalStrength.NEUTRAL,):
|
||
ts = report.trade_setup
|
||
dir_emoji = "📈" if ts.direction == "BUY" else "📉"
|
||
lines += [
|
||
"",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
f"🚀 *TRADE SETUP*",
|
||
f"{dir_emoji} Direction: *{ts.direction}*",
|
||
f"▶ Entry: `{ts.entry_price:.3f}`",
|
||
f"🛑 SL: `{ts.stop_loss:.3f}` ({ts.risk_pips:.1f} pips)",
|
||
f"🎯 TP1: `{ts.take_profit_1:.3f}` ({ts.reward_pips:.1f} pips)",
|
||
]
|
||
if ts.take_profit_2:
|
||
lines.append(f"🎯 TP2: `{ts.take_profit_2:.3f}`")
|
||
lines.append(f"⚖️ RR: 1:{ts.risk_reward:.1f}")
|
||
|
||
lines += ["", "━━━━━━━━━━━━━━━━━━━━",
|
||
"_Educational / research use only._"]
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ── Correlation filter ───────────────────────────────────────────────
|
||
|
||
def _detect_correlated_exposure(reports: List[SignalReport]) -> List[str]:
|
||
warnings = []
|
||
for group_info in CORRELATION_GROUPS.values():
|
||
group_pairs = group_info["pairs"]
|
||
group_label = group_info["label"]
|
||
active = [
|
||
r for r in reports
|
||
if r.symbol in group_pairs
|
||
and r.final_signal not in (SignalStrength.NEUTRAL,)
|
||
]
|
||
if len(active) >= 2:
|
||
directions = set()
|
||
for r in active:
|
||
if r.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
|
||
directions.add("BUY")
|
||
else:
|
||
directions.add("SELL")
|
||
if len(directions) == 1:
|
||
names = ", ".join(r.display_name for r in active)
|
||
warnings.append(
|
||
f"⚠️ Same exposure ({group_label}): {names} both {directions.pop()}"
|
||
)
|
||
return warnings
|
||
|
||
|
||
# ── Auto-scan summary format ─────────────────────────────────────────
|
||
|
||
def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
|
||
now = datetime.now(timezone.utc).strftime("%H:%M UTC")
|
||
lines = [
|
||
f"📊 *{tf.value} Auto-Scan* | {now}",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
]
|
||
for r in reports:
|
||
emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
|
||
sig = _signal_label(r.final_signal)
|
||
mtf = "✅" if r.mtf_agreement else "❌"
|
||
lines.append(
|
||
f"{emoji} *{r.display_name}* {sig} "
|
||
f"{_QUALITY_EMOJI.get(r.signal_quality, '')}{r.signal_quality} "
|
||
f"{r.final_score:+.1f} {r.final_confidence:.0f}% {mtf}MTF"
|
||
)
|
||
lines.append("━━━━━━━━━━━━━━━━━━━━")
|
||
|
||
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
|
||
if actionable:
|
||
names = ", ".join(r.display_name for r in actionable)
|
||
lines.append(f"🔔 Actionable: {names}")
|
||
else:
|
||
lines.append("No actionable signals this cycle.")
|
||
|
||
corr_warnings = _detect_correlated_exposure(reports)
|
||
for w in corr_warnings:
|
||
lines.append(w)
|
||
|
||
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",
|
||
}
|
||
|
||
|
||
_TF_DISPLAY = {"5min": "M5", "15min": "M15", "1h": "H1", "4h": "H4", "1d": "D1"}
|
||
_DASH_TFS = [Timeframe.M15, Timeframe.H1, Timeframe.H4]
|
||
|
||
|
||
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}", ""]
|
||
|
||
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(_DASH_TFS):
|
||
if tf.value in pair_data:
|
||
latest = pair_data[tf.value]
|
||
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 _DASH_TFS:
|
||
label = _TF_DISPLAY.get(tf.value, tf.value)
|
||
r = pair_data.get(tf.value)
|
||
if not r:
|
||
tf_parts.append(f"{label}: ⬜—")
|
||
continue
|
||
emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
|
||
sig = _SHORT_SIGNAL.get(r.final_signal, "Neutral")
|
||
tf_parts.append(f"{label}: {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("_🔄 live refresh_")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ── Live dashboard (edit-in-place) ───────────────────────────────────
|
||
|
||
_dashboard_msg_ids: Dict[int, int] = {}
|
||
_dashboard_task: Optional[asyncio.Task] = None
|
||
_last_dashboard_refresh: float = 0
|
||
_DASHBOARD_THROTTLE = 5
|
||
|
||
|
||
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, 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
|
||
|
||
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(5)
|
||
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:
|
||
rows = []
|
||
for i in range(0, len(SCAN_BUTTONS), 4):
|
||
rows.append([
|
||
InlineKeyboardButton(b["label"], callback_data=b["cb"])
|
||
for b in SCAN_BUTTONS[i:i + 4]
|
||
])
|
||
return InlineKeyboardMarkup(rows)
|
||
|
||
|
||
# ── Signal runner (shared data provider for cache reuse) ──────────────
|
||
|
||
def _run_signal_sync(
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
confirmation_tfs: Optional[List[Timeframe]] = None,
|
||
) -> SignalReport:
|
||
if confirmation_tfs is None:
|
||
confirmation_tfs = TF_CONFIRMATIONS.get(
|
||
timeframe, [Timeframe.M15, Timeframe.H4]
|
||
)
|
||
config = PlatformConfig()
|
||
config.primary_timeframe = timeframe
|
||
config.confirmation_timeframes = confirmation_tfs
|
||
aggregator = SignalAggregator(config)
|
||
aggregator.data_provider = _get_shared_dp()
|
||
return aggregator.analyze_pair(symbol=symbol)
|
||
|
||
|
||
async def run_signal(
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
confirmation_tfs: Optional[List[Timeframe]] = None,
|
||
) -> SignalReport:
|
||
loop = asyncio.get_event_loop()
|
||
return await loop.run_in_executor(
|
||
_executor, _run_signal_sync, symbol, timeframe, confirmation_tfs
|
||
)
|
||
|
||
|
||
# ── Candle-boundary scheduler ────────────────────────────────────────
|
||
|
||
_autoscan_chats: set[int] = set()
|
||
_autoscan_tasks: list[asyncio.Task] = []
|
||
|
||
|
||
def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
||
"""Seconds until the next candle close + buffer."""
|
||
now = datetime.now(timezone.utc)
|
||
total_minutes = now.hour * 60 + now.minute
|
||
elapsed_minutes = total_minutes % interval_minutes
|
||
elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
|
||
interval_seconds = interval_minutes * 60
|
||
remaining = interval_seconds - elapsed_seconds + buffer
|
||
if remaining <= buffer:
|
||
remaining += interval_seconds
|
||
return remaining
|
||
|
||
|
||
async def _scan_timeframe(
|
||
bot,
|
||
tf: Timeframe,
|
||
confirmation_tfs: List[Timeframe],
|
||
) -> None:
|
||
"""Scan all pairs for one timeframe and broadcast results."""
|
||
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}")
|
||
|
||
if not reports:
|
||
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")]
|
||
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}")
|
||
|
||
|
||
# ── WebSocket level-breach monitor ──────────────────────────────────
|
||
|
||
_level_cache: Dict[str, Dict[str, float]] = {}
|
||
_last_prices: Dict[str, float] = {}
|
||
_scan_cooldowns: Dict[str, float] = {}
|
||
|
||
_cfg = PlatformConfig()
|
||
_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
|
||
|
||
|
||
def _update_level_cache(reports: List[SignalReport]) -> None:
|
||
for r in reports:
|
||
if r.primary_analysis and r.primary_analysis.key_levels:
|
||
levels = {
|
||
k: v for k, v in r.primary_analysis.key_levels.items()
|
||
if k in _LEVEL_KEYS
|
||
}
|
||
if levels:
|
||
_level_cache[r.symbol] = levels
|
||
|
||
|
||
def _check_level_breach(symbol: str, price: float) -> Optional[str]:
|
||
levels = _level_cache.get(symbol)
|
||
if not levels:
|
||
return None
|
||
|
||
last = _last_prices.get(symbol)
|
||
if last is None:
|
||
return None
|
||
|
||
for name, level in levels.items():
|
||
if (last < level <= price) or (last > level >= price):
|
||
return name
|
||
|
||
return None
|
||
|
||
|
||
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:
|
||
return
|
||
|
||
_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} (alert disabled, cache updated)")
|
||
|
||
try:
|
||
report = await run_signal(pair_symbol, Timeframe.M15)
|
||
_update_level_cache([report])
|
||
_update_signal_stats([report], Timeframe.M15)
|
||
except Exception:
|
||
logger.exception(f"Breach scan failed for {display}")
|
||
|
||
|
||
async def _ws_monitor(bot) -> None:
|
||
backoff = 1
|
||
while _autoscan_chats:
|
||
try:
|
||
async with websockets.connect(_WS_URL) as ws:
|
||
subscribe = {
|
||
"action": "subscribe",
|
||
"params": {"symbols": ",".join(_WS_SYMBOLS)},
|
||
}
|
||
await ws.send(json.dumps(subscribe))
|
||
logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols")
|
||
backoff = 1
|
||
|
||
async for raw in ws:
|
||
if not _autoscan_chats:
|
||
break
|
||
|
||
msg = json.loads(raw)
|
||
if msg.get("event") != "price":
|
||
continue
|
||
|
||
ws_symbol = msg.get("symbol")
|
||
pair_symbol = _WS_TO_PAIR.get(ws_symbol)
|
||
if not pair_symbol:
|
||
continue
|
||
|
||
price = float(msg["price"])
|
||
breached = _check_level_breach(pair_symbol, price)
|
||
_last_prices[pair_symbol] = price
|
||
|
||
if breached:
|
||
asyncio.create_task(
|
||
_handle_breach(bot, ws_symbol, pair_symbol, price, breached)
|
||
)
|
||
|
||
except asyncio.CancelledError:
|
||
break
|
||
except Exception:
|
||
logger.exception(f"WebSocket error, reconnecting in {backoff}s")
|
||
await asyncio.sleep(backoff)
|
||
backoff = min(backoff * 2, 60)
|
||
|
||
logger.info("WebSocket monitor stopped")
|
||
|
||
|
||
async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
|
||
label = tf.value
|
||
try:
|
||
while _autoscan_chats:
|
||
wait = _seconds_until_boundary(interval_minutes, buffer=10)
|
||
logger.info(f"{label} auto-scan: next in {wait:.0f}s")
|
||
await asyncio.sleep(wait)
|
||
if not _autoscan_chats:
|
||
break
|
||
logger.info(f"{label} auto-scan: firing")
|
||
await _scan_timeframe(
|
||
bot, tf, AUTOSCAN_CONFIRMATIONS[tf]
|
||
)
|
||
except asyncio.CancelledError:
|
||
pass
|
||
finally:
|
||
logger.info(f"{label} auto-scan loop stopped")
|
||
|
||
|
||
_AUTOSCAN_LOOPS = [
|
||
(Timeframe.M5, 5),
|
||
(Timeframe.M15, 15),
|
||
(Timeframe.H1, 60),
|
||
(Timeframe.H4, 240),
|
||
]
|
||
|
||
|
||
_ws_task: Optional[asyncio.Task] = None
|
||
|
||
|
||
def _start_scheduler(bot) -> None:
|
||
global _ws_task, _dashboard_task
|
||
alive = [t for t in _autoscan_tasks if not t.done()]
|
||
if not alive:
|
||
_autoscan_tasks.clear()
|
||
for tf, interval in _AUTOSCAN_LOOPS:
|
||
_autoscan_tasks.append(
|
||
asyncio.create_task(_candle_loop(bot, tf, interval))
|
||
)
|
||
logger.info("Candle-boundary scheduler started (M5/M15/H1/H4)")
|
||
|
||
if _ws_task is None or _ws_task.done():
|
||
_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, _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
|
||
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 ──────────────────────────────────────────────────────────
|
||
|
||
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
scanning = "running" if _autoscan_chats else "off"
|
||
text = (
|
||
"👋 *ForexBot2 — Signal Intelligence*\n\n"
|
||
"Pick a pair and timeframe to scan.\n"
|
||
"Signals include a full trade setup with entry, SL and TP.\n\n"
|
||
f"Auto-scan: *{scanning}* — /autoscan to toggle\n"
|
||
"/status — Bot & WebSocket status\n"
|
||
"/help — All commands"
|
||
)
|
||
await update.message.reply_text(
|
||
text,
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
reply_markup=scan_keyboard(),
|
||
)
|
||
|
||
|
||
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
text = (
|
||
"📖 *ForexBot2 Commands*\n\n"
|
||
"/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"
|
||
"/clearcache — Wipe data cache, force fresh fetch\n"
|
||
"/help — This message\n\n"
|
||
"*Auto-scan* fires at candle close: M5/M15/H1/H4.\n"
|
||
"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)
|
||
|
||
|
||
async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
lines = ["📊 *ForexBot2 Status*\n"]
|
||
|
||
if _autoscan_chats:
|
||
lines.append(f"✅ Auto-scan: *running* ({len(_autoscan_chats)} chat)")
|
||
tasks_alive = sum(1 for t in _autoscan_tasks if not t.done())
|
||
lines.append(f" Candle loops: {tasks_alive}/{len(_AUTOSCAN_LOOPS)}")
|
||
else:
|
||
lines.append("⏹ Auto-scan: *stopped*")
|
||
|
||
if _ws_task and not _ws_task.done():
|
||
lines.append("📡 WebSocket: *connected*")
|
||
else:
|
||
lines.append("❌ WebSocket: *disconnected*")
|
||
|
||
if _last_prices:
|
||
lines.append("\n*Live Prices (WebSocket):*")
|
||
for ws_sym, pair in _WS_TO_PAIR.items():
|
||
price = _last_prices.get(pair)
|
||
display = PAIR_INFO.get(pair, {}).get("display", ws_sym)
|
||
if price is None:
|
||
lines.append(f" {display}: waiting...")
|
||
elif price < 10:
|
||
lines.append(f" {display}: `{price:.5f}`")
|
||
else:
|
||
lines.append(f" {display}: `{price:.3f}`")
|
||
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")
|
||
else:
|
||
lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan")
|
||
|
||
active_cooldowns = sum(
|
||
1 for v in _scan_cooldowns.values()
|
||
if time.time() - v < _WS_COOLDOWN
|
||
)
|
||
if active_cooldowns:
|
||
lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown")
|
||
|
||
await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
|
||
|
||
|
||
async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
dp = _get_shared_dp()
|
||
if not dp._cache_timestamps:
|
||
await update.message.reply_text("📭 Cache is empty — no data fetched yet.")
|
||
return
|
||
|
||
now = time.time()
|
||
lines = ["📦 *Data Cache Status*\n"]
|
||
|
||
for pair in SCAN_PAIRS:
|
||
display = PAIR_INFO.get(pair, {}).get("display", pair)
|
||
pair_lines = []
|
||
for tf in [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4, Timeframe.D1]:
|
||
cache_key = f"{pair}_{tf.value}"
|
||
ts = dp._cache_timestamps.get(cache_key)
|
||
if ts is None:
|
||
continue
|
||
age_min = (now - ts) / 60
|
||
df = dp._cache.get(cache_key)
|
||
last_bar = df.index[-1].strftime("%H:%M") if df is not None and not df.empty else "?"
|
||
fresh = "✅" if age_min < dp.CANDLE_SECONDS.get(tf, 3600) / 60 else "⚠️"
|
||
pair_lines.append(f" {tf.value:<5} {fresh} fetched `{age_min:.0f}m` ago | last bar `{last_bar}`")
|
||
if pair_lines:
|
||
lines.append(f"*{display}*")
|
||
lines.extend(pair_lines)
|
||
|
||
await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
|
||
|
||
|
||
async def cmd_clearcache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
dp = _get_shared_dp()
|
||
count = dp.clear_cache()
|
||
await update.message.reply_text(
|
||
f"🗑 Cache cleared ({count} entries wiped).\nNext scan will fetch fresh data from API.",
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
)
|
||
|
||
|
||
async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
chat_id = update.effective_chat.id
|
||
|
||
if chat_id in _autoscan_chats:
|
||
_autoscan_chats.discard(chat_id)
|
||
if not _autoscan_chats:
|
||
_stop_scheduler()
|
||
await update.message.reply_text(
|
||
"⏹ Auto-scan stopped.",
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
)
|
||
return
|
||
|
||
_autoscan_chats.add(chat_id)
|
||
_start_scheduler(context.bot)
|
||
|
||
next_m5 = _seconds_until_boundary(5, buffer=10)
|
||
next_m15 = _seconds_until_boundary(15, buffer=10)
|
||
next_h1 = _seconds_until_boundary(60, buffer=10)
|
||
next_h4 = _seconds_until_boundary(240, buffer=10)
|
||
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 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.",
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
)
|
||
|
||
|
||
async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
query = update.callback_query
|
||
await query.answer()
|
||
|
||
btn = SCAN_LOOKUP.get(query.data)
|
||
if not btn:
|
||
return
|
||
|
||
pair_display = btn["label"]
|
||
await query.edit_message_text(
|
||
f"⏳ Scanning {pair_display}…",
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
)
|
||
|
||
try:
|
||
report = await run_signal(btn["symbol"], btn["tf"])
|
||
text = format_signal_message(report)
|
||
except Exception as exc:
|
||
logger.exception("Signal scan failed")
|
||
text = f"❌ Scan failed: {exc}"
|
||
|
||
await query.edit_message_text(
|
||
text,
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
reply_markup=scan_keyboard(),
|
||
)
|
||
|
||
|
||
# ── Boot-time auto-start ──────────────────────────────────────────────
|
||
|
||
async def _post_init(app: Application) -> None:
|
||
"""Register command menu and start scheduler on boot."""
|
||
await app.bot.set_my_commands([
|
||
("start", "Show scan buttons"),
|
||
("autoscan", "Toggle auto-scan on/off"),
|
||
("status", "Bot status, WS connection, live prices"),
|
||
("cache", "Show data cache age per pair/TF"),
|
||
("clearcache", "Wipe data cache, force fresh fetch"),
|
||
("help", "All commands"),
|
||
])
|
||
|
||
if not CHAT_ID:
|
||
logger.info("TELEGRAM_CHAT_ID not set — auto-scan disabled on boot. "
|
||
"Use /autoscan in chat to enable.")
|
||
return
|
||
|
||
chat_id = int(CHAT_ID)
|
||
_autoscan_chats.add(chat_id)
|
||
|
||
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")
|
||
|
||
_start_scheduler(app.bot)
|
||
|
||
next_m5 = _seconds_until_boundary(5, buffer=10)
|
||
next_m15 = _seconds_until_boundary(15, buffer=10)
|
||
next_h1 = _seconds_until_boundary(60, buffer=10)
|
||
next_h4 = _seconds_until_boundary(240, buffer=10)
|
||
logger.info(
|
||
f"Auto-scan started for chat {chat_id} — "
|
||
f"M5 {next_m5:.0f}s, M15 {next_m15:.0f}s, H1 {next_h1:.0f}s, H4 {next_h4:.0f}s"
|
||
)
|
||
|
||
try:
|
||
await app.bot.send_message(
|
||
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"
|
||
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",
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
)
|
||
await _send_dashboard(app.bot, chat_id)
|
||
except Exception:
|
||
logger.exception("Failed to send boot message")
|
||
|
||
|
||
# ── Entry point ───────────────────────────────────────────────────────
|
||
|
||
def main() -> None:
|
||
if not TOKEN:
|
||
sys.exit("TELEGRAM_TOKEN not set")
|
||
|
||
logger.info("Starting ForexBot2 Telegram bot…")
|
||
|
||
app = (
|
||
Application.builder()
|
||
.token(TOKEN)
|
||
.post_init(_post_init)
|
||
.build()
|
||
)
|
||
|
||
app.add_handler(CommandHandler("start", cmd_start))
|
||
app.add_handler(CommandHandler("help", cmd_help))
|
||
app.add_handler(CommandHandler("status", cmd_status))
|
||
app.add_handler(CommandHandler("autoscan", cmd_autoscan))
|
||
app.add_handler(CommandHandler("cache", cmd_cache))
|
||
app.add_handler(CommandHandler("clearcache", cmd_clearcache))
|
||
app.add_handler(CallbackQueryHandler(cb_scan, pattern="^scan_"))
|
||
|
||
logger.info("Bot running — press Ctrl+C to stop")
|
||
app.run_polling(allowed_updates=Update.ALL_TYPES)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|