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

pulizia veloce

Browse Source
This commit is contained in:
Simone Garau
2025-09-25 18:24:39 +02:00
parent 995048831c
commit 5dbdb050ca
10 changed files with 1690 additions and 573 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -172,3 +172,6 @@ cython_debug/
# PyPI configuration file
.pypirc
# chroma db
./chroma_db/

255
docs/App_Architecture_Diagrams.md Normal file
View File

@@ -0,0 +1,255 @@
# 📊 Architettura e Flussi dell'App upo-appAI
## 🏗️ Diagramma Architettura Generale
```
┌─────────────────────────────────────────────────────────────────┐
│ 🌐 GRADIO UI │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ User Input │ │ Provider │ │ Style │ │
│ │ (Query) │ │ (Model) │ │ (Conservative/ │ │
│ │ │ │ │ │ Aggressive) │ │
│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │
└─────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 🔧 TOOL AGENT │
│ (Central Orchestrator) │
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 1. Collect Data │ │ 2. Analyze │ │ 3. Predict & │ │
│ │ │ │ Sentiment │ │ Recommend │ │
│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │
└─────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 📊 AGENT ECOSYSTEM │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐│
│ │ MARKET │ │ NEWS │ │ SOCIAL │ │ PREDICTOR ││
│ │ AGENT │ │ AGENT │ │ AGENT │ │ AGENT ││
│ │ │ │ │ │ │ │ ││
│ │ 📈 Coinbase │ │ 📰 News API │ │ 🐦 Social │ │ 🤖 LLM ││
│ │ 📊 CryptoCmp│ │ │ │ Media │ │ Analysis ││
│ │ 🟡 Binance │ │ │ │ │ │ ││
│ └─────────────┘ └─────────────┘ └─────────────┘ └───────────┘│
└─────────────────────────────────────────────────────────────────┘
```
## 🔄 Flusso di Esecuzione Dettagliato
```
👤 USER REQUEST
│
│ "Analizza Bitcoin con strategia aggressiva"
▼
┌─────────────────────────────────────────────────────────────┐
│ 🔧 TOOL AGENT │
│ │
│ def interact(query, provider, style): │
│ │ │
│ ├── 📊 market_data = market_agent.analyze(query) │
│ ├── 📰 news_sentiment = news_agent.analyze(query) │
│ ├── 🐦 social_sentiment = social_agent.analyze(query) │
│ │ │
│ └── 🤖 prediction = predictor_agent.predict(...) │
└─────────────────────────────────────────────────────────────┘
│
▼
📊 MARKET AGENT - Parallel Data Collection
┌─────────────────────────────────────────────────────────────┐
│ │
│ 🔍 Auto-detect Available Providers: │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Coinbase │ │ CryptoComp │ │ Binance │ │
│ │ REST │ │ API │ │ Mock │ │
│ │ │ │ │ │ │ │
│ │ ✅ Active │ │ ✅ Active │ │ ✅ Active │ │
│ │ $63,500 BTC │ │ $63,450 BTC │ │ $63,600 BTC │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ 📈 Aggregated Result: │
│ { │
│ "aggregated_data": { │
│ "BTC_USD": { │
│ "price": 63516.67, │
│ "confidence": 0.94, │
│ "sources_count": 3 │
│ } │
│ }, │
│ "individual_sources": {...}, │
│ "market_signals": {...} │
│ } │
└─────────────────────────────────────────────────────────────┘
│
▼
📰 NEWS AGENT + 🐦 SOCIAL AGENT
┌─────────────────────────────────────────────────────────────┐
│ │
│ 📰 News Sentiment: "Positive momentum, institutional │
│ adoption increasing..." │
│ │
│ 🐦 Social Sentiment: "Bullish sentiment on Reddit, │
│ Twitter mentions up 15%..." │
└─────────────────────────────────────────────────────────────┘
│
▼
🤖 PREDICTOR AGENT
┌─────────────────────────────────────────────────────────────┐
│ │
│ Input: │
│ ├── 📊 Market Data (aggregated + confidence) │
│ ├── 📰🐦 Combined Sentiment │
│ ├── 🎯 Style: "aggressive" │
│ └── 🤖 Provider: "openai/anthropic/google..." │
│ │
│ 🧠 LLM Processing: │
│ "Based on high confidence market data (0.94) showing │
│ $63,516 BTC with positive sentiment across news and │
│ social channels, aggressive strategy recommendation..." │
└─────────────────────────────────────────────────────────────┘
│
▼
📋 FINAL OUTPUT
┌─────────────────────────────────────────────────────────────┐
│ 📊 Market Data Summary │
│ 📰🐦 Sentiment Analysis │
│ 📈 Final Recommendation: │
│ "Strong BUY signal with 85% confidence..." │
└─────────────────────────────────────────────────────────────┘
```
## 🏛️ Architettura dei Provider (Market Agent)
```
┌─────────────────────────────────────────────────────────────────┐
│ 📊 MARKET AGENT │
│ │
│ 🔍 Provider Detection Logic: │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ def _setup_providers(): ││
│ │ ├── 🔑 Check CDP_API_KEY_NAME + CDP_API_PRIVATE_KEY ││
│ │ │ └── ✅ Setup Coinbase Advanced Trade ││
│ │ ├── 🔑 Check CRYPTOCOMPARE_API_KEY ││
│ │ │ └── ✅ Setup CryptoCompare ││
│ │ └── 🔑 Check BINANCE_API_KEY (future) ││
│ │ └── ✅ Setup Binance API ││
│ └─────────────────────────────────────────────────────────────┘│
│ │
│ 📡 Data Flow: │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Provider 1 │───▶│ │◀───│ Provider 2 │ │
│ │ Coinbase │ │ AGGREGATOR │ │ CryptoComp │ │
│ │ │ │ │ │ │ │
│ │ Real-time │ │ ┌─────────┐ │ │ Real-time │ │
│ │ Market Data │ │ │Confidence│ │ │ Market Data │ │
│ └─────────────┘ │ │Scoring │ │ └─────────────┘ │
│ │ │ │ │ │
│ ┌─────────────┐ │ │ Spread │ │ ┌─────────────┐ │
│ │ Provider 3 │───▶│ │Analysis │ │◀───│ Provider N │ │
│ │ Binance │ │ │ │ │ │ Future │ │
│ │ │ │ └─────────┘ │ │ │ │
│ │ Mock Data │ │ │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
## 🔧 Signers Architecture
```
┌─────────────────────────────────────────────────────────────────┐
│ 🔐 SIGNERS ECOSYSTEM │
│ │
│ 📁 src/app/signers/market_signers/ │
│ │ │
│ ├── 🏦 coinbase_rest_signer.py │
│ │ ├── 🔑 Uses: CDP_API_KEY_NAME + CDP_API_PRIVATE_KEY │
│ │ ├── 📡 RESTClient from coinbase.rest │
│ │ ├── 📊 get_asset_info() → Real Coinbase data │
│ │ └── 📈 get_multiple_assets() → Bulk data │
│ │ │
│ ├── 📊 cryptocompare_signer.py │
│ │ ├── 🔑 Uses: CRYPTOCOMPARE_API_KEY │
│ │ ├── 📡 Direct HTTP requests │
│ │ ├── 💰 get_crypto_prices() → Multi-currency │
│ │ └── 🏆 get_top_cryptocurrencies() → Market cap │
│ │ │
│ └── 🟡 binance_signer.py │
│ ├── 🔑 Uses: BINANCE_API_KEY (future) │
│ ├── 📡 Mock implementation │
│ ├── 🎭 Simulated market data │
│ └── 📈 Compatible interface │
└─────────────────────────────────────────────────────────────────┘
```
## 🚀 Future Enhancement: Async Flow
```
📱 USER REQUEST
│
▼
🔧 TOOL AGENT (async)
│
┌────────────────┼────────────────┐
│ │ │
▼ ▼ ▼
📊 Market 📰 News 🐦 Social
Agent (async) Agent (async) Agent (async)
│ │ │
┌────┼────┐ │ │
▼ ▼ ▼ │ │
Coinbase │ Binance │ │
CC │ │ │
▼▼▼ ▼ ▼
🔄 Parallel 📰 Sentiment 🐦 Sentiment
Aggregation Analysis Analysis
│ │ │
└────────────────┼────────────────┘
▼
🤖 PREDICTOR AGENT
(LLM Analysis)
│
▼
📋 FINAL RESULT
(JSON + Confidence)
```
## 📊 Data Flow Example
```
Input: "Analyze Bitcoin aggressive strategy"
│
├── 📊 Market Agent Output:
│ {
│ "aggregated_data": {
│ "BTC_USD": {"price": 63516.67, "confidence": 0.94}
│ },
│ "individual_sources": {
│ "coinbase": {"price": 63500, "volume": "1.2M"},
│ "cryptocompare": {"price": 63450, "volume": "N/A"},
│ "binance": {"price": 63600, "volume": "2.1M"}
│ },
│ "market_signals": {
│ "spread_analysis": "Low spread (0.24%) - healthy liquidity",
│ "price_divergence": "Max deviation: 0.24% - Normal range"
│ }
│ }
│
├── 📰 News Sentiment: "Positive institutional adoption news..."
├── 🐦 Social Sentiment: "Bullish Reddit sentiment, +15% mentions"
│
└── 🤖 Predictor Output:
"📈 Strong BUY recommendation based on:
- High confidence market data (94%)
- Positive news sentiment
- Bullish social indicators
- Low spread indicates healthy liquidity
Aggressive Strategy: Consider 15-20% portfolio allocation"
```
---
*Diagrammi creati: 2025-09-23*
*Sistema: upo-appAI Market Analysis Platform*

203
docs/Async_Implementation_Detail.md Normal file
View File

@@ -0,0 +1,203 @@
# 🚀 Diagramma Dettaglio: Implementazione Asincrona
## ⚡ Async Market Data Collection (Fase 3)
```
┌─────────────────────────────────────────────────────────────────┐
│ 🔧 TOOL AGENT │
│ │
│ async def interact(query, provider, style): │
│ │ │
│ ├── 📊 market_data = await market_agent.analyze_async() │
│ ├── 📰 news_data = await news_agent.analyze_async() │
│ ├── 🐦 social_data = await social_agent.analyze_async() │
│ │ │
│ └── 🤖 prediction = await predictor.predict_async(...) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 📊 MARKET AGENT - ASYNC IMPLEMENTATION │
│ │
│ async def analyze_async(self, query): │
│ symbols = extract_symbols(query) # ["BTC", "ETH"] │
│ │ │
│ └── 🔄 tasks = [ │
│ │ self._query_coinbase_async(symbols), │
│ │ self._query_cryptocompare_async(symbols), │
│ │ self._query_binance_async(symbols) │
│ │ ] │
│ │ │
│ └── 📊 results = await asyncio.gather(*tasks) │
│ │ │
│ ▼ │
│ 🧮 aggregate_results(results) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ⏱️ TIMING DIAGRAM │
│ │
│ Time: 0ms 500ms 1000ms 1500ms 2000ms │
│ │ │ │ │ │ │
│ 📡 Start all requests │
│ ├─────────────────────────────────────────┐ │
│ │ 🏦 Coinbase Request │ │
│ │ ✅ Response │ (1.2s) │
│ ├─────────────────────────────┐ │ │
│ │ 📊 CryptoCompare Request │ │ │
│ │ ✅ Response (0.8s) │ │
│ ├─────────────┐ │ │ │
│ │ 🟡 Binance │ │ │ │
│ │ ✅ Response (0.3s - mock) │ │ │
│ │ │ │ │ │
│ └─────────────┼───────────────┼───────────┘ │
│ │ │ │
│ Wait for all... │ │
│ │ │
│ 🧮 Aggregate (1.2s total) │
│ │
│ 📈 Performance Gain: │
│ Sequential: 1.2s + 0.8s + 0.3s = 2.3s │
│ Parallel: max(1.2s, 0.8s, 0.3s) = 1.2s │
│ Improvement: ~48% faster! 🚀 │
└─────────────────────────────────────────────────────────────────┘
```
## 🧮 Aggregation Algorithm Detail
```
┌─────────────────────────────────────────────────────────────────┐
│ 🔬 DATA AGGREGATION LOGIC │
│ │
│ def aggregate_market_data(results): │
│ │ │
│ ├── 📊 Input Data: │
│ │ ┌─────────────────────────────────────────────────┐ │
│ │ │ coinbase: {"BTC": 63500, "ETH": 4150} │ │
│ │ │ cryptocomp: {"BTC": 63450, "ETH": 4160} │ │
│ │ │ binance: {"BTC": 63600, "ETH": 4140} │ │
│ │ └─────────────────────────────────────────────────┘ │
│ │ │
│ ├── 🧮 Price Calculation: │
│ │ ┌─────────────────────────────────────────────────┐ │
│ │ │ BTC_prices = [63500, 63450, 63600] │ │
│ │ │ BTC_avg = 63516.67 │ │
│ │ │ BTC_std = 75.83 │ │
│ │ │ BTC_spread = (max-min)/avg = 0.24% │ │
│ │ └─────────────────────────────────────────────────┘ │
│ │ │
│ ├── 🎯 Confidence Scoring: │
│ │ ┌─────────────────────────────────────────────────┐ │
│ │ │ confidence = 1 - (std_dev / mean) │ │
│ │ │ if spread < 0.5%: confidence += 0.1 │ │
│ │ │ if sources >= 3: confidence += 0.05 │ │
│ │ │ BTC_confidence = 0.94 (excellent!) │ │
│ │ └─────────────────────────────────────────────────┘ │
│ │ │
│ └── 📈 Market Signals: │
│ ┌─────────────────────────────────────────────────┐ │
│ │ spread_analysis: │ │
│ │ "Low spread (0.24%) indicates healthy liq." │ │
│ │ volume_trend: │ │
│ │ "Combined volume: 4.1M USD" │ │
│ │ price_divergence: │ │
│ │ "Max deviation: 0.24% - Normal range" │ │
│ └─────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
## 🔄 Error Handling & Resilience
```
┌─────────────────────────────────────────────────────────────────┐
│ 🛡️ RESILIENCE STRATEGY │
│ │
│ Scenario 1: One Provider Fails │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 🏦 Coinbase: ✅ Success (BTC: $63500) │ │
│ │ 📊 CryptoComp: ❌ Timeout/Error │ │
│ │ 🟡 Binance: ✅ Success (BTC: $63600) │ │
│ │ │ │
│ │ Result: Continue with 2 sources │ │
│ │ Confidence: 0.89 (slightly reduced) │ │
│ │ Note: "CryptoCompare unavailable" │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ Scenario 2: Multiple Providers Fail │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 🏦 Coinbase: ❌ API Limit │ │
│ │ 📊 CryptoComp: ✅ Success (BTC: $63450) │ │
│ │ 🟡 Binance: ❌ Network Error │ │
│ │ │ │
│ │ Result: Single source data │ │
│ │ Confidence: 0.60 (low - warn user) │ │
│ │ Note: "Limited data - consider waiting" │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ Scenario 3: All Providers Fail │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 🏦 Coinbase: ❌ Maintenance │ │
│ │ 📊 CryptoComp: ❌ API Down │ │
│ │ 🟡 Binance: ❌ Rate Limit │ │
│ │ │ │
│ │ Result: Graceful degradation │ │
│ │ Message: "Market data temporarily unavailable" │ │
│ │ Fallback: Cached data (if available) │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
## 📊 JSON Output Schema
```json
{
"aggregated_data": {
"BTC_USD": {
"price": 63516.67,
"confidence": 0.94,
"sources_count": 3,
"last_updated": "2025-09-23T17:30:00Z"
},
"ETH_USD": {
"price": 4150.33,
"confidence": 0.91,
"sources_count": 3,
"last_updated": "2025-09-23T17:30:00Z"
}
},
"individual_sources": {
"coinbase": {
"BTC": {"price": 63500, "volume": "1.2M", "status": "online"},
"ETH": {"price": 4150, "volume": "25.6M", "status": "online"}
},
"cryptocompare": {
"BTC": {"price": 63450, "volume": "N/A", "status": "active"},
"ETH": {"price": 4160, "volume": "N/A", "status": "active"}
},
"binance": {
"BTC": {"price": 63600, "volume": "2.1M", "status": "mock"},
"ETH": {"price": 4140, "volume": "18.3M", "status": "mock"}
}
},
"market_signals": {
"spread_analysis": "Low spread (0.24%) indicates healthy liquidity",
"volume_trend": "Combined BTC volume: 3.3M USD (+12% from avg)",
"price_divergence": "Max deviation: 0.24% - Normal range",
"data_quality": "High - 3 sources, low variance",
"recommendation": "Data suitable for trading decisions"
},
"metadata": {
"query_time_ms": 1247,
"sources_queried": ["coinbase", "cryptocompare", "binance"],
"sources_successful": ["coinbase", "cryptocompare", "binance"],
"sources_failed": [],
"aggregation_method": "weighted_average",
"confidence_threshold": 0.75
}
}
```
---
*Diagramma dettaglio asincrono: 2025-09-23*
*Focus: Performance, Resilienza, Qualità Dati*

10
pyproject.toml
View File

@@ -37,5 +37,13 @@ dependencies = [
"coinbase-advanced-py",
"cryptocompare",
"cdp-sdk",
"python-binance"
"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"
]

2
src/app/agents/market_agent.py
View File

@@ -3,7 +3,7 @@ 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.coinbase_signer import CoinbaseCDPSigner
from app.signers.market_signers.cryptocompare_signer import CryptoCompareSigner
load_dotenv()

186
src/app/signers/market_signers/coinbase_cdp_signer.py
View File

@@ -1,186 +0,0 @@
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
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
}

368
src/app/signers/market_signers/coinbase_signer.py
View File

@@ -1,159 +1,243 @@
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
import os
import logging
from coinbase.rest import RESTClient
from typing import Dict, List, Any, Optional
logger = logging.getLogger(__name__)
class CoinbaseSigner:
class CoinbaseCDPSigner:
"""
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
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: str, secret_or_private_key: str, passphrase: str = None) -> None:
self.api_key = api_key
self.passphrase = passphrase
def __init__(self, api_key_name: str = None, api_private_key: str = None):
"""
Inizializza il Coinbase REST client.
# 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"""
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:
# 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,
self.client = RESTClient(
api_key=self.api_key_name,
api_secret=self.api_private_key
)
return private_key
self._configured = True
logger.info(f"✅ Coinbase REST Client configured with key: {self.api_key_name[:50]}...")
except Exception as e:
raise ValueError(f"Invalid private key format: {e}")
self._configured = False
logger.error(f"Failed to configure Coinbase REST Client: {e}")
raise ValueError(f"Failed to configure Coinbase REST Client: {e}")
@staticmethod
def _normalize_path(path: str) -> str:
if not path:
return "/"
return path if path.startswith("/") else f"/{path}"
def is_configured(self) -> bool:
"""Verifica se Coinbase REST client è configurato correttamente"""
return getattr(self, '_configured', False)
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())
)
def get_asset_info(self, asset_id: str) -> Dict[str, Any]:
"""
Ottiene informazioni su un asset specifico usando Coinbase Advanced Trade API.
# Converti signature in base64
cb_access_sign = base64.b64encode(signature).decode("utf-8")
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 {
"CB-ACCESS-KEY": self.api_key,
"CB-ACCESS-SIGN": cb_access_sign,
"CB-ACCESS-TIMESTAMP": ts,
"Content-Type": "application/json",
'Content-Type': 'application/json',
'User-Agent': 'upo-appAI/1.0-coinbase-advanced'
}
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[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 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)
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
}

70
src/app/utils/market_aggregator.py Normal file
View File

@@ -0,0 +1,70 @@
import statistics
from typing import Dict, List, Any
class MarketAggregator:
"""
Aggrega dati di mercato da più provider e genera segnali e analisi avanzate.
"""
@staticmethod
def aggregate(symbol: str, sources: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
prices = []
volumes = []
price_map = {}
for provider, data in sources.items():
price = data.get('price')
if price is not None:
prices.append(price)
price_map[provider] = price
volume = data.get('volume')
if volume is not None:
volumes.append(MarketAggregator._parse_volume(volume))
# Aggregated price (mean)
agg_price = statistics.mean(prices) if prices else None
# Spread analysis
spread = (max(prices) - min(prices)) / agg_price if prices and agg_price else 0
# Confidence
stddev = statistics.stdev(prices) if len(prices) > 1 else 0
confidence = max(0.5, 1 - (stddev / agg_price)) if agg_price else 0
if spread < 0.005:
confidence += 0.1
if len(prices) >= 3:
confidence += 0.05
confidence = min(confidence, 1.0)
# Volume trend
total_volume = sum(volumes) if volumes else None
# Price divergence
max_deviation = (max(prices) - min(prices)) / agg_price if prices and agg_price else 0
# Signals
signals = {
"spread_analysis": f"Low spread ({spread:.2%}) indicates healthy liquidity" if spread < 0.01 else f"Spread {spread:.2%} - check liquidity",
"volume_trend": f"Combined volume: {total_volume:.2f}" if total_volume else "Volume data not available",
"price_divergence": f"Max deviation: {max_deviation:.2%} - {'Normal range' if max_deviation < 0.01 else 'High divergence'}"
}
return {
"aggregated_data": {
f"{symbol}_USD": {
"price": round(agg_price, 2) if agg_price else None,
"confidence": round(confidence, 2),
"sources_count": len(prices)
}
},
"individual_sources": price_map,
"market_signals": signals
}
@staticmethod
def _parse_volume(volume: Any) -> float:
# Supporta stringhe tipo "1.2M" o numeri
if isinstance(volume, (int, float)):
return float(volume)
if isinstance(volume, str):
v = volume.upper().replace(' ', '')
if v.endswith('M'):
return float(v[:-1]) * 1_000_000
if v.endswith('K'):
return float(v[:-1]) * 1_000
try:
return float(v)
except Exception:
return 0.0
return 0.0

923
uv.lock generated
View File

File diff suppressed because it is too large Load Diff