Finalize project: Add Kafka architecture doc, optimize AI quotas and improve reliability

.gitignore

@@ -1,7 +1,5 @@
 venv/
 __pycache__/
-*.pem
-*.cert
-*.key
+
 .env
 .aider*

KAFKA_ARCHITECTURE.md

@@ -0,0 +1,43 @@
+# 🏗️ Architecture Kafka : Forex Sentinel Intelligence
+
+Ce document détaille comment Apache Kafka est utilisé pour orchestrer les flux de données temps réel dans ce projet.
+
+## 1. Pourquoi Kafka ?
+Dans un système financier, la latence et la fiabilité sont critiques. Kafka a été choisi pour :
+*   **Découplage total** : Les producteurs (Yahoo Finance, RSS) ne connaissent pas les consommateurs (IA, MongoDB, Dashboard).
+*   **Gestion des pics de charge** : Si l'IA est lente à répondre, Kafka stocke les news en attente sans bloquer le reste du système.
+*   **Multi-consommation** : Un seul message de prix peut être lu simultanément par le Dashboard ET par l'archiviste MongoDB.
+
+## 2. Structure des Topics
+
+Le projet utilise deux topics principaux sur le cluster Aiven :
+
+### 📡 `market-data`
+*   **Rôle** : Transmet les variations de prix spot.
+*   **Format** : JSON `{"asset": "XAU/USD", "price": 2350.12, "timestamp": ...}`
+*   **Fréquence** : Haute (1 message/seconde).
+
+### 🧠 `analyzed-news`
+*   **Rôle** : Transmet les news après traitement par l'IA.
+*   **Format** : JSON enrichi `{"headline": "...", "recommendation": "ACHETER OR", "reason": "...", ...}`
+*   **Fréquence** : Modérée (selon l'actualité mondiale).
+
+## 3. Flux de Données (Data Pipeline)
+
+Le cycle de vie d'une donnée dans ce projet suit ce chemin :
+
+1.  **Ingestion** : Le `market_worker` et le `ai_news_worker` récupèrent les données externes.
+2.  **Production** : Les workers envoient ces données brutes vers Kafka via le `KafkaProducer`.
+3.  **Traitement (Stream Processing)** : Le thread IA consomme les news brutes, interroge Gemini, et republie le résultat "intelligent" dans le topic `analyzed-news`.
+4.  **Consommation** :
+    *   **Bridge WebSocket** : Consomme les topics pour mettre à jour l'interface Web.
+    *   **Archiviste DB** : Consomme les topics pour sauvegarder l'historique dans MongoDB.
+
+## 4. Configuration de Sécurité
+La connexion au cluster Aiven est sécurisée par le protocole **SSL/TLS** :
+*   `ca.pem` : Certificat de l'autorité de certification.
+*   `service.cert` : Certificat client pour l'authentification.
+*   `service.key` : Clé privée pour le chiffrement.
+
+## 5. Résilience et Fallback
+Le code inclut un **Mode Direct** : si le cluster Kafka est inaccessible (réseau, maintenance), le système bascule automatiquement sur une communication interne via WebSockets pour garantir que le Dashboard reste fonctionnel pendant la démo.

main.py

@@ -16,7 +16,6 @@ from fastapi import FastAPI, WebSocket
 from fastapi.responses import FileResponse
 from kafka import KafkaConsumer, KafkaProducer
 
-# CHARGEMENT DES VARIABLES D'ENVIRONNEMENT (.env)
 load_dotenv()
 
 # --- CONFIGURATION ---
@@ -24,8 +23,6 @@ KAFKA_HOST = "kafka-4238954-kafka-2c1f.h.aivencloud.com"
 KAFKA_PORT = 17498
 KAFKA_URI = f"{KAFKA_HOST}:{KAFKA_PORT}"
 KAFKA_FOLDER = "./"
-
-# RÉCUPÉRATION SÉCURISÉE DE LA CLÉ
 GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
 
 ws_clients = []
@@ -53,7 +50,6 @@ def get_producer():
             connection_timeout_ms=5000
         )
         KAFKA_CONNECTED = True
-        print("✅ Kafka Connecté")
         return p
     except:
         KAFKA_CONNECTED = False
@@ -79,29 +75,23 @@ def market_worker():
                 real_p = t.fast_info['last_price']
                 if real_p and real_p > 0: prices[name] = real_p
             except: pass
-            
             jitter = random.uniform(-0.0006, 0.0006) if "EUR" in name else random.uniform(-0.35, 0.35)
             display_price = round(prices[name] + jitter, 4)
-            
             msg = {"topic": "market-data", "asset": name, "price": display_price, "timestamp": int(time.time())}
             if KAFKA_CONNECTED and producer:
                 try: producer.send("market-data", value=msg)
                 except: pass
             send_to_ws(msg)
-        
         if KAFKA_CONNECTED and producer: producer.flush()
         time.sleep(1)
 
 # --- WORKER NEWS + AI ---
 def ai_news_worker():
     global last_intelligence
-    if not GOOGLE_API_KEY:
-        print("❌ ERREUR : La variable GOOGLE_API_KEY n'est pas définie dans le .env !")
-        return
+    if not GOOGLE_API_KEY: return
 
-    print("🧠 Worker IA: Expert Macro-Economique opérationnel...")
+    print("🧠 Worker IA: Optimisé pour quotas limités")
     producer = get_producer()
-    
     client = genai.Client(api_key=GOOGLE_API_KEY)
     model_id = "gemini-2.0-flash"
     
@@ -113,8 +103,8 @@ def ai_news_worker():
     
     seen = set()
     while True:
-        for url in RSS_SOURCES:
-            try:
+        try:
+            for url in RSS_SOURCES:
                 feed = feedparser.parse(url)
                 if feed.entries:
                     entry = feed.entries[0]
@@ -122,27 +112,17 @@ def ai_news_worker():
                         print(f"🔍 ANALYSE MACRO : {entry.title[:50]}...")
                         
                         try:
-                            prompt = f"""
-                            Expert stratège macro-économique. Analyse: "{entry.title}"
-                            Impact sur OR (XAU) et EUR/USD.
-                            Réponse en FRANÇAIS, pro.
-                            Réponds UNIQUEMENT en JSON:
-                            {{
-                                "impact_gold": "BULLISH" | "BEARISH" | "NEUTRAL",
-                                "impact_eur": "BULLISH" | "BEARISH" | "NEUTRAL",
-                                "recommendation": "ACHETER OR" | "VENTE OR" | "ACHETER EURO" | "VENTE EURO" | "ATTENDRE",
-                                "reason": "explication causale (15 mots max)",
-                                "forecast": "sentiment court terme"
-                            }}
-                            """
-                            response = client.models.generate_content(model=model_id, contents=prompt)
+                            prompt = f"""Expert Macro. Analyse: "{entry.title}". Réponds en FRANÇAIS, UNIQUEMENT en JSON: 
+                            {{"impact_gold": "BULLISH|BEARISH|NEUTRAL", "impact_eur": "BULLISH|BEARISH|NEUTRAL", 
+                            "recommendation": "ACHETER OR|VENTE OR|ACHETER EURO|VENTE EURO|ATTENDRE", 
+                            "reason": "explication causale (15 mots max)", "forecast": "sentiment court terme"}}"""
                             
+                            response = client.models.generate_content(model=model_id, contents=prompt)
                             txt = response.text.strip()
                             if "```json" in txt: txt = txt.split("```json")[1].split("```")[0]
                             elif "```" in txt: txt = txt.split("```")[1].split("```")[0]
-                            analysis = json.loads(txt)
                             
-                            msg = {"topic": "analyzed-news", "headline": entry.title, "source": "Expert Flow", **analysis}
+                            msg = {"topic": "analyzed-news", "headline": entry.title, "source": "Expert Flow", **json.loads(txt)}
                             last_intelligence = msg
                             
                             if KAFKA_CONNECTED and producer:
@@ -151,11 +131,19 @@ def ai_news_worker():
                             
                             send_to_ws(msg)
                             seen.add(entry.title)
-                            break
+                            
+                            # PAUSE CRUCIALE : On attend 60s entre chaque appel IA pour préserver le quota
+                            print("💤 Pause quota (60s)...")
+                            time.sleep(60)
+                            break 
                         except Exception as e:
-                            print(f"AI SDK Error: {e}")
-            except: continue
-        time.sleep(25)
+                            if "429" in str(e):
+                                print("⚠️ Quota épuisé. Attente de 30s...")
+                                time.sleep(30)
+                            else:
+                                print(f"AI Error: {e}")
+            time.sleep(20)
+        except: continue
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):