Update app/app.py

app/app.py

@@ -1,167 +1,171 @@
 """
-The Sentinel Protocol — An Autonomous AI Market Analyst.
-Connects real-time news events to market data visualizations.
+The Sentinel TradeFlow Protocol - Main Application
+
+Orchestrates the 3-tier intelligence funnel:
+1. Ingests a (mock) real-time data stream.
+2. Filters events with the local Tier 1 SentimentEngine.
+3. Escalates high-conviction events to the Tier 2 GeminiAnalyzer.
+4. Generates actionable trade hypotheses with the Tier 3 Strategy Engine.
+5. Pushes final signals to the Command Center UI.
 """
 import asyncio
 import json
 import os
 from contextlib import asynccontextmanager
-from typing import Optional, Union, List
-import pandas as pd
+from typing import Optional, Dict
 
 import httpx
 from fastapi import FastAPI, Request, BackgroundTasks, Form
 from fastapi.responses import HTMLResponse, StreamingResponse
 from fastapi.templating import Jinja2Templates
-from markdown import markdown
-from newsapi import NewsApiClient
 
-# Import our modular service classes
-from .price_fetcher import PriceFetcher
+# Import our intelligence modules
 from .gemini_analyzer import GeminiAnalyzer
-from .chart_generator import generate_price_chart
+from .sentiment_engine import LocalSentimentFilter
+
+# --- Tier 3: The Strategist ---
+def generate_trade_hypothesis(analysis: dict) -> Optional[Dict]:
+    """A simple rules-based engine to generate an actionable signal."""
+    sentiment = analysis.get("sentiment")
+    impact = analysis.get("impact")
+    score = analysis.get("sentiment_score", 0.0)
+    
+    # High-conviction rules
+    if impact == "HIGH" and sentiment == "NEGATIVE" and score < -0.7:
+        return {"type": "HYPOTHETICAL SHORT", "confidence": "HIGH", "reason": "High impact, strongly negative news detected."}
+    if impact == "HIGH" and sentiment == "POSITIVE" and score > 0.7:
+        return {"type": "HYPOTHETICAL LONG", "confidence": "HIGH", "reason": "High impact, strongly positive news detected."}
+    
+    # Medium-conviction rules
+    if impact == "MEDIUM" and sentiment == "NEGATIVE" and score < -0.5:
+        return {"type": "HYPOTHETICAL SHORT", "confidence": "MEDIUM", "reason": "Medium impact, negative news."}
+    if impact == "MEDIUM" and sentiment == "POSITIVE" and score > 0.5:
+        return {"type": "HYPOTHETICAL LONG", "confidence": "MEDIUM", "reason": "Medium impact, positive news."}
+        
+    return None
+
+# --- Mock Real-Time Data Feed & Pipeline Orchestration ---
+async def real_time_intelligence_pipeline(app: FastAPI):
+    """Mocks a high-frequency WebSocket news feed and runs it through the 3-tier funnel."""
+    await asyncio.sleep(5) # Initial delay to let UI connect
+    print("🚀 [Pipeline] Real-time intelligence pipeline is active.")
+    
+    # A more realistic stream of headlines
+    mock_headlines = [
+        ("Coinbase reports minor outage, services restored.", 5),
+        ("New memecoin 'ShibaCat' gains 20% on low volume.", 3),
+        ("BREAKING: US Federal Reserve signals potential for surprise interest rate hike next month.", 8),
+        ("Ethereum developer announces successful testnet merge for upcoming 'Prague' upgrade.", 6),
+        ("CEO of major crypto fund says market is 'overheated'.", 4),
+        ("MASSIVE EXPLOIT: Cross-chain bridge 'Wormhole' drained of $150M in ETH and SOL.", 7),
+        ("BlackRock files updated S-1 form for its spot Bitcoin ETF.", 5),
+        ("Polygon announces major partnership with a leading gaming studio.", 4),
+    ]
+
+    for headline, delay in mock_headlines:
+        print(f"🔥 [Tier 1] Ingested: '{headline}'")
+        
+        # Tier 1 Analysis: Fast, local filtering
+        local_analysis = LocalSentimentFilter.analyze(headline)
+        
+        # Trigger Condition for Tier 2: Is sentiment strong enough?
+        if abs(local_analysis['score']) > 0.65 or local_analysis['label'].lower() != 'neutral':
+            print(f"⚡️ [Tier 2 Triggered] Event '{headline[:30]}...' escalated to Gemini. Reason: Local sentiment {local_analysis['label']} ({local_analysis['score']:.2f})")
+            
+            analyzer: GeminiAnalyzer = app.state.gemini_analyzer
+            gemini_analysis = await analyzer.analyze_text(headline)
+            
+            # Tier 3: Generate actionable signal
+            signal = generate_trade_hypothesis(gemini_analysis)
+            
+            if signal:
+                print(f"🎯 [Tier 3] Actionable Signal Generated: {signal['type']} with {signal['confidence']} confidence.")
+                final_payload = {"signal": signal, "analysis": gemini_analysis, "headline": headline}
+                await app.state.signal_queue.put(final_payload)
+        
+        await asyncio.sleep(delay)
+    print("✅ [Pipeline] Mock real-time feed complete.")
 
-# --- Lifespan Manager ---
+
+# --- Application Lifespan ---
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Manages application startup and shutdown events."""
     async with httpx.AsyncClient() as client:
-        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
-
-        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()
-
-        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]
+        app.state.signal_queue = asyncio.Queue()
+        
+        # Warm up the local model on startup
+        LocalSentimentFilter.analyze("Warming up FinBERT model...")
         
-        print("🚀 The Sentinel Protocol is online.")
+        # Start the intelligence pipeline as a background task
+        pipeline_task = asyncio.create_task(real_time_intelligence_pipeline(app))
+        
+        print("🚀 Sentinel TradeFlow Protocol is online.")
         yield
         
         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)
-
-async def run_periodic_news_analysis(app: FastAPI, interval: int):
-    """Fetches news, analyzes it, generates charts, and queues the full payload."""
-    while True:
-        print("📰 Sentinel is scanning for new market intelligence...")
+        pipeline_task.cancel()
         try:
-            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 headlines.get('articles', []):
-                title = article.get('title')
-                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"❌ News analysis pipeline failed: {e}")
-        await asyncio.sleep(interval)
-
-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:
-        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 = []
-            print("✅ Daily briefing synthesized.")
-        else:
-            print("ℹ️ No new intelligence to synthesize. Waiting for next cycle.")
-        await asyncio.sleep(interval)
+            await pipeline_task
+        except asyncio.CancelledError:
+            print("Intelligence pipeline successfully shut down.")
+        print("✅ Sentinel Protocol offline.")
 
-# --- App and Endpoints ---
-app = FastAPI(title="Sentinel Protocol", lifespan=lifespan)
+# --- App Initialization ---
+app = FastAPI(title="Sentinel TradeFlow", lifespan=lifespan)
 templates = Jinja2Templates(directory="templates")
 
 # --- HTML Rendering Helper ---
-def render_analysis_card(payload: dict) -> str:
-    s = payload
-    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()}"
-    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 ""
+def render_signal_card(payload: dict) -> str:
+    """Renders the final signal payload into a rich HTML card."""
+    signal = payload.get("signal", {})
+    analysis = payload.get("analysis", {})
+    headline = payload.get("headline", "N/A")
+    
+    signal_type = signal.get("type", "INFO")
+    confidence = signal.get("confidence", "N/A")
     
-    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>"""
+    # Dynamic styling based on signal
+    if "SHORT" in signal_type:
+        card_class = "signal-short"
+        icon = "📉"
+    elif "LONG" in signal_type:
+        card_class = "signal-long"
+        icon = "📈"
+    else:
+        card_class = ""
+        icon = "ℹ️"
+
+    return f"""
+    <div class="card {card_class}">
+        <header class="signal-header">
+            <span>{icon} {signal_type}</span>
+            <span>Confidence: <strong>{confidence}</strong></span>
+        </header>
+        <p class="headline"><strong>Source Headline:</strong> {headline}</p>
+        <p><strong>Sentinel's Assessment:</strong> {analysis.get('summary', 'N/A')}</p>
+        <div class="grid">
+            <div><strong>Impact:</strong> {analysis.get('impact')}</div>
+            <div><strong>Topic:</strong> {analysis.get('topic')}</div>
+            <div><strong>Entities:</strong> {', '.join(analysis.get('entities', []))}</div>
+        </div>
+    </div>
+    """
 
 # --- API Endpoints ---
 @app.get("/", response_class=HTMLResponse)
-async def serve_dashboard(request: Request):
+async def serve_command_center(request: Request):
     return templates.TemplateResponse("index.html", {"request": request})
 
-# ... (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/news/stream")
-async def news_stream(request: Request):
-    queue: asyncio.Queue = request.app.state.news_queue
+@app.get("/api/signals/stream")
+async def signal_stream(request: Request):
+    """SSE stream for pushing generated trade hypotheses to the UI."""
+    queue: asyncio.Queue = request.app.state.signal_queue
     async def event_generator():
         while True:
             payload = await queue.get()
-            html = render_analysis_card(payload)
+            html = render_signal_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")
-
-# 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>")
+            yield f"event: new_signal\ndata: {data_payload}\n\n"
+    return StreamingResponse(event_generator(), media_type="text/event-stream")