← Назад"""
Squeeze-VWAP Bot — Telegram Bot
==================================
Команды для управления и мониторинга.
Использует python-telegram-bot v20+ (async).
Отличия от WT Bot:
- /watchlist вместо /whitelist
- /pos показывает Z-VWAP + score
- /status показывает Squeeze-VWAP стратегию
"""
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from src.config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, MAX_POSITIONS
logger = logging.getLogger("bot")
class TelegramBot:
def __init__(self, manager=None, screener=None, exchange=None, tmm=None):
self.manager = manager
self.screener = screener
self.exchange = exchange
self.tmm = tmm # TMMClient
self.z_manager = None # ZatochkiManager — set from main.py
self.app = None
self._chat_id = TELEGRAM_CHAT_ID
async def start(self):
"""Init and start the bot."""
self.app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
self.app.add_handler(CommandHandler("start", self._cmd_start))
self.app.add_handler(CommandHandler("help", self._cmd_help))
self.app.add_handler(CommandHandler("positions", self._cmd_positions))
self.app.add_handler(CommandHandler("pos", self._cmd_positions))
self.app.add_handler(CommandHandler("pnl", self._cmd_pnl))
self.app.add_handler(CommandHandler("watchlist", self._cmd_watchlist))
self.app.add_handler(CommandHandler("wl", self._cmd_watchlist))
self.app.add_handler(CommandHandler("balance", self._cmd_balance))
self.app.add_handler(CommandHandler("bal", self._cmd_balance))
self.app.add_handler(CommandHandler("status", self._cmd_status))
self.app.add_handler(CommandHandler("scan", self._cmd_scan))
self.app.add_handler(CommandHandler("tmm", self._cmd_tmm))
# Zatochki commands
self.app.add_handler(CommandHandler("zstatus", self._cmd_zstatus))
self.app.add_handler(CommandHandler("zstats", self._cmd_zstats))
self.app.add_handler(CommandHandler("zpos", self._cmd_zstatus))
await self.app.initialize()
await self.app.start()
await self.app.updater.start_polling(drop_pending_updates=True)
logger.info("Telegram bot started")
async def stop(self):
"""Stop the bot."""
if self.app:
await self.app.updater.stop()
await self.app.stop()
await self.app.shutdown()
# ============================================================
# NOTIFY
# ============================================================
async def send_message(self, text):
"""Send message to chat."""
if self.app and self._chat_id:
try:
await self.app.bot.send_message(
chat_id=self._chat_id,
text=text,
parse_mode="Markdown",
)
except Exception as e:
logger.error(f"Send message error: {e}")
try:
await self.app.bot.send_message(
chat_id=self._chat_id,
text=text,
)
except Exception:
pass
# ============================================================
# COMMANDS
# ============================================================
async def _cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"🤖 Squeeze-VWAP Bot — Active\n"
"Strategy: Squeeze + Z-VWAP + Waddah 5m\n"
"SL 1.5% / Dynamic TP (Z→fair value)\n\n"
"Commands: /help"
)
async def _cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"📋 *Commands:*\n\n"
"/positions (/pos) — open positions\n"
"/pnl — PnL summary\n"
"/watchlist (/wl) — current watchlist\n"
"/balance (/bal) — USDT balance\n"
"/status — bot status\n"
"/scan — force market scan\n"
"/tmm — TMM journal summary\n",
parse_mode="Markdown",
)
async def _cmd_positions(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.manager:
await update.message.reply_text("No manager")
return
info = self.manager.get_positions_info()
try:
await update.message.reply_text(info, parse_mode="Markdown")
except Exception:
await update.message.reply_text(info)
async def _cmd_pnl(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.manager:
await update.message.reply_text("No data")
return
summary = self.manager.get_pnl_summary()
try:
await update.message.reply_text(summary, parse_mode="Markdown")
except Exception:
await update.message.reply_text(summary)
async def _cmd_watchlist(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.screener:
await update.message.reply_text("No screener")
return
wl = self.screener.get_watchlist()
if not wl:
await update.message.reply_text("📭 Watchlist empty")
return
lines = [f"📋 *Watchlist ({len(wl)}):*\n"]
for e in wl:
dir_str = "🟢 LONG" if e.get("direction") == 1 else "🔴 SHORT"
sqz = "SQZ" if e.get("is_squeeze") else ("REL" if e.get("squeeze_released") else "---")
lines.append(
f"*{e['symbol']}* {dir_str} score={e.get('score', '?')}/5\n"
f" Z={e.get('z_score', 0):+.2f} Sqz={sqz} "
f"Wad={'STR' if e.get('waddah_strong') else 'low'}"
)
text = "\n".join(lines)
try:
await update.message.reply_text(text, parse_mode="Markdown")
except Exception:
await update.message.reply_text(text)
async def _cmd_balance(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
if not self.exchange:
await update.message.reply_text("No exchange")
return
try:
balance = self.exchange.get_balance()
positions = len(self.manager.positions) if self.manager else 0
await update.message.reply_text(
f"💰 *Balance:* ${balance:.2f} USDT\n"
f"📊 Positions: {positions}/{MAX_POSITIONS}",
parse_mode="Markdown",
)
except Exception as e:
await update.message.reply_text(f"Error: {e}")
async def _cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
wl_count = len(self.screener.get_watchlist()) if self.screener else 0
pos_count = len(self.manager.positions) if self.manager else 0
await update.message.reply_text(
f"🤖 *Squeeze-VWAP Bot Status*\n\n"
f"Strategy: Squeeze + Z-VWAP + Waddah 5m\n"
f"SL: 1.5% | TP: Dynamic (Z→0.5) | Cap: 3%\n"
f"Leverage: 10x | Size: $5\n"
f"Time stop: 15min | BE: +0.7%\n\n"
f"Watchlist: {wl_count}\n"
f"Positions: {pos_count}/{MAX_POSITIONS}\n"
f"Status: ✅ Running",
parse_mode="Markdown",
)
async def _cmd_scan(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Force a market scan."""
if not self.screener:
await update.message.reply_text("No screener")
return
await update.message.reply_text("🔍 Scanning...")
try:
new = self.screener.run_scan()
wl = self.screener.get_watchlist()
await update.message.reply_text(
f"✅ Scan done\n"
f"New: {len(new)} | Watchlist: {len(wl)}"
)
except Exception as e:
await update.message.reply_text(f"❌ Scan error: {e}")
async def _cmd_tmm(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""TMM journal summary."""
if not self.tmm:
await update.message.reply_text("TMM not configured")
return
try:
summary = self.tmm.generate_summary()
await update.message.reply_text(summary, parse_mode="Markdown")
except Exception:
try:
summary = self.tmm.generate_summary()
await update.message.reply_text(summary)
except Exception as e:
await update.message.reply_text(f"TMM error: {e}")
# ============================================================
# ZATOCHKI COMMANDS
# ============================================================
async def _cmd_zstatus(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Zatochki positions."""
if not self.z_manager:
await update.message.reply_text("Zatochki not initialized")
return
info = self.z_manager.get_positions_info()
try:
await update.message.reply_text(info, parse_mode="Markdown")
except Exception:
await update.message.reply_text(info)
async def _cmd_zstats(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Zatochki trade stats."""
if not self.z_manager:
await update.message.reply_text("Zatochki not initialized")
return
stats = self.z_manager.get_stats()
try:
await update.message.reply_text(stats, parse_mode="Markdown")
except Exception:
await update.message.reply_text(stats)