""" Portfolio Core Module. Manages portfolio holdings, calculations (P&L, weights, values). """ import logging from dataclasses import dataclass, field from typing import List, Dict, Tuple, Optional from datetime import datetime logger = logging.getLogger(__name__) @dataclass class Holding: """Represents a single holding in the portfolio.""" ticker: str name: str asset_type: str # 'ETF', 'Stock', 'Cash' quantity: float avg_price: float current_price: Optional[float] = None current_value: float = 0.0 pnl_amount: float = 0.0 pnl_percent: float = 0.0 weight_pct: float = 0.0 last_updated: Optional[datetime] = None def calculate_values(self) -> None: """Calculate current value and P&L.""" if self.current_price is not None and self.current_price > 0: self.current_value = self.quantity * self.current_price invested = self.quantity * self.avg_price self.pnl_amount = self.current_value - invested if invested > 0: self.pnl_percent = (self.pnl_amount / invested) * 100 else: self.pnl_percent = 0.0 else: # Use avg_price if current_price not available self.current_value = self.quantity * self.avg_price self.pnl_amount = 0.0 self.pnl_percent = 0.0 def to_dict(self) -> Dict: """Convert holding to dictionary for display.""" return { 'ticker': self.ticker, 'name': self.name, 'asset_type': self.asset_type, 'quantity': self.quantity, 'avg_price': self.avg_price, 'current_price': self.current_price, 'current_value': self.current_value, 'pnl_amount': self.pnl_amount, 'pnl_percent': self.pnl_percent, 'weight_pct': self.weight_pct } class Portfolio: """Manages portfolio holdings and calculations.""" def __init__(self, db_manager): """ Initialize portfolio. Args: db_manager: DBManager instance """ self.db_manager = db_manager self.holdings: List[Holding] = [] self._total_value: float = 0.0 self._total_invested: float = 0.0 def load_holdings(self) -> List[Holding]: """ Load holdings from database. Returns: List of Holding objects """ holdings_data = self.db_manager.get_all_holdings() self.holdings = [] for data in holdings_data: holding = Holding( ticker=data['ticker'], name=data['name'] or data['ticker'], asset_type=data['asset_type'] or 'Unknown', quantity=data['quantity'], avg_price=data['avg_price'], current_price=data['current_price'], last_updated=datetime.fromisoformat(data['last_updated']) if data.get('last_updated') else None ) holding.calculate_values() self.holdings.append(holding) # Calculate weights self.calculate_weights() logger.info(f"Loaded {len(self.holdings)} holdings from database") return self.holdings def get_total_value(self) -> float: """ Calculate total portfolio value. Returns: Total current value of all holdings """ self._total_value = sum(h.current_value for h in self.holdings) return self._total_value def get_total_invested(self) -> float: """ Calculate total amount invested. Returns: Total amount invested (qty × avg_price for all holdings) """ self._total_invested = sum(h.quantity * h.avg_price for h in self.holdings) return self._total_invested def get_total_pnl(self) -> Tuple[float, float]: """ Calculate total P&L. Returns: Tuple of (pnl_amount, pnl_percent) """ total_value = self.get_total_value() total_invested = self.get_total_invested() pnl_amount = total_value - total_invested if total_invested > 0: pnl_percent = (pnl_amount / total_invested) * 100 else: pnl_percent = 0.0 return (pnl_amount, pnl_percent) def calculate_weights(self) -> None: """Calculate weight percentage for each holding.""" total_value = self.get_total_value() if total_value > 0: for holding in self.holdings: holding.weight_pct = (holding.current_value / total_value) * 100 else: for holding in self.holdings: holding.weight_pct = 0.0 logger.debug(f"Calculated weights for {len(self.holdings)} holdings") def update_prices(self, prices: Dict[str, float]) -> None: """ Update current prices for holdings. Args: prices: Dictionary mapping ticker -> price """ updated_count = 0 for holding in self.holdings: if holding.ticker in prices: old_price = holding.current_price holding.current_price = prices[holding.ticker] holding.calculate_values() # Update database self.db_manager.update_holding( holding.ticker, current_price=holding.current_price, current_value=holding.current_value ) updated_count += 1 logger.debug(f"Updated price for {holding.ticker}: {old_price} -> {holding.current_price}") # Recalculate weights after price updates self.calculate_weights() logger.info(f"Updated prices for {updated_count}/{len(self.holdings)} holdings") def add_holding( self, ticker: str, name: str, asset_type: str, quantity: float, avg_price: float ) -> None: """ Add a new holding to the portfolio. Args: ticker: Stock ticker symbol name: Full name asset_type: 'ETF', 'Stock', or 'Cash' quantity: Number of shares avg_price: Average purchase price """ # Add to database self.db_manager.add_holding(ticker, name, asset_type, quantity, avg_price) # Create holding object holding = Holding( ticker=ticker, name=name, asset_type=asset_type, quantity=quantity, avg_price=avg_price ) holding.calculate_values() self.holdings.append(holding) self.calculate_weights() logger.info(f"Added new holding: {ticker} ({quantity} @ {avg_price})") def get_holding(self, ticker: str) -> Optional[Holding]: """ Get a specific holding by ticker. Args: ticker: Ticker to find Returns: Holding object or None if not found """ for holding in self.holdings: if holding.ticker == ticker: return holding return None def update_holding_quantity(self, ticker: str, new_quantity: float, new_avg_price: float) -> None: """ Update holding quantity and average price. Args: ticker: Ticker to update new_quantity: New quantity new_avg_price: New average price """ holding = self.get_holding(ticker) if holding: holding.quantity = new_quantity holding.avg_price = new_avg_price holding.calculate_values() # Update database self.db_manager.update_holding( ticker, quantity=new_quantity, avg_price=new_avg_price, current_value=holding.current_value ) # Recalculate weights self.calculate_weights() logger.info(f"Updated holding {ticker}: qty={new_quantity}, avg_price={new_avg_price}") def get_holdings_summary(self) -> List[Dict]: """ Get holdings summary for table display. Returns: List of holdings as dictionaries """ return [h.to_dict() for h in self.holdings] def get_cash_position(self) -> float: """ Get total cash position. Returns: Total cash amount """ cash_holdings = [h for h in self.holdings if h.asset_type == 'Cash'] return sum(h.current_value for h in cash_holdings) def get_holdings_by_type(self, asset_type: str) -> List[Holding]: """ Get holdings filtered by asset type. Args: asset_type: 'ETF', 'Stock', or 'Cash' Returns: List of holdings matching the type """ return [h for h in self.holdings if h.asset_type == asset_type] def get_portfolio_summary(self) -> Dict: """ Get comprehensive portfolio summary. Returns: Dictionary with portfolio metrics """ total_value = self.get_total_value() total_invested = self.get_total_invested() pnl_amount, pnl_percent = self.get_total_pnl() cash = self.get_cash_position() return { 'total_value': total_value, 'total_invested': total_invested, 'pnl_amount': pnl_amount, 'pnl_percent': pnl_percent, 'cash': cash, 'holdings_count': len(self.holdings) }