| """ |
| 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}.") |
|
|
| |
| |
| 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." |
|
|
