Harwinder Sekhon
6e2fdbd684
- Credentials, alert quality, scan pairs/timeframes, log level and candle-loop intervals now resolve via app_config (env -> config.json -> code default) instead of hardcoded fallbacks. - _get_shared_dp() uses make_data_provider() for provider-registry dispatch. - main() refuses to start without a token/chat_id, raising SystemExit with an actionable message instead of a bare sys.exit string. - test_all.py Module 13: two new tests covering credential/quality/ loop wiring via a reload, and the startup guard. Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b
1303 lines
48 KiB
Python
Executable File
1303 lines
48 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: XAU/USD × 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.
|
||
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 XAU/USD.
|
||
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 prevents spam.
|
||
|
||
Owner lock: All commands and buttons only respond to TELEGRAM_CHAT_ID
|
||
(the owner). Everyone else gets a polite refusal — the bot
|
||
is a private control panel, not the product.
|
||
Channel: If TELEGRAM_CHANNEL_ID is set (-100xxxxxxxxxx), auto-scan
|
||
signal alerts and level-breach alerts are also posted to
|
||
that channel. Pair it with a Telegram Stars subscription
|
||
invite link to sell access — Telegram manages join/renewal/
|
||
kick; the bot just posts there (needs channel-admin +
|
||
post-messages). Manual scans, status and cache replies stay
|
||
owner-DM only. The channel feed runs even when the owner
|
||
toggles their own DM auto-scan off.
|
||
|
||
Usage:
|
||
python telegram_bot.py
|
||
TELEGRAM_TOKEN=<token> TELEGRAM_CHAT_ID=<id> \
|
||
TELEGRAM_CHANNEL_ID=<-100…> python telegram_bot.py
|
||
"""
|
||
|
||
import asyncio
|
||
import functools
|
||
import json
|
||
import logging
|
||
import logging.handlers
|
||
import os
|
||
import signal
|
||
import sys
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Optional, Set
|
||
|
||
import websockets
|
||
|
||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||
from telegram.constants import ParseMode
|
||
from telegram.error import BadRequest
|
||
from telegram.ext import (
|
||
Application,
|
||
CallbackQueryHandler,
|
||
CommandHandler,
|
||
ContextTypes,
|
||
)
|
||
|
||
from config import (
|
||
PlatformConfig, SignalStrength, Timeframe,
|
||
XAUUSD, PAIR_INFO,
|
||
)
|
||
from app_config import (
|
||
get_app_config, provider_api_key, telegram_token, telegram_chat_id,
|
||
telegram_channel_id,
|
||
)
|
||
from data_provider import ForexDataProvider, make_data_provider
|
||
from pair_signals import PAIR_PRODUCERS
|
||
from signal_aggregator import SignalAggregator, SignalReport
|
||
|
||
_APP = get_app_config()
|
||
|
||
# ── Force UTF-8 on Windows ────────────────────────────────────────────
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
_LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
|
||
os.makedirs(_LOG_DIR, exist_ok=True)
|
||
_log_formatter = logging.Formatter(
|
||
"%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%H:%M:%S"
|
||
)
|
||
_console_handler = logging.StreamHandler()
|
||
_console_handler.setFormatter(_log_formatter)
|
||
_file_handler = logging.handlers.TimedRotatingFileHandler(
|
||
os.path.join(_LOG_DIR, "telegram.log"),
|
||
when="midnight", backupCount=30, encoding="utf-8",
|
||
)
|
||
_file_handler.setFormatter(_log_formatter)
|
||
logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
|
||
handlers=[_console_handler, _file_handler])
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||
logger = logging.getLogger("telegram_bot")
|
||
|
||
TOKEN = telegram_token()
|
||
CHAT_ID = telegram_chat_id()
|
||
|
||
# Owner lock: only this user/chat may drive the bot (commands + buttons).
|
||
# The bot is the product's signal source — without the lock, anyone who
|
||
# finds it could /autoscan and receive paid signals for free in DM.
|
||
OWNER_ID: Optional[int] = int(CHAT_ID) if CHAT_ID else None
|
||
|
||
# Subscriber channel (paid, Stars-subscription invite link managed by
|
||
# Telegram). When set, auto-scan signal alerts and level-breach alerts
|
||
# are also posted there. Manual scans, status, cache and boot messages
|
||
# stay owner-DM only. Set TELEGRAM_CHANNEL_ID="" to disable the feed.
|
||
_CHANNEL_ID_RAW = telegram_channel_id()
|
||
CHANNEL_ID: Optional[int] = int(_CHANNEL_ID_RAW) if _CHANNEL_ID_RAW else None
|
||
|
||
# ── Pairs & buttons ───────────────────────────────────────────────────
|
||
# The scanned pair list is config-driven: PlatformConfig.scan_pairs is
|
||
# the single switch for single- vs multi-pair operation. Buttons, WS
|
||
# subscriptions, and scan loops all derive from it.
|
||
|
||
SCAN_PAIRS = list(_APP.scan_pairs)
|
||
|
||
_PAIRS_LABEL = " + ".join(PAIR_INFO[p]["display"] for p in SCAN_PAIRS)
|
||
|
||
_WS_TO_PAIR = {PAIR_INFO[p]["display"]: p for p in SCAN_PAIRS}
|
||
_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", "bb_upper", "bb_lower",
|
||
"kc_upper", "kc_lower",
|
||
)
|
||
|
||
_TF_BUTTON_LABELS = [
|
||
(Timeframe.M5, "5 Min"),
|
||
(Timeframe.M15, "15 Min"),
|
||
(Timeframe.H1, "1 Hour"),
|
||
(Timeframe.H4, "4 Hour"),
|
||
]
|
||
|
||
|
||
def _build_scan_buttons() -> List[dict]:
|
||
"""One row of TF buttons + a scan-all button per scanned pair."""
|
||
multi = len(SCAN_PAIRS) > 1
|
||
buttons = []
|
||
for symbol in SCAN_PAIRS:
|
||
slug = symbol.replace("/", "").lower()
|
||
base = PAIR_INFO[symbol]["display"].split("/")[0] # XAU
|
||
prefix = f"{base} " if multi else ""
|
||
for tf, label in _TF_BUTTON_LABELS:
|
||
buttons.append({
|
||
"cb": f"scan_{slug}_{tf.value}", "symbol": symbol,
|
||
"tf": tf, "label": f"{prefix}{label}",
|
||
})
|
||
buttons.append({
|
||
"cb": f"scan_{slug}_all", "symbol": symbol, "tf": None,
|
||
"label": f"Scan All TFs — {base}" if multi else "Scan All TFs",
|
||
})
|
||
return buttons
|
||
|
||
|
||
SCAN_BUTTONS = _build_scan_buttons()
|
||
|
||
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=4)
|
||
|
||
# ── Markdown escaping ─────────────────────────────────────────────────
|
||
|
||
def _escape_md(text: str) -> str:
|
||
for ch in ("*", "_", "`", "[", "]"):
|
||
text = text.replace(ch, "\\" + ch)
|
||
return text
|
||
|
||
|
||
def _fmt_price(price: float) -> str:
|
||
return f"{price:.2f}" if price >= 10 else f"{price:.5f}"
|
||
|
||
|
||
# ── 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 = make_data_provider(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}] *{_escape_md(ind.name)}*: {_escape_md(short)}"
|
||
|
||
|
||
def _live_price_line(symbol: str) -> Optional[str]:
|
||
"""Fresh WebSocket tick for this symbol, if we have one (<2 min old)."""
|
||
price = _last_prices.get(symbol)
|
||
ts = _last_price_ts.get(symbol, 0)
|
||
if price is None:
|
||
return None
|
||
age = time.time() - ts
|
||
if age > 120:
|
||
return None
|
||
return f"📡 Live: *{_fmt_price(price)}* ({age:.0f}s tick)"
|
||
|
||
|
||
_DISCLAIMER = (
|
||
"⚠️ _Not financial advice. For educational and research purposes only. "
|
||
"Trading forex/CFDs carries a high risk of loss — do your own analysis "
|
||
"and never risk money you cannot afford to lose._"
|
||
)
|
||
|
||
|
||
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")
|
||
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"💰 Close: *{_fmt_price(report.current_price)}*",
|
||
]
|
||
live = _live_price_line(report.symbol)
|
||
if live:
|
||
lines.append(live)
|
||
lines += [
|
||
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"_{_escape_md(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"⚠️ {_escape_md(w)}")
|
||
|
||
if report.trade_setup and sig not in (SignalStrength.NEUTRAL,):
|
||
ts = report.trade_setup
|
||
dir_emoji = "📈" if ts.direction == "BUY" else "📉"
|
||
risk_usd = ts.risk_pips * ts.pip_value
|
||
reward_usd = ts.reward_pips * ts.pip_value
|
||
lines += [
|
||
"",
|
||
"━━━━━━━━━━━━━━━━━━━━",
|
||
f"🚀 *TRADE SETUP*",
|
||
f"{dir_emoji} Direction: *{ts.direction}*",
|
||
f"▶ Entry: `{_fmt_price(ts.entry_price)}`",
|
||
f"🛑 SL: `{_fmt_price(ts.stop_loss)}` (${risk_usd:.2f} / {ts.risk_pips:.0f} pips)",
|
||
f"🎯 TP1: `{_fmt_price(ts.take_profit_1)}` (${reward_usd:.2f} / {ts.reward_pips:.0f} pips)",
|
||
]
|
||
if ts.take_profit_2:
|
||
lines.append(f"🎯 TP2: `{_fmt_price(ts.take_profit_2)}`")
|
||
lines.append(f"⚖️ RR: 1:{ts.risk_reward:.1f}")
|
||
|
||
# Trailing-stop management plan (matches backtested exit model:
|
||
# TP1 is advisory, the trail is the exit). Pullback setups run
|
||
# tighter stops and arm later.
|
||
sig_cfg = PlatformConfig().signals
|
||
is_pb = getattr(ts, "strategy", "GENERIC").endswith("_PB")
|
||
arm_r = sig_cfg.pb_trail_arm_r if is_pb else sig_cfg.trail_arm_r
|
||
dist_r = sig_cfg.pb_trail_dist_r if is_pb else sig_cfg.trail_dist_r
|
||
if arm_r > 0:
|
||
risk_dist = abs(ts.entry_price - ts.stop_loss)
|
||
if ts.direction == "BUY":
|
||
arm_price = ts.entry_price + arm_r * risk_dist
|
||
else:
|
||
arm_price = ts.entry_price - arm_r * risk_dist
|
||
trail_dist = dist_r * risk_dist
|
||
tag = " (pullback)" if is_pb else ""
|
||
lines.append(
|
||
f"🔄 Trail{tag}: once price hits `{_fmt_price(arm_price)}` "
|
||
f"(+{arm_r:.1f}R), trail SL ${trail_dist:.2f} "
|
||
f"behind best price — TP1 is a guide, let it run"
|
||
)
|
||
|
||
lines += ["", "━━━━━━━━━━━━━━━━━━━━", _DISCLAIMER]
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ── 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.")
|
||
|
||
lines += ["", _DISCLAIMER]
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ── Keyboard helper ───────────────────────────────────────────────────
|
||
|
||
def _scan_button_rows() -> List[List[InlineKeyboardButton]]:
|
||
"""Per pair: one row of TF buttons + one scan-all row."""
|
||
rows = []
|
||
for symbol in SCAN_PAIRS:
|
||
tf_buttons = [b for b in SCAN_BUTTONS
|
||
if b["symbol"] == symbol and b["tf"] is not None]
|
||
rows.append([InlineKeyboardButton(b["label"], callback_data=b["cb"])
|
||
for b in tf_buttons])
|
||
all_button = next((b for b in SCAN_BUTTONS
|
||
if b["symbol"] == symbol and b["tf"] is None), None)
|
||
if all_button:
|
||
rows.append([InlineKeyboardButton(all_button["label"],
|
||
callback_data=all_button["cb"])])
|
||
return rows
|
||
|
||
|
||
def scan_keyboard() -> InlineKeyboardMarkup:
|
||
return InlineKeyboardMarkup(_scan_button_rows())
|
||
|
||
|
||
def main_menu_keyboard() -> InlineKeyboardMarkup:
|
||
"""Full interactive menu: scans + settings + status in one keyboard."""
|
||
autoscan_label = "🔔 Auto-scan: ON" if _autoscan_chats else "🔕 Auto-scan: OFF"
|
||
quality_label = f"🎚 Alerts: {_alert_min_quality}+" if _alert_min_quality != "A" else "🎚 Alerts: A only"
|
||
rows = _scan_button_rows()
|
||
rows.append([
|
||
InlineKeyboardButton(autoscan_label, callback_data="menu_autoscan"),
|
||
InlineKeyboardButton(quality_label, callback_data="menu_quality"),
|
||
])
|
||
rows.append([
|
||
InlineKeyboardButton("📊 Status", callback_data="menu_status"),
|
||
InlineKeyboardButton("ℹ️ Help", callback_data="menu_help"),
|
||
])
|
||
return InlineKeyboardMarkup(rows)
|
||
|
||
|
||
async def _safe_edit(query, text: str, reply_markup=None) -> None:
|
||
"""edit_message_text that survives 'message is not modified' and
|
||
Markdown parse failures (falls back to plain text)."""
|
||
try:
|
||
await query.edit_message_text(
|
||
text, parse_mode=ParseMode.MARKDOWN, reply_markup=reply_markup
|
||
)
|
||
except BadRequest as exc:
|
||
if "not modified" in str(exc).lower():
|
||
return
|
||
logger.warning(f"Markdown edit failed ({exc}); retrying as plain text")
|
||
try:
|
||
await query.edit_message_text(text, reply_markup=reply_markup)
|
||
except BadRequest:
|
||
pass
|
||
|
||
|
||
# ── 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]
|
||
)
|
||
# Per-pair tuned configs (regime weights, gates, sessions) come from
|
||
# the pair_signals factories; a plain PlatformConfig would silently
|
||
# run every pair with XAU defaults.
|
||
factory = PAIR_PRODUCERS.get(symbol)
|
||
if factory is not None:
|
||
config = factory(timeframe, confirmation_tfs)
|
||
else:
|
||
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_running_loop()
|
||
return await loop.run_in_executor(
|
||
_executor, _run_signal_sync, symbol, timeframe, confirmation_tfs
|
||
)
|
||
|
||
|
||
# ── Alert gating (quality threshold + dedup) ─────────────────────────
|
||
|
||
_QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1}
|
||
_ALERT_QUALITY_CYCLE = ["C", "B", "A"] # C+ → B+ → A-only
|
||
_alert_min_quality = _APP.alert_min_quality
|
||
|
||
# Fingerprint of the last alert per (symbol, timeframe) so a persisting
|
||
# signal doesn't re-send an identical message every candle.
|
||
_last_alert_fp: Dict[str, tuple] = {}
|
||
_last_alert_ts: Dict[str, float] = {}
|
||
_ALERT_REARM_SECONDS = 3600 # identical signal re-alerts after 1h
|
||
|
||
|
||
def _passes_quality(quality: str) -> bool:
|
||
return _QUALITY_RANK.get(quality, 0) >= _QUALITY_RANK.get(_alert_min_quality, 1)
|
||
|
||
|
||
def _alert_fingerprint(report: SignalReport) -> tuple:
|
||
"""Alert identity for dedup. Includes a risk-relative entry bucket:
|
||
a persisting signal whose entry drifts a few ticks between candles
|
||
keeps the same fingerprint, while a setup that moved by half its risk
|
||
distance or more (new pullback, new swing anchor) re-alerts instead
|
||
of being swallowed for the full re-arm hour."""
|
||
setup = report.trade_setup
|
||
entry_bucket = None
|
||
if setup is not None:
|
||
sl_dist = abs(setup.entry_price - setup.stop_loss)
|
||
entry_bucket = round(setup.entry_price / max(sl_dist * 0.5, 1e-9))
|
||
return (
|
||
report.final_signal.name,
|
||
report.signal_quality,
|
||
setup.direction if setup else None,
|
||
entry_bucket,
|
||
)
|
||
|
||
|
||
def _should_alert(key: str, report: SignalReport) -> bool:
|
||
"""True when this signal is new/changed for the key, or re-armed.
|
||
|
||
Pure check — call _commit_alert only after the message actually went
|
||
out, so a failed Telegram send doesn't consume the alert for an hour.
|
||
"""
|
||
fp = _alert_fingerprint(report)
|
||
return not (
|
||
_last_alert_fp.get(key) == fp
|
||
and time.time() - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS
|
||
)
|
||
|
||
|
||
def _commit_alert(key: str, report: SignalReport) -> None:
|
||
_last_alert_fp[key] = _alert_fingerprint(report)
|
||
_last_alert_ts[key] = time.time()
|
||
|
||
|
||
# ── Candle-boundary scheduler ────────────────────────────────────────
|
||
|
||
_autoscan_chats: Set[int] = set()
|
||
_autoscan_tasks: list[asyncio.Task] = []
|
||
|
||
|
||
def _feed_active() -> bool:
|
||
"""Scan loops keep running while anyone is listening: owner DM
|
||
auto-scan chats or the subscriber channel."""
|
||
return bool(_autoscan_chats) or CHANNEL_ID is not None
|
||
|
||
|
||
def _signal_targets() -> List[int]:
|
||
"""Where signal/breach alerts go: owner auto-scan chats + channel."""
|
||
targets: List[int] = list(_autoscan_chats)
|
||
if CHANNEL_ID is not None:
|
||
targets.append(CHANNEL_ID)
|
||
return targets
|
||
|
||
|
||
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."""
|
||
tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
|
||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||
reports: List[SignalReport] = []
|
||
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
|
||
|
||
_update_level_cache(reports)
|
||
|
||
actionable = [
|
||
r for r in reports
|
||
if _passes_quality(r.signal_quality)
|
||
and _should_alert(f"{r.symbol}_{tf.value}", r)
|
||
]
|
||
sent_ok: Set[int] = set()
|
||
for chat_id in _signal_targets():
|
||
for i, report in enumerate(actionable):
|
||
detail = format_signal_message(report)
|
||
try:
|
||
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
|
||
sent_ok.add(i)
|
||
except Exception as exc:
|
||
logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
|
||
# Commit fingerprints only for alerts that reached at least one target;
|
||
# a fully failed send stays eligible for the next scan.
|
||
for i, report in enumerate(actionable):
|
||
if i in sent_ok:
|
||
_commit_alert(f"{report.symbol}_{tf.value}", report)
|
||
|
||
|
||
# ── 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()
|
||
_WS_API_KEY = provider_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 = {
|
||
k: v for k, v in r.primary_analysis.key_levels.items()
|
||
if k in _LEVEL_KEYS
|
||
}
|
||
if levels:
|
||
_level_cache[r.symbol] = levels
|
||
_level_cache_ts[r.symbol] = now
|
||
|
||
|
||
def _check_level_breach(symbol: str, price: float) -> Optional[str]:
|
||
levels = _level_cache.get(symbol)
|
||
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
|
||
|
||
for name, level in levels.items():
|
||
if (last < level <= price) or (last > level >= price):
|
||
return name
|
||
|
||
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:
|
||
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} — rescanning M15")
|
||
|
||
try:
|
||
report = await run_signal(pair_symbol, Timeframe.M15)
|
||
_update_level_cache([report])
|
||
except Exception:
|
||
logger.exception(f"Breach scan failed for {display}")
|
||
return
|
||
|
||
# Alert only when the rescan produced an actionable, changed signal —
|
||
# same gates as the auto-scanner, so a breach can't spam the channel.
|
||
if (report.final_signal == SignalStrength.NEUTRAL
|
||
or not _passes_quality(report.signal_quality)
|
||
or not _should_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)):
|
||
return
|
||
|
||
header = (
|
||
f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed "
|
||
f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
|
||
)
|
||
detail = header + format_signal_message(report)
|
||
sent = False
|
||
for chat_id in _signal_targets():
|
||
try:
|
||
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
|
||
sent = True
|
||
except Exception as exc:
|
||
logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}")
|
||
if sent:
|
||
_commit_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)
|
||
|
||
|
||
async def _ws_heartbeat(ws) -> None:
|
||
"""Twelve Data drops idle WS connections; heartbeat every 10s keeps it
|
||
alive through quiet periods (weekends, low-liquidity hours)."""
|
||
while True:
|
||
await asyncio.sleep(10)
|
||
await ws.send(json.dumps({"action": "heartbeat"}))
|
||
|
||
|
||
async def _ws_monitor(bot) -> None:
|
||
backoff = 1
|
||
while _feed_active():
|
||
hb_task: Optional[asyncio.Task] = None
|
||
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
|
||
hb_task = asyncio.create_task(_ws_heartbeat(ws))
|
||
|
||
async for raw in ws:
|
||
if not _feed_active():
|
||
break
|
||
|
||
try:
|
||
msg = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
logger.warning(f"WebSocket: malformed JSON, skipping")
|
||
continue
|
||
|
||
event = msg.get("event")
|
||
if event == "subscribe-status":
|
||
fails = msg.get("fails") or []
|
||
if fails:
|
||
logger.warning(f"WebSocket: failed to subscribe: {fails}")
|
||
continue
|
||
if event != "price":
|
||
continue
|
||
|
||
ws_symbol = msg.get("symbol")
|
||
pair_symbol = _WS_TO_PAIR.get(ws_symbol)
|
||
if not pair_symbol:
|
||
continue
|
||
|
||
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:
|
||
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
|
||
except Exception:
|
||
logger.exception(f"WebSocket error, reconnecting in {backoff}s")
|
||
await asyncio.sleep(backoff)
|
||
backoff = min(backoff * 2, 60)
|
||
finally:
|
||
if hb_task is not None:
|
||
hb_task.cancel()
|
||
|
||
logger.info("WebSocket monitor stopped")
|
||
|
||
|
||
async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
|
||
label = tf.value
|
||
try:
|
||
while _feed_active():
|
||
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 _feed_active():
|
||
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")
|
||
|
||
|
||
_TF_INTERVAL_MINUTES = {
|
||
Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30,
|
||
Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440,
|
||
}
|
||
_AUTOSCAN_LOOPS = [
|
||
(tf, _TF_INTERVAL_MINUTES[tf]) for tf in _APP.scan_timeframes
|
||
]
|
||
|
||
|
||
_ws_task: Optional[asyncio.Task] = None
|
||
|
||
|
||
def _start_scheduler(bot) -> None:
|
||
global _ws_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 (%s)",
|
||
"/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS))
|
||
|
||
if _ws_task is None or _ws_task.done():
|
||
_ws_task = asyncio.create_task(_ws_monitor(bot))
|
||
logger.info("WebSocket level-breach monitor started")
|
||
|
||
|
||
def _stop_scheduler() -> None:
|
||
global _ws_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
|
||
for t in list(_breach_tasks):
|
||
t.cancel()
|
||
_breach_tasks.clear()
|
||
logger.info("Scheduler and WebSocket monitor stopped")
|
||
|
||
|
||
# ── Handlers ──────────────────────────────────────────────────────────
|
||
|
||
def _owner_only(handler):
|
||
"""Reject commands/buttons from anyone but the owner.
|
||
|
||
Signals are sold via the subscriber channel; the bot itself is the
|
||
owner's control panel. Without this, any user who finds the bot
|
||
could /autoscan and receive the paid feed for free in their DM.
|
||
"""
|
||
@functools.wraps(handler)
|
||
async def wrapped(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
user = update.effective_user
|
||
chat = update.effective_chat
|
||
uid = user.id if user else None
|
||
if OWNER_ID is not None and (uid == OWNER_ID or (chat and chat.id == OWNER_ID)):
|
||
return await handler(update, context)
|
||
logger.warning(
|
||
f"Unauthorized access blocked: user={uid} "
|
||
f"chat={chat.id if chat else None} handler={handler.__name__}"
|
||
)
|
||
if update.callback_query:
|
||
await update.callback_query.answer("🔒 Not authorized.")
|
||
elif update.message:
|
||
await update.message.reply_text(
|
||
"🔒 This is a private control bot. Signals are published "
|
||
"in the subscriber channel."
|
||
)
|
||
return wrapped
|
||
|
||
|
||
@_owner_only
|
||
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
scanning = "running" if _autoscan_chats else "off"
|
||
text = (
|
||
f"👋 *ForexBot2 — {_PAIRS_LABEL} Signal Intelligence*\n\n"
|
||
"Pick a timeframe to scan gold.\n"
|
||
"Signals include a full trade setup with entry, SL and TP.\n\n"
|
||
f"Auto-scan: *{scanning}* — /autoscan to toggle\n"
|
||
"/scan — Quick scan buttons\n"
|
||
"/status — Bot & WebSocket status\n"
|
||
"/help — All commands"
|
||
)
|
||
await update.message.reply_text(
|
||
text,
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
reply_markup=main_menu_keyboard(),
|
||
)
|
||
|
||
|
||
_HELP_TEXT = (
|
||
"📖 *ForexBot2 Commands*\n\n"
|
||
"/start — Welcome & interactive menu\n"
|
||
"/scan — Quick scan menu\n"
|
||
"/autoscan — Toggle auto-scan on/off\n"
|
||
"/status — Bot status, WS connection, live prices\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"
|
||
"*Alerts* button sets the minimum quality grade that gets sent.\n"
|
||
"*WebSocket* streams live prices; key-level breaches trigger an "
|
||
"instant M15 rescan and alert (5-min cooldown).\n"
|
||
"*Channel feed*: signal + breach alerts mirror to the subscriber "
|
||
"channel when TELEGRAM_CHANNEL_ID is set; /autoscan only toggles "
|
||
"your DM copy."
|
||
)
|
||
|
||
|
||
@_owner_only
|
||
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
await update.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)
|
||
|
||
|
||
def _status_text() -> str:
|
||
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 CHANNEL_ID is not None:
|
||
lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)")
|
||
|
||
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)
|
||
age = time.time() - _last_price_ts.get(pair, 0)
|
||
if price is None:
|
||
lines.append(f" {display}: waiting...")
|
||
else:
|
||
lines.append(f" {display}: `{_fmt_price(price)}` ({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:
|
||
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")
|
||
|
||
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")
|
||
|
||
active_breach = sum(1 for t in _breach_tasks if not t.done())
|
||
if active_breach:
|
||
lines.append(f"*Breach tasks:* {active_breach} running")
|
||
|
||
lines.append(f"\n*Alert quality:* {_alert_min_quality}+ "
|
||
f"| re-arm {_ALERT_REARM_SECONDS // 60}m")
|
||
return "\n".join(lines)
|
||
|
||
|
||
@_owner_only
|
||
async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
await update.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN)
|
||
|
||
|
||
@_owner_only
|
||
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)
|
||
|
||
|
||
@_owner_only
|
||
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,
|
||
)
|
||
|
||
|
||
@_owner_only
|
||
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 _feed_active():
|
||
_stop_scheduler()
|
||
msg = "⏹ Auto-scan stopped."
|
||
else:
|
||
msg = ("⏹ Auto-scan stopped for this chat.\n"
|
||
"📢 Subscriber channel feed continues.")
|
||
await update.message.reply_text(msg, 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 {_PAIRS_LABEL} on M5/M15/H1/H4 at candle close.\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,
|
||
)
|
||
|
||
|
||
@_owner_only
|
||
async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
await update.message.reply_text(
|
||
f"🔍 *{_PAIRS_LABEL} — Pick a timeframe*",
|
||
parse_mode=ParseMode.MARKDOWN,
|
||
reply_markup=main_menu_keyboard(),
|
||
)
|
||
|
||
|
||
@_owner_only
|
||
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
|
||
|
||
if btn["tf"] is None:
|
||
await _cb_scan_all(query, btn["symbol"])
|
||
return
|
||
|
||
display = PAIR_INFO[btn["symbol"]]["display"]
|
||
await _safe_edit(query, f"⏳ Scanning {display} {btn['label']}…")
|
||
|
||
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: {_escape_md(str(exc))}"
|
||
|
||
await _safe_edit(query, text, reply_markup=main_menu_keyboard())
|
||
|
||
|
||
@_owner_only
|
||
async def cb_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
"""Interactive menu actions: autoscan toggle, quality cycle, status, help."""
|
||
global _alert_min_quality
|
||
query = update.callback_query
|
||
action = query.data
|
||
|
||
if action == "menu_autoscan":
|
||
chat_id = query.message.chat.id
|
||
if chat_id in _autoscan_chats:
|
||
_autoscan_chats.discard(chat_id)
|
||
if not _feed_active():
|
||
_stop_scheduler()
|
||
await query.answer("Auto-scan stopped")
|
||
else:
|
||
await query.answer("DM alerts off — channel feed continues")
|
||
else:
|
||
_autoscan_chats.add(chat_id)
|
||
_start_scheduler(context.bot)
|
||
await query.answer("Auto-scan started")
|
||
try:
|
||
await query.edit_message_reply_markup(reply_markup=main_menu_keyboard())
|
||
except BadRequest:
|
||
pass
|
||
|
||
elif action == "menu_quality":
|
||
i = _ALERT_QUALITY_CYCLE.index(_alert_min_quality) if _alert_min_quality in _ALERT_QUALITY_CYCLE else 0
|
||
_alert_min_quality = _ALERT_QUALITY_CYCLE[(i + 1) % len(_ALERT_QUALITY_CYCLE)]
|
||
await query.answer(f"Alerting {_alert_min_quality}+ signals")
|
||
try:
|
||
await query.edit_message_reply_markup(reply_markup=main_menu_keyboard())
|
||
except BadRequest:
|
||
pass
|
||
|
||
elif action == "menu_status":
|
||
await query.answer()
|
||
await query.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN)
|
||
|
||
elif action == "menu_help":
|
||
await query.answer()
|
||
await query.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)
|
||
|
||
|
||
async def _cb_scan_all(query, symbol: str = XAUUSD) -> None:
|
||
display = PAIR_INFO[symbol]["display"]
|
||
await _safe_edit(query, f"⏳ Scanning {display} across M5 / M15 / H1 / H4…")
|
||
|
||
tfs = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]
|
||
tasks = [run_signal(symbol, tf) for tf in tfs]
|
||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
reports = []
|
||
errors = []
|
||
for tf, result in zip(tfs, results):
|
||
if isinstance(result, Exception):
|
||
errors.append(f"❌ {tf.value}: {result}")
|
||
else:
|
||
reports.append(result)
|
||
|
||
if reports:
|
||
_update_level_cache(reports)
|
||
|
||
lines = [
|
||
f"📊 *{display} — All Timeframes*",
|
||
f"🕐 {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
|
||
"",
|
||
]
|
||
|
||
for r in reports:
|
||
emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
|
||
qe = _QUALITY_EMOJI.get(r.signal_quality, "")
|
||
tf_label = r.primary_analysis.timeframe if r.primary_analysis else "?"
|
||
sig = _signal_label(r.final_signal)
|
||
regime = r.market_regime.replace("_", " ").title()
|
||
lines.append(
|
||
f"{emoji} *{tf_label}* {sig} {qe}{r.signal_quality} "
|
||
f"Score: {r.final_score:+.1f} {r.final_confidence:.0f}%"
|
||
)
|
||
lines.append(f" _{_escape_md(regime)}_ | {'✅ MTF' if r.mtf_agreement else '❌ MTF'}")
|
||
if r.trade_setup and r.final_signal != SignalStrength.NEUTRAL:
|
||
ts = r.trade_setup
|
||
d = "📈" if ts.direction == "BUY" else "📉"
|
||
lines.append(
|
||
f" {d} {ts.direction} `{_fmt_price(ts.entry_price)}` "
|
||
f"SL `{_fmt_price(ts.stop_loss)}` TP `{_fmt_price(ts.take_profit_1)}` "
|
||
f"RR 1:{ts.risk_reward:.1f}"
|
||
)
|
||
lines.append("")
|
||
|
||
for err in errors:
|
||
lines.append(err)
|
||
|
||
lines.append("_Use buttons below for detailed single-TF view._")
|
||
lines += ["", _DISCLAIMER]
|
||
|
||
await _safe_edit(query, "\n".join(lines), reply_markup=main_menu_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", "Welcome & scan buttons"),
|
||
("scan", "Quick 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 CHAT_ID:
|
||
_autoscan_chats.add(int(CHAT_ID))
|
||
|
||
if not _feed_active():
|
||
logger.info("Neither TELEGRAM_CHAT_ID nor TELEGRAM_CHANNEL_ID set — "
|
||
"auto-scan disabled on boot. Use /autoscan in chat to enable.")
|
||
return
|
||
|
||
if CHANNEL_ID is not None:
|
||
logger.info(f"Subscriber channel feed enabled: {CHANNEL_ID}")
|
||
|
||
logger.info("Running initial scan to populate level cache and signal stats...")
|
||
try:
|
||
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)
|
||
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 (chats={sorted(_autoscan_chats)}, channel={CHANNEL_ID}) — "
|
||
f"M5 {next_m5:.0f}s, M15 {next_m15:.0f}s, H1 {next_h1:.0f}s, H4 {next_h4:.0f}s"
|
||
)
|
||
|
||
# Boot message goes to the owner DM only — the subscriber channel
|
||
# receives nothing but signal and breach alerts.
|
||
if OWNER_ID is not None:
|
||
channel_line = "📢 Subscriber channel feed active.\n" if CHANNEL_ID else ""
|
||
try:
|
||
await app.bot.send_message(
|
||
OWNER_ID,
|
||
f"🤖 *ForexBot2 started*\n\n"
|
||
f"Auto-scanning {_PAIRS_LABEL} on M5/M15/H1/H4.\n"
|
||
f"📡 Live price monitoring active.\n"
|
||
f"{channel_line}"
|
||
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,
|
||
)
|
||
except Exception:
|
||
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:
|
||
if not TOKEN or not CHAT_ID:
|
||
raise SystemExit(
|
||
"telegram_bot: missing Telegram credentials. Set telegram.token "
|
||
"and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and "
|
||
"TELEGRAM_CHAT_ID environment variables.")
|
||
|
||
logger.info("Starting ForexBot2 Telegram bot…")
|
||
|
||
app = (
|
||
Application.builder()
|
||
.token(TOKEN)
|
||
.post_init(_post_init)
|
||
.post_shutdown(_post_shutdown)
|
||
.build()
|
||
)
|
||
|
||
app.add_handler(CommandHandler("start", cmd_start))
|
||
app.add_handler(CommandHandler("scan", cmd_scan))
|
||
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_"))
|
||
app.add_handler(CallbackQueryHandler(cb_menu, pattern="^menu_"))
|
||
|
||
logger.info("Bot running — press Ctrl+C to stop")
|
||
app.run_polling(allowed_updates=Update.ALL_TYPES)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|