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

Merge branch '2-news-api' into 3-market-api

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-09-30 17:30:01 +02:00
parent 3074b58ea5 a5ef982e12
commit 754d43771d
9 changed files with 226 additions and 755 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
src/app/markets/__init__.py
View File

@@ -1,87 +1,30 @@
from .base import BaseWrapper
from app.markets.coinbase import CoinBaseWrapper
from app.markets.cryptocompare import CryptoCompareWrapper
from app.markets.binance import BinanceWrapper
from app.markets.binance_public import PublicBinanceAgent
from app.markets.error_handler import ProviderFallback, MarketAPIError, safe_execute
from .coinbase import CoinBaseWrapper
from .cryptocompare import CryptoCompareWrapper
from app.utils.wrapper_handler import WrapperHandler
from agno.utils.log import log_warning
import logging
__all__ = [ "MarketAPIs", "BaseWrapper", "CoinBaseWrapper", "CryptoCompareWrapper" ]
logger = logging.getLogger(__name__)
# TODO se si vuole usare un aggregatore di dati di mercato, si può aggiungere qui facendo una classe extra (simile a questa) che per ogni chiamata chiama tutti i wrapper e aggrega i risultati
class MarketAPIs(BaseWrapper):
"""
Classe per gestire le API di mercato disponibili.
Permette di ottenere un'istanza della prima API disponibile in base alla priorità specificata.
Supporta operazioni come ottenere informazioni su singoli prodotti, liste di prodotti e dati storici.
Usa un WrapperHandler per gestire più wrapper e tentare chiamate in modo resiliente.
"""
@staticmethod
def get_list_available_market_apis(currency: str = "USD") -> list[BaseWrapper]:
"""
Restituisce una lista di istanze delle API di mercato disponibili.
La priorità è data dall'ordine delle API nella lista wrappers.
1. CoinBase
2. CryptoCompare
:param currency: Valuta di riferimento (default "USD")
:return: Lista di istanze delle API di mercato disponibili
"""
wrapper_builders = [
CoinBaseWrapper,
CryptoCompareWrapper,
]
result = []
for wrapper in wrapper_builders:
try:
result.append(wrapper(currency=currency))
except Exception as e:
log_warning(f"{wrapper} cannot be initialized: {e}")
assert result, "No market API keys set in environment variables."
return result
def __init__(self, currency: str = "USD"):
"""
Inizializza la classe con la valuta di riferimento e la priorità dei provider.
Args:
currency: Valuta di riferimento (default "USD")
"""
self.currency = currency
self.wrappers = MarketAPIs.get_list_available_market_apis(currency=currency)
self.fallback_manager = ProviderFallback(self.wrappers)
# Metodi con fallback robusto tra provider multipli
def get_product(self, asset_id: str):
"""Ottiene informazioni su un prodotto con fallback automatico tra provider."""
try:
return self.fallback_manager.execute_with_fallback("get_product", asset_id)
except MarketAPIError as e:
logger.error(f"Failed to get product {asset_id}: {str(e)}")
raise
wrappers = [ CoinBaseWrapper, CryptoCompareWrapper ]
self.wrappers: WrapperHandler[BaseWrapper] = WrapperHandler.build_wrappers(wrappers)
def get_product(self, asset_id):
return self.wrappers.try_call(lambda w: w.get_product(asset_id))
def get_products(self, asset_ids: list):
"""Ottiene informazioni su più prodotti con fallback automatico tra provider."""
try:
return self.fallback_manager.execute_with_fallback("get_products", asset_ids)
except MarketAPIError as e:
logger.error(f"Failed to get products {asset_ids}: {str(e)}")
raise
return self.wrappers.try_call(lambda w: w.get_products(asset_ids))
def get_all_products(self):
"""Ottiene tutti i prodotti con fallback automatico tra provider."""
try:
return self.fallback_manager.execute_with_fallback("get_all_products")
except MarketAPIError as e:
logger.error(f"Failed to get all products: {str(e)}")
raise
def get_historical_prices(self, asset_id: str = "BTC"):
"""Ottiene prezzi storici con fallback automatico tra provider."""
try:
return self.fallback_manager.execute_with_fallback("get_historical_prices", asset_id)
except MarketAPIError as e:
logger.error(f"Failed to get historical prices for {asset_id}: {str(e)}")
raise
return self.wrappers.try_call(lambda w: w.get_all_products())
def get_historical_prices(self, asset_id = "BTC", limit: int = 100):
return self.wrappers.try_call(lambda w: w.get_historical_prices(asset_id, limit))

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

@@ -12,7 +12,7 @@ class BaseWrapper:
raise NotImplementedError
def get_all_products(self) -> list['ProductInfo']:
raise NotImplementedError
def get_historical_prices(self, asset_id: str = "BTC") -> list['Price']:
def get_historical_prices(self, asset_id: str = "BTC", limit: int = 100) -> list['Price']:
raise NotImplementedError
class ProductInfo(BaseModel):

68
src/app/markets/coinbase.py
View File

@@ -1,27 +1,36 @@
import os
from typing import Optional
from enum import Enum
from datetime import datetime, timedelta
from coinbase.rest import RESTClient
from .base import ProductInfo, BaseWrapper, Price
from .error_handler import retry_on_failure, handle_api_errors, MarketAPIError, RateLimitError
class Granularity(Enum):
UNKNOWN_GRANULARITY = 0
ONE_MINUTE = 60
FIVE_MINUTE = 300
FIFTEEN_MINUTE = 900
THIRTY_MINUTE = 1800
ONE_HOUR = 3600
TWO_HOUR = 7200
FOUR_HOUR = 14400
SIX_HOUR = 21600
ONE_DAY = 86400
class CoinBaseWrapper(BaseWrapper):
"""
Wrapper per le API di Coinbase Advanced Trade.
Wrapper per le API di Coinbase Advanced Trade.\n
Implementa l'interfaccia BaseWrapper per fornire accesso unificato
ai dati di mercato di Coinbase tramite le API REST.
La documentazione delle API è disponibile qui:
ai dati di mercato di Coinbase tramite le API REST.\n
https://docs.cdp.coinbase.com/api-reference/advanced-trade-api/rest-api/introduction
"""
def __init__(self, api_key: Optional[str] = None, api_private_key: Optional[str] = None, currency: str = "USD"):
if api_key is None:
api_key = os.getenv("COINBASE_API_KEY")
def __init__(self, currency: str = "USD"):
api_key = os.getenv("COINBASE_API_KEY")
assert api_key is not None, "API key is required"
if api_private_key is None:
api_private_key = os.getenv("COINBASE_API_SECRET")
api_private_key = os.getenv("COINBASE_API_SECRET")
assert api_private_key is not None, "API private key is required"
self.currency = currency
@@ -33,49 +42,30 @@ class CoinBaseWrapper(BaseWrapper):
def __format(self, asset_id: str) -> str:
return asset_id if '-' in asset_id else f"{asset_id}-{self.currency}"
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_product(self, asset_id: str) -> ProductInfo:
asset_id = self.__format(asset_id)
asset = self.client.get_product(asset_id)
return ProductInfo.from_coinbase(asset)
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_products(self, asset_ids: list[str]) -> list[ProductInfo]:
all_asset_ids = [self.__format(asset_id) for asset_id in asset_ids]
assets = self.client.get_products(product_ids=all_asset_ids)
if assets.products:
return [ProductInfo.from_coinbase_product(asset) for asset in assets.products]
return []
return [ProductInfo.from_coinbase(asset) for asset in assets.products]
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_all_products(self) -> list[ProductInfo]:
assets = self.client.get_products()
if assets.products:
return [ProductInfo.from_coinbase_product(asset) for asset in assets.products]
return []
return [ProductInfo.from_coinbase_product(asset) for asset in assets.products]
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_historical_prices(self, asset_id: str = "BTC") -> list[Price]:
def get_historical_prices(self, asset_id: str = "BTC", limit: int = 100) -> list[Price]:
asset_id = self.__format(asset_id)
# Get last 14 days of hourly data (14*24 = 336 candles, within 350 limit)
end_time = datetime.now()
start_time = end_time - timedelta(days=14)
# Convert to UNIX timestamps as strings (required by Coinbase API)
start_timestamp = str(int(start_time.timestamp()))
end_timestamp = str(int(end_time.timestamp()))
data = self.client.get_candles(
product_id=asset_id,
start=start_timestamp,
end=end_timestamp,
granularity="ONE_HOUR",
limit=350 # Explicitly set the limit
granularity=Granularity.ONE_HOUR.name,
start=str(int(start_time.timestamp())),
end=str(int(end_time.timestamp())),
limit=limit
)
if data.candles:
return [Price.from_coinbase(candle) for candle in data.candles]
return []
return [Price.from_coinbase(candle) for candle in data.candles]

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

@@ -12,9 +12,8 @@ class CryptoCompareWrapper(BaseWrapper):
La documentazione delle API è disponibile qui: https://developers.coindesk.com/documentation/legacy/Price/SingleSymbolPriceEndpoint
!!ATTENZIONE!! sembra essere una API legacy e potrebbe essere deprecata in futuro.
"""
def __init__(self, api_key: Optional[str] = None, currency: str = 'USD'):
if api_key is None:
api_key = os.getenv("CRYPTOCOMPARE_API_KEY")
def __init__(self, currency:str='USD'):
api_key = os.getenv("CRYPTOCOMPARE_API_KEY")
assert api_key is not None, "API key is required"
self.api_key = api_key
@@ -28,8 +27,6 @@ class CryptoCompareWrapper(BaseWrapper):
response = requests.get(f"{BASE_URL}{endpoint}", params=params)
return response.json()
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_product(self, asset_id: str) -> ProductInfo:
response = self.__request("/data/pricemultifull", params = {
"fsyms": asset_id,
@@ -38,8 +35,6 @@ class CryptoCompareWrapper(BaseWrapper):
data = response.get('RAW', {}).get(asset_id, {}).get(self.currency, {})
return ProductInfo.from_cryptocompare(data)
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_products(self, asset_ids: list[str]) -> list[ProductInfo]:
response = self.__request("/data/pricemultifull", params = {
"fsyms": ",".join(asset_ids),
@@ -52,42 +47,14 @@ class CryptoCompareWrapper(BaseWrapper):
assets.append(ProductInfo.from_cryptocompare(asset_data))
return assets
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_all_products(self) -> list[ProductInfo]:
"""
Workaround per CryptoCompare: utilizza una lista predefinita di asset popolari
poiché l'API non fornisce un endpoint per recuperare tutti i prodotti.
"""
# Lista di asset popolari supportati da CryptoCompare
popular_assets = [
"BTC", "ETH", "ADA", "DOT", "LINK", "LTC", "XRP", "BCH", "BNB", "SOL",
"MATIC", "AVAX", "ATOM", "UNI", "DOGE", "SHIB", "TRX", "ETC", "FIL", "XLM"
]
try:
# Utilizza get_products per recuperare i dati di tutti gli asset popolari
return self.get_products(popular_assets)
except Exception as e:
# Fallback: prova con un set ridotto di asset principali
main_assets = ["BTC", "ETH", "ADA", "DOT", "LINK"]
try:
return self.get_products(main_assets)
except Exception as fallback_error:
# Se anche il fallback fallisce, solleva l'errore originale con informazioni aggiuntive
raise NotImplementedError(
f"CryptoCompare get_all_products() workaround failed. "
f"Original error: {str(e)}, Fallback error: {str(fallback_error)}"
)
raise NotImplementedError("get_all_products is not supported by CryptoCompare API")
@retry_on_failure(max_retries=3, delay=1.0)
@handle_api_errors
def get_historical_prices(self, asset_id: str = "BTC", day_back: int = 10) -> list[Price]:
assert day_back <= 30, "day_back should be less than or equal to 30"
def get_historical_prices(self, asset_id: str, limit: int = 100) -> list[dict]:
response = self.__request("/data/v2/histohour", params = {
"fsym": asset_id,
"tsym": self.currency,
"limit": day_back * 24
"limit": limit-1 # because the API returns limit+1 items (limit + current)
})
data = response.get('Data', {}).get('Data', [])

642
tests/agents/test_market.py
View File

@@ -1,596 +1,68 @@
"""
Test suite completo per il sistema di mercato.
Questo modulo testa approfonditamente tutte le implementazioni di BaseWrapper
e verifica la conformità all'interfaccia definita in base.py.
"""
import os
import pytest
from unittest.mock import Mock, patch, MagicMock
from typing import Type, List
# Import delle classi da testare
from app.markets.base import BaseWrapper, ProductInfo, Price
from app.markets.coinbase import CoinBaseWrapper
from app.markets.cryptocompare import CryptoCompareWrapper
from app.markets.binance import BinanceWrapper
from app.markets.binance_public import PublicBinanceAgent
from app.agents.market_agent import MarketToolkit
from app.markets import MarketAPIs
class TestBaseWrapperInterface:
"""Test per verificare che tutte le implementazioni rispettino l'interfaccia BaseWrapper."""
def test_all_wrappers_extend_basewrapper(self):
"""Verifica che tutte le classi wrapper estendano BaseWrapper."""
wrapper_classes = [
CoinBaseWrapper,
CryptoCompareWrapper,
BinanceWrapper,
PublicBinanceAgent,
MarketAPIs
]
for wrapper_class in wrapper_classes:
assert issubclass(wrapper_class, BaseWrapper), f"{wrapper_class.__name__} deve estendere BaseWrapper"
def test_all_wrappers_implement_required_methods(self):
"""Verifica che tutte le classi implementino i metodi richiesti dall'interfaccia."""
wrapper_classes = [
CoinBaseWrapper,
CryptoCompareWrapper,
BinanceWrapper,
PublicBinanceAgent,
MarketAPIs
]
required_methods = ['get_product', 'get_products', 'get_all_products', 'get_historical_prices']
for wrapper_class in wrapper_classes:
for method in required_methods:
assert hasattr(wrapper_class, method), f"{wrapper_class.__name__} deve implementare {method}"
assert callable(getattr(wrapper_class, method)), f"{method} deve essere callable in {wrapper_class.__name__}"
class TestProductInfoModel:
"""Test per la classe ProductInfo e i suoi metodi di conversione."""
def test_productinfo_initialization(self):
"""Test inizializzazione di ProductInfo."""
product = ProductInfo()
assert product.id == ""
assert product.symbol == ""
assert product.price == 0.0
assert product.volume_24h == 0.0
assert product.status == ""
assert product.quote_currency == ""
def test_productinfo_with_data(self):
"""Test ProductInfo con dati specifici."""
product = ProductInfo(
id="BTC-USD",
symbol="BTC",
price=50000.0,
volume_24h=1000000.0,
status="TRADING",
quote_currency="USD"
)
assert product.id == "BTC-USD"
assert product.symbol == "BTC"
assert product.price == 50000.0
assert product.volume_24h == 1000000.0
assert product.status == "TRADING"
assert product.quote_currency == "USD"
def test_productinfo_from_cryptocompare(self):
"""Test conversione da dati CryptoCompare."""
mock_data = {
'FROMSYMBOL': 'BTC',
'TOSYMBOL': 'USD',
'PRICE': 50000.0,
'VOLUME24HOUR': 1000000.0
}
product = ProductInfo.from_cryptocompare(mock_data)
assert product.id == "BTC-USD"
assert product.symbol == "BTC"
assert product.price == 50000.0
assert product.volume_24h == 1000000.0
assert product.status == ""
def test_productinfo_from_binance(self):
"""Test conversione da dati Binance."""
ticker_data = {'symbol': 'BTCUSDT', 'price': '50000.0'}
ticker_24h_data = {'volume': '1000000.0'}
product = ProductInfo.from_binance(ticker_data, ticker_24h_data)
assert product.id == "BTCUSDT"
assert product.symbol == "BTC"
assert product.price == 50000.0
assert product.volume_24h == 1000000.0
assert product.status == "TRADING"
assert product.quote_currency == "USDT"
class TestPriceModel:
"""Test per la classe Price e i suoi metodi di conversione."""
def test_price_initialization(self):
"""Test inizializzazione di Price."""
price = Price()
assert price.high == 0.0
assert price.low == 0.0
assert price.open == 0.0
assert price.close == 0.0
assert price.volume == 0.0
assert price.time == ""
def test_price_with_data(self):
"""Test Price con dati specifici."""
price = Price(
high=51000.0,
low=49000.0,
open=50000.0,
close=50500.0,
volume=1000.0,
time="2024-01-01T00:00:00Z"
)
assert price.high == 51000.0
assert price.low == 49000.0
assert price.open == 50000.0
assert price.close == 50500.0
assert price.volume == 1000.0
assert price.time == "2024-01-01T00:00:00Z"
def test_price_from_cryptocompare(self):
"""Test conversione da dati CryptoCompare."""
mock_data = {
'high': 51000.0,
'low': 49000.0,
'open': 50000.0,
'close': 50500.0,
'volumeto': 1000.0,
'time': 1704067200
}
price = Price.from_cryptocompare(mock_data)
assert price.high == 51000.0
assert price.low == 49000.0
assert price.open == 50000.0
assert price.close == 50500.0
assert price.volume == 1000.0
assert price.time == "1704067200"
class TestCoinBaseWrapper:
"""Test specifici per CoinBaseWrapper."""
@pytest.mark.skipif(
not (os.getenv('COINBASE_API_KEY') and os.getenv('COINBASE_API_SECRET')),
reason="Credenziali Coinbase non configurate"
)
def test_coinbase_initialization_with_env_vars(self):
"""Test inizializzazione con variabili d'ambiente."""
wrapper = CoinBaseWrapper(currency="USD")
assert wrapper.currency == "USD"
assert wrapper.client is not None
@patch.dict(os.environ, {}, clear=True)
def test_coinbase_initialization_with_params(self):
"""Test inizializzazione con parametri espliciti quando non ci sono variabili d'ambiente."""
with pytest.raises(AssertionError, match="API key is required"):
CoinBaseWrapper(api_key=None, api_private_key=None)
@patch('app.markets.coinbase.RESTClient')
def test_coinbase_asset_formatting_behavior(self, mock_client):
"""Test comportamento di formattazione asset ID attraverso get_product."""
mock_response = Mock()
mock_response.product_id = "BTC-USD"
mock_response.base_currency_id = "BTC"
mock_response.price = "50000.0"
mock_response.volume_24h = "1000000.0"
mock_response.status = "TRADING"
mock_client_instance = Mock()
mock_client_instance.get_product.return_value = mock_response
mock_client.return_value = mock_client_instance
wrapper = CoinBaseWrapper(api_key="test", api_private_key="test")
# Test che entrambi i formati funzionino
wrapper.get_product("BTC")
wrapper.get_product("BTC-USD")
# Verifica che get_product sia stato chiamato con il formato corretto
assert mock_client_instance.get_product.call_count == 2
@patch('app.markets.coinbase.RESTClient')
def test_coinbase_get_product(self, mock_client):
"""Test get_product con mock."""
mock_response = Mock()
mock_response.product_id = "BTC-USD"
mock_response.base_currency_id = "BTC"
mock_response.price = "50000.0"
mock_response.volume_24h = "1000000.0"
mock_response.status = "TRADING"
mock_client_instance = Mock()
mock_client_instance.get_product.return_value = mock_response
mock_client.return_value = mock_client_instance
wrapper = CoinBaseWrapper(api_key="test", api_private_key="test")
product = wrapper.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
mock_client_instance.get_product.assert_called_once_with("BTC-USD")
class TestCryptoCompareWrapper:
"""Test specifici per CryptoCompareWrapper."""
@pytest.mark.skipif(
not os.getenv('CRYPTOCOMPARE_API_KEY'),
reason="CRYPTOCOMPARE_API_KEY non configurata"
)
def test_cryptocompare_initialization_with_env_var(self):
"""Test inizializzazione con variabile d'ambiente."""
wrapper = CryptoCompareWrapper(currency="USD")
assert wrapper.currency == "USD"
assert wrapper.api_key is not None
def test_cryptocompare_initialization_with_param(self):
"""Test inizializzazione con parametro esplicito."""
wrapper = CryptoCompareWrapper(api_key="test_key", currency="EUR")
assert wrapper.api_key == "test_key"
assert wrapper.currency == "EUR"
@patch('app.markets.cryptocompare.requests.get')
def test_cryptocompare_get_product(self, mock_get):
"""Test get_product con mock."""
mock_response = Mock()
mock_response.json.return_value = {
'RAW': {
'BTC': {
'USD': {
'FROMSYMBOL': 'BTC',
'TOSYMBOL': 'USD',
'PRICE': 50000.0,
'VOLUME24HOUR': 1000000.0
}
}
}
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
wrapper = CryptoCompareWrapper(api_key="test_key")
product = wrapper.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
assert product.price == 50000.0
def test_cryptocompare_get_all_products_workaround(self):
"""Test che get_all_products funzioni con il workaround implementato."""
wrapper = CryptoCompareWrapper(api_key="test_key")
# Il metodo ora dovrebbe restituire una lista di ProductInfo invece di sollevare NotImplementedError
products = wrapper.get_all_products()
assert isinstance(products, list)
# Verifica che la lista non sia vuota (dovrebbe contenere almeno alcuni asset popolari)
assert len(products) > 0
# Verifica che ogni elemento sia un ProductInfo
for product in products:
assert isinstance(product, ProductInfo)
class TestBinanceWrapper:
"""Test specifici per BinanceWrapper."""
def test_binance_initialization_without_credentials(self):
"""Test che l'inizializzazione fallisca senza credenziali."""
# Assicuriamoci che le variabili d'ambiente siano vuote per questo test
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(AssertionError, match="API key is required"):
BinanceWrapper(api_key=None, api_secret="test")
with pytest.raises(AssertionError, match="API secret is required"):
BinanceWrapper(api_key="test", api_secret=None)
@patch('app.markets.binance.Client')
def test_binance_symbol_formatting_behavior(self, mock_client):
"""Test comportamento di formattazione simbolo attraverso get_product."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
wrapper = BinanceWrapper(api_key="test", api_secret="test")
# Test che entrambi i formati funzionino
wrapper.get_product("BTC")
wrapper.get_product("BTCUSDT")
# Verifica che i metodi siano stati chiamati
assert mock_client_instance.get_symbol_ticker.call_count == 2
@patch('app.markets.binance.Client')
def test_binance_get_product(self, mock_client):
"""Test get_product con mock."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
wrapper = BinanceWrapper(api_key="test", api_secret="test")
product = wrapper.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
assert product.price == 50000.0
class TestPublicBinanceAgent:
"""Test specifici per PublicBinanceAgent."""
@patch('app.markets.binance_public.Client')
def test_public_binance_initialization(self, mock_client):
"""Test inizializzazione senza credenziali."""
agent = PublicBinanceAgent()
assert agent.client is not None
mock_client.assert_called_once_with()
@patch('app.markets.binance_public.Client')
def test_public_binance_symbol_formatting_behavior(self, mock_client):
"""Test comportamento di formattazione simbolo attraverso get_product."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
# Test che entrambi i formati funzionino
agent.get_product("BTC")
agent.get_product("BTCUSDT")
# Verifica che i metodi siano stati chiamati
assert mock_client_instance.get_symbol_ticker.call_count == 2
@patch('app.markets.binance_public.Client')
def test_public_binance_get_product(self, mock_client):
"""Test get_product con mock."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
product = agent.get_product("BTC")
assert isinstance(product, ProductInfo)
assert product.symbol == "BTC"
assert product.price == 50000.0
@patch('app.markets.binance_public.Client')
def test_public_binance_get_all_products(self, mock_client):
"""Test get_all_products restituisce asset principali."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {
'symbol': 'BTCUSDT',
'price': '50000.0'
}
mock_client_instance.get_ticker.return_value = {
'volume': '1000000.0'
}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
products = agent.get_all_products()
assert isinstance(products, list)
assert len(products) == 8 # Numero di asset principali definiti
for product in products:
assert isinstance(product, ProductInfo)
@patch('app.markets.binance_public.Client')
def test_public_binance_get_public_prices(self, mock_client):
"""Test metodo specifico get_public_prices."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.return_value = {'price': '50000.0'}
mock_client_instance.get_server_time.return_value = {'serverTime': 1704067200000}
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
prices = agent.get_public_prices(["BTCUSDT"])
assert isinstance(prices, dict)
assert 'BTC_USD' in prices
assert prices['BTC_USD'] == 50000.0
assert 'source' in prices
assert prices['source'] == 'binance_public'
@pytest.mark.limited # usa molte api calls e non voglio esaurire le chiavi api
class TestMarketAPIs:
"""Test per la classe MarketAPIs che aggrega i wrapper."""
def test_market_apis_initialization_no_providers(self):
"""Test che l'inizializzazione fallisca senza provider disponibili."""
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(AssertionError, match="No market API keys"):
MarketAPIs("USD")
@patch('app.markets.CoinBaseWrapper')
def test_market_apis_with_coinbase_only(self, mock_coinbase):
"""Test con solo Coinbase disponibile."""
mock_instance = Mock()
mock_coinbase.return_value = mock_instance
with patch('app.markets.CryptoCompareWrapper', side_effect=Exception("No API key")):
apis = MarketAPIs("USD")
assert len(apis.wrappers) == 1
assert apis.wrappers[0] == mock_instance
@patch('app.markets.CoinBaseWrapper')
@patch('app.markets.CryptoCompareWrapper')
def test_market_apis_delegation(self, mock_crypto, mock_coinbase):
"""Test che i metodi vengano delegati al primo wrapper disponibile."""
mock_coinbase_instance = Mock()
mock_crypto_instance = Mock()
mock_coinbase.return_value = mock_coinbase_instance
mock_crypto.return_value = mock_crypto_instance
apis = MarketAPIs("USD")
# Test delegazione get_product
apis.get_product("BTC")
mock_coinbase_instance.get_product.assert_called_once_with("BTC")
# Test delegazione get_products
apis.get_products(["BTC", "ETH"])
mock_coinbase_instance.get_products.assert_called_once_with(["BTC", "ETH"])
# Test delegazione get_all_products
apis.get_all_products()
mock_coinbase_instance.get_all_products.assert_called_once()
# Test delegazione get_historical_prices
apis.get_historical_prices("BTC")
mock_coinbase_instance.get_historical_prices.assert_called_once_with("BTC")
def test_wrapper_initialization(self):
market_wrapper = MarketAPIs("USD")
assert market_wrapper is not None
assert hasattr(market_wrapper, 'get_product')
assert hasattr(market_wrapper, 'get_products')
assert hasattr(market_wrapper, 'get_all_products')
assert hasattr(market_wrapper, 'get_historical_prices')
def test_wrapper_capabilities(self):
market_wrapper = MarketAPIs("USD")
capabilities = []
if hasattr(market_wrapper, 'get_product'):
capabilities.append('single_product')
if hasattr(market_wrapper, 'get_products'):
capabilities.append('multiple_products')
if hasattr(market_wrapper, 'get_historical_prices'):
capabilities.append('historical_data')
assert len(capabilities) > 0
class TestErrorHandling:
"""Test per la gestione degli errori in tutti i wrapper."""
@patch('app.markets.binance_public.Client')
def test_public_binance_error_handling(self, mock_client):
"""Test gestione errori in PublicBinanceAgent."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.side_effect = Exception("API Error")
mock_client.return_value = mock_client_instance
agent = PublicBinanceAgent()
product = agent.get_product("INVALID")
# Dovrebbe restituire un ProductInfo vuoto invece di sollevare eccezione
assert isinstance(product, ProductInfo)
assert product.id == "INVALID"
assert product.symbol == "INVALID"
@patch('app.markets.cryptocompare.requests.get')
def test_cryptocompare_network_error(self, mock_get):
"""Test gestione errori di rete in CryptoCompareWrapper."""
mock_get.side_effect = Exception("Network Error")
wrapper = CryptoCompareWrapper(api_key="test")
with pytest.raises(Exception):
wrapper.get_product("BTC")
@patch('app.markets.binance.Client')
def test_binance_api_error_in_get_products(self, mock_client):
"""Test gestione errori in BinanceWrapper.get_products."""
mock_client_instance = Mock()
mock_client_instance.get_symbol_ticker.side_effect = Exception("API Error")
mock_client.return_value = mock_client_instance
wrapper = BinanceWrapper(api_key="test", api_secret="test")
products = wrapper.get_products(["BTC", "ETH"])
# Dovrebbe restituire lista vuota invece di sollevare eccezione
assert isinstance(products, list)
assert len(products) == 0
def test_market_data_retrieval(self):
market_wrapper = MarketAPIs("USD")
btc_product = market_wrapper.get_product("BTC")
assert btc_product is not None
assert hasattr(btc_product, 'symbol')
assert hasattr(btc_product, 'price')
assert btc_product.price > 0
def test_market_toolkit_integration(self):
try:
toolkit = MarketToolkit()
assert toolkit is not None
assert hasattr(toolkit, 'market_agent')
assert toolkit.market_api is not None
class TestIntegrationScenarios:
"""Test di integrazione per scenari reali."""
def test_wrapper_method_signatures(self):
"""Verifica che tutti i wrapper abbiano le stesse signature dei metodi."""
wrapper_classes = [CoinBaseWrapper, CryptoCompareWrapper, BinanceWrapper, PublicBinanceAgent]
for wrapper_class in wrapper_classes:
# Verifica get_product
assert hasattr(wrapper_class, 'get_product')
# Verifica get_products
assert hasattr(wrapper_class, 'get_products')
# Verifica get_all_products
assert hasattr(wrapper_class, 'get_all_products')
# Verifica get_historical_prices
assert hasattr(wrapper_class, 'get_historical_prices')
def test_productinfo_consistency(self):
"""Test che tutti i metodi from_* di ProductInfo restituiscano oggetti consistenti."""
# Test from_cryptocompare
crypto_data = {
'FROMSYMBOL': 'BTC',
'TOSYMBOL': 'USD',
'PRICE': 50000.0,
'VOLUME24HOUR': 1000000.0
}
crypto_product = ProductInfo.from_cryptocompare(crypto_data)
# Test from_binance
binance_ticker = {'symbol': 'BTCUSDT', 'price': '50000.0'}
binance_24h = {'volume': '1000000.0'}
binance_product = ProductInfo.from_binance(binance_ticker, binance_24h)
# Verifica che entrambi abbiano gli stessi campi
assert hasattr(crypto_product, 'id')
assert hasattr(crypto_product, 'symbol')
assert hasattr(crypto_product, 'price')
assert hasattr(crypto_product, 'volume_24h')
assert hasattr(binance_product, 'id')
assert hasattr(binance_product, 'symbol')
assert hasattr(binance_product, 'price')
assert hasattr(binance_product, 'volume_24h')
def test_price_consistency(self):
"""Test che tutti i metodi from_* di Price restituiscano oggetti consistenti."""
# Test from_cryptocompare
crypto_data = {
'high': 51000.0,
'low': 49000.0,
'open': 50000.0,
'close': 50500.0,
'volumeto': 1000.0,
'time': 1704067200
}
crypto_price = Price.from_cryptocompare(crypto_data)
# Verifica che abbia tutti i campi richiesti
assert hasattr(crypto_price, 'high')
assert hasattr(crypto_price, 'low')
assert hasattr(crypto_price, 'open')
assert hasattr(crypto_price, 'close')
assert hasattr(crypto_price, 'volume')
assert hasattr(crypto_price, 'time')
tools = toolkit.tools
assert len(tools) > 0
except Exception as e:
print(f"MarketToolkit test failed: {e}")
# Non fail completamente - il toolkit potrebbe avere dipendenze specifiche
if __name__ == "__main__":
pytest.main([__file__, "-v"])
def test_provider_selection_mechanism(self):
potential_providers = 0
if os.getenv('CDP_API_KEY_NAME') and os.getenv('CDP_API_PRIVATE_KEY'):
potential_providers += 1
if os.getenv('CRYPTOCOMPARE_API_KEY'):
potential_providers += 1
def test_error_handling(self):
try:
market_wrapper = MarketAPIs("USD")
fake_product = market_wrapper.get_product("NONEXISTENT_CRYPTO_SYMBOL_12345")
assert fake_product is None or fake_product.price == 0
except Exception as e:
pass
def test_wrapper_currency_support(self):
market_wrapper = MarketAPIs("USD")
assert hasattr(market_wrapper, 'currency')
assert isinstance(market_wrapper.currency, str)
assert len(market_wrapper.currency) >= 3 # USD, EUR, etc.

7
tests/api/test_binance.py Normal file
View File

@@ -0,0 +1,7 @@
import pytest
@pytest.mark.market
@pytest.mark.api
class TestBinance:
# TODO fare dei test veri e propri
pass

54
tests/api/test_coinbase.py Normal file
View File

@@ -0,0 +1,54 @@
import os
import pytest
from app.markets import CoinBaseWrapper
@pytest.mark.market
@pytest.mark.api
@pytest.mark.skipif(not(os.getenv('COINBASE_API_KEY')) or not(os.getenv('COINBASE_API_SECRET')), reason="COINBASE_API_KEY or COINBASE_API_SECRET not set in environment variables")
class TestCoinBase:
def test_coinbase_init(self):
market = CoinBaseWrapper()
assert market is not None
assert hasattr(market, 'currency')
assert market.currency == "USD"
def test_coinbase_get_product(self):
market = CoinBaseWrapper()
product = market.get_product("BTC")
assert product is not None
assert hasattr(product, 'symbol')
assert product.symbol == "BTC"
assert hasattr(product, 'price')
assert product.price > 0
def test_coinbase_get_products(self):
market = CoinBaseWrapper()
products = market.get_products(["BTC", "ETH"])
assert products is not None
assert isinstance(products, list)
assert len(products) == 2
symbols = [p.symbol for p in products]
assert "BTC" in symbols
assert "ETH" in symbols
for product in products:
assert hasattr(product, 'price')
assert product.price > 0
def test_coinbase_invalid_product(self):
market = CoinBaseWrapper()
with pytest.raises(Exception):
_ = market.get_product("INVALID")
def test_coinbase_history(self):
market = CoinBaseWrapper()
history = market.get_historical_prices("BTC", limit=5)
assert history is not None
assert isinstance(history, list)
assert len(history) == 5
for entry in history:
assert hasattr(entry, 'time')
assert hasattr(entry, 'close')
assert hasattr(entry, 'high')
assert entry.close > 0
assert entry.high > 0

56
tests/api/test_cryptocompare.py Normal file
View File

@@ -0,0 +1,56 @@
import os
import pytest
from app.markets import CryptoCompareWrapper
@pytest.mark.market
@pytest.mark.api
@pytest.mark.skipif(not os.getenv('CRYPTOCOMPARE_API_KEY'), reason="CRYPTOCOMPARE_API_KEY not set in environment variables")
class TestCryptoCompare:
def test_cryptocompare_init(self):
market = CryptoCompareWrapper()
assert market is not None
assert hasattr(market, 'api_key')
assert market.api_key == os.getenv('CRYPTOCOMPARE_API_KEY')
assert hasattr(market, 'currency')
assert market.currency == "USD"
def test_cryptocompare_get_product(self):
market = CryptoCompareWrapper()
product = market.get_product("BTC")
assert product is not None
assert hasattr(product, 'symbol')
assert product.symbol == "BTC"
assert hasattr(product, 'price')
assert product.price > 0
def test_cryptocompare_get_products(self):
market = CryptoCompareWrapper()
products = market.get_products(["BTC", "ETH"])
assert products is not None
assert isinstance(products, list)
assert len(products) == 2
symbols = [p.symbol for p in products]
assert "BTC" in symbols
assert "ETH" in symbols
for product in products:
assert hasattr(product, 'price')
assert product.price > 0
def test_cryptocompare_invalid_product(self):
market = CryptoCompareWrapper()
with pytest.raises(Exception):
_ = market.get_product("INVALID")
def test_cryptocompare_history(self):
market = CryptoCompareWrapper()
history = market.get_historical_prices("BTC", limit=5)
assert history is not None
assert isinstance(history, list)
assert len(history) == 5
for entry in history:
assert hasattr(entry, 'time')
assert hasattr(entry, 'close')
assert hasattr(entry, 'high')
assert entry.close > 0
assert entry.high > 0

22
tests/conftest.py
View File

@@ -15,8 +15,7 @@ def pytest_configure(config:pytest.Config):
markers = [
("slow", "marks tests as slow (deselect with '-m \"not slow\"')"),
("api", "marks tests that require API access"),
("coinbase", "marks tests that require Coinbase credentials"),
("cryptocompare", "marks tests that require CryptoCompare credentials"),
("market", "marks tests that use market data"),
("gemini", "marks tests that use Gemini model"),
("ollama_gpt", "marks tests that use Ollama GPT model"),
("ollama_qwen", "marks tests that use Ollama Qwen model"),
@@ -30,24 +29,7 @@ def pytest_configure(config:pytest.Config):
config.addinivalue_line("markers", line)
def pytest_collection_modifyitems(config, items):
"""Modifica automaticamente gli item di test aggiungendogli marker basati sul nome"""
markers_to_add = {
"coinbase": pytest.mark.api,
"cryptocompare": pytest.mark.api,
"overview": pytest.mark.slow,
"analysis": pytest.mark.slow,
"gemini": pytest.mark.gemini,
"ollama_gpt": pytest.mark.ollama_gpt,
"ollama_qwen": pytest.mark.ollama_qwen,
}
for item in items:
name = item.name.lower()
for key, marker in markers_to_add.items():
if key in name:
item.add_marker(marker)
"""Modifica automaticamente degli item di test rimovendoli"""
# Rimuovo i test "limited" e "slow" se non richiesti esplicitamente
mark_to_remove = ['limited', 'slow']
for mark in mark_to_remove: