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