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

Telegram: subscriber channel feed, owner lock, full risk disclaimer

Browse Source 
- TELEGRAM_CHANNEL_ID mirrors auto-scan signal + breach alerts to a
  subscriber channel (Stars-subscription monetization; Telegram manages
  membership). Feed keeps running even when the owner's DM auto-scan
  is toggled off.
- _owner_only decorator locks all commands/buttons to TELEGRAM_CHAT_ID;
  others get a refusal pointing at the channel.
- Every signal-bearing message (detailed signal, scan summary, all-TF
  summary, breach alerts) now ends with a not-financial-advice /
  educational-use disclaimer via a single _DISCLAIMER constant.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Harwinder Sekhon
2026-07-11 18:37:50 -04:00
parent ec13cd4c07
commit 43d9064d1e
2 changed files with 143 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_bot.py
+139 -36
View File
@@ -18,12 +18,26 @@ WebSocket: Streams live ticks from TwelveData for XAU/USD.
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> 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
@@ -79,6 +93,18 @@ logger = logging.getLogger("telegram_bot")
TOKEN = os.environ.get("TELEGRAM_TOKEN", "8228251480:AAFW2OAyQ2idMicql2t_CIXk1r9-qfRTfNE")
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022")
# 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 = os.environ.get("TELEGRAM_CHANNEL_ID", "-1003942265371")
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
@@ -225,6 +251,13 @@ def _live_price_line(symbol: str) -> Optional[str]:
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, "⬜")
@@ -310,8 +343,7 @@ def format_signal_message(report: SignalReport) -> str:
f"behind best price — TP1 is a guide, let it run"
)
lines += ["", "━━━━━━━━━━━━━━━━━━━━",
"_Educational / research use only._"]
lines += ["", "━━━━━━━━━━━━━━━━━━━━", _DISCLAIMER]
return "\n".join(lines)
@@ -342,6 +374,8 @@ def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
else:
lines.append("No actionable signals this cycle.")
lines += ["", _DISCLAIMER]
return "\n".join(lines)
@@ -477,6 +511,20 @@ _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)
@@ -515,7 +563,7 @@ async def _scan_timeframe(
if _passes_quality(r.signal_quality)
and _should_alert(f"{r.symbol}_{tf.value}", r)
]
for chat_id in list(_autoscan_chats):
for chat_id in _signal_targets():
for report in actionable:
detail = format_signal_message(report)
try:
@@ -610,7 +658,7 @@ async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, le
f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
)
detail = header + format_signal_message(report)
for chat_id in list(_autoscan_chats):
for chat_id in _signal_targets():
try:
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception as exc:
@@ -627,7 +675,7 @@ async def _ws_heartbeat(ws) -> None:
async def _ws_monitor(bot) -> None:
backoff = 1
while _autoscan_chats:
while _feed_active():
hb_task: Optional[asyncio.Task] = None
try:
async with websockets.connect(_WS_URL) as ws:
@@ -641,7 +689,7 @@ async def _ws_monitor(bot) -> None:
hb_task = asyncio.create_task(_ws_heartbeat(ws))
async for raw in ws:
if not _autoscan_chats:
if not _feed_active():
break
try:
@@ -699,11 +747,11 @@ async def _ws_monitor(bot) -> None:
async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
label = tf.value
try:
while _autoscan_chats:
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 _autoscan_chats:
if not _feed_active():
break
logger.info(f"{label} auto-scan: firing")
await _scan_timeframe(
@@ -758,6 +806,35 @@ def _stop_scheduler() -> None:
# ── 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 = (
@@ -788,10 +865,14 @@ _HELP_TEXT = (
"*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)."
"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)
@@ -806,6 +887,9 @@ def _status_text() -> str:
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:
@@ -848,10 +932,12 @@ def _status_text() -> str:
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:
@@ -881,6 +967,7 @@ async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
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()
@@ -890,17 +977,19 @@ async def cmd_clearcache(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
)
@_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 _autoscan_chats:
if not _feed_active():
_stop_scheduler()
await update.message.reply_text(
"⏹ Auto-scan stopped.",
parse_mode=ParseMode.MARKDOWN,
)
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)
@@ -921,6 +1010,7 @@ async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
)
@_owner_only
async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text(
f"🔍 *{_PAIRS_LABEL} — Pick a timeframe*",
@@ -929,6 +1019,7 @@ async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
)
@_owner_only
async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query
await query.answer()
@@ -954,6 +1045,7 @@ async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
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
@@ -964,9 +1056,11 @@ async def cb_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
chat_id = query.message.chat.id
if chat_id in _autoscan_chats:
_autoscan_chats.discard(chat_id)
if not _autoscan_chats:
if not _feed_active():
_stop_scheduler()
await query.answer("Auto-scan stopped")
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)
@@ -1044,6 +1138,7 @@ async def _cb_scan_all(query, symbol: str = XAUUSD) -> None:
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())
@@ -1062,13 +1157,16 @@ async def _post_init(app: Application) -> None:
("help", "All commands"),
])
if not CHAT_ID:
logger.info("TELEGRAM_CHAT_ID not set — auto-scan disabled on boot. "
"Use /autoscan in chat to enable.")
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
chat_id = int(CHAT_ID)
_autoscan_chats.add(chat_id)
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:
@@ -1094,23 +1192,28 @@ async def _post_init(app: Application) -> None:
next_h1 = _seconds_until_boundary(60, buffer=10)
next_h4 = _seconds_until_boundary(240, buffer=10)
logger.info(
f"Auto-scan started for chat {chat_id} — "
f"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"
)
try:
await app.bot.send_message(
chat_id,
f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {_PAIRS_LABEL} on M5/M15/H1/H4.\n"
f"📡 Live price monitoring active.\n"
f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
f"/autoscan to stop",
parse_mode=ParseMode.MARKDOWN,
)
except Exception:
logger.exception("Failed to send boot message")
# 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 ─────────────────────────────────────────────────