Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Add demo scripts for Coinbase and CryptoCompare; remove outdated market agent demo

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-09-29 21:06:50 +02:00
parent 4615ebe63e
commit fc0ba1e37d
6 changed files with 80 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
demos/example.py → demos/agno_demo.py
View File

@@ -1,3 +1,9 @@
#### 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 agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.reasoning import ReasoningTools
@@ -8,9 +14,7 @@ try:
reasoning_agent = Agent(
model=Gemini(),
tools=[
ReasoningTools(),
],
tools=[ReasoningTools()],
instructions="Use tables to display data.",
markdown=True,
)

116
demos/cdp_market_demo.py
View File

@@ -1,116 +0,0 @@
#!/usr/bin/env python3
"""
Demo script per testare il MarketAgent aggiornato con Coinbase CDP
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from src.app.agents.market_agent import MarketAgent
import logging
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
def main():
print("🚀 Test MarketAgent con Coinbase CDP")
print("=" * 50)
# Inizializza l'agent
agent = MarketAgent()
# Verifica provider disponibili
providers = agent.get_available_providers()
print(f"📡 Provider disponibili: {providers}")
if not providers:
print("⚠️ Nessun provider configurato. Verifica il file .env")
print("\nPer Coinbase CDP, serve:")
print("CDP_API_KEY_NAME=your_key_name")
print("CDP_API_PRIVATE_KEY=your_private_key")
print("\nPer CryptoCompare, serve:")
print("CRYPTOCOMPARE_API_KEY=your_api_key")
return
# Mostra capabilities di ogni provider
for provider in providers:
capabilities = agent.get_provider_capabilities(provider)
print(f"🔧 {provider.upper()}: {capabilities}")
print("\n" + "=" * 50)
# Test ottenimento prezzo singolo
test_symbols = ["BTC", "ETH", "ADA"]
for symbol in test_symbols:
print(f"\n💰 Prezzo {symbol}:")
# Prova ogni provider
for provider in providers:
try:
price = agent.get_asset_price(symbol, provider)
if price:
print(f" {provider}: ${price:,.2f}")
else:
print(f" {provider}: N/A")
except Exception as e:
print(f" {provider}: Errore - {e}")
print("\n" + "=" * 50)
# Test market overview
print("\n📊 Market Overview:")
try:
overview = agent.get_market_overview(["BTC", "ETH", "ADA", "DOT"])
if overview["data"]:
print(f"📡 Fonte: {overview['source']}")
for crypto, prices in overview["data"].items():
if isinstance(prices, dict):
usd_price = prices.get("USD", "N/A")
eur_price = prices.get("EUR", "N/A")
if eur_price != "N/A":
print(f" {crypto}: ${usd_price} (€{eur_price})")
else:
print(f" {crypto}: ${usd_price}")
else:
print("⚠️ Nessun dato disponibile")
except Exception as e:
print(f"❌ Errore nel market overview: {e}")
print("\n" + "=" * 50)
# Test funzione analyze
print("\n🔍 Analisi mercato:")
try:
analysis = agent.analyze("Market overview")
print(analysis)
except Exception as e:
print(f"❌ Errore nell'analisi: {e}")
# Test specifico Coinbase CDP se disponibile
if 'coinbase' in providers:
print("\n" + "=" * 50)
print("\n🏦 Test specifico Coinbase CDP:")
try:
# Test asset singolo
btc_info = agent.get_coinbase_asset_info("BTC")
print(f"📈 BTC Info: {btc_info}")
# Test asset multipli
multi_assets = agent.get_coinbase_multiple_assets(["BTC", "ETH"])
print(f"📊 Multi Assets: {multi_assets}")
except Exception as e:
print(f"❌ Errore nel test Coinbase CDP: {e}")
if __name__ == "__main__":
main()

36
demos/coinbase_demo.py Normal file
View File

@@ -0,0 +1,36 @@
#### 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()

36
demos/cryptocompare_demo.py Normal file
View File

@@ -0,0 +1,36 @@
#### 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()

100
demos/market_agent_demo.py
View File

@@ -1,100 +0,0 @@
#!/usr/bin/env python3
"""
Esempio di utilizzo del MarketAgent unificato.
Questo script mostra come utilizzare il nuovo MarketAgent che supporta
multiple fonti di dati (Coinbase e CryptoCompare).
"""
import sys
from pathlib import Path
# Aggiungi il path src al PYTHONPATH
src_path = Path(__file__).parent / "src"
sys.path.insert(0, str(src_path))
from dotenv import load_dotenv
from app.agents.market_agent import MarketAgent
# Carica variabili d'ambiente
load_dotenv()
def main():
print("🚀 Market Agent Demo\n")
try:
# Inizializza il market agent (auto-configura i provider disponibili)
agent = MarketAgent()
# Mostra provider disponibili
providers = agent.get_available_providers()
print(f"📡 Available providers: {providers}")
if not providers:
print("❌ No providers configured. Please check your .env file.")
print("Required variables:")
print(" For Coinbase: COINBASE_API_KEY, COINBASE_SECRET, COINBASE_PASSPHRASE")
print(" For CryptoCompare: CRYPTOCOMPARE_API_KEY")
return
# Mostra le capacità di ogni provider
print("\n🔧 Provider capabilities:")
for provider in providers:
capabilities = agent.get_provider_capabilities(provider)
print(f" {provider}: {capabilities}")
# Ottieni panoramica del mercato
print("\n📊 Market Overview:")
overview = agent.get_market_overview(["BTC", "ETH", "ADA"])
print(f"Data source: {overview.get('source', 'Unknown')}")
for crypto, prices in overview.get('data', {}).items():
if isinstance(prices, dict):
usd = prices.get('USD', 'N/A')
eur = prices.get('EUR', 'N/A')
if eur != 'N/A':
print(f" {crypto}: ${usd} (€{eur})")
else:
print(f" {crypto}: ${usd}")
# Analisi completa del mercato
print("\n📈 Market Analysis:")
analysis = agent.analyze("comprehensive market analysis")
print(analysis)
# Test specifici per provider (se disponibili)
if 'cryptocompare' in providers:
print("\n🔸 CryptoCompare specific test:")
try:
btc_price = agent.get_single_crypto_price("BTC", "USD")
print(f" BTC price: ${btc_price}")
top_coins = agent.get_top_cryptocurrencies(5)
if top_coins.get('Data'):
print(" Top 5 cryptocurrencies by market cap:")
for coin in top_coins['Data'][:3]: # Show top 3
coin_info = coin.get('CoinInfo', {})
display = coin.get('DISPLAY', {}).get('USD', {})
name = coin_info.get('FullName', 'Unknown')
price = display.get('PRICE', 'N/A')
print(f" {name}: {price}")
except Exception as e:
print(f" CryptoCompare test failed: {e}")
if 'coinbase' in providers:
print("\n🔸 Coinbase specific test:")
try:
ticker = agent.get_coinbase_ticker("BTC-USD")
price = ticker.get('price', 'N/A')
volume = ticker.get('volume_24h', 'N/A')
print(f" BTC-USD: ${price} (24h volume: {volume})")
except Exception as e:
print(f" Coinbase test failed: {e}")
print("\n✅ Demo completed successfully!")
except Exception as e:
print(f"❌ Demo failed: {e}")
print("Make sure you have configured at least one provider in your .env file.")
if __name__ == "__main__":
main()

2
src/app/markets/cryptocompare.py
View File

@@ -49,7 +49,7 @@ class CryptoCompareWrapper(BaseWrapper):
def get_all_products(self) -> list[ProductInfo]:
raise NotImplementedError("CryptoCompare does not support fetching all assets")
def get_historical_prices(self, asset_id: str, day_back: int = 10) -> list[dict]:
def get_historical_prices(self, asset_id: str, day_back: int = 10) -> list[Price]:
assert day_back <= 30, "day_back should be less than or equal to 30"
response = self.__request("/data/v2/histohour", params = {
"fsym": asset_id,