Refactor Predictor and market data handling

- Added Predictor class with input preparation and instructions for financial strategy generation.
- Removed PredictorAgent class and integrated its functionality into the new Predictor module.
- Created a base market API wrapper and specific implementations for Coinbase and CryptoCompare.
- Introduced PublicBinanceAgent for fetching public prices from Binance.
- Refactored ToolAgent to utilize the new Predictor and market API wrappers for improved data handling and predictions.
- Updated models to streamline the selection of available LLM providers.
- Removed deprecated signer classes for Coinbase and CryptoCompare.
This commit is contained in:
2025-09-26 03:43:31 +02:00
parent 48502fc6c7
commit 148bff7cfd
18 changed files with 362 additions and 2324 deletions

View File

@@ -1,3 +1,7 @@
###########################################################################
# Configurazioni per i modelli di linguaggio
###############################################################################
# Alcune API sono a pagamento, altre hanno un piano gratuito con limiti di utilizzo
# Vedi https://docs.agno.com/examples/models per vedere tutti i modelli supportati
GOOGLE_API_KEY=
@@ -8,17 +12,18 @@ GOOGLE_API_KEY=
# wsl: /usr/share/ollama/.ollama
OLLAMA_MODELS_PATH=
###############################################################################
# Configurazioni per gli agenti di mercato
###############################################################################
# Coinbase CDP API per Market Agent
# Ottenibili da: https://portal.cdp.coinbase.com/access/api
CDP_API_KEY_NAME=organizations/your-org-id/apiKeys/your-key-id
CDP_API_PRIVATE_KEY=-----BEGIN EC PRIVATE KEY-----
YOUR_ACTUAL_PRIVATE_KEY_HERE
-----END EC PRIVATE KEY-----
CDP_API_KEY_NAME=
CDP_API_PRIVATE_KEY=
# CryptoCompare API per Market Agent (alternativa)
# Ottenibile da: https://www.cryptocompare.com/cryptopian/api-keys
CRYPTOCOMPARE_API_KEY=
CRYPTOCOMPARE_AUTH_METHOD=query
# Binance API per Market Agent (alternativa)
BINANCE_API_KEY=

View File

@@ -1,7 +1,7 @@
[project]
name = "upo-app-ai"
version = "0.1.0"
description = "Agente di Pianificazione Viaggi Intelligente"
description = "Agente di analisi e consulenza crypto"
requires-python = "==3.12.*"
# Qui ho messo alcune librerie utili utilizzate durante il corso.
@@ -18,32 +18,17 @@ dependencies = [
#"bs4",
# ✅ per fare una UI web semplice con input e output
"gradio",
# ✅ per la crittografia (necessaria per autenticazione Coinbase)
"cryptography",
# ❌ per l'elaborazione del linguaggio naturale in locale (https://huggingface.co/learn/llm-course/chapter1/3?fw=pt)
#"transformers",
# ❌ per fare chiamate a modelli indipendentemente dal modello specifico (astrae meglio rispetto a openai)
#"langchain",
# ✅ per costruire agenti (ovvero modelli che possono fare più cose tramite tool) https://github.com/agno-agi/agno
# altamente consigliata dato che ha anche tools integrati per fare scraping, calcoli e molto altro
# oltre a questa è necessario installare anche le librerie specifiche per i modelli che si vogliono usare
"agno",
# ✅ Modelli supportati e installati (aggiungere qui sotto quelli che si vogliono usare)
"google-genai",
"ollama",
"openai",
"anthropic",
"google",
# ✅ per interagire con API di exchange di criptovalute
"coinbase-advanced-py",
"cryptocompare",
"cdp-sdk",
"python-binance",
"langchain>=0.3.27",
"langchain-community>=0.3.21",
"langchain-chroma>=0.2.2",
"langchain-ollama>=0.3.7",
"chromadb",
"fastapi",
"uvicorn",
"pydantic"
]

View File

@@ -3,6 +3,7 @@ import gradio as gr
from dotenv import load_dotenv
from app.tool import ToolAgent
from app.models import Models
from app.agents.predictor import PredictorStyle
########################################
# MAIN APP & GRADIO INTERFACE
@@ -16,21 +17,29 @@ if __name__ == "__main__":
load_dotenv()
######################################
list_models = Models.available()
tool_agent = ToolAgent()
models = Models.availables()
styles = list(PredictorStyle)
models_names = [model.name for model in models]
styles_names = [style.value for style in styles]
print("✅ Modelli disponibili:", models_names)
print("✅ Stili disponibili:", styles_names)
tool_agent = ToolAgent(models, styles)
with gr.Blocks() as demo:
gr.Markdown("# 🤖 Agente di Analisi e Consulenza Crypto")
with gr.Row():
provider = gr.Dropdown(
choices=list_models,
value=list_models[0],
choices=models_names,
type="index",
label="Modello da usare"
)
provider.change(fn=tool_agent.choose_provider, inputs=provider, outputs=None)
style = gr.Dropdown(
choices=["conservative", "aggressive"],
value="conservative",
choices=styles_names,
type="index",
label="Stile di investimento"
)
@@ -38,7 +47,5 @@ if __name__ == "__main__":
output = gr.Textbox(label="Risultato analisi", lines=12)
analyze_btn = gr.Button("🔎 Analizza")
analyze_btn.click(fn=tool_agent.interact, inputs=[user_input, provider, style], outputs=output)
if __name__ == "__main__":
demo.launch()
analyze_btn.click(fn=tool_agent.interact, inputs=[user_input, style], outputs=output)
demo.launch()

39
src/app/agents/market.py Normal file
View File

@@ -0,0 +1,39 @@
from agno.tools import Toolkit
from app.markets import get_first_available_market_api
# TODO (?) in futuro fare in modo che la LLM faccia da sé per il mercato
# Non so se può essere utile, per ora lo lascio qui
# per ora mettiamo tutto statico e poi, se abbiamo API-Key senza limiti
# possiamo fare in modo di far scegliere alla LLM quale crypto proporre
# in base alle sue proprie chiamate API
class MarketToolkit(Toolkit):
def __init__(self):
self.market_agent = get_first_available_market_api("USD") # change currency if needed
super().__init__(
name="Market Toolkit",
tools=[
self.get_historical_data,
self.get_current_price,
],
)
def get_historical_data(self, symbol: str):
return self.market_agent.get_historical_prices(symbol)
def get_current_price(self, symbol: str):
return self.market_agent.get_products(symbol)
def prepare_inputs():
pass
def instructions():
return """
Utilizza questo strumento per ottenere dati di mercato storici e attuali per criptovalute specifiche.
Puoi richiedere i prezzi storici o il prezzo attuale di una criptovaluta specifica.
Esempio di utilizzo:
- get_historical_data("BTC")
- get_current_price("ETH")
"""

View File

@@ -1,266 +0,0 @@
from typing import Dict, List, Optional, Any
import requests
import logging
import os
from app.signers.market_signers.coinbase_signer import CoinbaseCDPSigner
from app.signers.market_signers.cryptocompare_signer import CryptoCompareSigner
logger = logging.getLogger(__name__)
class MarketAgent:
"""
Market Agent unificato che supporta multiple fonti di dati:
- Coinbase Advanced Trade API (dati di mercato reali)
- CryptoCompare (market data)
Auto-configura i provider basandosi sulle variabili d'ambiente disponibili.
"""
def __init__(self):
self.providers = {}
self._setup_providers()
if not self.providers:
logger.warning("No market data providers configured. Check your .env file.")
def _setup_providers(self):
"""Configura automaticamente i provider disponibili"""
# Setup Coinbase Advanced Trade API (nuovo formato)
cdp_api_key_name = os.getenv('CDP_API_KEY_NAME')
cdp_api_private_key = os.getenv('CDP_API_PRIVATE_KEY')
if cdp_api_key_name and cdp_api_private_key:
try:
signer = CoinbaseCDPSigner(cdp_api_key_name, cdp_api_private_key)
self.providers['coinbase'] = {
'type': 'coinbase_advanced_trade',
'signer': signer,
'capabilities': ['assets', 'market_data', 'trading', 'real_time_prices']
}
logger.info("✅ Coinbase Advanced Trade API provider configured")
except Exception as e:
logger.error(f"Failed to setup Coinbase Advanced Trade API provider: {e}")
# Setup CryptoCompare se la API key è disponibile
cryptocompare_key = os.getenv('CRYPTOCOMPARE_API_KEY')
if cryptocompare_key:
try:
auth_method = os.getenv('CRYPTOCOMPARE_AUTH_METHOD', 'query')
signer = CryptoCompareSigner(cryptocompare_key, auth_method)
self.providers['cryptocompare'] = {
'type': 'cryptocompare',
'signer': signer,
'base_url': 'https://min-api.cryptocompare.com',
'capabilities': ['prices', 'historical', 'top_coins']
}
logger.info("✅ CryptoCompare provider configured")
except Exception as e:
logger.error(f"Failed to setup CryptoCompare provider: {e}")
def get_available_providers(self) -> List[str]:
"""Restituisce la lista dei provider configurati"""
return list(self.providers.keys())
def get_provider_capabilities(self, provider: str) -> List[str]:
"""Restituisce le capacità di un provider specifico"""
if provider in self.providers:
return self.providers[provider]['capabilities']
return []
# === COINBASE CDP METHODS ===
def get_coinbase_asset_info(self, symbol: str) -> Dict:
"""Ottiene informazioni su un asset da Coinbase CDP"""
if 'coinbase' not in self.providers:
raise ValueError("Coinbase provider not configured")
signer = self.providers['coinbase']['signer']
return signer.get_asset_info(symbol)
def get_coinbase_multiple_assets(self, symbols: List[str]) -> Dict:
"""Ottiene informazioni su multipli asset da Coinbase CDP"""
if 'coinbase' not in self.providers:
raise ValueError("Coinbase provider not configured")
signer = self.providers['coinbase']['signer']
return signer.get_multiple_assets(symbols)
def get_asset_price(self, symbol: str, provider: str = None) -> Optional[float]:
"""
Ottiene il prezzo di un asset usando il provider specificato o il primo disponibile
"""
if provider == 'coinbase' and 'coinbase' in self.providers:
try:
asset_info = self.get_coinbase_asset_info(symbol)
return float(asset_info.get('price', 0))
except Exception as e:
logger.error(f"Error getting {symbol} price from Coinbase: {e}")
return None
elif provider == 'cryptocompare' and 'cryptocompare' in self.providers:
try:
return self.get_single_crypto_price(symbol)
except Exception as e:
logger.error(f"Error getting {symbol} price from CryptoCompare: {e}")
return None
# Auto-select provider
if 'cryptocompare' in self.providers:
try:
return self.get_single_crypto_price(symbol)
except Exception:
pass
if 'coinbase' in self.providers:
try:
asset_info = self.get_coinbase_asset_info(symbol)
return float(asset_info.get('price', 0))
except Exception:
pass
return None
# === CRYPTOCOMPARE METHODS ===
def _cryptocompare_request(self, endpoint: str, params: Dict = None) -> Dict:
"""Esegue una richiesta CryptoCompare autenticata"""
if 'cryptocompare' not in self.providers:
raise ValueError("CryptoCompare provider not configured")
provider = self.providers['cryptocompare']
request_data = provider['signer'].prepare_request(
provider['base_url'], endpoint, params
)
response = requests.get(
request_data['url'],
headers=request_data['headers'],
timeout=10
)
response.raise_for_status()
return response.json()
def get_crypto_prices(self, from_symbols: List[str], to_symbols: List[str] = None) -> Dict:
"""Ottiene prezzi da CryptoCompare"""
if to_symbols is None:
to_symbols = ["USD", "EUR"]
params = {
"fsyms": ",".join(from_symbols),
"tsyms": ",".join(to_symbols)
}
return self._cryptocompare_request("/data/pricemulti", params)
def get_single_crypto_price(self, from_symbol: str, to_symbol: str = "USD") -> float:
"""Ottiene il prezzo di una singola crypto da CryptoCompare"""
params = {
"fsym": from_symbol,
"tsyms": to_symbol
}
data = self._cryptocompare_request("/data/price", params)
return data.get(to_symbol, 0.0)
def get_top_cryptocurrencies(self, limit: int = 10, to_symbol: str = "USD") -> Dict:
"""Ottiene le top crypto per market cap da CryptoCompare"""
params = {
"limit": str(limit),
"tsym": to_symbol
}
return self._cryptocompare_request("/data/top/mktcapfull", params)
# === UNIFIED INTERFACE ===
def get_market_overview(self, symbols: List[str] = None) -> Dict:
"""
Ottiene una panoramica del mercato usando il miglior provider disponibile
"""
if symbols is None:
symbols = ["BTC", "ETH", "ADA"]
result = {
"timestamp": None,
"data": {},
"source": None,
"providers_used": []
}
# Prova CryptoCompare per prezzi multipli (più completo)
if 'cryptocompare' in self.providers:
try:
prices = self.get_crypto_prices(symbols, ["USD", "EUR"])
result["data"] = prices
result["source"] = "cryptocompare"
result["providers_used"].append("cryptocompare")
logger.info("Market overview retrieved from CryptoCompare")
except Exception as e:
logger.warning(f"CryptoCompare failed, trying fallback: {e}")
# Fallback a Coinbase Advanced Trade se CryptoCompare fallisce
if not result["data"] and 'coinbase' in self.providers:
try:
# Usa il nuovo metodo Advanced Trade per ottenere multipli asset
coinbase_data = self.get_coinbase_multiple_assets(symbols)
if coinbase_data:
# Trasforma i dati Advanced Trade nel formato standard
formatted_data = {}
for symbol in symbols:
if symbol in coinbase_data:
formatted_data[symbol] = {
"USD": coinbase_data[symbol].get("price", 0)
}
result["data"] = formatted_data
result["source"] = "coinbase_advanced_trade"
result["providers_used"].append("coinbase")
logger.info("Market overview retrieved from Coinbase Advanced Trade API")
except Exception as e:
logger.error(f"Coinbase Advanced Trade fallback failed: {e}")
return result
def analyze(self, query: str) -> str:
"""
Analizza il mercato usando tutti i provider disponibili
"""
if not self.providers:
return "⚠️ Nessun provider di dati di mercato configurato. Controlla il file .env."
try:
# Ottieni panoramica del mercato
overview = self.get_market_overview(["BTC", "ETH", "ADA", "DOT"])
if not overview["data"]:
return "⚠️ Impossibile recuperare dati di mercato da nessun provider."
# Formatta i risultati
result_lines = [
f"📊 **Market Analysis** (via {overview['source'].upper()})\n"
]
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":
result_lines.append(f"**{crypto}**: ${usd_price} (€{eur_price})")
else:
result_lines.append(f"**{crypto}**: ${usd_price}")
# Aggiungi info sui provider
providers_info = f"\n🔧 **Available providers**: {', '.join(self.get_available_providers())}"
result_lines.append(providers_info)
return "\n".join(result_lines)
except Exception as e:
logger.error(f"Market analysis failed: {e}")
return f"⚠️ Errore nell'analisi del mercato: {e}"

View File

@@ -0,0 +1,67 @@
import json
from enum import Enum
from app.markets.base import ProductInfo
class PredictorStyle(Enum):
CONSERVATIVE = "Conservativo"
AGGRESSIVE = "Aggressivo"
# TODO (?) Change sentiment to a more structured format or merge it with data analysis (change then also the prompt)
def prepare_inputs(data: list[ProductInfo], style: PredictorStyle, sentiment: str) -> str:
return json.dumps({
"data": [(product.symbol, f"{product.price:.2f}") for product in data],
"style": style.value,
"sentiment": sentiment
})
def instructions() -> str:
return """
Sei un **Consulente Finanziario Algoritmico (CFA) Specializzato in Criptovalute**. Il tuo compito è agire come un sistema esperto di gestione del rischio e allocazione di portafoglio.
**Istruzione Principale:** Analizza l'Input fornito in formato JSON. La tua strategia deve essere **logica, misurabile e basata esclusivamente sui dati e sullo stile di rischio/rendimento richiesto**.
## Input Dati (Formato JSON)
Ti sarà passato un singolo blocco JSON contenente i seguenti campi obbligatori:
1. **"data":** *Array di tuple (stringa)*. Rappresenta i dati di mercato in tempo reale o recenti. Ogni tupla è `[Nome_Asset, Prezzo_Corrente]`. **Esempio:** `[["BTC", "60000.00"], ["ETH", "3500.00"]]`.
2. **"style":** *Stringa ENUM (solo "conservativo" o "aggressivo")*. Definisce l'approccio alla volatilità e all'allocazione.
3. **"sentiment":** *Stringa descrittiva*. Riassume il sentiment di mercato estratto da fonti sociali e notizie. **Esempio:** `"Sentiment estremamente positivo, alta FOMO per le altcoin."`.
## Regole di Logica dello Stile di Investimento
- **Stile "Aggressivo":**
- **Obiettivo:** Massimizzazione del rendimento, accettando Volatilità Massima.
- **Allocazione:** Maggiore focus su **Asset a Media/Bassa Capitalizzazione (Altcoin)** o su criptovalute che mostrano un'elevata Momentum di crescita, anche se il rischio di ribasso è superiore. L'allocazione su BTC/ETH deve rimanere una base (ancoraggio) ma non dominare il portafoglio.
- **Correlazione Sentiment:** Sfrutta il sentiment positivo per allocazioni ad alto beta (più reattive ai cambiamenti di mercato).
- **Stile "Conservativo":**
- **Obiettivo:** Preservazione del capitale, minimizzazione della Volatilità.
- **Allocazione:** Maggioranza del capitale allocata in **Asset a Larga Capitalizzazione (Blue Chip: BTC e/o ETH)**. Eventuali allocazioni su Altcoin devono essere minime e su progetti con utilità comprovata e rischio regolatorio basso.
- **Correlazione Sentiment:** Utilizza il sentiment positivo come conferma per un'esposizione maggiore, ma ignora segnali eccessivi di "FOMO" (Fear Of Missing Out) per evitare asset speculativi.
## Requisiti di Formato dell'Output
L'output deve essere formattato in modo rigoroso, composto da **due sezioni distinte**: la Strategia e il Dettaglio del Portafoglio.
### 1. Strategia Sintetica
Fornisci una **descrizione operativa** della strategia. Deve essere:
- Estremamente concisa.
- Contenuta in un **massimo di 5 frasi totali**.
### 2. Dettaglio del Portafoglio
Presenta l'allocazione del portafoglio come una **lista puntata**.
- La somma delle percentuali deve essere **esattamente 100%**.
- Per **ogni Asset allocato**, devi fornire:
- **Nome dell'Asset** (es. BTC, ETH, SOL, ecc.)
- **Percentuale di Allocazione** (X%)
- **Motivazione Tecnica/Sintetica** (Massimo 1 frase chiara) che giustifichi l'allocazione in base ai *Dati di Mercato*, allo *Style* e al *Sentiment*.
**Formato Esempio di Output (da seguire fedelmente):**
[Strategia sintetico-operativa in massimo 5 frasi...]
- **Asset_A:** X% (Motivazione: [Massimo una frase che collega dati, stile e allocazione])
- **Asset_B:** Y% (Motivazione: [Massimo una frase che collega dati, stile e allocazione])
- **Asset_C:** Z% (Motivazione: [Massimo una frase che collega dati, stile e allocazione])
...
"""

View File

@@ -1,169 +0,0 @@
import json
import os
from typing import Any
import anthropic
import requests
from google import genai
from openai import OpenAI
class PredictorAgent:
def __init__(self):
# Ollama via HTTP locale
self.providers = {
"ollama": {"type": "ollama", "host": os.getenv("OLLAMA_HOST", "http://localhost:11434"), "model": "gpt-oss:latest"}
}
# OpenAI
openai_key = os.getenv("OPENAI_API_KEY")
if openai_key:
client = OpenAI(api_key=openai_key)
self.providers["openai"] = {
"type": "openai",
"client": client,
"model": "gpt-4o-mini"
}
# Anthropic
anthropic_key = os.getenv("ANTHROPIC_API_KEY")
if anthropic_key:
client = anthropic.Anthropic(api_key=anthropic_key)
self.providers["anthropic"] = {
"type": "anthropic",
"client": client,
"model": "claude-3-haiku-20240307"
}
# Google Gemini
google_key = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")
if google_key:
client = genai.Client(api_key=google_key)
self.providers["google"] = {"type": "google", "client": client, "model": "gemini-1.5-flash"}
# DeepSeek
deepseek_key = os.getenv("DEEPSEEK_API_KEY")
if deepseek_key:
self.providers["deepseek"] = {"type": "deepseek", "api_key": deepseek_key, "model": "deepseek-chat"}
print("✅ Providers attivi:", list(self.providers.keys()))
def predict(self, data, sentiment, style="conservative", provider="mock"):
provider = provider.lower()
if provider == "mock" or provider not in self.providers:
return self._predict_mock(style)
prompt = f"""
Sei un consulente finanziario crypto.
Dati di mercato: {data}
Sentiment estratto: {sentiment}
Stile richiesto: {style}
Fornisci una strategia di investimento chiara e breve (max 5 frasi),
con percentuali di portafoglio e motivazioni sintetiche."""
cfg: Any = self.providers[provider]
try:
if cfg["type"] == "ollama":
return self._predict_ollama_http(prompt, cfg["host"], cfg["model"])
elif cfg["type"] == "openai":
return self._predict_openai(prompt, cfg["client"], cfg["model"])
elif cfg["type"] == "anthropic":
return self._predict_anthropic(prompt, cfg["client"], cfg["model"])
elif cfg["type"] == "google":
return self._predict_google(prompt, cfg["client"], cfg["model"])
elif cfg["type"] == "deepseek":
return self._predict_deepseek(prompt, cfg["api_key"], cfg["model"])
return None
except Exception as e:
return f"⚠️ Provider {provider} non riconosciuto: {e}"
@staticmethod
def _predict_ollama_http(prompt, host, model):
url = host.rstrip("/") + "/api/generate"
payload = {"model": model, "prompt": prompt, "max_tokens": 300}
r = requests.post(url, json=payload, timeout=60)
r.raise_for_status()
try:
data = r.json()
if isinstance(data, dict):
for key in ("text", "generated", "content"):
if key in data:
return str(data[key])
if "choices" in data and isinstance(data["choices"], list) and data["choices"]:
c = data["choices"][0]
if "text" in c:
return c["text"]
return json.dumps(data)
except ValueError:
return r.text
@staticmethod
def _predict_openai(prompt, client, model):
resp = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Sei un consulente finanziario crypto."},
{"role": "user", "content": prompt}
],
max_tokens=300,
temperature=0.7
)
return resp.choices[0].message.content.strip()
@staticmethod
def _predict_anthropic(prompt, client, model):
response = client.completions.create(
model=model,
prompt=prompt,
max_tokens=300,
temperature=0.7
)
return response.completion.strip()
@staticmethod
def _predict_google(prompt, client, model):
response = client.models.generate_content(
model=model,
contents=prompt,
config={
"temperature": 0.7, # Controlla la creatività (0.0-1.0)
"max_output_tokens": 300, # Numero massimo di token nella risposta
"top_p": 0.9, # Nucleus sampling (opzionale)
"top_k": 40, # Top-k sampling (opzionale)
"candidate_count": 1, # Numero di risposte candidate (di solito 1)
"stop_sequences": [] # Sequenze che fermano la generazione (opzionale)
}
)
# Gestisce il caso in cui response.text sia None
result = getattr(response, 'text', None)
if result is None:
return "⚠️ Google API ha restituito una risposta vuota"
return result.strip()
@staticmethod
def _predict_deepseek(prompt, api_key, model):
url = "https://api.deepseek.ai/v1/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 300,
"temperature": 0.7
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"].strip()
@staticmethod
def _predict_mock(style):
if style.lower() in ("aggressive", "aggr"):
return (
"🚀 Strategia aggressiva (mock): "
"30% BTC, 20% ETH, 50% altcoins emergenti. "
"Motivo: alta volatilità + potenziale upside."
)
return (
"🛡️ Strategia conservativa (mock): "
"60% BTC, 30% ETH, 10% stablecoins. "
"Motivo: protezione da volatilità + focus su asset solidi."
)

View File

@@ -0,0 +1,29 @@
import os
from app.markets.base import BaseWrapper
from app.markets.coinbase import CoinBaseWrapper
from app.markets.cryptocompare import CryptoCompareWrapper
# TODO Dare la priorità in base alla qualità del servizio
# TODO Aggiungere altri wrapper se necessario
def get_first_available_market_api(currency:str = "USD") -> BaseWrapper:
"""
Restituisce il primo wrapper disponibile in base alle configurazioni del file .env e alle chiavi API presenti.
La priorità è data a Coinbase, poi a CryptoCompare.
Se non sono presenti chiavi API, restituisce una eccezione.
:param currency: Valuta di riferimento (default "USD")
:return: Lista di istanze di wrapper
"""
wrappers = []
api_key = os.getenv("COINBASE_API_KEY")
api_secret = os.getenv("COINBASE_API_SECRET")
if api_key and api_secret:
wrappers.append(CoinBaseWrapper(api_key=api_key, api_private_key=api_secret, currency=currency))
api_key = os.getenv("CRYPTOCOMPARE_API_KEY")
if api_key:
wrappers.append(CryptoCompareWrapper(api_key=api_key, currency=currency))
assert wrappers, "No valid API keys set in environment variables."
return wrappers[0]

67
src/app/markets/base.py Normal file
View File

@@ -0,0 +1,67 @@
from coinbase.rest.types.product_types import Candle, GetProductResponse
class BaseWrapper:
def get_product(self, asset_id: str) -> 'ProductInfo':
raise NotImplementedError
def get_products(self, asset_ids: list[str]) -> list['ProductInfo']:
raise NotImplementedError
def get_all_products(self) -> list['ProductInfo']:
raise NotImplementedError
def get_historical_prices(self, asset_id: str = "BTC") -> list['Price']:
raise NotImplementedError
class ProductInfo:
id: str
symbol: str
price: float
volume_24h: float
status: str
quote_currency: str
def from_coinbase(product_data: GetProductResponse) -> 'ProductInfo':
product = ProductInfo()
product.id = product_data.product_id
product.symbol = product_data.base_currency_id
product.price = float(product_data.price)
product.volume_24h = float(product_data.volume_24h) if product_data.volume_24h else 0
# TODO Check what status means in Coinbase
product.status = product_data.status
return product
def from_cryptocompare(asset_data: dict) -> 'ProductInfo':
product = ProductInfo()
product.id = asset_data['FROMSYMBOL'] + '-' + asset_data['TOSYMBOL']
product.symbol = asset_data['FROMSYMBOL']
product.price = float(asset_data['PRICE'])
product.volume_24h = float(asset_data['VOLUME24HOUR'])
product.status = "" # Cryptocompare does not provide status
return product
class Price:
high: float
low: float
open: float
close: float
volume: float
time: str
def from_coinbase(candle_data: Candle) -> 'Price':
price = Price()
price.high = float(candle_data.high)
price.low = float(candle_data.low)
price.open = float(candle_data.open)
price.close = float(candle_data.close)
price.volume = float(candle_data.volume)
price.time = str(candle_data.start)
return price
def from_cryptocompare(price_data: dict) -> 'Price':
price = Price()
price.high = float(price_data['high'])
price.low = float(price_data['low'])
price.open = float(price_data['open'])
price.close = float(price_data['close'])
price.volume = float(price_data['volumeto'])
price.time = str(price_data['time'])
return price

View File

@@ -5,13 +5,13 @@ class PublicBinanceAgent:
def __init__(self):
# Client pubblico (senza credenziali)
self.client = Client()
def get_public_prices(self):
"""Ottiene prezzi pubblici"""
try:
btc_price = self.client.get_symbol_ticker(symbol="BTCUSDT")
eth_price = self.client.get_symbol_ticker(symbol="ETHUSDT")
return {
'BTC_USD': float(btc_price['price']),
'ETH_USD': float(eth_price['price']),

View File

@@ -0,0 +1,35 @@
from coinbase.rest import RESTClient
from app.markets.base import ProductInfo, BaseWrapper, Price
class CoinBaseWrapper(BaseWrapper):
def __init__(self, api_key:str, api_private_key:str, currency: str = "USD"):
assert api_key is not None, "API key is required"
assert api_private_key is not None, "API private key is required"
self.currency = currency
self.client: RESTClient = RESTClient(
api_key=api_key,
api_secret=api_private_key
)
def __format(self, asset_id: str) -> str:
return asset_id if '-' in asset_id else f"{asset_id}-{self.currency}"
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)
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(all_asset_ids)
return [ProductInfo.from_coinbase(asset) for asset in assets.products]
def get_all_products(self) -> list[ProductInfo]:
assets = self.client.get_products()
return [ProductInfo.from_coinbase(asset) for asset in assets.products]
def get_historical_prices(self, asset_id: str = "BTC") -> list[Price]:
asset_id = self.__format(asset_id)
data = self.client.get_candles(product_id=asset_id)
return [Price.from_coinbase(candle) for candle in data.candles]

View File

@@ -0,0 +1,54 @@
import requests
from app.markets.base import ProductInfo, BaseWrapper, Price
BASE_URL = "https://min-api.cryptocompare.com"
class CryptoCompareWrapper(BaseWrapper):
def __init__(self, api_key:str, currency:str='USD'):
assert api_key is not None, "API key is required"
self.api_key = api_key
self.currency = currency
def __request(self, endpoint: str, params: dict = None) -> dict:
if params is None:
params = {}
params['api_key'] = self.api_key
response = requests.get(f"{BASE_URL}{endpoint}", params=params)
return response.json()
def get_product(self, asset_id: str) -> ProductInfo:
response = self.__request("/data/pricemultifull", params = {
"fsyms": asset_id,
"tsyms": self.currency
})
data = response.get('RAW', {}).get(asset_id, {}).get(self.currency, {})
return ProductInfo.from_cryptocompare(data)
def get_products(self, asset_ids: list[str]) -> list[ProductInfo]:
response = self.__request("/data/pricemultifull", params = {
"fsyms": ",".join(asset_ids),
"tsyms": self.currency
})
assets = []
data = response.get('RAW', {})
for asset_id in asset_ids:
asset_data = data.get(asset_id, {}).get(self.currency, {})
assets.append(ProductInfo.from_cryptocompare(asset_data))
return assets
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]:
assert day_back <= 30, "day_back should be less than or equal to 30"
response = self.__request("/data/v2/histohour", params = {
"fsym": asset_id,
"tsym": self.currency,
"limit": day_back * 24
})
data = response.get('Data', {}).get('Data', [])
prices = [Price.from_cryptocompare(price_data) for price_data in data]
return prices

View File

@@ -14,16 +14,14 @@ class Models(Enum):
"""
GEMINI = "gemini-2.0-flash" # API online
GEMINI_PRO = "gemini-2.0-pro" # API online, più costoso ma migliore
OLLAMA = "llama3.1" # little and fast (7b) but not so good
OLLAMA_GPT = "gpt-oss" # a bit big (13b) but very good (almost like gemini API)
OLLAMA_QWEN = "qwen3:8b" # good
MOCK = "mock"
OLLAMA = "llama3.1" # + fast (7b) - very very bad
OLLAMA_GPT = "gpt-oss" # + good - slow (13b) - doesn't follow instructions
OLLAMA_QWEN = "qwen3:8b" # + good + fast (8b), - doesn't follow instructions
def available() -> list[str]:
def availables() -> list['Models']:
"""
Controlla quali provider di modelli LLM hanno le loro API keys disponibili
come variabili d'ambiente e ritorna una lista di provider disponibili.
Se nessuna API key è disponibile, ritorna solo 'mock' come opzione.
L'ordine di preferenza è:
1. Gemini (Google)
2. Ollama (locale)
@@ -37,11 +35,8 @@ class Models(Enum):
availables.append(Models.OLLAMA_GPT)
availables.append(Models.OLLAMA_QWEN)
return [Models.MOCK, *availables]
def __str__(self) -> str:
# Per semplificare la visualizzazione del modello come stringa.
return self.name
assert availables, "No valid model API keys set in environment variables."
return availables
def get_model(self, instructions:str) -> BaseModel:
"""
@@ -51,13 +46,10 @@ class Models(Enum):
Raise ValueError se il modello non è supportato.
"""
name = self.value
if self in {Models.GEMINI}:
if self in {Models.GEMINI, Models.GEMINI_PRO}:
return Gemini(name, instructions=instructions)
elif self in {Models.OLLAMA, Models.OLLAMA_GPT, Models.OLLAMA_QWEN}:
return Ollama(name, instructions=instructions)
elif self in {Models.MOCK}:
from agno.models.base import Model
return Model(name, instructions=instructions)
raise ValueError(f"Modello non supportato: {self}")

View File

@@ -1,243 +0,0 @@
import os
import logging
from coinbase.rest import RESTClient
from typing import Dict, List, Any, Optional
logger = logging.getLogger(__name__)
class CoinbaseCDPSigner:
"""
Signer per Coinbase Advanced Trade API.
Utilizza le credenziali CDP per accedere alle vere API di market data di Coinbase.
"""
def __init__(self, api_key_name: str = None, api_private_key: str = None):
"""
Inizializza il Coinbase REST client.
Args:
api_key_name: Nome della API key (da CDP_API_KEY_NAME)
api_private_key: Private key (da CDP_API_PRIVATE_KEY)
"""
self.api_key_name = api_key_name or os.getenv('CDP_API_KEY_NAME')
self.api_private_key = api_private_key or os.getenv('CDP_API_PRIVATE_KEY')
if not self.api_key_name or not self.api_private_key:
raise ValueError("CDP_API_KEY_NAME and CDP_API_PRIVATE_KEY are required")
# Configura Coinbase REST client
try:
self.client = RESTClient(
api_key=self.api_key_name,
api_secret=self.api_private_key
)
self._configured = True
logger.info(f"✅ Coinbase REST Client configured with key: {self.api_key_name[:50]}...")
except Exception as e:
self._configured = False
logger.error(f"Failed to configure Coinbase REST Client: {e}")
raise ValueError(f"Failed to configure Coinbase REST Client: {e}")
def is_configured(self) -> bool:
"""Verifica se Coinbase REST client è configurato correttamente"""
return getattr(self, '_configured', False)
def get_asset_info(self, asset_id: str) -> Dict[str, Any]:
"""
Ottiene informazioni su un asset specifico usando Coinbase Advanced Trade API.
Args:
asset_id: ID dell'asset (es. "BTC", "ETH")
Returns:
Dict con informazioni sull'asset
"""
if not self.is_configured():
return {
'asset_id': asset_id,
'error': 'Coinbase REST Client not configured',
'success': False
}
try:
# Prova con USD prima, poi EUR se non funziona
product_id = f"{asset_id.upper()}-USD"
product_data = self.client.get_product(product_id)
return {
'asset_id': asset_id,
'symbol': product_data.product_id,
'price': float(product_data.price),
'volume_24h': float(product_data.volume_24h) if product_data.volume_24h else 0,
'status': product_data.status,
'base_currency': product_data.base_currency_id,
'quote_currency': product_data.quote_currency_id,
'success': True,
'source': 'coinbase_advanced_trade'
}
except Exception as e:
logger.error(f"Error getting asset info for {asset_id}: {e}")
return {
'asset_id': asset_id,
'error': str(e),
'success': False
}
def get_multiple_assets(self, asset_ids: List[str]) -> Dict[str, Any]:
"""
Ottiene informazioni su multipli asset.
Args:
asset_ids: Lista di ID degli asset
Returns:
Dict con informazioni sugli asset
"""
if not self.is_configured():
return {
'error': 'Coinbase REST Client not configured',
'success': False
}
results = {}
for asset_id in asset_ids:
asset_info = self.get_asset_info(asset_id)
if asset_info.get('success'):
results[asset_id] = asset_info
return results
def get_all_products(self) -> Dict[str, Any]:
"""
Ottiene lista di tutti i prodotti disponibili su Coinbase.
"""
if not self.is_configured():
return {
'error': 'Coinbase REST Client not configured',
'success': False
}
try:
products = self.client.get_products()
products_data = []
for product in products.products:
if product.status == "online": # Solo prodotti attivi
products_data.append({
'product_id': product.product_id,
'price': float(product.price) if product.price else 0,
'volume_24h': float(product.volume_24h) if product.volume_24h else 0,
'status': product.status,
'base_currency': product.base_currency_id,
'quote_currency': product.quote_currency_id
})
return {
'products': products_data,
'total': len(products_data),
'success': True
}
except Exception as e:
logger.error(f"Error getting products: {e}")
return {
'error': str(e),
'success': False
}
def get_market_trades(self, symbol: str = "BTC-USD", limit: int = 10) -> Dict[str, Any]:
"""
Ottiene gli ultimi trade di mercato per un prodotto.
Args:
symbol: Simbolo del prodotto (es. "BTC-USD")
limit: Numero massimo di trade da restituire
Returns:
Dict con i trade
"""
if not self.is_configured():
return {
'error': 'Coinbase REST Client not configured',
'success': False
}
try:
trades = self.client.get_market_trades(product_id=symbol, limit=limit)
trades_data = []
for trade in trades.trades:
trades_data.append({
'trade_id': trade.trade_id,
'price': float(trade.price),
'size': float(trade.size),
'time': trade.time,
'side': trade.side
})
return {
'symbol': symbol,
'trades': trades_data,
'count': len(trades_data),
'success': True
}
except Exception as e:
logger.error(f"Error getting market trades for {symbol}: {e}")
return {
'symbol': symbol,
'error': str(e),
'success': False
}
# Metodi di compatibilità con l'interfaccia precedente
def build_headers(self, method: str, request_path: str, body: Optional[Dict] = None) -> Dict[str, str]:
"""
Metodo di compatibilità - Coinbase REST client gestisce internamente l'autenticazione.
"""
return {
'Content-Type': 'application/json',
'User-Agent': 'upo-appAI/1.0-coinbase-advanced'
}
def sign_request(self, method: str, request_path: str, body: Optional[Dict] = None) -> Dict[str, Any]:
"""
Metodo di compatibilità - Coinbase REST client gestisce internamente l'autenticazione.
"""
return {
'method': method,
'path': request_path,
'body': body or {},
'headers': self.build_headers(method, request_path, body),
'coinbase_configured': self.is_configured()
}
def test_connection(self) -> Dict[str, Any]:
"""
Testa la connessione Coinbase con dati reali.
"""
try:
if not self.is_configured():
return {
'success': False,
'error': 'Coinbase REST Client not configured'
}
# Test con BTC-USD
test_asset = self.get_asset_info('BTC')
return {
'success': test_asset.get('success', False),
'client_configured': True,
'test_asset': test_asset.get('asset_id'),
'test_price': test_asset.get('price'),
'message': 'Coinbase Advanced Trade API is working with real data'
}
except Exception as e:
return {
'success': False,
'error': str(e),
'client_configured': False
}

View File

@@ -1,135 +0,0 @@
import time
from typing import Any, Dict, Optional
from urllib.parse import urlencode
class CryptoCompareSigner:
"""Genera le intestazioni e parametri di autenticazione per CryptoCompare API.
CryptoCompare utilizza un'autenticazione semplice basata su API key che può essere
passata come parametro nella query string o nell'header Authorization.
Contratto:
- Input: api_key, metodo di autenticazione (query o header)
- Output: dict di header e parametri per la richiesta
- Errori: solleva ValueError se api_key non è fornita
"""
def __init__(self, api_key: str, auth_method: str = "query") -> None:
"""
Inizializza il signer per CryptoCompare.
Args:
api_key: La chiave API di CryptoCompare
auth_method: Metodo di autenticazione ("query" o "header")
- "query": aggiunge api_key come parametro URL
- "header": aggiunge api_key nell'header Authorization
"""
if not api_key:
raise ValueError("API key è richiesta per CryptoCompare")
self.api_key = api_key
self.auth_method = auth_method.lower()
if self.auth_method not in ("query", "header"):
raise ValueError("auth_method deve essere 'query' o 'header'")
def build_headers(self, include_timestamp: bool = False) -> Dict[str, str]:
"""
Costruisce gli header per la richiesta CryptoCompare.
Args:
include_timestamp: Se includere un timestamp nell'header (opzionale)
Returns:
Dict con gli header necessari
"""
headers = {
"Content-Type": "application/json",
"User-Agent": "upo-appAI/1.0"
}
# Se si usa autenticazione via header
if self.auth_method == "header":
headers["Authorization"] = f"Apikey {self.api_key}"
# Aggiungi timestamp se richiesto (utile per debugging)
if include_timestamp:
headers["X-Request-Timestamp"] = str(int(time.time()))
return headers
def build_url_params(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Costruisce i parametri URL includendo l'API key se necessario.
Args:
params: Parametri aggiuntivi per la query
Returns:
Dict con tutti i parametri per l'URL
"""
if params is None:
params = {}
# Se si usa autenticazione via query string
if self.auth_method == "query":
params["api_key"] = self.api_key
return params
def build_full_url(self, base_url: str, endpoint: str, params: Optional[Dict[str, Any]] = None) -> str:
"""
Costruisce l'URL completo con tutti i parametri.
Args:
base_url: URL base dell'API (es. "https://min-api.cryptocompare.com")
endpoint: Endpoint specifico (es. "/data/pricemulti")
params: Parametri aggiuntivi per la query
Returns:
URL completo con parametri
"""
base_url = base_url.rstrip("/")
endpoint = endpoint if endpoint.startswith("/") else f"/{endpoint}"
url_params = self.build_url_params(params)
if url_params:
query_string = urlencode(url_params)
return f"{base_url}{endpoint}?{query_string}"
else:
return f"{base_url}{endpoint}"
def prepare_request(self,
base_url: str,
endpoint: str,
params: Optional[Dict[str, Any]] = None,
include_timestamp: bool = False) -> Dict[str, Any]:
"""
Prepara tutti i componenti per una richiesta CryptoCompare.
Args:
base_url: URL base dell'API
endpoint: Endpoint specifico
params: Parametri per la query
include_timestamp: Se includere timestamp negli header
Returns:
Dict con url, headers e params pronti per la richiesta
"""
return {
"url": self.build_full_url(base_url, endpoint, params),
"headers": self.build_headers(include_timestamp),
"params": self.build_url_params(params) if self.auth_method == "query" else params or {}
}
# Alias per compatibilità con il pattern Coinbase
def sign_request(self,
endpoint: str,
params: Optional[Dict[str, Any]] = None,
base_url: str = "https://min-api.cryptocompare.com") -> Dict[str, Any]:
"""
Alias per prepare_request per mantenere compatibilità con il pattern del CoinbaseSigner.
"""
return self.prepare_request(base_url, endpoint, params)

View File

@@ -1,27 +1,36 @@
from app.agents.market_agent import MarketAgent
from app.agents.news_agent import NewsAgent
from app.agents.social_agent import SocialAgent
from app.agents.predictor_agent import PredictorAgent
from app.agents import predictor
from app.agents.predictor import PredictorStyle
from app.markets import get_first_available_market_api
from app.models import Models
class ToolAgent:
def __init__(self, available_models: list[Models]):
self.market_agent = MarketAgent()
self.news_agent = NewsAgent()
self.social_agent = SocialAgent()
self.predictor_agent = PredictorAgent()
def __init__(self, available_models: list[Models], all_styles: list[PredictorStyle]):
self.available_models = available_models
self.all_styles = all_styles
def interact(self, query: str, provider: str, style: str):
"""
Funzione principale che coordina gli agenti per rispondere alla richiesta dell'utente.
"""
self.market = get_first_available_market_api(currency="USD")
self.choose_provider(0) # Default to the first model
def choose_provider(self, index: int):
# TODO Utilizzare AGNO per gestire i modelli... è molto più semplice e permette di cambiare modello facilmente
# TODO https://docs.agno.com/introduction
# Inoltre permette di creare dei team e workflow di agenti più facilmente
chosen_model = self.available_models[index]
self.predictor = chosen_model.get_agent(predictor.instructions())
self.news_agent = NewsAgent()
self.social_agent = SocialAgent()
def interact(self, query: str, style_index: int):
"""
Funzione principale che coordina gli agenti per rispondere alla richiesta dell'utente.
"""
# Step 1: raccolta analisi
market_data = self.market_agent.analyze(query)
cryptos = ["BTC", "ETH", "XRP", "LTC", "BCH"] # TODO rendere dinamico in futuro
market_data = self.market.get_products(cryptos)
news_sentiment = self.news_agent.analyze(query)
social_sentiment = self.social_agent.analyze(query)
@@ -29,11 +38,15 @@ class ToolAgent:
sentiment = f"{news_sentiment}\n{social_sentiment}"
# Step 3: previsione
prediction = self.predictor_agent.predict(
inputs = predictor.prepare_inputs(
data=market_data,
sentiment=sentiment,
style=style,
provider=provider
style=self.all_styles[style_index],
sentiment=sentiment
)
return f"{market_data}\n{sentiment}\n\n📈 Consiglio finale:\n{prediction}"
prediction = self.predictor.run(inputs)
#output = prediction.content.split("</think>")[-1] # remove thinking steps and reasoning from the final output
output = prediction.content
market_data = "\n".join([f"{product.symbol}: {product.price}" for product in market_data])
return f"{market_data}\n{sentiment}\n\n📈 Consiglio finale:\n{output}"

1442
uv.lock generated

File diff suppressed because it is too large Load Diff