"""
AlphaPulse Bot โ AI Commentary (OpenRouter)
"""
import re
import asyncio
import logging
import requests
from config import AI_RETRY_COUNT, AI_RETRY_DELAY
logger = logging.getLogger(__name__)
# โโ Bot Personality System Prompt โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
SYSTEM_PROMPT = (
"You are AlphaPulse โ a sharp crypto analyst with a dry wit. "
"You've traded through every cycle since 2017, seen 3 bear markets, "
"and survived the FTX collapse with your portfolio intact. "
"Your style: confident but not arrogant, data-driven, occasionally sarcastic. "
"You use $TICKER notation for coins. "
"You never use emojis in your text (the template handles that). "
"You never say 'buckle up', 'strap in', 'DYOR', 'not financial advice', 'to the moon'. "
"You sound like a smart trader talking to other traders, not a hype influencer. "
"Keep it concise โ Telegram readers scroll fast."
)
class AICommentary:
def __init__(self, api_key: str):
self.api_key = api_key
self.url = "https://openrouter.ai/api/v1/chat/completions"
async def generate(self, title: str, source: str, post_format: str,
time_theme: dict, prices: dict = None) -> str | None:
price_ctx = ''
if prices:
arrow = 'โฒ' if prices['btc_change'] > 0 else 'โผ'
price_ctx = (
f"\nMarket context: BTC ${prices['btc_price']:,.0f} "
f"({arrow}{abs(prices['btc_change']):.1f}% 24h)"
)
prompts = {
'hot': (
f"React to this breaking news in 1-2 punchy sentences. "
f"What should traders do RIGHT NOW? Mention $TICKER if relevant.\n"
f"News: {title}{price_ctx}"
),
'analysis': (
f"Analyze the market impact of this news. 2-3 sentences, "
f"connect it to the bigger picture. Who benefits, who gets rekt?\n"
f"News: {title}{price_ctx}"
),
'sarcasm': (
f"React to this crypto news with dry humor. "
f"You've seen this movie before โ what happens next? 1-2 sentences.\n"
f"News: {title}"
),
'facts': (
f"Strip this to the essential facts only. What changed, "
f"what number moved, who did what. 1-2 sentences, zero opinions.\n"
f"News: {title}"
),
'signal': (
f"Is this bullish or bearish? Pick ONE side and give a concrete reason. "
f"End with a specific level to watch if possible.\n"
f"News: {title}{price_ctx}"
),
'thread': (
f"3 key takeaways from this news. Use 1. 2. 3. format. "
f"Each point max 1 sentence. Last point should be actionable.\n"
f"News: {title}{price_ctx}"
),
'hot_take': (
f"Contrarian take: what is everyone getting wrong about this? "
f"Be provocative but back it up. 1-2 sentences.\n"
f"News: {title}"
),
'deep_dive': (
f"What second-order effect is everyone missing? "
f"Connect this news to something unexpected. 2 sentences max.\n"
f"News: {title}{price_ctx}"
),
}
prompt = prompts.get(post_format, prompts['facts'])
if time_theme:
prompt += f"\n[Time: {time_theme['theme']}]"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": "deepseek/deepseek-v3.2",
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": prompt},
],
"temperature": 0.85,
"max_tokens": 200,
}
for attempt in range(1, AI_RETRY_COUNT + 1):
try:
resp = await asyncio.to_thread(
requests.post, self.url, headers=headers, json=payload, timeout=30
)
resp.raise_for_status()
content = resp.json()['choices'][0]['message']['content'].strip()
content = re.sub(r'^["\']+|["\']+$', '', content)
return content[:500]
except Exception as e:
logger.warning(f"AI attempt {attempt}/{AI_RETRY_COUNT} failed: {e}")
if attempt < AI_RETRY_COUNT:
await asyncio.sleep(AI_RETRY_DELAY)
logger.error("AI generation failed after all retries")
return None
async def generate_watchlist(self, trending: list, gainers: list,
funding_rates: list, prices: dict = None) -> tuple[list, str]:
"""Generate Monday watchlist: 5 coins to watch + AI commentary.
Returns (coins_list, ai_summary).
"""
# Build candidate coins from multiple signals
seen = set()
candidates = []
# From trending
for c in (trending or [])[:5]:
sym = c.get('symbol', '').upper()
if sym and sym not in seen:
seen.add(sym)
candidates.append({'symbol': sym, 'reason': 'Trending on CoinGecko'})
# From top gainers (momentum)
for c in (gainers or [])[:5]:
sym = c.get('symbol', '').upper()
pct = c.get('price_change_percentage_24h', 0) or 0
if sym and sym not in seen:
seen.add(sym)
candidates.append({'symbol': sym, 'reason': f'Top gainer +{pct:.1f}%'})
# From extreme funding rates
for r in (funding_rates or [])[:5]:
sym = r.get('symbol', '').upper()
rate = r.get('rate_pct', 0)
if sym and sym not in seen and abs(rate) > 0.03:
seen.add(sym)
direction = "high long bias" if rate > 0 else "high short bias"
candidates.append({'symbol': sym, 'reason': f'Extreme funding ({direction})'})
coins = candidates[:5]
if not coins:
return [], ""
# Generate AI summary
coin_list_str = ", ".join(c['symbol'] for c in coins)
price_ctx = ""
if prices:
price_ctx = f" BTC at ${prices['btc_price']:,.0f}."
prompt = (
f"Weekly watchlist coins: {coin_list_str}.{price_ctx} "
f"Write ONE sentence (max 25 words) โ what's the theme connecting these picks? "
f"Be specific about catalysts, not generic market vibes."
)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": "deepseek/deepseek-v3.2",
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": prompt},
],
"temperature": 0.8,
"max_tokens": 100,
}
summary = ""
try:
resp = await asyncio.to_thread(
requests.post, self.url, headers=headers, json=payload, timeout=30
)
resp.raise_for_status()
summary = resp.json()['choices'][0]['message']['content'].strip()
summary = re.sub(r'^["\']+|["\']+$', '', summary)[:200]
except Exception as e:
logger.warning(f"Watchlist AI summary failed: {e}")
return coins, summary
async def generate_history_reflection(self, events: list, prices: dict = None) -> str:
"""Generate a short AI reflection on historical crypto events."""
events_text = "; ".join(f"{e['year']}: {e['event']}" for e in events[:3])
price_ctx = ""
if prices:
price_ctx = f" BTC is currently at ${prices['btc_price']:,.0f}."
prompt = (
f"Crypto history for today: {events_text}.{price_ctx} "
f"Write ONE witty reflection (max 20 words) connecting then vs now. "
f"Reference a specific number or fact. No cliches."
)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": "deepseek/deepseek-v3.2",
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": prompt},
],
"temperature": 0.9,
"max_tokens": 80,
}
try:
resp = await asyncio.to_thread(
requests.post, self.url, headers=headers, json=payload, timeout=30
)
resp.raise_for_status()
content = resp.json()['choices'][0]['message']['content'].strip()
content = re.sub(r'^["\']+|["\']+$', '', content)
return content[:200]
except Exception as e:
logger.warning(f"History reflection AI failed: {e}")
return ""
@staticmethod
def fear_greed_text(value: int, classification: str) -> str:
texts = {
'Extreme Fear': f"๐ฑ Extreme Fear ({value}). Bottom or going lower?",
'Fear': f"๐ฐ Fear zone ({value}). Blood in the streets โ opportunity?",
'Neutral': f"๐ Neutral ({value}). Everyone's waiting for something.",
'Greed': f"๐ค Greed ({value}). FOMO is real. Pullback soon?",
'Extreme Greed': f"๐ Extreme Greed ({value}). Euphoria rarely ends well...",
}
text = texts.get(classification, f"Fear & Greed: {value} ({classification})")
if value <= 20:
text += "\n๐ก Historical buying zone"
elif value >= 75:
text += "\nโ ๏ธ Consider taking profits"
return text