โ ะะฐะทะฐะด"""
AlphaPulse Bot โ Telegram Poster (v7.0 โ card-style templates)
"""
import asyncio
import logging
from datetime import timedelta
from telegram import Bot
from telegram.error import TelegramError, TimedOut
from config import CONFIG, QUICK_ACTIONS_KEYBOARD
from utils import now_van, build_hashtags
from templates import (
card_news, card_price_snapshot, card_trending,
card_top_movers, card_weekly_digest, card_fear_greed,
card_funding_rates, card_watchlist, card_long_short,
card_history, card_whale_alerts,
)
logger = logging.getLogger(__name__)
class TelegramPoster:
FORMAT_EMOJIS = {
'hot': '๐ฅ',
'analysis': '๐ง ',
'sarcasm': '๐',
'facts': '๐',
'signal': '๐',
'fear_greed':'๐๏ธ',
'thread': '๐งต',
'hot_take': 'โก',
'deep_dive': '๐ฌ',
}
FORMAT_TITLES = {
'hot': 'BREAKING',
'analysis': 'MARKET ANALYSIS',
'sarcasm': 'MARKET VIBES',
'facts': 'KEY FACTS',
'signal': 'TRADE SIGNAL',
'thread': 'KEY POINTS',
'hot_take': 'HOT TAKE',
'deep_dive': 'DEEP DIVE',
}
def __init__(self, token: str, channel_id: str):
self.bot = Bot(token=token)
self.channel_id = channel_id
self.post_counter = 0
async def _send(self, chat_id=None, **kwargs) -> bool:
"""Send with 3-attempt retry. TimedOut is NOT retried."""
target = chat_id or self.channel_id
for attempt in range(1, 4):
try:
await self.bot.send_message(chat_id=target, **kwargs)
self.post_counter += 1
return True
except TimedOut:
logger.warning("Telegram TimedOut โ message likely delivered, skipping retry")
self.post_counter += 1
return True
except TelegramError as e:
logger.warning(f"Telegram attempt {attempt}/3: {e}")
if attempt < 3:
await asyncio.sleep(5)
logger.error("Telegram send failed after 3 retries")
return False
async def post(self, text: str, url: str, source: str,
fmt: str = 'regular', add_affiliate: bool = False,
chat_id: str = None, title: str = '') -> bool:
"""Post a news item with card-style formatting."""
emoji = self.FORMAT_EMOJIS.get(fmt, '๐ฐ')
if fmt == 'fear_greed':
from ai import AICommentary
message = card_fear_greed(0, '', text)
else:
hashtags = build_hashtags(title, source) if title else ''
message = card_news(emoji, text, source, url, hashtags,
add_affiliate=add_affiliate)
keyboard = QUICK_ACTIONS_KEYBOARD if fmt not in ('fear_greed',) else None
success = await self._send(
chat_id=chat_id, text=message,
parse_mode='HTML', disable_web_page_preview=True,
reply_markup=keyboard,
)
if success:
logger.info(f"Posted [{fmt}]: {source}")
return success
async def post_price_snapshot(self, prices: dict, fg: dict = None,
dominance: dict = None,
sparklines: dict = None) -> bool:
"""Daily price card with sparklines and progress bar."""
message = card_price_snapshot(prices, fg, dominance, sparklines)
logger.info("Posted price snapshot")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_trending(self, coins: list) -> bool:
"""Morning trending coins card."""
if not coins:
return False
date_str = now_van().strftime('%b %d, %Y')
message = card_trending(coins, date_str)
logger.info("Posted trending coins")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_top_movers(self, gainers: list, losers: list) -> bool:
"""Top movers card."""
if not gainers and not losers:
return False
message = card_top_movers(gainers, losers)
logger.info("Posted top movers")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_weekly_digest(self, prices: dict, fg: dict = None,
gainers: list = None) -> bool:
"""Sunday weekly summary card."""
now = now_van()
week_start = (now - timedelta(days=6)).strftime('%b %d')
week_end = now.strftime('%b %d')
week_range = f"{week_start} โ {week_end}"
message = card_weekly_digest(prices, fg, gainers, week_range)
logger.info("Posted weekly digest")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_fear_greed(self, value: int, classification: str,
comment: str) -> bool:
"""Standalone Fear & Greed card."""
message = card_fear_greed(value, classification, comment)
logger.info("Posted fear & greed")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_funding_rates(self, rates: list) -> bool:
"""Funding rates card."""
if not rates:
return False
message = card_funding_rates(rates)
logger.info("Posted funding rates")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_watchlist(self, coins: list, ai_commentary: str,
date_str: str) -> bool:
"""Monday watchlist card."""
if not coins:
return False
message = card_watchlist(coins, ai_commentary, date_str)
logger.info("Posted watchlist")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_history(self, events: list, date_str: str,
ai_reflection: str = '') -> bool:
"""This Day in Crypto card."""
if not events:
return False
message = card_history(events, date_str, ai_reflection)
logger.info("Posted 'This Day in Crypto'")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_long_short(self, positions: list) -> bool:
"""Market positioning card."""
if not positions:
return False
message = card_long_short(positions)
logger.info("Posted long/short positioning")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_whale_alerts(self, alerts: list) -> bool:
"""Whale watch card."""
if not alerts:
return False
message = card_whale_alerts(alerts)
logger.info("Posted whale alerts")
return await self._send(
text=message, parse_mode='HTML', disable_web_page_preview=True
)
async def post_poll(self, question: str, options: list) -> bool:
try:
await self.bot.send_poll(
chat_id=self.channel_id,
question=question,
options=options,
is_anonymous=True,
allows_multiple_answers=False,
)
logger.info("Posted poll")
self.post_counter += 1
return True
except TelegramError as e:
logger.error(f"Poll error: {e}")
return False