From c30e89f82cbcfc11f81452402aa5ad96e4a05aa5 Mon Sep 17 00:00:00 2001 From: Pascal Date: Wed, 4 Jun 2025 14:35:14 +0200 Subject: [PATCH] cleaning unused files --- services/drip_service/__init__.py | 4 - services/drip_service/logger.py | 23 -- services/drip_service/models.py | 46 --- services/drip_service/service.py | 455 ----------------------- services/nav_erosion_service/__init__.py | 8 - services/nav_erosion_service/models.py | 31 -- services/nav_erosion_service/service.py | 209 ----------- 7 files changed, 776 deletions(-) delete mode 100644 services/drip_service/__init__.py delete mode 100644 services/drip_service/logger.py delete mode 100644 services/drip_service/models.py delete mode 100644 services/drip_service/service.py delete mode 100644 services/nav_erosion_service/__init__.py delete mode 100644 services/nav_erosion_service/models.py delete mode 100644 services/nav_erosion_service/service.py diff --git a/services/drip_service/__init__.py b/services/drip_service/__init__.py deleted file mode 100644 index 6f00234..0000000 --- a/services/drip_service/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -from .service import DripService -from .models import DripConfig, DripResult, MonthlyData, PortfolioAllocation - -__all__ = ['DripService', 'DripConfig', 'DripResult', 'MonthlyData', 'PortfolioAllocation'] \ No newline at end of file diff --git a/services/drip_service/logger.py b/services/drip_service/logger.py deleted file mode 100644 index 900e36c..0000000 --- a/services/drip_service/logger.py +++ /dev/null @@ -1,23 +0,0 @@ -import logging -import sys - -def setup_logger(): - # Create logger - logger = logging.getLogger('drip_service') - logger.setLevel(logging.DEBUG) - - # Create console handler with formatting - console_handler = logging.StreamHandler(sys.stdout) - console_handler.setLevel(logging.DEBUG) - - # Create formatter - formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') - console_handler.setFormatter(formatter) - - # Add handler to logger - logger.addHandler(console_handler) - - return logger - -# Create logger instance -logger = setup_logger() \ No newline at end of file diff --git a/services/drip_service/models.py b/services/drip_service/models.py deleted file mode 100644 index 6434058..0000000 --- a/services/drip_service/models.py +++ /dev/null @@ -1,46 +0,0 @@ -from dataclasses import dataclass -from typing import Dict, List, Optional -from datetime import datetime - -@dataclass -class PortfolioAllocation: - ticker: str - price: float - yield_annual: float - initial_shares: float - initial_allocation: float - distribution: str - -@dataclass -class MonthlyData: - month: int - total_value: float - monthly_income: float - cumulative_income: float - shares: Dict[str, float] - prices: Dict[str, float] - yields: Dict[str, float] - -@dataclass -class DripConfig: - months: int - erosion_type: str - erosion_level: Dict - dividend_frequency: Dict[str, int] = None - - def __post_init__(self): - if self.dividend_frequency is None: - self.dividend_frequency = { - "Monthly": 12, - "Quarterly": 4, - "Semi-Annually": 2, - "Annually": 1, - "Unknown": 12 # Default to monthly if unknown - } - -@dataclass -class DripResult: - monthly_data: List[MonthlyData] - final_portfolio_value: float - total_income: float - total_shares: Dict[str, float] \ No newline at end of file diff --git a/services/drip_service/service.py b/services/drip_service/service.py deleted file mode 100644 index 3ef8082..0000000 --- a/services/drip_service/service.py +++ /dev/null @@ -1,455 +0,0 @@ -from typing import Dict, List, Optional, Tuple, Any -import pandas as pd -import numpy as np -import traceback -from dataclasses import dataclass, field -from enum import Enum -from .models import PortfolioAllocation, MonthlyData, DripConfig, DripResult -from ..nav_erosion_service import NavErosionService -from .logger import logger - -class DistributionFrequency(Enum): - """Enum for distribution frequencies""" - MONTHLY = ("Monthly", 12) - QUARTERLY = ("Quarterly", 4) - SEMI_ANNUALLY = ("Semi-Annually", 2) - ANNUALLY = ("Annually", 1) - UNKNOWN = ("Unknown", 12) - - def __init__(self, name: str, payments_per_year: int): - self.display_name = name - self.payments_per_year = payments_per_year - -@dataclass -class TickerData: - """Data structure for individual ticker information""" - ticker: str - price: float - annual_yield: float - shares: float - allocation_pct: float - distribution_freq: DistributionFrequency - - @property - def market_value(self) -> float: - return self.price * self.shares - - @property - def monthly_yield(self) -> float: - return self.annual_yield / 12 - - @property - def distribution_yield(self) -> float: - return self.annual_yield / self.distribution_freq.payments_per_year - -class DripService: - """Enhanced DRIP calculation service with improved performance and accuracy""" - - def __init__(self) -> None: - self.DISTRIBUTION_FREQUENCIES = {freq.display_name: freq for freq in DistributionFrequency} - self.nav_erosion_service = NavErosionService() - - def calculate_drip_growth(self, portfolio_df: pd.DataFrame, config: DripConfig) -> DripResult: - """ - Calculate DRIP growth for a portfolio over a specified period with enhanced accuracy. - - Args: - portfolio_df: DataFrame containing portfolio allocation - config: DripConfig object with simulation parameters - - Returns: - DripResult object containing the simulation results - """ - try: - # Validate inputs - self._validate_inputs(portfolio_df, config) - - # Get erosion data from nav_erosion_service - erosion_data = self.nav_erosion_service.analyze_etf_erosion_risk(portfolio_df["Ticker"].tolist()) - logger.info(f"Erosion data results: {erosion_data.results}") - - # Initialize erosion rates dictionary - erosion_rates = {} - - # Use erosion rates from nav_erosion_service - for ticker in portfolio_df["Ticker"]: - # Find the result for this ticker in erosion_data.results - result = next((r for r in erosion_data.results if r.ticker == ticker), None) - - if result: - erosion_rates[ticker] = { - "nav": result.monthly_nav_erosion_rate, - "yield": result.monthly_yield_erosion_rate - } - logger.info(f"=== EROSION RATE DEBUG ===") - logger.info(f"Ticker: {ticker}") - logger.info(f"Erosion rates from nav_erosion_service:") - logger.info(f" NAV: {erosion_rates[ticker]['nav']:.4%}") - logger.info(f" Yield: {erosion_rates[ticker]['yield']:.4%}") - logger.info(f"=== END EROSION RATE DEBUG ===\n") - else: - # Use default erosion rates if not found - erosion_rates[ticker] = { - "nav": 0.05, # 5% per month (very high, for test) - "yield": 0.07 # 7% per month (very high, for test) - } - logger.info(f"=== EROSION RATE DEBUG ===") - logger.info(f"Ticker: {ticker}") - logger.info(f"Using default erosion rates:") - logger.info(f" NAV: {erosion_rates[ticker]['nav']:.4%}") - logger.info(f" Yield: {erosion_rates[ticker]['yield']:.4%}") - logger.info(f"=== END EROSION RATE DEBUG ===\n") - - # Log the final erosion rates dictionary - logger.info(f"Final erosion rates dictionary: {erosion_rates}") - - # Initialize portfolio data - ticker_data = self._initialize_ticker_data(portfolio_df) - - # Pre-calculate distribution schedule for performance - distribution_schedule = self._create_distribution_schedule(ticker_data, config.months) - - # Initialize simulation state - simulation_state = self._initialize_simulation_state(ticker_data) - monthly_data: List[MonthlyData] = [] - - # Run monthly simulation - for month in range(1, config.months + 1): - logger.info(f"\n=== Starting Month {month} ===") - logger.info(f"Initial state for month {month}:") - for ticker in ticker_data.keys(): - logger.info(f" {ticker}:") - logger.info(f" Price: ${simulation_state['current_prices'][ticker]:.2f}") - logger.info(f" Yield: {simulation_state['current_yields'][ticker]:.2%}") - logger.info(f" Shares: {simulation_state['current_shares'][ticker]:.4f}") - - month_result = self._simulate_month( - month, - simulation_state, - ticker_data, - erosion_rates, - distribution_schedule - ) - monthly_data.append(month_result) - - logger.info(f"Final state for month {month}:") - for ticker in ticker_data.keys(): - logger.info(f" {ticker}:") - logger.info(f" Price: ${simulation_state['current_prices'][ticker]:.2f}") - logger.info(f" Yield: {simulation_state['current_yields'][ticker]:.2%}") - logger.info(f" Shares: {simulation_state['current_shares'][ticker]:.4f}") - logger.info(f"=== End Month {month} ===\n") - - # Calculate final results - return self._create_drip_result(monthly_data, simulation_state) - - except Exception as e: - logger.error(f"Error calculating DRIP growth: {str(e)}") - logger.error(traceback.format_exc()) - raise - - def _validate_inputs(self, portfolio_df: pd.DataFrame, config: DripConfig) -> None: - """Validate input parameters""" - required_columns = ["Ticker", "Price", "Yield (%)", "Shares"] - missing_columns = [col for col in required_columns if col not in portfolio_df.columns] - - if missing_columns: - raise ValueError(f"Missing required columns: {missing_columns}") - - if config.months <= 0: - raise ValueError("Months must be positive") - - if portfolio_df.empty: - raise ValueError("Portfolio DataFrame is empty") - - def _initialize_ticker_data(self, portfolio_df: pd.DataFrame) -> Dict[str, TickerData]: - """Initialize ticker data with validation""" - ticker_data = {} - - for _, row in portfolio_df.iterrows(): - ticker = row["Ticker"] - - # Handle distribution frequency - dist_period = row.get("Distribution Period", "Monthly") - dist_freq = self.DISTRIBUTION_FREQUENCIES.get(dist_period, DistributionFrequency.MONTHLY) - - ticker_data[ticker] = TickerData( - ticker=ticker, - price=max(0.01, float(row["Price"])), # Prevent zero/negative prices - annual_yield=max(0.0, float(row["Yield (%)"] / 100)), # Convert to decimal - shares=max(0.0, float(row["Shares"])), - allocation_pct=float(row.get("Allocation (%)", 0) / 100), - distribution_freq=dist_freq - ) - - return ticker_data - - def _create_distribution_schedule(self, ticker_data: Dict[str, TickerData], total_months: int) -> Dict[str, List[int]]: - """Pre-calculate which months each ticker pays distributions""" - schedule = {} - - for ticker, data in ticker_data.items(): - distribution_months = [] - freq = data.distribution_freq - - for month in range(1, total_months + 1): - if self._is_distribution_month(month, freq): - distribution_months.append(month) - - schedule[ticker] = distribution_months - - return schedule - - def _initialize_simulation_state(self, ticker_data: Dict[str, TickerData]) -> Dict[str, Any]: - """Initialize simulation state variables""" - return { - 'current_shares': {ticker: data.shares for ticker, data in ticker_data.items()}, - 'current_prices': {ticker: data.price for ticker, data in ticker_data.items()}, - 'current_yields': {ticker: data.annual_yield for ticker, data in ticker_data.items()}, - 'cumulative_income': 0.0 - } - - def _simulate_month( - self, - month: int, - state: Dict[str, Any], - ticker_data: Dict[str, TickerData], - erosion_rates: Dict[str, Dict[str, float]], - distribution_schedule: Dict[str, List[int]] - ) -> MonthlyData: - """Simulate a single month with improved accuracy""" - - # Debug logging for erosion rates - logger.info(f"\n=== EROSION RATES DEBUG ===") - logger.info(f"Erosion rates dictionary: {erosion_rates}") - for ticker, rates in erosion_rates.items(): - logger.info(f" {ticker}:") - logger.info(f" nav: {rates['nav']:.4%}") - logger.info(f" yield: {rates['yield']:.4%}") - logger.info(f"=== END EROSION RATES DEBUG ===\n") - - # Apply erosion first - for ticker, rates in erosion_rates.items(): - if ticker in state['current_prices']: - # Get monthly erosion rates (already in decimal form) - monthly_nav_erosion = rates['nav'] - monthly_yield_erosion = rates['yield'] - - # Get current values - old_price = state['current_prices'][ticker] - old_yield = state['current_yields'][ticker] - - # Debug logging - logger.info(f"\n=== EROSION CALCULATION DEBUG ===") - logger.info(f"Ticker: {ticker}") - logger.info(f"Raw erosion rates from nav_erosion_service:") - logger.info(f" monthly_nav_erosion: {monthly_nav_erosion:.4%}") - logger.info(f" monthly_yield_erosion: {monthly_yield_erosion:.4%}") - logger.info(f"Current values:") - logger.info(f" old_price: ${old_price:.4f}") - logger.info(f" old_yield: {old_yield:.4%}") - - # Calculate new values - new_price = old_price * (1 - monthly_nav_erosion) - new_yield = old_yield * (1 - monthly_yield_erosion) - - logger.info(f"Calculated new values:") - logger.info(f" new_price = ${old_price:.4f} * (1 - {monthly_nav_erosion:.4%})") - logger.info(f" new_price = ${old_price:.4f} * {1 - monthly_nav_erosion:.4f}") - logger.info(f" new_price = ${new_price:.4f}") - logger.info(f" new_yield = {old_yield:.4%} * (1 - {monthly_yield_erosion:.4%})") - logger.info(f" new_yield = {old_yield:.4%} * {1 - monthly_yield_erosion:.4f}") - logger.info(f" new_yield = {new_yield:.4%}") - - # Apply the new values with bounds checking - state['current_prices'][ticker] = max(0.01, new_price) # Prevent zero/negative prices - state['current_yields'][ticker] = max(0.0, new_yield) # Prevent negative yields - - logger.info(f"Final values after bounds checking:") - logger.info(f" final_price: ${state['current_prices'][ticker]:.4f}") - logger.info(f" final_yield: {state['current_yields'][ticker]:.4%}") - logger.info(f"=== END EROSION CALCULATION DEBUG ===\n") - - # Log the actual erosion being applied - logger.info(f"Applied erosion to {ticker}:") - logger.info(f" NAV: {monthly_nav_erosion:.4%} -> New price: ${state['current_prices'][ticker]:.2f}") - logger.info(f" Yield: {monthly_yield_erosion:.4%} -> New yield: {state['current_yields'][ticker]:.2%}") - - # Calculate monthly income from distributions using eroded values - monthly_income = self._calculate_monthly_distributions( - month, state, ticker_data, distribution_schedule - ) - - # Update cumulative income - state['cumulative_income'] += monthly_income - - # Reinvest dividends (DRIP) - self._reinvest_dividends(month, state, distribution_schedule) - - # Calculate total portfolio value with bounds checking - total_value = 0.0 - for ticker in ticker_data.keys(): - shares = state['current_shares'][ticker] - price = state['current_prices'][ticker] - if shares > 0 and price > 0: - total_value += shares * price - - return MonthlyData( - month=month, - total_value=total_value, - monthly_income=monthly_income, - cumulative_income=state['cumulative_income'], - shares=state['current_shares'].copy(), - prices=state['current_prices'].copy(), - yields=state['current_yields'].copy() - ) - - def _calculate_monthly_distributions( - self, - month: int, - state: Dict[str, Any], - ticker_data: Dict[str, TickerData], - distribution_schedule: Dict[str, List[int]] - ) -> float: - """Calculate distributions for the current month""" - monthly_income = 0.0 - - for ticker, data in ticker_data.items(): - if month in distribution_schedule[ticker]: - shares = state['current_shares'][ticker] - price = state['current_prices'][ticker] - yield_rate = state['current_yields'][ticker] - - # Calculate distribution amount using annual yield divided by payments per year - distribution_yield = yield_rate / data.distribution_freq.payments_per_year - distribution_amount = shares * price * distribution_yield - monthly_income += distribution_amount - - return monthly_income - - def _reinvest_dividends( - self, - month: int, - state: Dict[str, Any], - distribution_schedule: Dict[str, List[int]] - ) -> None: - """Reinvest dividends for tickers that distributed in this month""" - - for ticker, distribution_months in distribution_schedule.items(): - if month in distribution_months: - shares = state['current_shares'][ticker] - price = state['current_prices'][ticker] - yield_rate = state['current_yields'][ticker] - - # Calculate dividend income using the correct distribution frequency - freq = self.DISTRIBUTION_FREQUENCIES.get(ticker, DistributionFrequency.MONTHLY) - dividend_income = shares * price * (yield_rate / freq.payments_per_year) - - # Purchase additional shares - if price > 0: - new_shares = dividend_income / price - state['current_shares'][ticker] += new_shares - - def _is_distribution_month(self, month: int, frequency: DistributionFrequency) -> bool: - """Check if current month is a distribution month""" - if frequency == DistributionFrequency.MONTHLY: - return True - elif frequency == DistributionFrequency.QUARTERLY: - return month % 3 == 0 - elif frequency == DistributionFrequency.SEMI_ANNUALLY: - return month % 6 == 0 - elif frequency == DistributionFrequency.ANNUALLY: - return month % 12 == 0 - else: - return True # Default to monthly for unknown - - def _create_drip_result(self, monthly_data: List[MonthlyData], state: Dict[str, Any]) -> DripResult: - """Create final DRIP result object""" - if not monthly_data: - raise ValueError("No monthly data generated") - - final_data = monthly_data[-1] - - return DripResult( - monthly_data=monthly_data, - final_portfolio_value=final_data.total_value, - total_income=final_data.cumulative_income, - total_shares=state['current_shares'].copy() - ) - - # Utility methods for analysis and comparison - def calculate_drip_vs_no_drip_comparison( - self, - portfolio_df: pd.DataFrame, - config: DripConfig - ) -> Dict[str, Any]: - """Calculate comparison between DRIP and no-DRIP scenarios""" - - # Calculate DRIP scenario - drip_result = self.calculate_drip_growth(portfolio_df, config) - - # Calculate no-DRIP scenario (dividends not reinvested) - no_drip_result = self._calculate_no_drip_scenario(portfolio_df, config) - - # Calculate comparison metrics - drip_advantage = drip_result.final_portfolio_value - no_drip_result['final_value'] - percentage_advantage = (drip_advantage / no_drip_result['final_value']) * 100 - - return { - 'drip_final_value': drip_result.final_portfolio_value, - 'no_drip_final_value': no_drip_result['final_value'], - 'drip_advantage': drip_advantage, - 'percentage_advantage': percentage_advantage, - 'total_dividends_reinvested': drip_result.total_income, - 'cash_dividends_no_drip': no_drip_result['total_dividends'] - } - - def _calculate_no_drip_scenario(self, portfolio_df: pd.DataFrame, config: DripConfig) -> Dict[str, float]: - """Calculate scenario where dividends are not reinvested""" - ticker_data = self._initialize_ticker_data(portfolio_df) - erosion_data = self.nav_erosion_service.analyze_etf_erosion_risk(portfolio_df["Ticker"].tolist()) - erosion_rates = { - result.ticker: { - "nav": result.monthly_nav_erosion_rate, - "yield": result.monthly_yield_erosion_rate - } - for result in erosion_data.results - } - state = self._initialize_simulation_state(ticker_data) - - total_dividends = 0.0 - - for month in range(1, config.months + 1): - # Calculate dividends but don't reinvest - monthly_dividends = self._calculate_monthly_distributions( - month, state, ticker_data, - self._create_distribution_schedule(ticker_data, config.months) - ) - total_dividends += monthly_dividends - - # Apply erosion - for ticker, rates in erosion_rates.items(): - if ticker in state['current_prices']: - # Get monthly erosion rates (already in decimal form) - monthly_nav_erosion = rates['nav'] - monthly_yield_erosion = rates['yield'] - - # Apply NAV erosion (decrease price) - old_price = state['current_prices'][ticker] - new_price = old_price * (1 - monthly_nav_erosion) - state['current_prices'][ticker] = max(0.01, new_price) # Prevent zero/negative prices - - # Apply yield erosion (decrease yield) - old_yield = state['current_yields'][ticker] - new_yield = old_yield * (1 - monthly_yield_erosion) - state['current_yields'][ticker] = max(0.0, new_yield) # Prevent negative yields - - final_value = sum( - state['current_shares'][ticker] * state['current_prices'][ticker] - for ticker in ticker_data.keys() - ) - - return { - 'final_value': final_value, - 'total_dividends': total_dividends - } \ No newline at end of file diff --git a/services/nav_erosion_service/__init__.py b/services/nav_erosion_service/__init__.py deleted file mode 100644 index ef3a1b2..0000000 --- a/services/nav_erosion_service/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -""" -Nav Erosion Service package -""" - -from .service import NavErosionService -from .models import NavErosionResult - -__all__ = ['NavErosionService', 'NavErosionResult'] \ No newline at end of file diff --git a/services/nav_erosion_service/models.py b/services/nav_erosion_service/models.py deleted file mode 100644 index 62265bd..0000000 --- a/services/nav_erosion_service/models.py +++ /dev/null @@ -1,31 +0,0 @@ -from dataclasses import dataclass -from typing import Dict, List, Optional -from datetime import datetime - -@dataclass -class NavErosionConfig: - max_erosion_level: int = 9 - max_monthly_erosion: float = 1 - (0.1)**(1/12) # ~17.54% monthly for 90% annual erosion - use_per_ticker: bool = False - global_nav_rate: float = 0 - per_ticker_rates: Dict[str, float] = None - -@dataclass -class NavErosionResult: - ticker: str - nav_erosion_rate: float - monthly_erosion_rate: float - annual_erosion_rate: float - risk_level: int # 0-9 scale - risk_explanation: str - max_drawdown: float - volatility: float - is_new_etf: bool - etf_age_years: Optional[float] - -@dataclass -class NavErosionAnalysis: - results: List[NavErosionResult] - portfolio_nav_risk: float # Average risk level - portfolio_erosion_rate: float # Weighted average erosion rate - risk_summary: str \ No newline at end of file diff --git a/services/nav_erosion_service/service.py b/services/nav_erosion_service/service.py deleted file mode 100644 index 3383fbd..0000000 --- a/services/nav_erosion_service/service.py +++ /dev/null @@ -1,209 +0,0 @@ -from typing import Dict, List, Tuple -from .models import NavErosionConfig, NavErosionResult, NavErosionAnalysis -from enum import Enum -from dataclasses import dataclass -import streamlit as st - -class ETFType(Enum): - INCOME = "Income" - GROWTH = "Growth" - BALANCED = "Balanced" - -@dataclass -class NavErosionResult: - """Result of NAV erosion analysis for a single ETF""" - ticker: str - nav_erosion_rate: float # Annual NAV erosion rate - yield_erosion_rate: float # Annual yield erosion rate - monthly_nav_erosion_rate: float # Monthly NAV erosion rate - monthly_yield_erosion_rate: float # Monthly yield erosion rate - risk_level: int - risk_explanation: str - max_drawdown: float - volatility: float - is_new_etf: bool - etf_age_years: float - -@dataclass -class NavErosionAnalysis: - """Complete NAV erosion analysis results""" - results: List[NavErosionResult] - portfolio_nav_risk: float = 0.0 - portfolio_erosion_rate: float = 0.0 - risk_summary: str = "" - -class NavErosionService: - def __init__(self): - self.NAV_COMPONENT_WEIGHTS = { - 'drawdown': 0.4, - 'volatility': 0.3, - 'sharpe': 0.15, - 'sortino': 0.15 - } - - # Default erosion rates based on risk level (0-9) - self.RISK_TO_EROSION = { - 0: 0.01, # 1% annual - 1: 0.02, # 2% annual - 2: 0.03, # 3% annual - 3: 0.04, # 4% annual - 4: 0.05, # 5% annual - 5: 0.06, # 6% annual - 6: 0.07, # 7% annual - 7: 0.08, # 8% annual - 8: 0.09, # 9% annual - 9: 0.10 # 10% annual - } - - def analyze_etf_erosion_risk(self, tickers: List[str]) -> NavErosionAnalysis: - """Analyze NAV erosion risk for a list of ETFs""" - results = [] - - print("\n=== NAV EROSION SERVICE DEBUG ===") - print(f"Session state keys: {st.session_state.keys()}") - print(f"Erosion level from session state: {st.session_state.get('erosion_level')}") - - for ticker in tickers: - # Get erosion rates from session state - erosion_level = st.session_state.get('erosion_level', {'nav': 5.0, 'yield': 5.0}) - annual_nav_erosion = erosion_level['nav'] / 100 # Convert from percentage to decimal - annual_yield_erosion = erosion_level['yield'] / 100 # Convert from percentage to decimal - - # Convert annual rates to monthly - monthly_nav_erosion = 1 - (1 - annual_nav_erosion) ** (1/12) - monthly_yield_erosion = 1 - (1 - annual_yield_erosion) ** (1/12) - - print(f"\n=== NAV EROSION SERVICE DEBUG ===") - print(f"Ticker: {ticker}") - print(f"Session State Values:") - print(f" Annual NAV Erosion: {annual_nav_erosion:.4%}") - print(f" Annual Yield Erosion: {annual_yield_erosion:.4%}") - print(f" Monthly NAV Erosion: {monthly_nav_erosion:.4%}") - print(f" Monthly Yield Erosion: {monthly_yield_erosion:.4%}") - print(f"=== END NAV EROSION SERVICE DEBUG ===\n") - - result = NavErosionResult( - ticker=ticker, - nav_erosion_rate=annual_nav_erosion, - yield_erosion_rate=annual_yield_erosion, - monthly_nav_erosion_rate=monthly_nav_erosion, - monthly_yield_erosion_rate=monthly_yield_erosion, - risk_level=5, # Arbitrary risk level - risk_explanation="Using erosion rates from session state", - max_drawdown=0.2, - volatility=0.25, - is_new_etf=False, - etf_age_years=1.0 - ) - results.append(result) - print(f"Created NavErosionResult for {ticker}:") - print(f" monthly_nav_erosion_rate: {result.monthly_nav_erosion_rate:.4%}") - print(f" monthly_yield_erosion_rate: {result.monthly_yield_erosion_rate:.4%}") - - # Calculate portfolio-level metrics - portfolio_nav_risk = sum(r.risk_level for r in results) / len(results) - portfolio_erosion_rate = sum(r.nav_erosion_rate for r in results) / len(results) - - analysis = NavErosionAnalysis( - results=results, - portfolio_nav_risk=portfolio_nav_risk, - portfolio_erosion_rate=portfolio_erosion_rate, - risk_summary="Portfolio has moderate NAV erosion risk" - ) - - print("\nFinal NavErosionAnalysis:") - for r in analysis.results: - print(f" {r.ticker}:") - print(f" monthly_nav_erosion_rate: {r.monthly_nav_erosion_rate:.4%}") - print(f" monthly_yield_erosion_rate: {r.monthly_yield_erosion_rate:.4%}") - print("=== END NAV EROSION SERVICE DEBUG ===\n") - - return analysis - - def _calculate_nav_risk(self, etf_data: Dict, etf_type: ETFType) -> Tuple[float, Dict]: - """Calculate NAV risk components with ETF-type specific adjustments""" - components = {} - - # Base risk calculation with ETF-type specific thresholds - if etf_data.get('max_drawdown') is not None: - if etf_type == ETFType.INCOME: - # Income ETFs typically have lower drawdowns - if etf_data['max_drawdown'] > 0.25: - components['drawdown'] = 7 - elif etf_data['max_drawdown'] > 0.15: - components['drawdown'] = 5 - elif etf_data['max_drawdown'] > 0.10: - components['drawdown'] = 3 - else: - components['drawdown'] = 2 - elif etf_type == ETFType.GROWTH: - # Growth ETFs typically have higher drawdowns - if etf_data['max_drawdown'] > 0.35: - components['drawdown'] = 7 - elif etf_data['max_drawdown'] > 0.25: - components['drawdown'] = 5 - elif etf_data['max_drawdown'] > 0.15: - components['drawdown'] = 3 - else: - components['drawdown'] = 2 - else: # BALANCED - # Balanced ETFs have moderate drawdowns - if etf_data['max_drawdown'] > 0.30: - components['drawdown'] = 7 - elif etf_data['max_drawdown'] > 0.20: - components['drawdown'] = 5 - elif etf_data['max_drawdown'] > 0.12: - components['drawdown'] = 3 - else: - components['drawdown'] = 2 - else: - components['drawdown'] = 4 # Default medium risk if no data - - # Rest of the method remains unchanged - if etf_data.get('volatility') is not None: - if etf_data['volatility'] > 0.40: - components['volatility'] = 7 - elif etf_data['volatility'] > 0.25: - components['volatility'] = 5 - elif etf_data['volatility'] > 0.15: - components['volatility'] = 3 - else: - components['volatility'] = 2 - else: - components['volatility'] = 4 - - if etf_data.get('sharpe_ratio') is not None: - if etf_data['sharpe_ratio'] >= 2.0: - components['sharpe'] = 1 - elif etf_data['sharpe_ratio'] >= 1.5: - components['sharpe'] = 2 - elif etf_data['sharpe_ratio'] >= 1.0: - components['sharpe'] = 3 - elif etf_data['sharpe_ratio'] >= 0.5: - components['sharpe'] = 4 - else: - components['sharpe'] = 5 - else: - components['sharpe'] = 4 - - if etf_data.get('sortino_ratio') is not None: - if etf_data['sortino_ratio'] >= 2.0: - components['sortino'] = 1 - elif etf_data['sortino_ratio'] >= 1.5: - components['sortino'] = 2 - elif etf_data['sortino_ratio'] >= 1.0: - components['sortino'] = 3 - elif etf_data['sortino_ratio'] >= 0.5: - components['sortino'] = 4 - else: - components['sortino'] = 5 - else: - components['sortino'] = 4 - - # Calculate weighted NAV risk - nav_risk = sum( - components[component] * weight - for component, weight in self.NAV_COMPONENT_WEIGHTS.items() - ) - - return nav_risk, components \ No newline at end of file