Update app/app.py

app/app.py

@@ -1,85 +1,78 @@
 """
-CryptoSentinel Pro — High-performance FastAPI application.
+CryptoSentinel Co-Pilot — Your AI-Powered Market Analyst.
 
-This is the main entry point that orchestrates the entire application.
-- Integrates an asynchronous PriceFetcher for live market data.
-- Integrates a sophisticated GeminiAnalyzer for deep text analysis.
-- Implements an automated pipeline to fetch, analyze, and stream top crypto news.
-- Serves the interactive frontend and provides all necessary API endpoints.
+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.
 """
 import asyncio
 import json
 import os
 from contextlib import asynccontextmanager
-from typing import Optional, Union
+from typing import Optional, Union, List
 
 import httpx
-# =================== FIX APPLIED HERE (1 of 2) ===================
 from fastapi import FastAPI, Request, BackgroundTasks, Form
-# =================================================================
 from fastapi.responses import HTMLResponse, StreamingResponse
 from fastapi.templating import Jinja2Templates
-from pydantic import BaseModel, constr
+from markdown import markdown
+from newsapi import NewsApiClient
 
-# Import our modular, asynchronous service classes
+# Import our modular service classes
 from .price_fetcher import PriceFetcher
 from .gemini_analyzer import GeminiAnalyzer
-from newsapi import NewsApiClient
 
 # --- Application Lifespan for Resource Management ---
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """
-    Manages application startup and shutdown events using the modern
-    lifespan context manager.
-    """
+    """Manages application startup and shutdown events."""
     async with httpx.AsyncClient() as client:
-        # Instantiate and store all services in the application state.
+        # 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"))
 
-        # Create separate queues for the two real-time feeds
+        # 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.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 for periodic updates.
-        price_task = asyncio.create_task(
-            run_periodic_updates(app.state.price_fetcher, interval_seconds=30)
-        )
-        news_task = asyncio.create_task(
-            run_periodic_news_analysis(app, interval_seconds=900)  # Run every 15 minutes
-        )
-        
-        print("🚀 CryptoSentinel Pro started successfully.")
+        # 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.")
         yield
         
         print("⏳ Shutting down background tasks...")
         price_task.cancel()
         news_task.cancel()
+        briefing_task.cancel()
         try:
-            await asyncio.gather(price_task, news_task)
+            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):
-    """A robust asyncio background task that periodically updates prices."""
+    """Periodically updates crypto prices."""
     while True:
         await fetcher.update_prices_async()
         await asyncio.sleep(interval_seconds)
 
 async def run_periodic_news_analysis(app: FastAPI, interval_seconds: int):
-    """Fetches, analyzes, and queues top crypto news periodically."""
+    """Periodically fetches, analyzes, and queues top crypto news."""
     while True:
-        print("📰 Fetching latest crypto news for automated analysis...")
+        print("📰 Fetching latest crypto news...")
         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
+                language='en', sort_by='publishedAt', page_size=5
             )
             analyzer: GeminiAnalyzer = app.state.gemini_analyzer
             for article in top_headlines.get('articles', []):
@@ -88,14 +81,28 @@ async def run_periodic_news_analysis(app: FastAPI, interval_seconds: int):
                     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)
         except Exception as e:
-            print(f"❌ Error during news fetching or analysis: {e}")
-            
+            print(f"❌ Error during news analysis: {e}")
         await asyncio.sleep(interval_seconds)
 
+async def run_daily_briefing_generation(app: FastAPI, interval_seconds: int):
+    """Periodically generates and caches the daily market briefing."""
+    while True:
+        await asyncio.sleep(interval_seconds) # Wait first, then generate
+        print("📝 Generating Daily Market 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.")
+        else:
+            print("ℹ️ No new news items to analyze for briefing. Skipping.")
+
 # --- FastAPI App Initialization ---
 
-app = FastAPI(title="CryptoSentinel Pro", lifespan=lifespan)
+app = FastAPI(title="CryptoSentinel Co-Pilot", lifespan=lifespan)
 templates = Jinja2Templates(directory="templates")
 
 # --- HTML Rendering Helper ---
@@ -107,19 +114,7 @@ def render_analysis_card(payload: dict, is_news: bool = False) -> str:
         text_to_show = f'<a href="{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>
-    """
+    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>"""
 
 # --- API Endpoints ---
 
@@ -129,33 +124,24 @@ async def serve_dashboard(request: Request):
 
 @app.get("/api/prices", response_class=HTMLResponse)
 async def get_prices_fragment(request: Request):
-    price_fetcher: PriceFetcher = request.app.state.price_fetcher
-    prices = price_fetcher.get_current_prices()
-    html_fragment = "".join(
-        f"<div><strong>{coin.capitalize()}:</strong> ${price:,.2f}</div>" if isinstance(price, (int, float))
-        else f"<div><strong>{coin.capitalize()}:</strong> {price}</div>"
-        for coin, price in prices.items()
-    )
-    return HTMLResponse(content=html_fragment)
-
-# =================== FIX APPLIED HERE (2 of 2) ===================
+    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(...)  # Tell FastAPI to expect a form field named 'text'
-):
-# =================================================================
-    """Queues user-submitted text for a full Gemini-powered analysis."""
+async def analyze_sentiment(request: Request, background_tasks: BackgroundTasks, text: str = Form(...)):
     analyzer: GeminiAnalyzer = request.app.state.gemini_analyzer
-
-    async def analysis_task_wrapper():
-        analysis_result = await analyzer.analyze_text(text) # Use the 'text' variable directly
-        await request.app.state.sentiment_queue.put(analysis_result)
-
-    background_tasks.add_task(analysis_task_wrapper)
+    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>")
 
+@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
@@ -164,8 +150,7 @@ async def sentiment_stream(request: Request):
             payload = await queue.get()
             html = render_analysis_card(payload)
             data_payload = html.replace('\n', '')
-            sse_message = f"event: sentiment_update\ndata: {data_payload}\n\n"
-            yield sse_message
+            yield f"event: sentiment_update\ndata: {data_payload}\n\n"
     return StreamingResponse(event_generator(), media_type="text/event-stream")
 
 @app.get("/api/news/stream")
@@ -176,6 +161,5 @@ async def news_stream(request: Request):
             payload = await queue.get()
             html = render_analysis_card(payload, is_news=True)
             data_payload = html.replace('\n', '')
-            sse_message = f"event: news_update\ndata: {data_payload}\n\n"
-            yield sse_message
+            yield f"event: news_update\ndata: {data_payload}\n\n"
     return StreamingResponse(event_generator(), media_type="text/event-stream")