Update app/app.py

app/app.py

@@ -1,17 +1,13 @@
 """
-CryptoSentinel Co-Pilot — Your AI-Powered Market Analyst.
-
-This application provides a multi-faceted view of the crypto market by:
-- Tracking live prices.
-- Proactively fetching, analyzing, and synthesizing top news into a daily briefing.
-- Offering on-demand deep analysis of any text.
-- Streaming all insights to a dynamic frontend in real-time.
+The Sentinel Protocol — An Autonomous AI Market Analyst.
+Connects real-time news events to market data visualizations.
 """
 import asyncio
 import json
 import os
 from contextlib import asynccontextmanager
 from typing import Optional, Union, List
+import pandas as pd
 
 import httpx
 from fastapi import FastAPI, Request, BackgroundTasks, Form
@@ -23,143 +19,149 @@ from newsapi import NewsApiClient
 # Import our modular service classes
 from .price_fetcher import PriceFetcher
 from .gemini_analyzer import GeminiAnalyzer
+from .chart_generator import generate_price_chart
 
-# --- Application Lifespan for Resource Management ---
-
+# --- Lifespan Manager ---
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Manages application startup and shutdown events."""
     async with httpx.AsyncClient() as client:
-        # Instantiate and store services in the application state
         app.state.price_fetcher = PriceFetcher(client=client, coins=["bitcoin", "ethereum", "dogecoin"])
         app.state.gemini_analyzer = GeminiAnalyzer(client=client)
         app.state.news_api = NewsApiClient(api_key=os.getenv("NEWS_API_KEY"))
+        app.state.coingecko_client = client # For fetching chart data
 
-        # State for caching and queues
-        app.state.daily_briefing_cache = "### Briefing Unavailable\nGenerating the first daily briefing, please check back in a few minutes."
+        app.state.daily_briefing_cache = "### Briefing Unavailable\nSentinel is gathering initial intelligence..."
         app.state.analyzed_news_today: List[dict] = []
         app.state.sentiment_queue: asyncio.Queue = asyncio.Queue()
         app.state.news_queue: asyncio.Queue = asyncio.Queue()
 
-        # Create cancellable background tasks
-        price_task = asyncio.create_task(run_periodic_updates(app.state.price_fetcher, 60))
-        news_task = asyncio.create_task(run_periodic_news_analysis(app, 900))  # 15 mins
-        briefing_task = asyncio.create_task(run_daily_briefing_generation(app, 3600)) # Every hour
-
-        print("🚀 CryptoSentinel Co-Pilot started.")
+        tasks = [
+            run_periodic_updates(app.state.price_fetcher, 60),
+            run_periodic_news_analysis(app, 900),  # 15 mins
+            run_daily_briefing_generation(app, 3600) # Every hour
+        ]
+        background_tasks = [asyncio.create_task(task) for task in tasks]
+        
+        print("🚀 The Sentinel Protocol is online.")
         yield
         
-        print("⏳ Shutting down background tasks...")
-        price_task.cancel()
-        news_task.cancel()
-        briefing_task.cancel()
-        try:
-            await asyncio.gather(price_task, news_task, briefing_task, return_exceptions=True)
-        except asyncio.CancelledError:
-            print("Background tasks cancelled successfully.")
-        print("✅ Shutdown complete.")
-
-async def run_periodic_updates(fetcher: PriceFetcher, interval_seconds: int):
-    """Periodically updates crypto prices."""
+        print("⏳ Shutting down Sentinel protocols...")
+        for task in background_tasks:
+            task.cancel()
+        await asyncio.gather(*background_tasks, return_exceptions=True)
+        print("✅ Sentinel Protocol offline.")
+
+# --- Background Tasks ---
+async def run_periodic_updates(fetcher: PriceFetcher, interval: int):
     while True:
         await fetcher.update_prices_async()
-        await asyncio.sleep(interval_seconds)
+        await asyncio.sleep(interval)
 
-async def run_periodic_news_analysis(app: FastAPI, interval_seconds: int):
-    """Periodically fetches, analyzes, and queues top crypto news."""
+async def run_periodic_news_analysis(app: FastAPI, interval: int):
+    """Fetches news, analyzes it, generates charts, and queues the full payload."""
     while True:
-        print("📰 Fetching latest crypto news...")
+        print("📰 Sentinel is scanning for new market intelligence...")
         try:
-            top_headlines = app.state.news_api.get_everything(
-                q='bitcoin OR ethereum OR crypto OR blockchain',
-                language='en', sort_by='publishedAt', page_size=5
-            )
+            headlines = app.state.news_api.get_everything(q='bitcoin OR ethereum OR solana', language='en', sort_by='publishedAt', page_size=5)
             analyzer: GeminiAnalyzer = app.state.gemini_analyzer
-            for article in top_headlines.get('articles', []):
+            
+            for article in headlines.get('articles', []):
                 title = article.get('title')
-                if title and "[Removed]" not in title:
-                    analysis = await analyzer.analyze_text(title)
-                    analysis['url'] = article.get('url')
-                    await app.state.news_queue.put(analysis)
-                    app.state.analyzed_news_today.append(analysis)
+                if not title or "[Removed]" in title:
+                    continue
+                
+                analysis = await analyzer.analyze_text(title)
+                if not analysis or analysis.get('sentiment') == 'ERROR':
+                    continue
+
+                analysis['url'] = article.get('url')
+                
+                # --- CHART GENERATION LOGIC ---
+                chart_base64 = ""
+                main_entity = analysis.get('entities', [])[0].lower() if analysis.get('entities') else None
+                if main_entity in ["bitcoin", "ethereum", "solana"]: # Add more as needed
+                    event_time = pd.to_datetime(article.get('publishedAt'))
+                    # Fetch last 24h of price data for charting
+                    cg_url = f"https://api.coingecko.com/api/v3/coins/{main_entity}/market_chart"
+                    params = {'vs_currency': 'usd', 'days': '1'}
+                    resp = await app.state.coingecko_client.get(cg_url, params=params)
+                    if resp.status_code == 200:
+                        price_data = resp.json().get('prices', [])
+                        chart_base64 = generate_price_chart(price_data, event_time, main_entity.capitalize())
+                
+                analysis['chart'] = chart_base64
+                await app.state.news_queue.put(analysis)
+                app.state.analyzed_news_today.append(analysis)
+
         except Exception as e:
-            print(f"❌ Error during news analysis: {e}")
-        await asyncio.sleep(interval_seconds)
+            print(f"❌ News analysis pipeline failed: {e}")
+        await asyncio.sleep(interval)
 
-async def run_daily_briefing_generation(app: FastAPI, interval_seconds: int):
-    """Periodically generates and caches the daily market briefing."""
+async def run_daily_briefing_generation(app: FastAPI, interval: int):
+    """Generates the daily briefing."""
+    await asyncio.sleep(120) # Initial delay to gather some news first
     while True:
-        await asyncio.sleep(interval_seconds) # Wait first, then generate
-        print("📝 Generating Daily Market Briefing...")
+        print("📝 Sentinel is synthesizing the Daily Briefing...")
         if app.state.analyzed_news_today:
             analyzer: GeminiAnalyzer = app.state.gemini_analyzer
             briefing = await analyzer.generate_daily_briefing(app.state.analyzed_news_today)
             app.state.daily_briefing_cache = briefing
-            app.state.analyzed_news_today = [] # Clear list for the next cycle
-            print("✅ Daily briefing generated and cached.")
+            app.state.analyzed_news_today = []
+            print("✅ Daily briefing synthesized.")
         else:
-            print("ℹ️ No new news items to analyze for briefing. Skipping.")
-
-# --- FastAPI App Initialization ---
+            print("ℹ️ No new intelligence to synthesize. Waiting for next cycle.")
+        await asyncio.sleep(interval)
 
-app = FastAPI(title="CryptoSentinel Co-Pilot", lifespan=lifespan)
+# --- App and Endpoints ---
+app = FastAPI(title="Sentinel Protocol", lifespan=lifespan)
 templates = Jinja2Templates(directory="templates")
 
 # --- HTML Rendering Helper ---
-def render_analysis_card(payload: dict, is_news: bool = False) -> str:
+def render_analysis_card(payload: dict) -> str:
     s = payload
-    text_to_show = s.get('summary', 'Analysis failed or not available.')
-    if is_news:
-        url = s.get('url', '#')
-        text_to_show = f'<a href="{url}" target="_blank" rel="noopener noreferrer">{s.get("summary", "N/A")}</a>'
+    summary_link = f'<a href="{s.get("url", "#")}" target="_blank" rel="noopener noreferrer">{s.get("summary", "N/A")}</a>'
     impact_class = f"impact-{s.get('impact', 'low').lower()}"
     sentiment_class = f"sentiment-{s.get('sentiment', 'neutral').lower()}"
-    return f"""<div class="card {impact_class}"><blockquote>{text_to_show}</blockquote><div class="grid"><div><strong>Sentiment:</strong> <span class="{sentiment_class}">{s.get('sentiment')} ({s.get('sentiment_score', 0):.2f})</span></div><div><strong>Impact:</strong> {s.get('impact')}</div></div><div class="grid"><div><strong>Topic:</strong> {s.get('topic')}</div><div><strong>Entities:</strong> {', '.join(s.get('entities', []))}</div></div></div>"""
+    chart_html = f'<img src="{s.get("chart")}" alt="Price chart" style="width:100%; margin-top:1rem; border-radius:var(--pico-border-radius);">' if s.get("chart") else ""
+    
+    return f"""<div class="card {impact_class}"><blockquote>{summary_link}</blockquote>{chart_html}<div class="grid"><div><strong>Sentiment:</strong> <span class="{sentiment_class}">{s.get('sentiment')} ({s.get('sentiment_score', 0):.2f})</span></div><div><strong>Impact:</strong> {s.get('impact')}</div></div><div class="grid"><div><strong>Topic:</strong> {s.get('topic')}</div><div><strong>Entities:</strong> {', '.join(s.get('entities', []))}</div></div></div>"""
 
 # --- API Endpoints ---
-
 @app.get("/", response_class=HTMLResponse)
 async def serve_dashboard(request: Request):
     return templates.TemplateResponse("index.html", {"request": request})
 
-@app.get("/api/prices", response_class=HTMLResponse)
-async def get_prices_fragment(request: Request):
-    prices = request.app.state.price_fetcher.get_current_prices()
-    html = "".join(f"<div><strong>{k.capitalize()}:</strong> ${v:,.2f}</div>" if isinstance(v, (int, float)) else f"<div><strong>{k.capitalize()}:</strong> {v}</div>" for k, v in prices.items())
-    return HTMLResponse(content=html)
-
-@app.post("/api/sentiment")
-async def analyze_sentiment(request: Request, background_tasks: BackgroundTasks, text: str = Form(...)):
-    analyzer: GeminiAnalyzer = request.app.state.gemini_analyzer
-    async def task_wrapper():
-        analysis = await analyzer.analyze_text(text)
-        await request.app.state.sentiment_queue.put(analysis)
-    background_tasks.add_task(task_wrapper)
-    return HTMLResponse(content="<small>✅ Queued for deep analysis...</small>")
-
+# ... (keep /api/prices and /api/sentiment as they were) ...
 @app.get("/api/briefing", response_class=HTMLResponse)
 async def get_daily_briefing(request: Request):
     briefing_html = markdown(request.app.state.daily_briefing_cache)
     return HTMLResponse(content=briefing_html)
 
-@app.get("/api/sentiment/stream")
-async def sentiment_stream(request: Request):
-    queue: asyncio.Queue = request.app.state.sentiment_queue
-    async def event_generator():
-        while True:
-            payload = await queue.get()
-            html = render_analysis_card(payload)
-            data_payload = html.replace('\n', '')
-            yield f"event: sentiment_update\ndata: {data_payload}\n\n"
-    return StreamingResponse(event_generator(), media_type="text/event-stream")
-
 @app.get("/api/news/stream")
 async def news_stream(request: Request):
     queue: asyncio.Queue = request.app.state.news_queue
     async def event_generator():
         while True:
             payload = await queue.get()
-            html = render_analysis_card(payload, is_news=True)
+            html = render_analysis_card(payload)
             data_payload = html.replace('\n', '')
             yield f"event: news_update\ndata: {data_payload}\n\n"
-    return StreamingResponse(event_generator(), media_type="text/event-stream")
+    return StreamingResponse(event_generator(), media_type="text/event-stream")
+
+# Add back the other endpoints that were removed for brevity
+@app.get("/api/prices", response_class=HTMLResponse)
+async def get_prices_fragment(request: Request):
+    prices = request.app.state.price_fetcher.get_current_prices()
+    html = "".join(f"<div><strong>{k.capitalize()}:</strong> ${v:,.2f}</div>" if isinstance(v, (int, float)) else f"<div><strong>{k.capitalize()}:</strong> {v}</div>" for k, v in prices.items())
+    return HTMLResponse(content=html)
+
+@app.post("/api/sentiment")
+async def analyze_sentiment(request: Request, background_tasks: BackgroundTasks, text: str = Form(...)):
+    analyzer: GeminiAnalyzer = request.app.state.gemini_analyzer
+    async def task_wrapper():
+        analysis = await analyzer.analyze_text(text)
+        # We can reuse the news queue for manual analysis results
+        await request.app.state.news_queue.put(analysis)
+    background_tasks.add_task(task_wrapper)
+    return HTMLResponse(content="<small>✅ Analysis protocol initiated...</small>")