refactor: simplify product info and price creation in YFinanceWrapper
This commit is contained in:
@@ -3,63 +3,30 @@ from agno.tools.yfinance import YFinanceTools
|
||||
from .base import BaseWrapper, ProductInfo, Price
|
||||
|
||||
|
||||
def create_product_info(symbol: str, stock_data: dict) -> ProductInfo:
|
||||
def create_product_info(stock_data: dict[str, str]) -> ProductInfo:
|
||||
"""
|
||||
Converte i dati di YFinanceTools in ProductInfo.
|
||||
"""
|
||||
product = ProductInfo()
|
||||
|
||||
# ID univoco per yfinance
|
||||
product.id = f"yfinance_{symbol}"
|
||||
product.symbol = symbol
|
||||
|
||||
# Estrai il prezzo corrente - gestisci diversi formati
|
||||
if 'currentPrice' in stock_data:
|
||||
product.price = float(stock_data['currentPrice'])
|
||||
elif 'regularMarketPrice' in stock_data:
|
||||
product.price = float(stock_data['regularMarketPrice'])
|
||||
elif 'Current Stock Price' in stock_data:
|
||||
# Formato: "254.63 USD" - estrai solo il numero
|
||||
price_str = stock_data['Current Stock Price'].split()[0]
|
||||
try:
|
||||
product.price = float(price_str)
|
||||
except ValueError:
|
||||
product.price = 0.0
|
||||
else:
|
||||
product.price = 0.0
|
||||
|
||||
# Volume 24h
|
||||
if 'volume' in stock_data:
|
||||
product.volume_24h = float(stock_data['volume'])
|
||||
elif 'regularMarketVolume' in stock_data:
|
||||
product.volume_24h = float(stock_data['regularMarketVolume'])
|
||||
else:
|
||||
product.volume_24h = 0.0
|
||||
|
||||
# Status basato sulla disponibilità dei dati
|
||||
product.id = stock_data.get('Symbol', '')
|
||||
product.symbol = product.id.split('-')[0] # Rimuovi il suffisso della valuta per le crypto
|
||||
product.price = float(stock_data.get('Current Stock Price', f"0.0 USD").split(" ")[0]) # prende solo il numero
|
||||
product.volume_24h = 0.0 # YFinance non fornisce il volume 24h direttamente
|
||||
product.status = "trading" if product.price > 0 else "offline"
|
||||
|
||||
# Valuta (default USD)
|
||||
product.quote_currency = stock_data.get('currency', 'USD') or 'USD'
|
||||
|
||||
product.quote_currency = product.id.split('-')[0] # La valuta è la parte dopo il '-'
|
||||
return product
|
||||
|
||||
|
||||
def create_price_from_history(hist_data: dict, timestamp: str) -> Price:
|
||||
def create_price_from_history(hist_data: dict[str, str]) -> Price:
|
||||
"""
|
||||
Converte i dati storici di YFinanceTools in Price.
|
||||
"""
|
||||
price = Price()
|
||||
|
||||
if timestamp in hist_data:
|
||||
day_data = hist_data[timestamp]
|
||||
price.high = float(day_data.get('High', 0.0))
|
||||
price.low = float(day_data.get('Low', 0.0))
|
||||
price.open = float(day_data.get('Open', 0.0))
|
||||
price.close = float(day_data.get('Close', 0.0))
|
||||
price.volume = float(day_data.get('Volume', 0.0))
|
||||
price.time = timestamp
|
||||
|
||||
price.high = float(hist_data.get('High', 0.0))
|
||||
price.low = float(hist_data.get('Low', 0.0))
|
||||
price.open = float(hist_data.get('Open', 0.0))
|
||||
price.close = float(hist_data.get('Close', 0.0))
|
||||
price.volume = float(hist_data.get('Volume', 0.0))
|
||||
price.time = hist_data.get('Timestamp', '')
|
||||
return price
|
||||
|
||||
|
||||
@@ -72,143 +39,46 @@ class YFinanceWrapper(BaseWrapper):
|
||||
|
||||
def __init__(self, currency: str = "USD"):
|
||||
self.currency = currency
|
||||
# Inizializza YFinanceTools - non richiede parametri specifici
|
||||
self.tool = YFinanceTools()
|
||||
|
||||
def _format_symbol(self, asset_id: str) -> str:
|
||||
"""
|
||||
Formatta il simbolo per yfinance.
|
||||
Per crypto, aggiunge '-USD' se non presente.
|
||||
Per crypto, aggiunge '-' e la valuta (es. BTC -> BTC-USD).
|
||||
"""
|
||||
asset_id = asset_id.upper()
|
||||
|
||||
# Se è già nel formato corretto (es: BTC-USD), usa così
|
||||
if '-' in asset_id:
|
||||
return asset_id
|
||||
|
||||
# Per crypto singole (BTC, ETH), aggiungi -USD
|
||||
if asset_id in ['BTC', 'ETH', 'ADA', 'SOL', 'DOT', 'LINK', 'UNI', 'AAVE']:
|
||||
return f"{asset_id}-USD"
|
||||
|
||||
# Per azioni, usa il simbolo così com'è
|
||||
return asset_id
|
||||
return f"{asset_id}-{self.currency}" if '-' not in asset_id else asset_id
|
||||
|
||||
def get_product(self, asset_id: str) -> ProductInfo:
|
||||
"""
|
||||
Recupera le informazioni di un singolo prodotto.
|
||||
"""
|
||||
symbol = self._format_symbol(asset_id)
|
||||
|
||||
# Usa YFinanceTools per ottenere i dati
|
||||
try:
|
||||
# Ottieni le informazioni base dello stock
|
||||
stock_info = self.tool.get_company_info(symbol)
|
||||
|
||||
# Se il risultato è una stringa JSON, parsala
|
||||
if isinstance(stock_info, str):
|
||||
try:
|
||||
stock_data = json.loads(stock_info)
|
||||
except json.JSONDecodeError:
|
||||
# Se non è JSON valido, prova a ottenere solo il prezzo
|
||||
price_data_str = self.tool.get_current_stock_price(symbol)
|
||||
if price_data_str and price_data_str.replace('.', '').replace('-', '').isdigit():
|
||||
price = float(price_data_str)
|
||||
stock_data = {'currentPrice': price, 'currency': 'USD'}
|
||||
else:
|
||||
raise Exception("Dati non validi")
|
||||
else:
|
||||
stock_data = stock_info
|
||||
|
||||
return create_product_info(symbol, stock_data)
|
||||
|
||||
except Exception as e:
|
||||
# Fallback: prova a ottenere solo il prezzo
|
||||
try:
|
||||
price_data_str = self.tool.get_current_stock_price(symbol)
|
||||
if price_data_str and price_data_str.replace('.', '').replace('-', '').isdigit():
|
||||
price = float(price_data_str)
|
||||
minimal_data = {
|
||||
'currentPrice': price,
|
||||
'currency': 'USD'
|
||||
}
|
||||
return create_product_info(symbol, minimal_data)
|
||||
else:
|
||||
raise Exception("Prezzo non disponibile")
|
||||
except Exception:
|
||||
# Se tutto fallisce, restituisci un prodotto vuoto
|
||||
product = ProductInfo()
|
||||
product.symbol = symbol
|
||||
product.status = "offline"
|
||||
return product
|
||||
stock_info = json.loads(stock_info)
|
||||
return create_product_info(stock_info)
|
||||
|
||||
def get_products(self, asset_ids: list[str]) -> list[ProductInfo]:
|
||||
"""
|
||||
Recupera le informazioni di multiple assets.
|
||||
"""
|
||||
products = []
|
||||
|
||||
for asset_id in asset_ids:
|
||||
try:
|
||||
product = self.get_product(asset_id)
|
||||
products.append(product)
|
||||
except Exception as e:
|
||||
# Se un asset non è disponibile, continua con gli altri
|
||||
continue
|
||||
|
||||
return products
|
||||
|
||||
def get_all_products(self) -> list[ProductInfo]:
|
||||
"""
|
||||
Recupera tutti i prodotti disponibili.
|
||||
Restituisce una lista predefinita di asset popolari.
|
||||
"""
|
||||
# Lista di asset popolari (azioni, ETF, crypto)
|
||||
popular_assets = [
|
||||
'BTC', 'ETH', 'ADA', 'SOL', 'DOT',
|
||||
'AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN',
|
||||
'SPY', 'QQQ', 'VTI', 'GLD', 'VIX'
|
||||
]
|
||||
|
||||
return self.get_products(popular_assets)
|
||||
raise NotImplementedError("YFinanceWrapper does not support get_all_products due to API limitations.")
|
||||
|
||||
def get_historical_prices(self, asset_id: str = "BTC", limit: int = 100) -> list[Price]:
|
||||
"""
|
||||
Recupera i dati storici di prezzo per un asset.
|
||||
"""
|
||||
symbol = self._format_symbol(asset_id)
|
||||
|
||||
try:
|
||||
# Determina il periodo appropriato in base al limite
|
||||
if limit <= 7:
|
||||
period = "1d"
|
||||
interval = "15m"
|
||||
elif limit <= 30:
|
||||
period = "5d"
|
||||
interval = "1h"
|
||||
elif limit <= 90:
|
||||
period = "1mo"
|
||||
interval = "1d"
|
||||
else:
|
||||
period = "3mo"
|
||||
interval = "1d"
|
||||
|
||||
# Ottieni i dati storici
|
||||
hist_data = self.tool.get_historical_stock_prices(symbol, period=period, interval=interval)
|
||||
|
||||
if isinstance(hist_data, str):
|
||||
days = limit // 24 + 1 # Arrotonda per eccesso
|
||||
hist_data = self.tool.get_historical_stock_prices(symbol, period=f"{days}d", interval="1h")
|
||||
hist_data = json.loads(hist_data)
|
||||
|
||||
# Il formato dei dati è {timestamp: {Open: x, High: y, Low: z, Close: w, Volume: v}}
|
||||
timestamps = sorted(hist_data.keys())[-limit:]
|
||||
|
||||
prices = []
|
||||
timestamps = sorted(hist_data.keys())[-limit:] # Prendi gli ultimi 'limit' timestamp
|
||||
|
||||
for timestamp in timestamps:
|
||||
price = create_price_from_history(hist_data, timestamp)
|
||||
if price.close > 0: # Solo se ci sono dati validi
|
||||
temp = hist_data[timestamp]
|
||||
temp['Timestamp'] = timestamp
|
||||
price = create_price_from_history(temp)
|
||||
prices.append(price)
|
||||
|
||||
return prices
|
||||
|
||||
except Exception as e:
|
||||
# Se fallisce, restituisci lista vuota
|
||||
return []
|
||||
@@ -1,4 +1,3 @@
|
||||
import os
|
||||
import pytest
|
||||
from app.markets import YFinanceWrapper
|
||||
|
||||
@@ -14,17 +13,6 @@ class TestYFinance:
|
||||
assert hasattr(market, 'tool')
|
||||
assert market.tool is not None
|
||||
|
||||
def test_yfinance_get_product(self):
|
||||
market = YFinanceWrapper()
|
||||
product = market.get_product("AAPL")
|
||||
assert product is not None
|
||||
assert hasattr(product, 'symbol')
|
||||
assert product.symbol == "AAPL"
|
||||
assert hasattr(product, 'price')
|
||||
assert product.price > 0
|
||||
assert hasattr(product, 'status')
|
||||
assert product.status == "trading"
|
||||
|
||||
def test_yfinance_get_crypto_product(self):
|
||||
market = YFinanceWrapper()
|
||||
product = market.get_product("BTC")
|
||||
@@ -37,49 +25,21 @@ class TestYFinance:
|
||||
|
||||
def test_yfinance_get_products(self):
|
||||
market = YFinanceWrapper()
|
||||
products = market.get_products(["AAPL", "GOOGL"])
|
||||
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 "AAPL" in symbols
|
||||
assert "GOOGL" in symbols
|
||||
assert "BTC" in symbols
|
||||
assert "ETH" in symbols
|
||||
for product in products:
|
||||
assert hasattr(product, 'price')
|
||||
assert product.price > 0
|
||||
|
||||
def test_yfinance_get_all_products(self):
|
||||
market = YFinanceWrapper()
|
||||
products = market.get_all_products()
|
||||
assert products is not None
|
||||
assert isinstance(products, list)
|
||||
assert len(products) > 0
|
||||
# Dovrebbe contenere asset popolari
|
||||
symbols = [p.symbol for p in products]
|
||||
assert "AAPL" in symbols # Apple dovrebbe essere nella lista
|
||||
for product in products:
|
||||
assert hasattr(product, 'symbol')
|
||||
assert hasattr(product, 'price')
|
||||
|
||||
def test_yfinance_invalid_product(self):
|
||||
market = YFinanceWrapper()
|
||||
# Per YFinance, un prodotto invalido dovrebbe restituire un prodotto offline
|
||||
product = market.get_product("INVALIDSYMBOL123")
|
||||
assert product is not None
|
||||
assert product.status == "offline"
|
||||
|
||||
def test_yfinance_history(self):
|
||||
market = YFinanceWrapper()
|
||||
history = market.get_historical_prices("AAPL", 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
|
||||
with pytest.raises(Exception):
|
||||
_ = market.get_product("INVALIDSYMBOL123")
|
||||
|
||||
def test_yfinance_crypto_history(self):
|
||||
market = YFinanceWrapper()
|
||||
@@ -91,3 +51,5 @@ class TestYFinance:
|
||||
assert hasattr(entry, 'time')
|
||||
assert hasattr(entry, 'close')
|
||||
assert entry.close > 0
|
||||
assert hasattr(entry, 'open')
|
||||
assert entry.open > 0
|
||||
Reference in New Issue
Block a user