enhanced_portfolio_system.py

"""
Enhanced Portfolio Management System
Author: Enhanced by AI Assistant
Date: 2025-06-08

This enhanced version includes comprehensive use of:
- Classes and objects with inheritance
- Dictionaries and lists for data management
- Functions for modular code organization
- Loops for data processing
- File I/O operations
- Database functionality
- Data visualization
- Exception handling
"""

from datetime import datetime, timedelta
import json
import sqlite3
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple
import os
import csv
from dataclasses import dataclass
from enum import Enum

# --- Enums and Data Classes ---

class InvestmentType(Enum):
    """Enumeration for investment types"""
    STOCK = "stock"
    BOND = "bond"
    ETF = "etf"
    MUTUAL_FUND = "mutual_fund"

@dataclass
class MarketData:
    """Data class for market information"""
    symbol: str
    date: str
    open_price: float
    high_price: float
    low_price: float
    close_price: float
    volume: int

# --- Enhanced Base Classes ---

class Investment:
    """
    Base class for all investment types.
    Demonstrates inheritance and polymorphism.
    """
    
    def __init__(self, purchase_id: str, symbol: str, shares: int, 
                 purchase_price: float, current_value: float, 
                 purchase_date: str, investment_type: InvestmentType):
        self.purchase_id = purchase_id
        self.symbol = symbol.upper()
        self.shares = shares
        self.purchase_price = purchase_price
        self.current_value = current_value
        self.purchase_date = purchase_date
        self.investment_type = investment_type
        self._validate_data()
    
    def _validate_data(self):
        """Validate investment data"""
        if self.shares <= 0:
            raise ValueError("Shares must be positive")
        if self.purchase_price <= 0:
            raise ValueError("Purchase price must be positive")
        if self.current_value < 0:
            raise ValueError("Current value cannot be negative")
    
    def earnings_loss(self) -> float:
        """
        Calculate total earnings or loss for this investment.
        Formula: (current_value - purchase_price) × number_of_shares
        """
        return (self.current_value - self.purchase_price) * self.shares
    
    def yearly_earnings_loss_rate(self, current_date: Optional[datetime] = None) -> float:
        """
        Calculate yearly earnings/loss rate as a percentage.
        Formula: ((((current_value - purchase_price) / purchase_price) / 
                  (current_date - purchase_date))) × 100
        """
        if current_date is None:
            current_date = datetime.now()
        
        purchase_date_obj = datetime.strptime(self.purchase_date, "%m/%d/%Y")
        days_held = (current_date - purchase_date_obj).days
        
        if days_held <= 0:
            return 0.0
        
        years_held = days_held / 365.25
        price_change_rate = (self.current_value - self.purchase_price) / self.purchase_price
        yearly_rate = (price_change_rate / years_held) * 100
        
        return yearly_rate
    
    def get_total_value(self) -> float:
        """Get total current value of the investment"""
        return self.current_value * self.shares
    
    def get_total_cost(self) -> float:
        """Get total cost basis of the investment"""
        return self.purchase_price * self.shares
    
    def to_dict(self) -> Dict:
        """Convert investment to dictionary for serialization"""
        return {
            'purchase_id': self.purchase_id,
            'symbol': self.symbol,
            'shares': self.shares,
            'purchase_price': self.purchase_price,
            'current_value': self.current_value,
            'purchase_date': self.purchase_date,
            'investment_type': self.investment_type.value,
            'earnings_loss': self.earnings_loss(),
            'yearly_rate': self.yearly_earnings_loss_rate(),
            'total_value': self.get_total_value(),
            'total_cost': self.get_total_cost()
        }

class Stock(Investment):
    """
    Enhanced Stock class with additional functionality
    """
    
    def __init__(self, purchase_id: str, symbol: str, shares: int, 
                 purchase_price: float, current_value: float, 
                 purchase_date: str, sector: str = "Unknown", 
                 dividend_yield: float = 0.0):
        super().__init__(purchase_id, symbol, shares, purchase_price, 
                        current_value, purchase_date, InvestmentType.STOCK)
        self.sector = sector
        self.dividend_yield = dividend_yield
    
    def annual_dividend_income(self) -> float:
        """Calculate annual dividend income"""
        return self.get_total_value() * (self.dividend_yield / 100)
    
    def to_dict(self) -> Dict:
        """Enhanced dictionary representation for stocks"""
        base_dict = super().to_dict()
        base_dict.update({
            'sector': self.sector,
            'dividend_yield': self.dividend_yield,
            'annual_dividend': self.annual_dividend_income()
        })
        return base_dict

class Bond(Investment):
    """
    Enhanced Bond class with additional bond-specific features
    """
    
    def __init__(self, purchase_id: str, symbol: str, shares: int, 
                 purchase_price: float, current_value: float, 
                 purchase_date: str, coupon_rate: float, 
                 maturity_date: str, credit_rating: str = "NR"):
        super().__init__(purchase_id, symbol, shares, purchase_price, 
                        current_value, purchase_date, InvestmentType.BOND)
        self.coupon_rate = coupon_rate
        self.maturity_date = maturity_date
        self.credit_rating = credit_rating
    
    def annual_coupon_payment(self) -> float:
        """Calculate annual coupon payment"""
        return self.get_total_cost() * (self.coupon_rate / 100)
    
    def years_to_maturity(self) -> float:
        """Calculate years until maturity"""
        maturity_date_obj = datetime.strptime(self.maturity_date, "%m/%d/%Y")
        current_date = datetime.now()
        days_to_maturity = (maturity_date_obj - current_date).days
        return max(0, days_to_maturity / 365.25)
    
    def to_dict(self) -> Dict:
        """Enhanced dictionary representation for bonds"""
        base_dict = super().to_dict()
        base_dict.update({
            'coupon_rate': self.coupon_rate,
            'maturity_date': self.maturity_date,
            'credit_rating': self.credit_rating,
            'annual_coupon': self.annual_coupon_payment(),
            'years_to_maturity': self.years_to_maturity()
        })
        return base_dict

# --- Enhanced Portfolio Management Classes ---

class PortfolioAnalytics:
    """
    Class for portfolio analytics and calculations
    Demonstrates extensive use of functions and mathematical operations
    """
    
    @staticmethod
    def calculate_portfolio_metrics(investments: List[Investment]) -> Dict:
        """Calculate comprehensive portfolio metrics"""
        if not investments:
            return {}
        
        total_value = sum(inv.get_total_value() for inv in investments)
        total_cost = sum(inv.get_total_cost() for inv in investments)
        total_earnings = sum(inv.earnings_loss() for inv in investments)
        
        # Calculate weighted average yearly return
        weighted_returns = []
        weights = []
        
        for investment in investments:
            weight = investment.get_total_value() / total_value if total_value > 0 else 0
            yearly_return = investment.yearly_earnings_loss_rate()
            weighted_returns.append(yearly_return * weight)
            weights.append(weight)
        
        avg_yearly_return = sum(weighted_returns)
        
        # Calculate portfolio diversity metrics
        sectors = {}
        investment_types = {}
        
        for investment in investments:
            # Count by investment type
            inv_type = investment.investment_type.value
            investment_types[inv_type] = investment_types.get(inv_type, 0) + investment.get_total_value()
            
            # Count by sector (for stocks)
            if isinstance(investment, Stock):
                sector = investment.sector
                sectors[sector] = sectors.get(sector, 0) + investment.get_total_value()
        
        return {
            'total_value': total_value,
            'total_cost': total_cost,
            'total_earnings': total_earnings,
            'total_return_percentage': (total_earnings / total_cost * 100) if total_cost > 0 else 0,
            'average_yearly_return': avg_yearly_return,
            'number_of_investments': len(investments),
            'sectors': sectors,
            'investment_types': investment_types,
            'best_performer': max(investments, key=lambda x: x.yearly_earnings_loss_rate()) if investments else None,
            'worst_performer': min(investments, key=lambda x: x.yearly_earnings_loss_rate()) if investments else None
        }
    
    @staticmethod
    def calculate_risk_metrics(investments: List[Investment]) -> Dict:
        """Calculate risk-related metrics"""
        if not investments:
            return {}
        
        returns = [inv.yearly_earnings_loss_rate() for inv in investments]
        
        # Calculate standard deviation of returns (volatility)
        mean_return = sum(returns) / len(returns)
        variance = sum((r - mean_return) ** 2 for r in returns) / len(returns)
        volatility = variance ** 0.5
        
        # Calculate Sharpe ratio (assuming risk-free rate of 2%)
        risk_free_rate = 2.0
        sharpe_ratio = (mean_return - risk_free_rate) / volatility if volatility > 0 else 0
        
        return {
            'volatility': volatility,
            'sharpe_ratio': sharpe_ratio,
            'max_return': max(returns) if returns else 0,
            'min_return': min(returns) if returns else 0,
            'return_range': max(returns) - min(returns) if returns else 0
        }

class Portfolio:
    """
    Enhanced Portfolio class with comprehensive functionality
    Demonstrates extensive use of dictionaries, lists, and functions
    """
    
    def __init__(self, portfolio_id: str, name: str, description: str = ""):
        self.portfolio_id = portfolio_id
        self.name = name
        self.description = description
        self.investments: Dict[str, Investment] = {}  # Dictionary of investments by symbol
        self.creation_date = datetime.now().strftime("%m/%d/%Y")
        self.analytics = PortfolioAnalytics()
    
    def add_investment(self, investment: Investment) -> bool:
        """Add an investment to the portfolio"""
        try:
            key = f"{investment.symbol}_{investment.purchase_id}"
            self.investments[key] = investment
            return True
        except Exception as e:
            print(f"Error adding investment: {e}")
            return False
    
    def remove_investment(self, symbol: str, purchase_id: str) -> bool:
        """Remove an investment from the portfolio"""
        key = f"{symbol.upper()}_{purchase_id}"
        if key in self.investments:
            del self.investments[key]
            return True
        return False
    
    def get_investments_by_type(self, investment_type: InvestmentType) -> List[Investment]:
        """Get all investments of a specific type"""
        return [inv for inv in self.investments.values() 
                if inv.investment_type == investment_type]
    
    def get_investments_by_symbol(self, symbol: str) -> List[Investment]:
        """Get all investments for a specific symbol"""
        return [inv for inv in self.investments.values() 
                if inv.symbol == symbol.upper()]
    
    def get_top_performers(self, count: int = 5) -> List[Investment]:
        """Get top performing investments by yearly return rate"""
        sorted_investments = sorted(self.investments.values(), 
                                  key=lambda x: x.yearly_earnings_loss_rate(), 
                                  reverse=True)
        return sorted_investments[:count]
    
    def get_bottom_performers(self, count: int = 5) -> List[Investment]:
        """Get worst performing investments by yearly return rate"""
        sorted_investments = sorted(self.investments.values(), 
                                  key=lambda x: x.yearly_earnings_loss_rate())
        return sorted_investments[:count]
    
    def get_portfolio_summary(self) -> Dict:
        """Get comprehensive portfolio summary"""
        investments_list = list(self.investments.values())
        basic_metrics = self.analytics.calculate_portfolio_metrics(investments_list)
        risk_metrics = self.analytics.calculate_risk_metrics(investments_list)
        
        summary = {
            'portfolio_info': {
                'id': self.portfolio_id,
                'name': self.name,
                'description': self.description,
                'creation_date': self.creation_date
            },
            'basic_metrics': basic_metrics,
            'risk_metrics': risk_metrics,
            'holdings_count': len(self.investments),
            'symbols': list(set(inv.symbol for inv in investments_list))
        }
        
        return summary
    
    def to_dict(self) -> Dict:
        """Convert entire portfolio to dictionary"""
        return {
            'portfolio_info': {
                'id': self.portfolio_id,
                'name': self.name,
                'description': self.description,
                'creation_date': self.creation_date
            },
            'investments': {key: inv.to_dict() for key, inv in self.investments.items()},
            'summary': self.get_portfolio_summary()
        }

class Investor:
    """
    Enhanced Investor class with multiple portfolios support
    """
    
    def __init__(self, investor_id: str, name: str, email: str, 
                 phone: str = "", address: str = ""):
        self.investor_id = investor_id
        self.name = name
        self.email = email
        self.phone = phone
        self.address = address
        self.portfolios: Dict[str, Portfolio] = {}  # Dictionary of portfolios
        self.registration_date = datetime.now().strftime("%m/%d/%Y")
    
    def create_portfolio(self, portfolio_id: str, name: str, description: str = "") -> Portfolio:
        """Create a new portfolio"""
        portfolio = Portfolio(portfolio_id, name, description)
        self.portfolios[portfolio_id] = portfolio
        return portfolio
    
    def get_portfolio(self, portfolio_id: str) -> Optional[Portfolio]:
        """Get a specific portfolio"""
        return self.portfolios.get(portfolio_id)
    
    def get_all_investments(self) -> List[Investment]:
        """Get all investments across all portfolios"""
        all_investments = []
        for portfolio in self.portfolios.values():
            all_investments.extend(portfolio.investments.values())
        return all_investments
    
    def get_total_portfolio_value(self) -> float:
        """Get total value across all portfolios"""
        return sum(inv.get_total_value() for inv in self.get_all_investments())
    
    def get_investor_summary(self) -> Dict:
        """Get comprehensive investor summary"""
        all_investments = self.get_all_investments()
        analytics = PortfolioAnalytics()
        
        return {
            'investor_info': {
                'id': self.investor_id,
                'name': self.name,
                'email': self.email,
                'phone': self.phone,
                'address': self.address,
                'registration_date': self.registration_date
            },
            'portfolio_count': len(self.portfolios),
            'total_investments': len(all_investments),
            'total_value': self.get_total_portfolio_value(),
            'metrics': analytics.calculate_portfolio_metrics(all_investments),
            'risk_metrics': analytics.calculate_risk_metrics(all_investments)
        }

if __name__ == "__main__":
    print("Enhanced Portfolio Management System - Core Classes Loaded")
    print("=" * 60)
    
    # Demonstration of the enhanced classes
    print("Creating sample investor and portfolio...")
    
    # Create investor
    investor = Investor("INV001", "John Doe", "john.doe@email.com", 
                       "555-1234", "123 Main St, City, State")
    
    # Create portfolio
    portfolio = investor.create_portfolio("PORT001", "Main Portfolio", 
                                        "Primary investment portfolio")
    
    # Create sample investments
    stock1 = Stock("STK001", "AAPL", 100, 150.0, 175.0, "01/15/2024", 
                   "Technology", 0.5)
    stock2 = Stock("STK002", "GOOGL", 50, 2500.0, 2750.0, "02/20/2024", 
                   "Technology", 0.0)
    bond1 = Bond("BND001", "US10Y", 10, 1000.0, 980.0, "03/10/2024", 
                 4.5, "03/10/2034", "AAA")
    
    # Add investments to portfolio
    portfolio.add_investment(stock1)
    portfolio.add_investment(stock2)
    portfolio.add_investment(bond1)
    
    # Display summary
    summary = investor.get_investor_summary()
    print(f"Investor: {summary['investor_info']['name']}")
    print(f"Total Portfolio Value: ${summary['total_value']:,.2f}")
    print(f"Total Investments: {summary['total_investments']}")
    print(f"Average Yearly Return: {summary['metrics']['average_yearly_return']:.2f}%")
    
    print("\nCore classes implementation complete!")