โ ะะฐะทะฐะด#!/usr/bin/env python3
"""
AlphaPulseXP Telegram Bot v2.0 - Enhanced Crypto Content
RSS + Social + On-chain aggregator with AI commentary & interactive features
"""
import os
import sys
import logging
import sqlite3
import asyncio
import re
import random
import json
import feedparser
import requests
from datetime import datetime, timedelta
from pathlib import Path
from time import sleep
from telegram import Bot, Poll
from telegram.error import TelegramError
# Configuration
CONFIG = {
'bot_token': '8716831731:AAGtulzWTeqXzAgMTdbJvvf22upCkvDAoeA',
'channel_id': '@alphapulsexp',
'openrouter_api_key': os.environ.get('OPENROUTER_API_KEY', ''),
'cryptopanic_api_key': os.environ.get('CRYPTOPANIC_API_KEY', ''),
'posting_schedule': [9, 13, 17, 21], # UTC hours
'max_posts_per_cycle': 1, # One post per scheduled time
'enable_polls': True,
'poll_frequency': 3, # Every Nth post is a poll
}
# Quality RSS Sources
RSS_SOURCES = {
'CoinDesk': 'https://www.coindesk.com/arc/outboundfeeds/rss/',
'Cointelegraph': 'https://cointelegraph.com/rss',
'Decrypt': 'https://decrypt.co/feed',
}
# Post format weights (probability)
POST_FORMATS = {
'hot': 0.15, # ๐ฅ Urgent/breaking
'analysis': 0.25, # ๐ง Market analysis
'sarcasm': 0.20, # ๐ Sarcastic/meme
'facts': 0.20, # ๐ Just numbers
'signal': 0.15, # ๐๐ป Bull/bear signal
'fear_greed': 0.05, # ๐ Fear & Greed (once per day)
}
# Time-based themes
TIME_THEMES = {
9: {"emoji": "๐
", "theme": "Asia Wrap", "style": "night_moves"},
13: {"emoji": "โ", "theme": "Europe Session", "style": "midday_update"},
17: {"emoji": "๐", "theme": "Daily Digest", "style": "summary"},
21: {"emoji": "๐", "theme": "US Close", "style": "wrap_up"},
}
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('bot.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class Database:
def __init__(self, db_path='posts.db'):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_db()
def _init_db(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS posted (
id TEXT PRIMARY KEY,
title TEXT,
url TEXT,
source TEXT,
post_type TEXT,
posted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS daily_stats (
date TEXT PRIMARY KEY,
fear_greed_posted INTEGER DEFAULT 0
)
''')
self.conn.commit()
def is_posted(self, post_id):
cursor = self.conn.cursor()
cursor.execute('SELECT id FROM posted WHERE id = ?', (post_id,))
return cursor.fetchone() is not None
def mark_posted(self, post_id, title, url, source, post_type='regular'):
cursor = self.conn.cursor()
cursor.execute('INSERT OR REPLACE INTO posted (id, title, url, source, post_type) VALUES (?, ?, ?, ?, ?)',
(post_id, title, url, source, post_type))
self.conn.commit()
def was_fear_greed_posted_today(self):
today = datetime.utcnow().strftime('%Y-%m-%d')
cursor = self.conn.cursor()
cursor.execute('SELECT fear_greed_posted FROM daily_stats WHERE date = ?', (today,))
result = cursor.fetchone()
return result and result[0] == 1
def mark_fear_greed_posted(self):
today = datetime.utcnow().strftime('%Y-%m-%d')
cursor = self.conn.cursor()
cursor.execute('INSERT OR REPLACE INTO daily_stats (date, fear_greed_posted) VALUES (?, 1)', (today,))
self.conn.commit()
class ContentFetcher:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
def fetch_rss(self):
"""Fetch from quality RSS sources"""
posts = []
for source_name, url in RSS_SOURCES.items():
try:
logger.info(f"Fetching {source_name}...")
feed = feedparser.parse(url)
for entry in feed.entries[:5]:
post_id = f"rss_{source_name}_{hash(entry.link)}"
posts.append({
'id': post_id,
'title': entry.title,
'url': entry.link,
'source': source_name,
'published': entry.get('published', ''),
'summary': entry.get('summary', '')[:200],
})
sleep(1)
except Exception as e:
logger.error(f"RSS fetch error {source_name}: {e}")
return posts
def fetch_fear_greed_index(self):
"""Fetch Crypto Fear & Greed Index"""
try:
url = "https://api.alternative.me/fng/?limit=1"
response = self.session.get(url, timeout=30)
data = response.json()
if data and 'data' in data and len(data['data']) > 0:
item = data['data'][0]
return {
'id': f"fng_{item['timestamp']}",
'value': int(item['value']),
'classification': item['value_classification'],
'timestamp': item['timestamp'],
}
except Exception as e:
logger.error(f"Fear & Greed fetch error: {e}")
return None
def fetch_reddit_crypto(self):
"""Fetch hot posts from r/CryptoCurrency"""
try:
url = "https://www.reddit.com/r/CryptoCurrency/hot.json?limit=10"
headers = {'User-Agent': 'AlphaPulseBot/2.0'}
response = self.session.get(url, headers=headers, timeout=30)
data = response.json()
posts = []
for post in data.get('data', {}).get('children', []):
post_data = post['data']
if not post_data.get('is_self') and post_data.get('score', 0) > 50:
posts.append({
'id': f"reddit_{post_data['id']}",
'title': post_data['title'],
'url': f"https://reddit.com{post_data['permalink']}",
'source': 'Reddit',
'score': post_data['score'],
})
return posts
except Exception as e:
logger.error(f"Reddit fetch error: {e}")
return []
class AICommentary:
def __init__(self, api_key):
self.api_key = api_key
self.url = "https://openrouter.ai/api/v1/chat/completions"
def generate_commentary(self, title, source, post_format, time_context):
"""Generate commentary based on format"""
format_prompts = {
'hot': f"""Breaking crypto news. Write like it's urgent and exciting but NOT cringe.
Use ๐ฅ emoji. 1-2 sentences. Be real, mention $TICKER if relevant.
News: {title}""",
'analysis': f"""Explain what this crypto news MEANS for the market.
What's the implication? 2-3 sentences. Be smart, not hype-y.
News: {title}""",
'sarcasm': f"""React to this crypto news with mild sarcasm/irony.
Like a tired trader who's seen it all. 1-2 sentences. Can be funny.
News: {title}""",
'facts': f"""Just the key facts about this crypto development.
No opinions. 1-2 sentences. What changed?
News: {title}""",
'signal': f"""Is this bullish ๐ or bearish ๐ป for crypto? Pick ONE and explain briefly why.
Don't be neutral. 1-2 sentences.
News: {title}""",
}
prompt = format_prompts.get(post_format, format_prompts['facts'])
if time_context:
prompt += f"\n\nContext: It's {time_context['emoji']} {time_context['theme']} time."
try:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek/deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.8,
"max_tokens": 150
}
response = requests.post(self.url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
content = data['choices'][0]['message']['content'].strip()
content = re.sub(r'^["\']+|["\']+$', '', content)
if len(content) > 400:
content = content[:400] + "..."
return content
except Exception as e:
logger.error(f"AI generation error: {e}")
return None
def generate_fear_greed_commentary(self, value, classification):
"""Generate commentary for Fear & Greed Index"""
prompts = {
'Extreme Fear': "Crypto Fear & Greed is at {value} (Extreme Fear ๐ฑ). Is this the bottom or are we going lower?",
'Fear': "Fear & Greed at {value} โ Fear zone. Blood in the streets or opportunity?",
'Neutral': "Markets are neutral at {value}. Everyone's waiting for something to happen.",
'Greed': "Greed index at {value}. FOMO is real. Are we due for a pullback?",
'Extreme Greed': "EXTREME GREED at {value} ๐ค. Euphoria everywhere. This rarely ends well...",
}
base_text = prompts.get(classification, f"Fear & Greed Index: {value} ({classification})")
base_text = base_text.format(value=value)
if value <= 20:
base_text += "\n\n๐ก Historical buying zone"
elif value >= 75:
base_text += "\n\nโ ๏ธ Consider taking profits"
return base_text
class TelegramPoster:
def __init__(self, token, channel_id):
self.bot = Bot(token=token)
self.channel_id = channel_id
self.post_counter = 0
def get_format_emoji(self, post_format):
emojis = {
'hot': '๐ฅ',
'analysis': '๐ง ',
'sarcasm': '๐',
'facts': '๐',
'signal': '๐',
'fear_greed': '๐๏ธ',
}
return emojis.get(post_format, '๐ฐ')
async def post(self, text, url, source, post_format='regular'):
"""Post to Telegram channel"""
try:
emoji = self.get_format_emoji(post_format)
if post_format == 'fear_greed':
message = f"{emoji} <b>Crypto Fear & Greed Index</b>\n\n{text}"
else:
message = f"{emoji} {text}\n\nโ {source}"
if url:
message += f"\n๐ <a href='{url}'>Read more</a>"
await self.bot.send_message(
chat_id=self.channel_id,
text=message,
parse_mode='HTML',
disable_web_page_preview=False
)
logger.info(f"Posted [{post_format}] to channel: {source}")
self.post_counter += 1
return True
except TelegramError as e:
logger.error(f"Telegram error: {e}")
return False
async def post_poll(self, question, options):
"""Post a poll to channel"""
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 to channel")
self.post_counter += 1
return True
except TelegramError as e:
logger.error(f"Poll error: {e}")
return False
class AlphaPulseBot:
def __init__(self):
self.db = Database()
self.fetcher = ContentFetcher()
self.ai = AICommentary(CONFIG['openrouter_api_key'])
self.telegram = TelegramPoster(CONFIG['bot_token'], CONFIG['channel_id'])
self.cycle_count = 0
def select_post_format(self, current_hour):
"""Select post format with time-based adjustments"""
# Check if fear & greed was posted today
if current_hour == 17 and not self.db.was_fear_greed_posted_today():
if random.random() < 0.3: # 30% chance at evening
return 'fear_greed'
# Weighted random selection
formats = list(POST_FORMATS.keys())
weights = list(POST_FORMATS.values())
return random.choices(formats, weights=weights)[0]
def get_poll_question(self):
"""Get random poll question"""
polls = [
("BTC by end of week?", ["๐ Up", "โฌ๏ธ Down", "๐ Crab"]),
("What moves the market most?", ["๐ฐ News", "๐ณ Whales", "๐ Tech", "๐ฃ๏ธ Social"]),
("Your current position?", ["๐ Bull", "๐ป Bear", "๐ฆ Sideways"]),
("Which sector pumps next?", ["BTC", "ETH/L2s", "Solana", "Memecoins", "AI coins"]),
("Reason for latest dip?", ["๐ Liquidation", "๐ฑ Panic", "๐ฐ FUD", "Just crypto things"]),
]
return random.choice(polls)
async def run_cycle(self):
"""One posting cycle"""
now = datetime.utcnow()
current_hour = now.hour
self.cycle_count += 1
# Check if it's a poll post
if CONFIG['enable_polls'] and self.cycle_count % CONFIG['poll_frequency'] == 0:
question, options = self.get_poll_question()
await self.telegram.post_poll(question, options)
return
# Select post format
post_format = self.select_post_format(current_hour)
# Handle Fear & Greed Index
if post_format == '