Demos (#42)

* Rimossi old docs & demos
* Aggiornata la documentazione dell'architettura dell'app
* Aggiunti nuovi script demo per i provider di mercato
* Fix problems with socials
* Fix Dockerfile per dipendenze di X

demos/agno_agent.py

@@ -8,9 +8,7 @@ try:
 
     reasoning_agent = Agent(
         model=Gemini(),
-        tools=[
-            ReasoningTools(),
-        ],
+        tools=[ReasoningTools()],
         instructions="Use tables to display data.",
         markdown=True,
     )

demos/api_market_providers.py

@@ -0,0 +1,12 @@
+from dotenv import load_dotenv
+from app.api.tools import MarketAPIsTool
+
+def main():
+    api = MarketAPIsTool()
+    prices_aggregated = api.get_historical_prices_aggregated("BTC", limit=5)
+    for price in prices_aggregated:
+        print(f"== [{price.timestamp}] {price.low:.2f} - {price.high:.2f} ==")
+
+if __name__ == "__main__":
+    load_dotenv()
+    main()

demos/api_news_providers.py

@@ -0,0 +1,16 @@
+from dotenv import load_dotenv
+from app.api.tools import NewsAPIsTool
+
+def main():
+    api = NewsAPIsTool()
+    articles_aggregated = api.get_latest_news_aggregated(query="bitcoin", limit=2)
+    for provider, articles in articles_aggregated.items():
+        print("===================================")
+        print(f"Provider: {provider}")
+        for article in articles:
+            print(f"== [{article.time}] {article.title} ==")
+            print(f"   {article.description}")
+
+if __name__ == "__main__":
+    load_dotenv()
+    main()

demos/api_socials_providers.py

@@ -0,0 +1,17 @@
+from dotenv import load_dotenv
+from app.api.tools import SocialAPIsTool
+
+def main():
+    api = SocialAPIsTool()
+    articles_aggregated = api.get_top_crypto_posts_aggregated(limit_per_wrapper=2)
+    for provider, posts in articles_aggregated.items():
+        print("===================================")
+        print(f"Provider: {provider}")
+        for post in posts:
+            print(f"== [{post.time}] - {post.title} ==")
+            print(f"   {post.description}")
+            print(f"   {len(post.comments)}")
+
+if __name__ == "__main__":
+    load_dotenv()
+    main()

demos/market_providers_api_demo.py

@@ -1,353 +0,0 @@
-#!/usr/bin/env python3
-"""
-Demo Completo per Market Data Providers
-========================================
-
-Questo script dimostra l'utilizzo di tutti i wrapper che implementano BaseWrapper:
-- CoinBaseWrapper (richiede credenziali)
-- CryptoCompareWrapper (richiede API key)
-- BinanceWrapper (richiede credenziali)
-- PublicBinanceAgent (accesso pubblico)
-- YFinanceWrapper (accesso gratuito a dati azionari e crypto)
-
-Lo script effettua chiamate GET a diversi provider e visualizza i dati
-in modo strutturato con informazioni dettagliate su timestamp, stato
-delle richieste e formattazione tabellare.
-"""
-
-import sys
-import os
-from pathlib import Path
-from datetime import datetime
-from typing import Dict, List, Optional, Any
-import traceback
-
-# Aggiungi il path src al PYTHONPATH
-project_root = Path(__file__).parent.parent
-sys.path.insert(0, str(project_root / "src"))
-
-from dotenv import load_dotenv
-from app.api.markets import ( 
-    CoinBaseWrapper, 
-    CryptoCompareWrapper, 
-    BinanceWrapper,
-    YFinanceWrapper,
-    MarketWrapper
-)
-
-# Carica variabili d'ambiente
-load_dotenv()
-
-class DemoFormatter:
-    """Classe per formattare l'output del demo in modo strutturato."""
-
-    @staticmethod
-    def print_header(title: str, char: str = "=", width: int = 80):
-        """Stampa un'intestazione formattata."""
-        print(f"\n{char * width}")
-        print(f"{title:^{width}}")
-        print(f"{char * width}")
-
-    @staticmethod
-    def print_subheader(title: str, char: str = "-", width: int = 60):
-        """Stampa una sotto-intestazione formattata."""
-        print(f"\n{char * width}")
-        print(f" {title}")
-        print(f"{char * width}")
-
-    @staticmethod
-    def print_request_info(provider_name: str, method: str, timestamp: datetime, 
-                          status: str, error: Optional[str] = None):
-        """Stampa informazioni sulla richiesta."""
-        print(f"🕒 Timestamp: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
-        print(f"🏷️  Provider: {provider_name}")
-        print(f"🔧 Method: {method}")
-        print(f"📊 Status: {status}")
-        if error:
-            print(f"❌ Error: {error}")
-        print()
-
-    @staticmethod
-    def print_product_table(products: List[Any], title: str = "Products"):
-        """Stampa una tabella di prodotti."""
-        if not products:
-            print(f"📋 {title}: Nessun prodotto trovato")
-            return
-
-        print(f"📋 {title} ({len(products)} items):")
-        print(f"{'Symbol':<15} {'ID':<20} {'Price':<12} {'Quote':<10} {'Status':<10}")
-        print("-" * 67)
-
-        for product in products[:10]:  # Mostra solo i primi 10
-            symbol = getattr(product, 'symbol', 'N/A')
-            product_id = getattr(product, 'id', 'N/A')
-            price = getattr(product, 'price', 0.0)
-            quote = getattr(product, 'quote_currency', 'N/A')
-            status = getattr(product, 'status', 'N/A')
-
-            # Tronca l'ID se troppo lungo
-            if len(product_id) > 18:
-                product_id = product_id[:15] + "..."
-
-            price_str = f"${price:.2f}" if price > 0 else "N/A"
-
-            print(f"{symbol:<15} {product_id:<20} {price_str:<12} {quote:<10} {status:<10}")
-
-        if len(products) > 10:
-            print(f"... e altri {len(products) - 10} prodotti")
-        print()
-
-    @staticmethod
-    def print_prices_table(prices: List[Any], title: str = "Historical Prices"):
-        """Stampa una tabella di prezzi storici."""
-        if not prices:
-            print(f"💰 {title}: Nessun prezzo trovato")
-            return
-
-        print(f"💰 {title} ({len(prices)} entries):")
-        print(f"{'Time':<12} {'Open':<12} {'High':<12} {'Low':<12} {'Close':<12} {'Volume':<15}")
-        print("-" * 75)
-
-        for price in prices[:5]:  # Mostra solo i primi 5
-            time_str = getattr(price, 'time', 'N/A')
-            # Il time è già una stringa, non serve strftime
-            if len(time_str) > 10:
-                time_str = time_str[:10]  # Tronca se troppo lungo
-
-            open_price = f"${getattr(price, 'open', 0):.2f}"
-            high_price = f"${getattr(price, 'high', 0):.2f}"
-            low_price = f"${getattr(price, 'low', 0):.2f}"
-            close_price = f"${getattr(price, 'close', 0):.2f}"
-            volume = f"{getattr(price, 'volume', 0):,.0f}"
-
-            print(f"{time_str:<12} {open_price:<12} {high_price:<12} {low_price:<12} {close_price:<12} {volume:<15}")
-
-        if len(prices) > 5:
-            print(f"... e altri {len(prices) - 5} prezzi")
-        print()
-
-class ProviderTester:
-    """Classe per testare i provider di market data."""
-
-    def __init__(self):
-        self.formatter = DemoFormatter()
-        self.test_symbols = ["BTC", "ETH", "ADA"]
-
-    def test_provider(self, wrapper: MarketWrapper, provider_name: str) -> Dict[str, Any]:
-        """Testa un provider specifico con tutti i metodi disponibili."""
-        results: Dict[str, Any] = {
-            "provider_name": provider_name,
-            "tests": {},
-            "overall_status": "SUCCESS"
-        }
-
-        self.formatter.print_subheader(f"🔍 Testing {provider_name}")
-
-        # Test get_product
-        for symbol in self.test_symbols:
-            timestamp = datetime.now()
-            try:
-                product = wrapper.get_product(symbol)
-                self.formatter.print_request_info(
-                    provider_name, f"get_product({symbol})", timestamp, "✅ SUCCESS"
-                )
-                if product:
-                    print(f"📦 Product: {product.symbol} (ID: {product.id})")
-                    print(f"   Price: ${product.price:.2f}, Quote: {product.currency}")
-                    print(f"   Volume 24h: {product.volume_24h:,.2f}")
-                else:
-                    print(f"📦 Product: Nessun prodotto trovato per {symbol}")
-
-                results["tests"][f"get_product_{symbol}"] = "SUCCESS"
-
-            except Exception as e:
-                error_msg = str(e)
-                self.formatter.print_request_info(
-                    provider_name, f"get_product({symbol})", timestamp, "❌ ERROR", error_msg
-                )
-                results["tests"][f"get_product_{symbol}"] = f"ERROR: {error_msg}"
-                results["overall_status"] = "PARTIAL"
-
-        # Test get_products
-        timestamp = datetime.now()
-        try:
-            products = wrapper.get_products(self.test_symbols)
-            self.formatter.print_request_info(
-                provider_name, f"get_products({self.test_symbols})", timestamp, "✅ SUCCESS"
-            )
-            self.formatter.print_product_table(products, f"{provider_name} Products")
-            results["tests"]["get_products"] = "SUCCESS"
-
-        except Exception as e:
-            error_msg = str(e)
-            self.formatter.print_request_info(
-                provider_name, f"get_products({self.test_symbols})", timestamp, "❌ ERROR", error_msg
-            )
-            results["tests"]["get_products"] = f"ERROR: {error_msg}"
-            results["overall_status"] = "PARTIAL"
-
-        # Test get_historical_prices
-        timestamp = datetime.now()
-        try:
-            prices = wrapper.get_historical_prices("BTC")
-            self.formatter.print_request_info(
-                provider_name, "get_historical_prices(BTC)", timestamp, "✅ SUCCESS"
-            )
-            self.formatter.print_prices_table(prices, f"{provider_name} BTC Historical Prices")
-            results["tests"]["get_historical_prices"] = "SUCCESS"
-
-        except Exception as e:
-            error_msg = str(e)
-            self.formatter.print_request_info(
-                provider_name, "get_historical_prices(BTC)", timestamp, "❌ ERROR", error_msg
-            )
-            results["tests"]["get_historical_prices"] = f"ERROR: {error_msg}"
-            results["overall_status"] = "PARTIAL"
-
-        return results
-
-def check_environment_variables() -> Dict[str, bool]:
-    """Verifica la presenza delle variabili d'ambiente necessarie."""
-    env_vars = {
-        "COINBASE_API_KEY": bool(os.getenv("COINBASE_API_KEY")),
-        "COINBASE_API_SECRET": bool(os.getenv("COINBASE_API_SECRET")),
-        "CRYPTOCOMPARE_API_KEY": bool(os.getenv("CRYPTOCOMPARE_API_KEY")),
-        "BINANCE_API_KEY": bool(os.getenv("BINANCE_API_KEY")),
-        "BINANCE_API_SECRET": bool(os.getenv("BINANCE_API_SECRET")),
-    }
-    return env_vars
-
-def initialize_providers() -> Dict[str, MarketWrapper]:
-    """Inizializza tutti i provider disponibili."""
-    providers: Dict[str, MarketWrapper] = {}
-    env_vars = check_environment_variables()
-
-    # CryptoCompareWrapper
-    if env_vars["CRYPTOCOMPARE_API_KEY"]:
-        try:
-            providers["CryptoCompare"] = CryptoCompareWrapper()
-            print("✅ CryptoCompareWrapper inizializzato con successo")
-        except Exception as e:
-            print(f"❌ Errore nell'inizializzazione di CryptoCompareWrapper: {e}")
-    else:
-        print("⚠️ CryptoCompareWrapper saltato: CRYPTOCOMPARE_API_KEY non trovata")
-
-    # CoinBaseWrapper
-    if env_vars["COINBASE_API_KEY"] and env_vars["COINBASE_API_SECRET"]:
-        try:
-            providers["CoinBase"] = CoinBaseWrapper()
-            print("✅ CoinBaseWrapper inizializzato con successo")
-        except Exception as e:
-            print(f"❌ Errore nell'inizializzazione di CoinBaseWrapper: {e}")
-    else:
-        print("⚠️ CoinBaseWrapper saltato: credenziali Coinbase non complete")
-
-    # BinanceWrapper
-    try:
-        providers["Binance"] = BinanceWrapper()
-        print("✅ BinanceWrapper inizializzato con successo")
-    except Exception as e:
-        print(f"❌ Errore nell'inizializzazione di BinanceWrapper: {e}")
-
-    # YFinanceWrapper (sempre disponibile - dati azionari e crypto gratuiti)
-    try:
-        providers["YFinance"] = YFinanceWrapper()
-        print("✅ YFinanceWrapper inizializzato con successo")
-    except Exception as e:
-        print(f"❌ Errore nell'inizializzazione di YFinanceWrapper: {e}")
-    return providers
-
-def print_summary(results: List[Dict[str, Any]]):
-    """Stampa un riassunto finale dei risultati."""
-    formatter = DemoFormatter()
-    formatter.print_header("📊 RIASSUNTO FINALE", "=", 80)
-
-    total_providers = len(results)
-    successful_providers = sum(1 for r in results if r["overall_status"] == "SUCCESS")
-    partial_providers = sum(1 for r in results if r["overall_status"] == "PARTIAL")
-
-    print(f"🔢 Provider testati: {total_providers}")
-    print(f"✅ Provider completamente funzionanti: {successful_providers}")
-    print(f"⚠️ Provider parzialmente funzionanti: {partial_providers}")
-    print(f"❌ Provider non funzionanti: {total_providers - successful_providers - partial_providers}")
-
-    print("\n📋 Dettaglio per provider:")
-    for result in results:
-        provider_name = result["provider_name"]
-        status = result["overall_status"]
-        status_icon = "✅" if status == "SUCCESS" else "⚠️" if status == "PARTIAL" else "❌"
-
-        print(f"\n{status_icon} {provider_name}:")
-        for test_name, test_result in result["tests"].items():
-            test_icon = "✅" if test_result == "SUCCESS" else "❌"
-            print(f"   {test_icon} {test_name}: {test_result}")
-
-def main():
-    """Funzione principale del demo."""
-    formatter = DemoFormatter()
-
-    # Intestazione principale
-    formatter.print_header("🚀 DEMO COMPLETO MARKET DATA PROVIDERS", "=", 80)
-
-    print(f"🕒 Avvio demo: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
-    print("📝 Questo demo testa tutti i wrapper BaseWrapper disponibili")
-    print("🔍 Ogni test include timestamp, stato della richiesta e dati formattati")
-
-    # Verifica variabili d'ambiente
-    formatter.print_subheader("🔐 Verifica Configurazione")
-    env_vars = check_environment_variables()
-
-    print("Variabili d'ambiente:")
-    for var_name, is_present in env_vars.items():
-        status = "✅ Presente" if is_present else "❌ Mancante"
-        print(f"  {var_name}: {status}")
-
-    # Inizializza provider
-    formatter.print_subheader("🏗️ Inizializzazione Provider")
-    providers = initialize_providers()
-
-    if not providers:
-        print("❌ Nessun provider disponibile. Verifica la configurazione.")
-        return
-
-    print(f"\n🎯 Provider disponibili per il test: {list(providers.keys())}")
-
-    # Testa ogni provider
-    formatter.print_header("🧪 ESECUZIONE TEST PROVIDER", "=", 80)
-
-    tester = ProviderTester()
-    all_results: List[Dict[str, Any]] = []
-
-    for provider_name, wrapper in providers.items():
-        try:
-            result = tester.test_provider(wrapper, provider_name)
-            all_results.append(result)
-        except Exception as e:
-            print(f"❌ Errore critico nel test di {provider_name}: {e}")
-            traceback.print_exc()
-            all_results.append({
-                "provider_name": provider_name,
-                "tests": {},
-                "overall_status": "CRITICAL_ERROR",
-                "error": str(e)
-            })
-
-    # Stampa riassunto finale
-    print_summary(all_results)
-
-    # Informazioni aggiuntive
-    formatter.print_header("ℹ️ INFORMAZIONI AGGIUNTIVE", "=", 80)
-    print("📚 Documentazione:")
-    print("   - BaseWrapper: src/app/markets/base.py")
-    print("   - Test completi: tests/agents/test_market.py")
-    print("   - Configurazione: .env")
-
-    print("\n🔧 Per abilitare tutti i provider:")
-    print("   1. Configura le credenziali nel file .env")
-    print("   2. Segui la documentazione di ogni provider")
-    print("   3. Riavvia il demo")
-
-    print(f"\n🏁 Demo completato: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
-
-if __name__ == "__main__":
-    main()

demos/news_api.py

@@ -1,18 +0,0 @@
-#### FOR ALL FILES OUTSIDE src/ FOLDER ####
-import sys
-import os
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../src')))
-###########################################
-
-from dotenv import load_dotenv
-from app.api.news import NewsApiWrapper
-
-def main():
-    api = NewsApiWrapper()
-    articles = api.get_latest_news(query="bitcoin", limit=5)
-    assert len(articles) > 0
-    print("ok")
-
-if __name__ == "__main__":
-    load_dotenv()
-    main()

demos/ollama_demo.py

@@ -1,20 +1,4 @@
-#!/usr/bin/env python3
-"""
-Demo di Ollama (Python) – mostra:
-    1. Elenco dei modelli disponibili
-    2. Generazione di testo semplice
-    3. Chat con streaming
-    4. Calcolo di embeddings
-    5. Esempio (opzionale) di function calling / tools
-
-Uso:
-    python ollama_demo.py
-
-Requisiti:
-    pip install ollama
-    Avviare il server Ollama (es. 'ollama serve' o l'app desktop) e avere i modelli già pullati.
-"""
-
+from typing import Any
 import ollama
 
 # Configurazione modelli
@@ -33,8 +17,8 @@ def list_models():
             print("  (Nessun modello trovato)")
             return
         for m in models:
-            name = getattr(m, 'model', None) or (m.get('model') if isinstance(m, dict) else 'sconosciuto')
-            details = getattr(m, 'details', None)
+            name = getattr(m, 'model', None) or (m.get('model') if isinstance(m, dict) else 'sconosciuto')  # type: ignore
+            details = getattr(m, 'details', None)  # type: ignore
             fmt = getattr(details, 'format', None) if details else 'unknown'
             print(f"  • {name} – {fmt}")
     except Exception as e:
@@ -46,7 +30,7 @@ def list_models():
 def generate_text(model: str, prompt: str, max_tokens: int = 200) -> str:
     """Genera testo dal modello indicato."""
     print(f"\n[2] Generazione testo con '{model}'")
-    response = ollama.chat(
+    response = ollama.chat(  # type: ignore
         model=model,
         messages=[{"role": "user", "content": prompt}]
     )
@@ -57,10 +41,10 @@ def generate_text(model: str, prompt: str, max_tokens: int = 200) -> str:
 
 # 3. Chat con streaming --------------------------------------------------------
 
-def chat_streaming(model: str, messages: list) -> str:
+def chat_streaming(model: str, messages: list[dict[str, str]]) -> str:
     """Esegue una chat mostrando progressivamente la risposta."""
     print(f"\n[3] Chat (streaming) con '{model}'")
-    stream = ollama.chat(model=model, messages=messages, stream=True)
+    stream = ollama.chat(model=model, messages=messages, stream=True)  # type: ignore
     full = ""
     for chunk in stream:
         if 'message' in chunk and 'content' in chunk['message']:
@@ -91,7 +75,7 @@ def get_embedding(model: str, text: str):
 def try_tools(model: str):
     """Esempio di function calling; se non supportato mostra messaggio informativo."""
     print(f"\n[5] Function calling / tools con '{model}'")
-    tools = [
+    tools: list[dict[str, Any]] = [
         {
             "type": "function",
             "function": {
@@ -109,7 +93,7 @@ def try_tools(model: str):
         }
     ]
     try:
-        response = ollama.chat(
+        response = ollama.chat(  # type: ignore
             model=model,
             messages=[{"role": "user", "content": "Che tempo fa a Milano?"}],
             tools=tools