Harwinder Sekhon
e609b433b0
New signal_log.py logs each SignalReport to logs/signals.db via a fail-safe hook in SignalAggregator.analyze_pair — the live-only entry point used by the bot, daemon, CLI and dashboard. Backtester calls analyze_from_dataframes directly, so live==backtest parity is preserved (test-locked). Rows carry signal/grade/price/spread, MTF + trade-setup + reasons/warnings JSON, and news-at-scan-time (calendar events + scored headlines JSON via NewsEngine class-level caches). Config: signal_log block (enabled, path; env SIGNAL_LOG_ENABLED/PATH), source tagged per consumer (telegram/oanda_daemon/cli/dashboard). 381 tests pass (24 modules).
1691 lines
66 KiB
Python
Executable File
1691 lines
66 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 the streaming provider — config.json
|
||
`data_providers.streaming` when set (e.g. tiingo), else the
|
||
active provider if it streams; auto-disabled otherwise.
|
||
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.
|
||
News: /news replies with the net XAU/USD news sentiment score
|
||
(display-only — news never gates signals).
|
||
|
||
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 logging
|
||
import logging.handlers
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Optional, Set
|
||
|
||
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, telegram_token, telegram_chat_id,
|
||
telegram_channel_id,
|
||
)
|
||
from providers import BaseDataProvider, make_data_provider
|
||
from agent_panel import attach_panel, panel_health
|
||
from llm_client import get_llm_client
|
||
from news_engine import FundamentalContext, NewsEngine
|
||
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="%Y-%m-%d %H:%M:%S",
|
||
)
|
||
_console_handler = logging.StreamHandler()
|
||
_console_handler.setFormatter(_log_formatter)
|
||
# 3-hour rotation × 240 backups = 30 days retention. Handlers sit on the
|
||
# ROOT logger (force=True guarantees ownership even if something logged
|
||
# first), so every module logger — data_provider, news_engine,
|
||
# signal_aggregator, websockets, telegram.ext — lands in the same file.
|
||
_file_handler = logging.handlers.TimedRotatingFileHandler(
|
||
os.path.join(_LOG_DIR, "telegram.log"),
|
||
when="h", interval=3, backupCount=240, encoding="utf-8",
|
||
)
|
||
_file_handler.setFormatter(_log_formatter)
|
||
logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
|
||
handlers=[_console_handler, _file_handler],
|
||
force=True)
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||
# Never let DEBUG runs write credentials into the 30-day rotating log:
|
||
# the websockets client logger emits the connect URI (Twelve Data puts
|
||
# the API key in the query string) and frame contents (Tiingo's auth
|
||
# frame carries the key) — 2026-08-04 audit, finding K47.
|
||
logging.getLogger("websockets").setLevel(logging.INFO)
|
||
logging.getLogger("websockets.client").setLevel(logging.INFO)
|
||
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.json-driven (`scanning.pairs`, read via
|
||
# app_config) and 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)
|
||
|
||
_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[BaseDataProvider] = None
|
||
|
||
|
||
def _get_shared_dp() -> BaseDataProvider:
|
||
global _shared_dp
|
||
if _shared_dp is None:
|
||
_shared_dp = make_data_provider(PlatformConfig())
|
||
return _shared_dp
|
||
|
||
|
||
_streaming_dp: Optional[BaseDataProvider] = None
|
||
|
||
|
||
def _get_streaming_dp() -> BaseDataProvider:
|
||
"""Provider for the tick stream: config.json data_providers.streaming
|
||
when set (a SEPARATE instance — the candle cache stays on the active
|
||
provider), else the shared active provider. Reads app config at call
|
||
time, not the import-time _APP snapshot, so tests can swap it via
|
||
set_app_config()."""
|
||
global _streaming_dp
|
||
if _streaming_dp is None:
|
||
app = get_app_config()
|
||
if app.streaming_provider is None or \
|
||
app.streaming_provider == app.active_provider:
|
||
_streaming_dp = _get_shared_dp()
|
||
else:
|
||
_streaming_dp = make_data_provider(
|
||
PlatformConfig(), app.streaming_provider)
|
||
return _streaming_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": "❌",
|
||
}
|
||
|
||
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)
|
||
|
||
|
||
# Informational fixed-pip target shown on every trade setup (display-only,
|
||
# not part of the backtested exit model).
|
||
_INFO_TP_PIPS = 100
|
||
|
||
_IND_SHORT = {
|
||
"ema_trend": "EMA", "rsi": "RSI", "macd": "MACD", "bollinger": "BB",
|
||
"adx": "ADX", "roc": "ROC", "keltner": "KC", "divergence": "Div",
|
||
"candlestick": "Cndl",
|
||
}
|
||
|
||
_IND_ARROW = {
|
||
SignalStrength.STRONG_BUY: "⇈",
|
||
SignalStrength.BUY: "↑",
|
||
SignalStrength.NEUTRAL: "→",
|
||
SignalStrength.SELL: "↓",
|
||
SignalStrength.STRONG_SELL: "⇊",
|
||
}
|
||
|
||
|
||
def _indicator_summary(report: SignalReport) -> str:
|
||
"""One-line condensed indicator readout: 'EMA↑ RSI↑ MACD→ …'."""
|
||
if not report.primary_analysis:
|
||
return ""
|
||
parts = []
|
||
for name, ind in report.primary_analysis.indicators.items():
|
||
if name.startswith("_") or ind.weight == 0:
|
||
continue
|
||
label = _IND_SHORT.get(name, name)
|
||
parts.append(f"{label}{_IND_ARROW.get(ind.signal, '→')}")
|
||
return " ".join(parts)
|
||
|
||
|
||
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 ✅" if report.mtf_agreement else "MTF ❌"
|
||
tf = report.primary_analysis.timeframe if report.primary_analysis else ""
|
||
|
||
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: list[str] = [
|
||
f"{emoji} *{label} · {report.display_name} · {tf}* {qe}{report.signal_quality}",
|
||
f"🕐 {now} · 💰 {_fmt_price(report.current_price)}",
|
||
]
|
||
live = _live_price_line(report.symbol)
|
||
if live:
|
||
lines.append(live)
|
||
lines += [
|
||
f"Score *{report.final_score:+.1f}* · Conf *{report.final_confidence:.0f}%* · {mtf_tag}",
|
||
f"📊 Regime: {regime_str} · Vol: {vol_str}",
|
||
]
|
||
if report.action_recommendation:
|
||
lines.append(f"_{_escape_md(report.action_recommendation)}_")
|
||
|
||
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 += [
|
||
"",
|
||
"🚀 *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}p)",
|
||
f"🎯 TP1 `{_fmt_price(ts.take_profit_1)}` (${reward_usd:.2f} / {ts.reward_pips:.0f}p)",
|
||
]
|
||
if ts.take_profit_2:
|
||
lines.append(f"🎯 TP2 `{_fmt_price(ts.take_profit_2)}`")
|
||
|
||
# Informational fixed-pip target: where a 100-pip TP would sit for
|
||
# this setup (display-only — not part of the backtested exit model).
|
||
info_dist = _INFO_TP_PIPS * ts.pip_value
|
||
if ts.direction == "BUY":
|
||
info_price = ts.entry_price + info_dist
|
||
else:
|
||
info_price = ts.entry_price - info_dist
|
||
risk_dist = abs(ts.entry_price - ts.stop_loss)
|
||
info_r = f" · ≈{info_dist / risk_dist:.1f}R" if risk_dist > 0 else ""
|
||
lines.append(
|
||
f"🎯 TP {_INFO_TP_PIPS}p: `{_fmt_price(info_price)}` (${info_dist:.2f}{info_r})"
|
||
)
|
||
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). Params are stamped on
|
||
# the setup by the producing config so per-TF factory overrides
|
||
# (M15 arm 0.70, M5 dist 0.85) are what subscribers see; only a
|
||
# legacy unstamped setup falls back to platform defaults.
|
||
is_pb = getattr(ts, "strategy", "GENERIC").endswith("_PB")
|
||
arm_r = getattr(ts, "trail_arm_r", None)
|
||
dist_r = getattr(ts, "trail_dist_r", None)
|
||
if arm_r is None or dist_r is None:
|
||
sig_cfg = PlatformConfig().signals
|
||
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:
|
||
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}: arm `{_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"
|
||
)
|
||
|
||
extra: list[str] = []
|
||
mtf = _mtf_line(report)
|
||
if mtf:
|
||
extra.append(f"📈 MTF: {mtf}")
|
||
ind_summary = _indicator_summary(report)
|
||
if ind_summary:
|
||
extra.append(f"📋 {ind_summary}")
|
||
for w in report.warnings:
|
||
extra.append(f"⚠️ {_escape_md(w)}")
|
||
if report.news_summary:
|
||
extra.append(f"📰 {_escape_md(report.news_summary)}")
|
||
if report.panel_verdict:
|
||
mark = {"AGREE": "✅", "OPPOSE": "⛔", "NEUTRAL": "➖"}.get(
|
||
report.panel_verdict, "")
|
||
extra.append(
|
||
f"🤖 AI Panel {mark} *{_escape_md(report.panel_verdict)}* "
|
||
f"({report.panel_confidence:.0f}%) — "
|
||
f"{_escape_md(report.panel_rationale)}")
|
||
if report.panel_takes:
|
||
votes = " · ".join(
|
||
f"{t['persona']}: {t['stance']}" for t in report.panel_takes)
|
||
extra.append(f" _{_escape_md(votes)}_")
|
||
if extra:
|
||
lines += [""] + extra
|
||
|
||
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, log_source="telegram")
|
||
|
||
|
||
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] = {}
|
||
_last_alert_setup: Dict[str, tuple] = {} # key → (entry, risk) of last ALERTED setup
|
||
_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: signal, quality, setup direction.
|
||
|
||
Entry movement is checked separately in _should_alert against the
|
||
last ALERTED entry/risk (a stable reference). Deriving a price bucket
|
||
from the CURRENT SL distance here made the fingerprint churn on
|
||
normal ATR/anchor drift — near-duplicate alerts every candle."""
|
||
setup = report.trade_setup
|
||
return (
|
||
report.final_signal.name,
|
||
report.signal_quality,
|
||
setup.direction if setup else None,
|
||
)
|
||
|
||
|
||
def _should_alert(key: str, report: SignalReport) -> bool:
|
||
"""True when this signal is new/changed for the key, or re-armed.
|
||
|
||
A persisting identical signal dedups for the re-arm hour, but a setup
|
||
whose entry moved >= half the last-alerted risk distance (new
|
||
pullback/swing anchor) re-alerts immediately.
|
||
|
||
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)
|
||
if _last_alert_fp.get(key) != fp:
|
||
return True
|
||
if time.time() - _last_alert_ts.get(key, 0) >= _ALERT_REARM_SECONDS:
|
||
return True
|
||
setup = report.trade_setup
|
||
last = _last_alert_setup.get(key)
|
||
if setup is not None and last is not None:
|
||
last_entry, last_risk = last
|
||
if abs(setup.entry_price - last_entry) >= 0.5 * max(last_risk, 1e-9):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _commit_alert(key: str, report: SignalReport) -> None:
|
||
_last_alert_fp[key] = _alert_fingerprint(report)
|
||
_last_alert_ts[key] = time.time()
|
||
setup = report.trade_setup
|
||
if setup is not None:
|
||
_last_alert_setup[key] = (
|
||
setup.entry_price, abs(setup.entry_price - setup.stop_loss)
|
||
)
|
||
else:
|
||
_last_alert_setup.pop(key, None)
|
||
|
||
|
||
# Per-key locks serializing the _should_alert → send → _commit_alert
|
||
# window. The auto-scan loop and a concurrent breach handler share the
|
||
# "XAU/USD_15min" dedup key; without the lock both pass _should_alert
|
||
# while the other's send is still awaiting Telegram, and the same signal
|
||
# goes out twice (2026-08-04 audit H36). Per-key rather than one global
|
||
# lock so a slow send on one timeframe cannot head-of-line-block another
|
||
# timeframe's alert. Keys are symbol × TF, so the dict stays tiny.
|
||
_alert_locks: Dict[str, asyncio.Lock] = {}
|
||
|
||
|
||
def _alert_lock(key: str) -> asyncio.Lock:
|
||
lock = _alert_locks.get(key)
|
||
if lock is None:
|
||
lock = _alert_locks[key] = asyncio.Lock()
|
||
return lock
|
||
|
||
|
||
async def _send_alert(bot, chat_id: int, text: str) -> bool:
|
||
"""Markdown alert send with a plain-text fallback on parse failures,
|
||
mirroring _safe_edit (2026-08-04 audit H35 — alert paths previously
|
||
dropped the message on a BadRequest with only a warning logged).
|
||
Returns True when a message reached the chat by either attempt."""
|
||
try:
|
||
await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
|
||
return True
|
||
except BadRequest as exc:
|
||
logger.warning(f"Markdown alert failed for chat {chat_id} ({exc}); "
|
||
f"retrying as plain text")
|
||
try:
|
||
await bot.send_message(chat_id, text)
|
||
return True
|
||
except Exception as exc2:
|
||
logger.warning(f"Failed to send alert to chat {chat_id}: {exc2}")
|
||
return False
|
||
except Exception as exc:
|
||
logger.warning(f"Failed to send alert to chat {chat_id}: {exc}")
|
||
return False
|
||
|
||
|
||
# ── 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
|
||
# NOTE: remaining > buffer always (elapsed < interval by construction);
|
||
# a call at the exact boundary instant targets the NEXT boundary —
|
||
# steady-state loops wake at boundary+buffer, so only a process
|
||
# started in the boundary microsecond skips one scan (2026-08-04
|
||
# audit I42; the old `remaining <= buffer` guard was unreachable).
|
||
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):
|
||
# gather(return_exceptions=True) means there is no active
|
||
# exception here — logger.exception would log "NoneType: None"
|
||
# instead of the real stack; pass the instance explicitly.
|
||
logger.error(
|
||
f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}",
|
||
exc_info=result,
|
||
)
|
||
else:
|
||
reports.append(result)
|
||
|
||
if not reports:
|
||
return
|
||
|
||
_update_level_cache(reports)
|
||
|
||
candidates = [
|
||
r for r in reports
|
||
if r.final_signal != SignalStrength.NEUTRAL # same gate as breach path
|
||
and _passes_quality(r.signal_quality)
|
||
]
|
||
for report in candidates:
|
||
key = f"{report.symbol}_{tf.value}"
|
||
# The lock spans check → send → commit so a concurrent breach
|
||
# handler on the same key cannot double-alert (H36).
|
||
async with _alert_lock(key):
|
||
if not _should_alert(key, report):
|
||
continue
|
||
# Panel AFTER the quality gate + dedup decision so no LLM
|
||
# spend on suppressed alerts. Runs on the scan executor to
|
||
# keep the event loop free; this key's alert lock is held
|
||
# for the panel's wall time (≤~25 s), which only delays a
|
||
# same-key duplicate that _should_alert would drop anyway.
|
||
await asyncio.get_running_loop().run_in_executor(
|
||
_executor, attach_panel, report)
|
||
detail = format_signal_message(report)
|
||
delivered = False
|
||
for chat_id in _signal_targets():
|
||
if await _send_alert(bot, chat_id, detail):
|
||
delivered = True
|
||
# Commit only when the alert reached at least one target;
|
||
# a fully failed send stays eligible for the next scan.
|
||
if delivered:
|
||
_commit_alert(key, report)
|
||
|
||
|
||
# ── WebSocket level-breach monitor ──────────────────────────────────
|
||
|
||
# Level cache keyed symbol → timeframe → {level_name: price}, with a
|
||
# matching per-TF write timestamp. Keyed per TF (2026-08-04 audit H34)
|
||
# because every scan path writes here — the M5/M15/H1/H4 boundary loops,
|
||
# the breach rescan, the manual Scan-All button — and a single flat dict
|
||
# meant the last writer wiped the other timeframes' swing/EMA levels
|
||
# within minutes. Breach checks scan the union of all fresh TF entries.
|
||
_level_cache: Dict[str, Dict[str, Dict[str, float]]] = {}
|
||
_level_cache_ts: Dict[str, Dict[str, float]] = {}
|
||
_LEVEL_CACHE_TTL = 3600
|
||
_last_prices: Dict[str, float] = {}
|
||
_last_price_ts: Dict[str, float] = {}
|
||
_scan_cooldowns: Dict[str, float] = {}
|
||
# Breaches are edge-triggered (only the crossing tick fires), so one that
|
||
# lands while the symbol is inside the rescan cooldown would otherwise be
|
||
# dropped forever (H33b). Park it here; the tick loop fires it once the
|
||
# cooldown expires. symbol → (price, level_name, recorded_at).
|
||
_pending_breach: Dict[str, tuple] = {}
|
||
# A failed breach rescan shrinks the just-set cooldown to this retry
|
||
# window instead of silencing the symbol for the full cooldown (H33a).
|
||
_BREACH_RETRY_SECONDS = 30
|
||
|
||
_cfg = PlatformConfig()
|
||
_WS_COOLDOWN = _cfg.ws_cooldown_seconds
|
||
|
||
_breach_tasks: Set[asyncio.Task] = set()
|
||
|
||
|
||
def _update_level_cache(reports: List[SignalReport]) -> None:
|
||
"""Each report updates only its own timeframe's slot — an M5 scan can
|
||
no longer overwrite the H1/H4 swing levels (H34)."""
|
||
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:
|
||
tf = r.primary_analysis.timeframe
|
||
_level_cache.setdefault(r.symbol, {})[tf] = levels
|
||
_level_cache_ts.setdefault(r.symbol, {})[tf] = now
|
||
|
||
|
||
def _check_level_breach(symbol: str, price: float) -> Optional[str]:
|
||
"""Scan the union of every fresh timeframe's levels; returns
|
||
'<tf> <level_name>' (e.g. '15min swing_high_1') on a crossing so the
|
||
alert label says which timeframe's level broke."""
|
||
per_tf = _level_cache.get(symbol)
|
||
if not per_tf:
|
||
return None
|
||
|
||
last = _last_prices.get(symbol)
|
||
if last is None:
|
||
return None
|
||
|
||
now = time.time()
|
||
ts_map = _level_cache_ts.get(symbol, {})
|
||
for tf, levels in per_tf.items():
|
||
if now - ts_map.get(tf, 0) > _LEVEL_CACHE_TTL:
|
||
continue # stale TF entry — skip, fresher TFs still count
|
||
for name, level in levels.items():
|
||
if (last < level <= price) or (last > level >= price):
|
||
return f"{tf} {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, pair_symbol: str, price: float, level_name: str) -> None:
|
||
now = time.time()
|
||
if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN:
|
||
# Park the crossing instead of dropping it — price holding beyond
|
||
# the level never re-fires the edge trigger, so without this the
|
||
# breach would be lost forever (H33b). The tick loop fires it
|
||
# once the cooldown expires; a newer crossing overwrites it.
|
||
_pending_breach[pair_symbol] = (price, level_name, now)
|
||
return
|
||
|
||
# Early-set on purpose: this is what stops concurrent breach tasks
|
||
# from stacking rescans. The failure path below rolls it back.
|
||
_scan_cooldowns[pair_symbol] = now
|
||
_pending_breach.pop(pair_symbol, None) # this crossing is being handled
|
||
display = PAIR_INFO.get(pair_symbol, {}).get("display", pair_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}")
|
||
# Shrink the just-set cooldown to a short retry window so a
|
||
# transient fetch error doesn't silence the symbol for the full
|
||
# cooldown (H33a); the floor bounds the retry rate while the
|
||
# provider stays down. Re-park the breach so the tick loop
|
||
# retries it (setdefault: a newer crossing recorded meanwhile
|
||
# wins over this stale one).
|
||
retry = min(_BREACH_RETRY_SECONDS, _WS_COOLDOWN)
|
||
_scan_cooldowns[pair_symbol] = now - _WS_COOLDOWN + retry
|
||
_pending_breach.setdefault(pair_symbol, (price, level_name, now))
|
||
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)):
|
||
return
|
||
|
||
key = f"{pair_symbol}_{Timeframe.M15.value}"
|
||
# Lock spans check → send → commit; without it this handler and an
|
||
# in-flight M15 auto-scan both pass _should_alert and double-alert
|
||
# the same signal (H36).
|
||
async with _alert_lock(key):
|
||
if not _should_alert(key, report):
|
||
return
|
||
|
||
# Same post-gate panel attach as the auto-scan path (see
|
||
# _scan_timeframe for the spend/lock rationale).
|
||
await asyncio.get_running_loop().run_in_executor(
|
||
_executor, attach_panel, report)
|
||
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():
|
||
if await _send_alert(bot, chat_id, detail):
|
||
sent = True
|
||
if sent:
|
||
_commit_alert(key, report)
|
||
|
||
|
||
def _breach_task_done(task: asyncio.Task) -> None:
|
||
"""Fire-and-forget breach tasks must still surface their failures —
|
||
without this, an exception sits unretrieved until GC and never
|
||
reaches the log file."""
|
||
_breach_tasks.discard(task)
|
||
if not task.cancelled() and task.exception() is not None:
|
||
logger.error("Breach alert task failed", exc_info=task.exception())
|
||
|
||
|
||
async def _ws_monitor(bot) -> None:
|
||
"""Level-breach tick monitor. The provider owns the connection and
|
||
protocol (`iter_ticks`); this loop owns reconnect/backoff and breach
|
||
dispatch. Providers without streaming quotes simply disable it."""
|
||
dp = _get_streaming_dp()
|
||
if not dp.supports_streaming:
|
||
logger.info(
|
||
"Data provider %r has no streaming quotes — "
|
||
"level-breach monitor disabled", dp.name)
|
||
return
|
||
if getattr(dp, "_api_key", "unused") is None:
|
||
# Guard, not a traceback loop: without a key every connect
|
||
# attempt would raise and re-log a full traceback every ≤60s.
|
||
logger.warning(
|
||
"Streaming provider %r has no API key — "
|
||
"level-breach monitor disabled", dp.name)
|
||
return
|
||
|
||
backoff = 1
|
||
while _feed_active():
|
||
got_tick = False
|
||
try:
|
||
async for pair_symbol, price in dp.iter_ticks(SCAN_PAIRS):
|
||
if not got_tick:
|
||
# Reset backoff only once the connection actually
|
||
# produced data — resetting on connect let a
|
||
# rejected key or accept-then-close server pin the
|
||
# backoff at minimum forever (H38a/H38b).
|
||
got_tick = True
|
||
backoff = 1
|
||
if not _feed_active():
|
||
break
|
||
|
||
breached = _check_level_breach(pair_symbol, price)
|
||
_last_prices[pair_symbol] = price
|
||
_last_price_ts[pair_symbol] = time.time()
|
||
|
||
if breached:
|
||
# A fresh crossing supersedes any parked one.
|
||
_pending_breach.pop(pair_symbol, None)
|
||
task = asyncio.create_task(
|
||
_handle_breach(bot, pair_symbol, price, breached)
|
||
)
|
||
_breach_tasks.add(task)
|
||
task.add_done_callback(_breach_task_done)
|
||
elif pair_symbol in _pending_breach:
|
||
pend = _pending_breach[pair_symbol]
|
||
if time.time() - pend[2] > _LEVEL_CACHE_TTL:
|
||
# Ancient parked breach (feed gap/weekend) — an
|
||
# alert about an hours-old crossing would only
|
||
# mislead; the level cache it came from is stale
|
||
# anyway.
|
||
_pending_breach.pop(pair_symbol)
|
||
elif (time.time() - _scan_cooldowns.get(pair_symbol, 0)
|
||
>= _WS_COOLDOWN):
|
||
# Crossing parked during the cooldown: fire it now
|
||
# that the window expired — the edge trigger alone
|
||
# would have dropped it forever (H33b).
|
||
_pending_breach.pop(pair_symbol)
|
||
task = asyncio.create_task(
|
||
_handle_breach(bot, pair_symbol, pend[0], pend[1])
|
||
)
|
||
_breach_tasks.add(task)
|
||
task.add_done_callback(_breach_task_done)
|
||
|
||
_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)
|
||
else:
|
||
# Clean server close: the async-for ended without raising.
|
||
# This must back off and log like the error path — silently
|
||
# reconnecting with zero delay hot-loops against a server
|
||
# that accepts and immediately closes (H38a).
|
||
if _feed_active():
|
||
logger.warning(
|
||
f"WebSocket closed cleanly, 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 _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")
|
||
try:
|
||
await _scan_timeframe(
|
||
bot, tf, AUTOSCAN_CONFIRMATIONS[tf]
|
||
)
|
||
except Exception:
|
||
# One failed cycle must not kill the loop for good —
|
||
# log it and wait for the next candle boundary.
|
||
logger.exception(f"{label} auto-scan: cycle failed")
|
||
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
|
||
# Per-TF respawn (2026-08-04 audit H37): drop finished loop tasks and
|
||
# recreate only the missing ones. The old all-or-nothing alive check
|
||
# left a single dead TF loop unmonitored forever while its siblings
|
||
# lived — and with the channel feed active, /autoscan never reaches
|
||
# _stop_scheduler, so nothing short of a process restart recovered it.
|
||
_autoscan_tasks[:] = [t for t in _autoscan_tasks if not t.done()]
|
||
alive_names = {t.get_name() for t in _autoscan_tasks}
|
||
started = []
|
||
for tf, interval in _AUTOSCAN_LOOPS:
|
||
name = f"candle_loop_{tf.name}"
|
||
if name in alive_names:
|
||
continue
|
||
_autoscan_tasks.append(
|
||
asyncio.create_task(_candle_loop(bot, tf, interval), name=name)
|
||
)
|
||
started.append(tf.name)
|
||
if started:
|
||
logger.info("Candle-boundary scheduler started (%s)",
|
||
"/".join(started))
|
||
|
||
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"
|
||
"/news — Net news sentiment score for XAU/USD\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 _fmt_age(ts: Optional[float]) -> str:
|
||
"""Compact age like '45s', '12m', '3.2h' for status lines."""
|
||
if not ts:
|
||
return "?"
|
||
s = time.time() - ts
|
||
if s < 90:
|
||
return f"{s:.0f}s"
|
||
if s < 5400:
|
||
return f"{s / 60:.0f}m"
|
||
return f"{s / 3600:.1f}h"
|
||
|
||
|
||
def _news_health_lines() -> List[str]:
|
||
"""News / calendar / DeepSeek source health for /status.
|
||
|
||
Reads the class-level `NewsEngine._source_health` written by the
|
||
scan-path fetches plus the LLM client's health stamps — display
|
||
only, never triggers a fetch. ⚠️ marks a degraded dependency;
|
||
details live in the rotating log (transition-logged WARNINGs)."""
|
||
lines = ["\n*News & AI:*"]
|
||
# Snapshot: a scan thread may insert a source's FIRST entry while we
|
||
# iterate (dict-changed-size RuntimeError); entry mutation is safe.
|
||
for _ in range(3):
|
||
try:
|
||
sh = dict(NewsEngine._source_health)
|
||
break
|
||
except RuntimeError:
|
||
continue
|
||
else:
|
||
sh = {}
|
||
if not sh:
|
||
lines.append(" _no news cycle yet (fills after first scan)_")
|
||
else:
|
||
cal = sh.get("calendar")
|
||
ff = sh.get("calendar:forexfactory")
|
||
if cal:
|
||
if NewsEngine.source_is_failing(cal):
|
||
lines.append(f" ⚠️ Calendar: unavailable — {cal['detail']}"
|
||
f" ({_fmt_age(cal['err'])})")
|
||
elif NewsEngine.source_is_failing(ff):
|
||
lines.append(f" ⚠️ Calendar: {cal['extra']} — ForexFactory"
|
||
f" down: {ff['detail']} ({_fmt_age(ff['err'])})")
|
||
else:
|
||
lines.append(f" ✅ Calendar: {cal['extra']}"
|
||
f" ({_fmt_age(cal['ok'])} ago)")
|
||
heads = sh.get("headlines")
|
||
if heads:
|
||
failing = NewsEngine.source_is_failing(heads)
|
||
mark = "⚠️" if failing else "✅"
|
||
detail = f" — {heads['detail']}" if failing else ""
|
||
ts = heads["err"] if failing else heads["ok"]
|
||
lines.append(f" {mark} Headlines: {heads['extra']}{detail}"
|
||
f" ({_fmt_age(ts)} ago)")
|
||
for name, entry in sorted(sh.items()):
|
||
if name.startswith("rss:") and NewsEngine.source_is_failing(entry):
|
||
lines.append(f" ⚠️ {name[4:]}: {entry['detail']}"
|
||
f" ({_fmt_age(entry['err'])})")
|
||
client = get_llm_client()
|
||
if not client.available:
|
||
lines.append(" DeepSeek: off — keyword scoring only")
|
||
else:
|
||
# The client's counter only resets on the first call of a new
|
||
# UTC day — show 0 (not yesterday's total, or a phantom
|
||
# cap-exhausted warning) until that reset happens.
|
||
today = datetime.now(timezone.utc).date()
|
||
used = client._count if client._count_day == today else 0
|
||
cap = client._daily_cap
|
||
scored = len(NewsEngine._llm_sent_cache)
|
||
err = client.last_error
|
||
if used >= cap:
|
||
lines.append(f" ⚠️ DeepSeek: daily cap {cap} exhausted —"
|
||
" keyword fallback until UTC midnight")
|
||
elif err and err[0] > client.last_ok:
|
||
lines.append(f" ⚠️ DeepSeek: last call failed ({err[1]},"
|
||
f" {_fmt_age(err[0])}) — keyword fallback")
|
||
else:
|
||
lines.append(f" ✅ DeepSeek: {used}/{cap} calls today,"
|
||
f" {scored} headlines scored")
|
||
if get_app_config().agent_panel_settings.get("enabled", False):
|
||
ph = panel_health()
|
||
lines.append(f" 🤖 AI panel: {ph['used_today']}/{ph['cap']}"
|
||
" panels today")
|
||
return lines
|
||
|
||
|
||
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():
|
||
provider = _streaming_dp.name if _streaming_dp else "?"
|
||
lines.append(f"📡 WebSocket: *connected* ({provider})")
|
||
else:
|
||
lines.append("❌ WebSocket: *disconnected*")
|
||
|
||
if _last_prices:
|
||
lines.append("\n*Live Prices (WebSocket):*")
|
||
for pair in SCAN_PAIRS:
|
||
price = _last_prices.get(pair)
|
||
display = PAIR_INFO.get(pair, {}).get("display", pair)
|
||
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(levels) for per_tf in _level_cache.values()
|
||
for levels in per_tf.values()
|
||
)
|
||
if cached_levels:
|
||
all_ts = [
|
||
ts for per_tf in _level_cache_ts.values()
|
||
for ts in per_tf.values()
|
||
]
|
||
oldest_age = time.time() - min(all_ts) if all_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.extend(_news_health_lines())
|
||
|
||
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)
|
||
|
||
|
||
def _news_text(context: FundamentalContext, engine: NewsEngine) -> str:
|
||
net, desc = engine.get_pair_sentiment(XAUUSD, context)
|
||
if round(net, 2) == 0:
|
||
net = 0.0 # a tiny negative mean would render as "-0.00"
|
||
lines = [
|
||
"📰 *XAU/USD News Score*\n",
|
||
f"*Net score:* `{net:+.2f}` — {desc}",
|
||
]
|
||
# A cold/failed feed yields 0.00 "Neutral" — flag it so a data outage
|
||
# can't be mistaken for a genuinely neutral read.
|
||
if not context.news_sentiment:
|
||
lines.append("_no scored headlines yet — score defaults to 0_")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _news_report_sync() -> str:
|
||
engine = NewsEngine(PlatformConfig().news)
|
||
return _news_text(engine.get_fundamental_context(), engine)
|
||
|
||
|
||
@_owner_only
|
||
async def cmd_news(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
msg = await update.message.reply_text("⏳ Fetching news…")
|
||
loop = asyncio.get_running_loop()
|
||
try:
|
||
text = await loop.run_in_executor(_executor, _news_report_sync)
|
||
except Exception as exc:
|
||
logger.exception("News fetch failed")
|
||
text = f"❌ News fetch failed: {_escape_md(str(exc))}"
|
||
await msg.edit_text(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"),
|
||
("news", "XAU/USD net news sentiment score"),
|
||
("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)
|
||
n_levels = sum(
|
||
len(levels) for per_tf in _level_cache.values()
|
||
for levels in per_tf.values()
|
||
)
|
||
logger.info(f"Level cache populated: {n_levels} 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 ───────────────────────────────────────────────────────
|
||
|
||
async def _on_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
"""python-telegram-bot routes every unhandled handler exception here;
|
||
without a registered error handler PTB only half-logs them."""
|
||
logger.error(
|
||
"Unhandled error in Telegram handler (update=%s)",
|
||
getattr(update, "update_id", update),
|
||
exc_info=context.error,
|
||
)
|
||
|
||
|
||
def _install_excepthooks() -> None:
|
||
"""Route uncaught exceptions from the main thread and worker threads
|
||
into the rotating log file instead of bare stderr, so a crash right
|
||
before exit is still recorded."""
|
||
def hook(exc_type, exc, tb):
|
||
if issubclass(exc_type, KeyboardInterrupt):
|
||
sys.__excepthook__(exc_type, exc, tb)
|
||
return
|
||
logger.critical("Uncaught exception", exc_info=(exc_type, exc, tb))
|
||
|
||
def thread_hook(args):
|
||
if issubclass(args.exc_type, SystemExit):
|
||
return
|
||
name = args.thread.name if args.thread else "?"
|
||
logger.critical(
|
||
f"Uncaught exception in thread {name}",
|
||
exc_info=(args.exc_type, args.exc_value, args.exc_traceback))
|
||
|
||
sys.excepthook = hook
|
||
threading.excepthook = thread_hook
|
||
|
||
|
||
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.")
|
||
|
||
_install_excepthooks()
|
||
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("news", cmd_news))
|
||
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_"))
|
||
app.add_error_handler(_on_error)
|
||
|
||
logger.info("Bot running — press Ctrl+C to stop")
|
||
app.run_polling(allowed_updates=Update.ALL_TYPES)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|