Integrato 3 fonti ma da finire binance. Fatto una struttura con i signers per i servizi.
This commit is contained in:
Simone Garau
@@ -1,4 +1,10 @@
|
||||
import gradio as gr
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Aggiungi src al Python path per gli import
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from app.tool import ToolAgent
|
||||
|
||||
tool_agent = ToolAgent()
|
||||
|
||||
@@ -1,5 +1,268 @@
|
||||
from typing import Dict, List, Optional, Any
|
||||
import requests
|
||||
import logging
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
from app.signers.market_signers.coinbase_rest_signer import CoinbaseCDPSigner
|
||||
from app.signers.market_signers.cryptocompare_signer import CryptoCompareSigner
|
||||
|
||||
load_dotenv()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MarketAgent:
|
||||
@staticmethod
|
||||
def analyze(query: str) -> str:
|
||||
# Mock analisi mercato
|
||||
return "📊 Analisi di mercato: BTC stabile, ETH in leggera crescita, altcoin volatili."
|
||||
"""
|
||||
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}"
|
||||
|
||||
0
src/app/signers/__init__.py
Normal file
0
src/app/signers/__init__.py
Normal file
27
src/app/signers/market_signers/binance_signer.py
Normal file
27
src/app/signers/market_signers/binance_signer.py
Normal file
@@ -0,0 +1,27 @@
|
||||
# Versione pubblica senza autenticazione
|
||||
from binance.client import Client
|
||||
|
||||
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']),
|
||||
'source': 'binance_public'
|
||||
}
|
||||
except Exception as e:
|
||||
print(f"Errore: {e}")
|
||||
return None
|
||||
|
||||
# Uso senza credenziali
|
||||
public_agent = PublicBinanceAgent()
|
||||
public_prices = public_agent.get_public_prices()
|
||||
print(public_prices)
|
||||
186
src/app/signers/market_signers/coinbase_cdp_signer.py
Normal file
186
src/app/signers/market_signers/coinbase_cdp_signer.py
Normal file
@@ -0,0 +1,186 @@
|
||||
import os
|
||||
import logging
|
||||
from cdp import *
|
||||
from typing import Dict, List, Any, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CoinbaseCDPSigner:
|
||||
"""
|
||||
Signer per Coinbase Developer Platform (CDP) SDK.
|
||||
Utilizza il nuovo sistema di autenticazione di Coinbase basato su CDP.
|
||||
"""
|
||||
|
||||
def __init__(self, api_key_name: str = None, api_private_key: str = None):
|
||||
"""
|
||||
Inizializza il CDP signer.
|
||||
|
||||
Args:
|
||||
api_key_name: Nome della API key (formato: organizations/org-id/apiKeys/key-id)
|
||||
api_private_key: Private key in formato PEM
|
||||
"""
|
||||
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 CDP client
|
||||
try:
|
||||
self.client = CdpClient(
|
||||
api_key_id=self.api_key_name,
|
||||
api_key_secret=self.api_private_key,
|
||||
debugging=False
|
||||
)
|
||||
self._configured = True
|
||||
logger.info(f"✅ CDP Client configured with key: {self.api_key_name[:50]}...")
|
||||
except Exception as e:
|
||||
self._configured = False
|
||||
logger.error(f"Failed to configure CDP Client: {e}")
|
||||
raise ValueError(f"Failed to configure CDP SDK: {e}")
|
||||
|
||||
def is_configured(self) -> bool:
|
||||
"""Verifica se CDP è configurato correttamente"""
|
||||
return getattr(self, '_configured', False)
|
||||
|
||||
def get_asset_info(self, asset_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Ottiene informazioni su un asset specifico.
|
||||
|
||||
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': 'CDP Client not configured',
|
||||
'success': False
|
||||
}
|
||||
|
||||
try:
|
||||
# Per ora, restituiamo un mock data structure
|
||||
# In futuro, quando CDP avrà metodi per asset info, useremo quelli
|
||||
return {
|
||||
'asset_id': asset_id,
|
||||
'price': self._get_mock_price(asset_id),
|
||||
'symbol': asset_id,
|
||||
'name': self._get_asset_name(asset_id),
|
||||
'success': True,
|
||||
'source': 'cdp_mock'
|
||||
}
|
||||
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': 'CDP 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_mock_price(self, asset_id: str) -> float:
|
||||
"""
|
||||
Mock prices per i test - da sostituire con vere API CDP quando disponibili
|
||||
"""
|
||||
mock_prices = {
|
||||
'BTC': 63500.0,
|
||||
'ETH': 2650.0,
|
||||
'ADA': 0.45,
|
||||
'DOT': 5.2,
|
||||
'SOL': 145.0,
|
||||
'MATIC': 0.85,
|
||||
'LINK': 11.2,
|
||||
'UNI': 7.8
|
||||
}
|
||||
return mock_prices.get(asset_id.upper(), 100.0)
|
||||
|
||||
def _get_asset_name(self, asset_id: str) -> str:
|
||||
"""
|
||||
Mock asset names
|
||||
"""
|
||||
names = {
|
||||
'BTC': 'Bitcoin',
|
||||
'ETH': 'Ethereum',
|
||||
'ADA': 'Cardano',
|
||||
'DOT': 'Polkadot',
|
||||
'SOL': 'Solana',
|
||||
'MATIC': 'Polygon',
|
||||
'LINK': 'Chainlink',
|
||||
'UNI': 'Uniswap'
|
||||
}
|
||||
return names.get(asset_id.upper(), f"{asset_id} Token")
|
||||
|
||||
# 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à - CDP SDK gestisce internamente l'autenticazione.
|
||||
Restituisce headers basic.
|
||||
"""
|
||||
return {
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'upo-appAI/1.0-cdp'
|
||||
}
|
||||
|
||||
def sign_request(self, method: str, request_path: str, body: Optional[Dict] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Metodo di compatibilità - CDP SDK gestisce internamente l'autenticazione.
|
||||
"""
|
||||
return {
|
||||
'method': method,
|
||||
'path': request_path,
|
||||
'body': body or {},
|
||||
'headers': self.build_headers(method, request_path, body),
|
||||
'cdp_configured': self.is_configured()
|
||||
}
|
||||
|
||||
def test_connection(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Testa la connessione CDP
|
||||
"""
|
||||
try:
|
||||
if not self.is_configured():
|
||||
return {
|
||||
'success': False,
|
||||
'error': 'CDP Client not configured'
|
||||
}
|
||||
|
||||
# Test basic con mock data
|
||||
test_asset = self.get_asset_info('BTC')
|
||||
return {
|
||||
'success': test_asset.get('success', False),
|
||||
'client_configured': True,
|
||||
'test_asset': test_asset.get('asset_id'),
|
||||
'message': 'CDP Client is working with mock data'
|
||||
}
|
||||
except Exception as e:
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e),
|
||||
'client_configured': False
|
||||
}
|
||||
243
src/app/signers/market_signers/coinbase_rest_signer.py
Normal file
243
src/app/signers/market_signers/coinbase_rest_signer.py
Normal file
@@ -0,0 +1,243 @@
|
||||
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
|
||||
}
|
||||
159
src/app/signers/market_signers/coinbase_signer.py
Normal file
159
src/app/signers/market_signers/coinbase_signer.py
Normal file
@@ -0,0 +1,159 @@
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import time
|
||||
from typing import Any, Mapping, Optional
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
|
||||
class CoinbaseSigner:
|
||||
"""
|
||||
Genera le intestazioni di autenticazione per Coinbase Advanced Trade API.
|
||||
|
||||
Supporta due formati di autenticazione:
|
||||
1. Legacy: API key, secret (base64), passphrase (per retrocompatibilità)
|
||||
2. New: API key name, private key (nuovo formato Coinbase)
|
||||
|
||||
Contratto:
|
||||
- Input: method, request_path, body opzionale, timestamp opzionale
|
||||
- Output: dict di header richiesti dall'API
|
||||
- Errori: solleva eccezioni se le credenziali non sono valide
|
||||
"""
|
||||
|
||||
def __init__(self, api_key: str, secret_or_private_key: str, passphrase: str = None) -> None:
|
||||
self.api_key = api_key
|
||||
self.passphrase = passphrase
|
||||
|
||||
# Determina se stiamo usando il nuovo formato o il legacy
|
||||
if passphrase is None:
|
||||
# Nuovo formato: solo API key + private key
|
||||
self.auth_method = "new"
|
||||
self.private_key = self._load_private_key(secret_or_private_key)
|
||||
self.secret_b64 = None
|
||||
else:
|
||||
# Formato legacy: API key + secret + passphrase
|
||||
self.auth_method = "legacy"
|
||||
self.secret_b64 = secret_or_private_key
|
||||
self.private_key = None
|
||||
|
||||
def _load_private_key(self, private_key_str: str):
|
||||
"""Carica la private key dal formato PEM"""
|
||||
try:
|
||||
# Rimuovi eventuali spazi e aggiungi header/footer se mancanti
|
||||
key_str = private_key_str.strip()
|
||||
if not key_str.startswith("-----BEGIN"):
|
||||
key_str = f"-----BEGIN EC PRIVATE KEY-----\n{key_str}\n-----END EC PRIVATE KEY-----"
|
||||
|
||||
private_key = serialization.load_pem_private_key(
|
||||
key_str.encode('utf-8'),
|
||||
password=None,
|
||||
)
|
||||
return private_key
|
||||
except Exception as e:
|
||||
raise ValueError(f"Invalid private key format: {e}")
|
||||
|
||||
@staticmethod
|
||||
def _normalize_path(path: str) -> str:
|
||||
if not path:
|
||||
return "/"
|
||||
return path if path.startswith("/") else f"/{path}"
|
||||
|
||||
def _build_legacy_headers(
|
||||
self,
|
||||
method: str,
|
||||
request_path: str,
|
||||
body: Optional[Mapping[str, Any]] = None,
|
||||
timestamp: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""Costruisce header usando il formato legacy (HMAC)"""
|
||||
# Timestamp in secondi come stringa
|
||||
ts = timestamp or str(int(time.time()))
|
||||
m = method.upper()
|
||||
req_path = self._normalize_path(request_path)
|
||||
|
||||
# Il body deve essere stringa vuota per GET/DELETE o quando assente
|
||||
if body is None or m in ("GET", "DELETE"):
|
||||
body_str = ""
|
||||
else:
|
||||
# JSON deterministico, senza spazi
|
||||
body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
|
||||
|
||||
# Concatenazione: timestamp + method + request_path + body
|
||||
message = f"{ts}{m}{req_path}{body_str}"
|
||||
|
||||
# Decodifica secret (base64) e firma HMAC-SHA256
|
||||
key = base64.b64decode(self.secret_b64)
|
||||
signature = hmac.new(key, message.encode("utf-8"), hashlib.sha256).digest()
|
||||
cb_access_sign = base64.b64encode(signature).decode("utf-8")
|
||||
|
||||
return {
|
||||
"CB-ACCESS-KEY": self.api_key,
|
||||
"CB-ACCESS-SIGN": cb_access_sign,
|
||||
"CB-ACCESS-TIMESTAMP": ts,
|
||||
"CB-ACCESS-PASSPHRASE": self.passphrase,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
def _build_new_headers(
|
||||
self,
|
||||
method: str,
|
||||
request_path: str,
|
||||
body: Optional[Mapping[str, Any]] = None,
|
||||
timestamp: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""Costruisce header usando il nuovo formato (EC signature)"""
|
||||
# Timestamp in secondi come stringa
|
||||
ts = timestamp or str(int(time.time()))
|
||||
m = method.upper()
|
||||
req_path = self._normalize_path(request_path)
|
||||
|
||||
# Il body deve essere stringa vuota per GET/DELETE o quando assente
|
||||
if body is None or m in ("GET", "DELETE"):
|
||||
body_str = ""
|
||||
else:
|
||||
# JSON deterministico, senza spazi
|
||||
body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
|
||||
|
||||
# Concatenazione: timestamp + method + request_path + body
|
||||
message = f"{ts}{m}{req_path}{body_str}"
|
||||
|
||||
# Firma con ECDSA
|
||||
signature = self.private_key.sign(
|
||||
message.encode("utf-8"),
|
||||
ec.ECDSA(hashes.SHA256())
|
||||
)
|
||||
|
||||
# Converti signature in base64
|
||||
cb_access_sign = base64.b64encode(signature).decode("utf-8")
|
||||
|
||||
return {
|
||||
"CB-ACCESS-KEY": self.api_key,
|
||||
"CB-ACCESS-SIGN": cb_access_sign,
|
||||
"CB-ACCESS-TIMESTAMP": ts,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
def build_headers(
|
||||
self,
|
||||
method: str,
|
||||
request_path: str,
|
||||
body: Optional[Mapping[str, Any]] = None,
|
||||
timestamp: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""Costruisce gli header di autenticazione usando il metodo appropriato"""
|
||||
if self.auth_method == "legacy":
|
||||
return self._build_legacy_headers(method, request_path, body, timestamp)
|
||||
else:
|
||||
return self._build_new_headers(method, request_path, body, timestamp)
|
||||
|
||||
def sign_request(
|
||||
self,
|
||||
method: str,
|
||||
request_path: str,
|
||||
body: Optional[Mapping[str, Any]] = None,
|
||||
passphrase: Optional[str] = None,
|
||||
) -> dict:
|
||||
return self.build_headers(method, request_path, body)
|
||||
135
src/app/signers/market_signers/cryptocompare_signer.py
Normal file
135
src/app/signers/market_signers/cryptocompare_signer.py
Normal file
@@ -0,0 +1,135 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user