diff --git a/demos/api_market_providers.py b/demos/api_market_providers.py new file mode 100644 index 0000000..a9ba754 --- /dev/null +++ b/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() diff --git a/demos/api_news_providers.py b/demos/api_news_providers.py new file mode 100644 index 0000000..4ac2f01 --- /dev/null +++ b/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() diff --git a/demos/api_socials_providers.py b/demos/api_socials_providers.py new file mode 100644 index 0000000..fb9da6e --- /dev/null +++ b/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() diff --git a/demos/coinbase_demo.py b/demos/coinbase_demo.py deleted file mode 100644 index 8b6bd41..0000000 --- a/demos/coinbase_demo.py +++ /dev/null @@ -1,36 +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.markets import CoinBaseWrapper - -def main(): - print("Demo Coinbase CDP") - - print("=== Chiavi API ===") - print(f" COINBASE_API_KEY: {os.getenv('COINBASE_API_KEY') is not None}") - print(f" COINBASE_API_SECRET: {os.getenv('COINBASE_API_SECRET') is not None}") - - # Inizializza le API - coinbase = CoinBaseWrapper() - - # ottenimento prezzo attuale - print("=== Demo prezzo attuale ===") - test_symbols = ["BTC", "ETH", "ADA"] - for symbol in test_symbols: - info = coinbase.get_product(symbol) - print(f" {symbol}: ${info.price:,.2f}") - - # ottenimento prezzi storici - print("\n=== Demo prezzi storici ===") - test_symbols = ["BTC", "ETH"] - for symbol in test_symbols: - prices = coinbase.get_historical_prices(symbol) - print(f" {symbol}: {" ".join([f'${entry["price"]:,.2f}' for entry in prices[:5]])}") # mostra solo i primi 5 - -if __name__ == "__main__": - load_dotenv() - main() diff --git a/demos/cryptocompare_demo.py b/demos/cryptocompare_demo.py deleted file mode 100644 index f929b87..0000000 --- a/demos/cryptocompare_demo.py +++ /dev/null @@ -1,36 +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.markets.cryptocompare import CryptoCompareWrapper - -def main(): - print("Demo CryptoCompare") - - print("=== Chiavi API ===") - print(f" CRYPTOCOMPARE_API_KEY: {os.getenv('CRYPTOCOMPARE_API_KEY') is not None}") - - # Inizializza le API - cryptocompare = CryptoCompareWrapper() - - # ottenimento prezzo attuale - print("=== Demo prezzo attuale ===") - test_symbols = ["BTC", "ETH", "ADA"] - for symbol in test_symbols: - info = cryptocompare.get_product(symbol) - print(f" {symbol}: ${info.price:,.2f}") - - # ottenimento prezzi storici - print("=== Demo prezzi storici ===") - test_symbols = ["BTC", "ETH"] - for symbol in test_symbols: - prices = cryptocompare.get_historical_prices(symbol) - prices = [f'[${entry.high:,.2f}-${entry.low:,.2f}]' for entry in prices] - print(f" {symbol}: {" ".join(prices[:5])}") # mostra solo i primi 5 - -if __name__ == "__main__": - load_dotenv() - main() \ No newline at end of file diff --git a/demos/market_providers_api_demo.py b/demos/market_providers_api_demo.py deleted file mode 100644 index c9ab116..0000000 --- a/demos/market_providers_api_demo.py +++ /dev/null @@ -1,348 +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.core.markets import MarketWrapper -from app.api.markets import * - -# 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() \ No newline at end of file diff --git a/demos/news_api.py b/demos/news_api.py deleted file mode 100644 index ef71974..0000000 --- a/demos/news_api.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/demos/ollama_demo.py b/demos/ollama_demo.py index 1e52f5f..8a327a6 100644 --- a/demos/ollama_demo.py +++ b/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