""" Basic Analytics Module. Provides portfolio analytics: allocations, risk metrics, performance summary. """ import logging from typing import Dict, List, Any logger = logging.getLogger(__name__) # Geography mapping for holdings (hardcoded for MVP) GEOGRAPHY_MAP = { "VWCE.MI": "World", "MEUD.MI": "Europe", "SWDA.MI": "World", "EIMI.MI": "Emerging", "AGGH.MI": "Global", "AZM.MI": "Italy", "ENI.MI": "Italy", "ISP.MI": "Italy", "UCG.MI": "Italy", "TIT.MI": "Italy", "RACE.MI": "Italy", "CASH": "Cash", # Add more mappings as needed } class AnalyticsEngine: """Provides portfolio analytics and metrics.""" def __init__(self, portfolio, db_manager): """ Initialize analytics engine. Args: portfolio: Portfolio instance db_manager: DBManager instance """ self.portfolio = portfolio self.db_manager = db_manager def get_allocation_by_type(self) -> Dict[str, float]: """ Calculate allocation by asset type. Returns: Dictionary mapping asset_type -> weight percentage Example: {'ETF': 89.9, 'Stock': 3.4, 'Cash': 6.7} """ total_value = self.portfolio.get_total_value() if total_value == 0: return {} allocation = {} for holding in self.portfolio.holdings: asset_type = holding.asset_type if asset_type not in allocation: allocation[asset_type] = 0.0 allocation[asset_type] += (holding.current_value / total_value) * 100 logger.debug(f"Allocation by type: {allocation}") return allocation def get_allocation_by_geography(self) -> Dict[str, float]: """ Calculate allocation by geography. Uses GEOGRAPHY_MAP to map tickers to regions. Returns: Dictionary mapping geography -> weight percentage Example: {'World': 70.2, 'Europe': 19.6, 'Italy': 3.4, 'Cash': 6.7} """ total_value = self.portfolio.get_total_value() if total_value == 0: return {} allocation = {} for holding in self.portfolio.holdings: # Map ticker to geography geography = GEOGRAPHY_MAP.get(holding.ticker, "Other") if geography not in allocation: allocation[geography] = 0.0 allocation[geography] += (holding.current_value / total_value) * 100 logger.debug(f"Allocation by geography: {allocation}") return allocation def get_risk_metrics(self) -> Dict[str, Any]: """ Calculate risk metrics. Returns: Dictionary with risk metrics: - max_single_holding: (ticker, weight_pct) - stock_concentration: total % in stocks - etf_concentration: total % in ETFs - cash_concentration: total % in cash - concentration_warning: warning level ('ok', 'warning', 'danger') Example: { 'max_single_holding': ('VWCE.MI', 70.2), 'stock_concentration': 3.4, 'etf_concentration': 89.9, 'cash_concentration': 6.7, 'concentration_warning': 'danger' } """ allocation_by_type = self.get_allocation_by_type() # Find max single holding max_holding = None max_weight = 0.0 for holding in self.portfolio.holdings: if holding.weight_pct > max_weight: max_weight = holding.weight_pct max_holding = holding.ticker # Concentration by type stock_conc = allocation_by_type.get('Stock', 0.0) etf_conc = allocation_by_type.get('ETF', 0.0) cash_conc = allocation_by_type.get('Cash', 0.0) # Determine warning level if max_weight > 60: warning = 'danger' # Red elif max_weight > 40: warning = 'warning' # Yellow else: warning = 'ok' # Green/normal metrics = { 'max_single_holding': (max_holding, max_weight), 'stock_concentration': stock_conc, 'etf_concentration': etf_conc, 'cash_concentration': cash_conc, 'concentration_warning': warning } logger.debug(f"Risk metrics: {metrics}") return metrics def get_top_holdings(self, n: int = 5) -> List[Dict]: """ Get top N holdings by value. Args: n: Number of top holdings to return Returns: List of holding dictionaries sorted by current_value (descending) Example: [ {'ticker': 'VWCE.MI', 'current_value': 126240, 'weight_pct': 70.2}, {'ticker': 'MEUD.MI', 'current_value': 35325, 'weight_pct': 19.6}, ... ] """ # Sort holdings by current value (descending) sorted_holdings = sorted( self.portfolio.holdings, key=lambda h: h.current_value, reverse=True ) # Take top N top_n = sorted_holdings[:n] # Convert to dictionaries result = [ { 'ticker': h.ticker, 'name': h.name, 'asset_type': h.asset_type, 'current_value': h.current_value, 'weight_pct': h.weight_pct, 'pnl_amount': h.pnl_amount, 'pnl_percent': h.pnl_percent } for h in top_n ] logger.debug(f"Top {n} holdings: {[h['ticker'] for h in result]}") return result def get_total_dividends(self) -> float: """ Get total dividends received. Returns: Total dividend amount """ transactions = self.db_manager.get_transactions(tx_type='DIVIDEND') total = sum(tx['amount'] for tx in transactions) logger.debug(f"Total dividends: {total}") return total def get_performance_summary(self) -> Dict[str, Any]: """ Get comprehensive performance summary. Returns: Dictionary with performance metrics: - total_invested - current_value - total_pnl_amount - total_pnl_percent - total_dividends - cash_position - unrealized_pnl - total_return (includes dividends) Example: { 'total_invested': 166469.0, 'current_value': 179705.0, 'total_pnl_amount': 13236.0, 'total_pnl_percent': 7.9, 'total_dividends': 520.0, 'cash_position': 12000.0, 'unrealized_pnl': 13236.0, 'total_return': 13756.0 # P&L + dividends } """ total_invested = self.portfolio.get_total_invested() current_value = self.portfolio.get_total_value() pnl_amount, pnl_percent = self.portfolio.get_total_pnl() total_dividends = self.get_total_dividends() cash_position = self.portfolio.get_cash_position() # Total return includes unrealized P&L + dividends total_return = pnl_amount + total_dividends summary = { 'total_invested': total_invested, 'current_value': current_value, 'total_pnl_amount': pnl_amount, 'total_pnl_percent': pnl_percent, 'total_dividends': total_dividends, 'cash_position': cash_position, 'unrealized_pnl': pnl_amount, 'total_return': total_return } logger.debug(f"Performance summary: {summary}") return summary def get_allocation_table(self) -> List[Dict]: """ Get allocation data formatted for table display. Returns: List of allocation items with type/geography/value/weight Example: [ {'category': 'ETF', 'value': 161565, 'weight_pct': 89.9}, {'category': 'Stock', 'value': 6140, 'weight_pct': 3.4}, {'category': 'Cash', 'value': 12000, 'weight_pct': 6.7} ] """ total_value = self.portfolio.get_total_value() allocation_by_type = self.get_allocation_by_type() result = [] for asset_type, weight_pct in allocation_by_type.items(): value = (weight_pct / 100) * total_value result.append({ 'category': asset_type, 'value': value, 'weight_pct': weight_pct }) # Sort by value (descending) result.sort(key=lambda x: x['value'], reverse=True) logger.debug(f"Allocation table: {len(result)} items") return result def get_diversification_score(self) -> Dict[str, Any]: """ Calculate simple diversification score. Returns: Dictionary with diversification metrics: - holdings_count: number of holdings - effective_holdings: effective number (weighted by allocation) - diversification_score: 0-100 (higher is better) - assessment: 'Poor', 'Fair', 'Good', 'Excellent' Example: { 'holdings_count': 4, 'effective_holdings': 2.1, 'diversification_score': 42, 'assessment': 'Fair' } """ holdings_count = len(self.portfolio.holdings) # Calculate effective number of holdings (Herfindahl-Hirschman Index) hhi = sum((h.weight_pct / 100) ** 2 for h in self.portfolio.holdings) effective_holdings = 1 / hhi if hhi > 0 else 0 # Diversification score (0-100) # Based on effective holdings: 1 holding = 0, 10+ holdings = 100 score = min(100, (effective_holdings - 1) * 11.11) # Assessment if score >= 75: assessment = 'Excellent' elif score >= 50: assessment = 'Good' elif score >= 25: assessment = 'Fair' else: assessment = 'Poor' result = { 'holdings_count': holdings_count, 'effective_holdings': effective_holdings, 'diversification_score': score, 'assessment': assessment } logger.debug(f"Diversification score: {result}") return result