""" AI assistant (premium) — answers questions using the terminal's OWN live data. Uses Groq's free API (OpenAI-compatible). Set GROQ_API_KEY. The model only ever sees a compact snapshot of the current dashboard state, and a strict system prompt keeps it educational and NON-advisory (no buy/sell calls). """ import os import json import requests import logging logger = logging.getLogger(__name__) GROQ_URL = "https://api.groq.com/openai/v1/chat/completions" _MODEL = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile") _SYSTEM = """You are the assistant for "Sentiments Analyzer", a crypto sentiment analytics tool. Answer ONLY from the DASHBOARD DATA given below. Rules: - Be concise (2–5 sentences), neutral and educational. Explain what the data means. - NEVER give financial advice. Do not tell the user to buy, sell, hold, or enter/ exit anything, and do not predict prices as certainties. If asked for advice or "should I buy/sell", say you can't give financial advice and describe what the data shows instead. - Ground every answer in the provided data. If the data doesn't cover the question, say so plainly — do not invent numbers. - Use plain language. You may mention the sentiment index, narratives, the divergence read, the track record and prices that appear in the data. - When asked WHICH news or WHY the mood moved, point to the specific headlines listed under "Most bearish / bullish headlines" and briefly say what they imply. If no headlines are listed, say the feed shows no strong single driver. """ def ai_configured() -> bool: return bool(os.getenv("GROQ_API_KEY")) def build_context(cache: dict) -> str: """Compact snapshot of the current dashboard for the model.""" if not cache: return "No data available yet." lines = [] s = (cache.get("sentiment_24h") or {}) o = (s.get("overall") or {}) if o: lines.append(f"Sentiment index: {o.get('index')}/100 ({o.get('label')}), " f"basis={o.get('index_basis','?')}. Bullish {o.get('bull_prob')}, " f"bearish {o.get('bear_prob')}, volatility {o.get('vol_prob')}, " f"confidence {o.get('confidence')}.") lines.append(f"Split of stories: {o.get('bullish_pct')}% bullish, " f"{o.get('neutral_pct')}% neutral, {o.get('bearish_pct')}% bearish.") if s.get("trend"): lines.append(f"24h trend: {s.get('trend')} ({s.get('trend_delta_index')} index pts).") dv = cache.get("divergence") or {} if dv.get("state"): lines.append(f"Sentiment-vs-price: {dv.get('headline')} — index {dv.get('index_change')} pts, " f"BTC {dv.get('price_change')}% over {dv.get('window_h')}h. {dv.get('note')}") narr = cache.get("narratives") or [] if narr: top = ", ".join(f"{n.get('name')} ({n.get('net_score')}, " f"{(n.get('sentiment') or {}).get('label','?') if isinstance(n.get('sentiment'),dict) else n.get('label','?')})" for n in narr[:8]) lines.append(f"Top narratives (name, score): {top}.") # Actual headlines driving the mood — so the model can name WHICH news moved # sentiment, sorted by how strongly negative / positive each item scored. feed = [n for n in (cache.get("news_feed") or []) if n.get("title")] if feed: neg = sorted(feed, key=lambda x: x.get("compound", 0))[:5] pos = sorted(feed, key=lambda x: x.get("compound", 0), reverse=True)[:5] if neg and (neg[0].get("compound", 0) < 0): lines.append("Most bearish headlines (title, source, score): " + " | ".join(f"\"{n.get('title')}\" ({n.get('source')}, {round(n.get('compound',0),2)})" for n in neg if n.get("compound", 0) < 0)) if pos and (pos[0].get("compound", 0) > 0): lines.append("Most bullish headlines (title, source, score): " + " | ".join(f"\"{n.get('title')}\" ({n.get('source')}, {round(n.get('compound',0),2)})" for n in pos if n.get("compound", 0) > 0)) sc = cache.get("scorecard") or {} if sc.get("evaluated_calls"): lines.append(f"Track record: {sc.get('evaluated_calls')} matured calls, " f"win-rate {sc.get('win_rate')}% (right direction at 24h), " f"TP-hit rate {sc.get('signal_hit_rate')}% " f"(touched >={sc.get('target_pct',1)}% in 24h).") ac = sc.get("active_call") if ac: lines.append(f"Live call: {ac.get('direction')} since {ac.get('started')}, " f"now {ac.get('current_return')}%, peak {ac.get('peak_return')}%, " f"{ac.get('hours_left')}h left.") prices = cache.get("coin_prices") or [] if prices: px = ", ".join(f"{p.get('symbol','?').upper()} ${p.get('price_usd')}" for p in prices[:6]) lines.append(f"Prices: {px}.") fg = cache.get("fear_greed_current") or {} if fg: lines.append(f"Fear & Greed: {fg.get('value')} ({fg.get('value_classification','')}).") return "\n".join(lines) or "No data available yet." def ask(question: str, cache: dict) -> tuple[bool, str]: key = os.getenv("GROQ_API_KEY") if not key: return False, "The AI assistant isn't configured yet (missing GROQ_API_KEY)." question = (question or "").strip()[:500] if not question: return False, "Ask a question first." context = build_context(cache) payload = { "model": _MODEL, "messages": [ {"role": "system", "content": _SYSTEM + "\n\nDASHBOARD DATA:\n" + context}, {"role": "user", "content": question}, ], "temperature": 0.3, "max_tokens": 400, } try: r = requests.post(GROQ_URL, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, data=json.dumps(payload), timeout=30) if r.status_code == 200: return True, r.json()["choices"][0]["message"]["content"].strip() logger.warning(f"[AI] Groq error {r.status_code}: {r.text[:200]}") return False, "The AI assistant is temporarily unavailable. Try again shortly." except Exception as exc: logger.warning(f"[AI] request failed: {exc}") return False, "Couldn't reach the AI service. Try again shortly."