diff --git a/ETF_Portal/services/data_service.py b/ETF_Portal/services/data_service.py index 17cd198..d96a8f6 100644 --- a/ETF_Portal/services/data_service.py +++ b/ETF_Portal/services/data_service.py @@ -273,23 +273,28 @@ class DataService: earliest_ttm = monthly_div[-12:].sum() latest_ttm = monthly_div[-1:].sum() if earliest_ttm > 0: - dividend_trend = (latest_ttm / earliest_ttm - 1) + dividend_trend = float((latest_ttm / earliest_ttm - 1)) else: - dividend_trend = 0 + dividend_trend = 0.0 else: # If less than 12 months of data, use the average - dividend_trend = monthly_div.mean() if not monthly_div.empty else 0 + dividend_trend = float(monthly_div.mean()) if not monthly_div.empty else 0.0 else: # Try to get dividend trend from info - dividend_rate = info.get('dividendRate', 0) - five_year_avg = info.get('fiveYearAvgDividendYield', 0) + dividend_rate = float(info.get('dividendRate', 0)) + five_year_avg = float(info.get('fiveYearAvgDividendYield', 0)) if dividend_rate > 0 and five_year_avg > 0: - dividend_trend = (dividend_rate / five_year_avg - 1) + dividend_trend = float((dividend_rate / five_year_avg - 1)) else: - dividend_trend = 0 + dividend_trend = 0.0 except Exception as e: logger.warning(f"Error calculating dividend trend for {ticker}: {str(e)}") - dividend_trend = 0 + dividend_trend = 0.0 + + # Ensure dividend_trend is a valid float + dividend_trend = float(dividend_trend) if dividend_trend is not None else 0.0 + if not isinstance(dividend_trend, (int, float)) or pd.isna(dividend_trend): + dividend_trend = 0.0 # Calculate ETF age inception_date = info.get('fundInceptionDate') @@ -303,7 +308,6 @@ class DataService: age_years = None # Ensure all values are valid numbers and properly formatted - dividend_trend = float(dividend_trend) if dividend_trend is not None else 0.0 volatility = float(volatility) if volatility is not None else 0.0 max_drawdown = float(max_drawdown) if max_drawdown is not None else 0.0 sharpe_ratio = float(sharpe_ratio) if sharpe_ratio is not None else 0.0 diff --git a/ETF_Portal/services/nav_erosion_service.py b/ETF_Portal/services/nav_erosion_service.py new file mode 100644 index 0000000..295559a --- /dev/null +++ b/ETF_Portal/services/nav_erosion_service.py @@ -0,0 +1,218 @@ +""" +NAV Erosion Service for analyzing ETF erosion risk +""" + +import pandas as pd +import numpy as np +from typing import List, Dict, Optional +from dataclasses import dataclass +import logging +from .data_service import DataService + +logger = logging.getLogger(__name__) + +@dataclass +class ErosionRiskResult: + """Result of erosion risk analysis for a single ETF""" + ticker: str + nav_erosion_risk: float # 0-9 scale + yield_erosion_risk: float # 0-9 scale + estimated_nav_erosion: float # Annual percentage + estimated_yield_erosion: float # Annual percentage + nav_risk_explanation: str + yield_risk_explanation: str + etf_age_years: Optional[float] = None + max_drawdown: Optional[float] = None + volatility: Optional[float] = None + sharpe_ratio: Optional[float] = None + sortino_ratio: Optional[float] = None + dividend_trend: Optional[float] = None + +@dataclass +class ErosionRiskAnalysis: + """Complete erosion risk analysis results""" + results: List[ErosionRiskResult] + timestamp: pd.Timestamp + +class NavErosionService: + """Service for analyzing ETF NAV and yield erosion risk""" + + def __init__(self): + self.data_service = DataService() + + def analyze_etf_erosion_risk(self, tickers: List[str]) -> ErosionRiskAnalysis: + """ + Analyze erosion risk for a list of ETFs + + Args: + tickers: List of ETF tickers to analyze + + Returns: + ErosionRiskAnalysis object containing results for each ETF + """ + results = [] + + for ticker in tickers: + try: + # Get ETF data + etf_data = self.data_service.get_etf_data(ticker) + if not etf_data: + logger.warning(f"No data available for {ticker}") + continue + + # Calculate NAV erosion risk + nav_risk = self._calculate_nav_erosion_risk(etf_data) + + # Calculate yield erosion risk + yield_risk = self._calculate_yield_erosion_risk(etf_data) + + # Create result object + result = ErosionRiskResult( + ticker=ticker, + nav_erosion_risk=nav_risk['score'], + yield_erosion_risk=yield_risk['score'], + estimated_nav_erosion=nav_risk['estimated_erosion'], + estimated_yield_erosion=yield_risk['estimated_erosion'], + nav_risk_explanation=nav_risk['explanation'], + yield_risk_explanation=yield_risk['explanation'], + etf_age_years=etf_data.get('age_years'), + max_drawdown=etf_data.get('max_drawdown'), + volatility=etf_data.get('volatility'), + sharpe_ratio=etf_data.get('sharpe_ratio'), + sortino_ratio=etf_data.get('sortino_ratio'), + dividend_trend=etf_data.get('dividend_trend') + ) + + results.append(result) + + except Exception as e: + logger.error(f"Error analyzing {ticker}: {str(e)}") + continue + + return ErosionRiskAnalysis( + results=results, + timestamp=pd.Timestamp.now() + ) + + def _calculate_nav_erosion_risk(self, etf_data: Dict) -> Dict: + """ + Calculate NAV erosion risk score and explanation + + Args: + etf_data: Dictionary containing ETF data + + Returns: + Dictionary with risk score, estimated erosion, and explanation + """ + try: + # Get relevant metrics + volatility = float(etf_data.get('volatility', 0)) + max_drawdown = float(etf_data.get('max_drawdown', 0)) + sharpe_ratio = float(etf_data.get('sharpe_ratio', 0)) + sortino_ratio = float(etf_data.get('sortino_ratio', 0)) + age_years = float(etf_data.get('age_years', 0)) + + # Calculate risk score components + volatility_score = min(9, int(volatility * 20)) # Scale volatility to 0-9 + drawdown_score = min(9, int(max_drawdown * 20)) # Scale drawdown to 0-9 + risk_adjusted_score = min(9, int((2 - min(2, max(0, sharpe_ratio))) * 4.5)) # Scale Sharpe to 0-9 + age_score = min(9, int((5 - min(5, age_years)) * 1.8)) # Scale age to 0-9 + + # Calculate final risk score (weighted average) + risk_score = ( + volatility_score * 0.3 + + drawdown_score * 0.3 + + risk_adjusted_score * 0.2 + + age_score * 0.2 + ) + + # Estimate annual erosion based on risk score + estimated_erosion = risk_score * 0.01 # 1% per risk point + + # Generate explanation + explanation = [] + if volatility_score > 6: + explanation.append(f"High volatility ({volatility:.1%})") + if drawdown_score > 6: + explanation.append(f"Large drawdowns ({max_drawdown:.1%})") + if risk_adjusted_score > 6: + explanation.append(f"Poor risk-adjusted returns (Sharpe: {sharpe_ratio:.2f})") + if age_score > 6: + explanation.append(f"New ETF ({age_years:.1f} years)") + + explanation = ", ".join(explanation) if explanation else "Moderate risk profile" + + return { + 'score': float(risk_score), + 'estimated_erosion': float(estimated_erosion), + 'explanation': explanation + } + + except Exception as e: + logger.error(f"Error calculating NAV erosion risk: {str(e)}") + return { + 'score': 5.0, # Default to middle risk + 'estimated_erosion': 0.05, # Default to 5% + 'explanation': "Unable to calculate precise risk" + } + + def _calculate_yield_erosion_risk(self, etf_data: Dict) -> Dict: + """ + Calculate yield erosion risk score and explanation + + Args: + etf_data: Dictionary containing ETF data + + Returns: + Dictionary with risk score, estimated erosion, and explanation + """ + try: + # Get relevant metrics + dividend_trend = float(etf_data.get('dividend_trend', 0)) + volatility = float(etf_data.get('volatility', 0)) + max_drawdown = float(etf_data.get('max_drawdown', 0)) + age_years = float(etf_data.get('age_years', 0)) + + # Calculate risk score components + trend_score = min(9, int((1 - min(1, max(-1, dividend_trend))) * 4.5)) # Scale trend to 0-9 + volatility_score = min(9, int(volatility * 20)) # Scale volatility to 0-9 + drawdown_score = min(9, int(max_drawdown * 20)) # Scale drawdown to 0-9 + age_score = min(9, int((5 - min(5, age_years)) * 1.8)) # Scale age to 0-9 + + # Calculate final risk score (weighted average) + risk_score = ( + trend_score * 0.3 + + volatility_score * 0.3 + + drawdown_score * 0.2 + + age_score * 0.2 + ) + + # Estimate annual erosion based on risk score + estimated_erosion = risk_score * 0.01 # 1% per risk point + + # Generate explanation + explanation = [] + if trend_score > 6: + explanation.append(f"Declining dividends ({dividend_trend:.1%})") + if volatility_score > 6: + explanation.append(f"High volatility ({volatility:.1%})") + if drawdown_score > 6: + explanation.append(f"Large drawdowns ({max_drawdown:.1%})") + if age_score > 6: + explanation.append(f"New ETF ({age_years:.1f} years)") + + explanation = ", ".join(explanation) if explanation else "Moderate risk profile" + + return { + 'score': float(risk_score), + 'estimated_erosion': float(estimated_erosion), + 'explanation': explanation + } + + except Exception as e: + logger.error(f"Error calculating yield erosion risk: {str(e)}") + return { + 'score': 5.0, # Default to middle risk + 'estimated_erosion': 0.05, # Default to 5% + 'explanation': "Unable to calculate precise risk" + } \ No newline at end of file