diff --git a/ETF_Portal/services/nav_erosion_service/service.py b/ETF_Portal/services/nav_erosion_service/service.py index 6dce8b9..5b1d9bc 100644 --- a/ETF_Portal/services/nav_erosion_service/service.py +++ b/ETF_Portal/services/nav_erosion_service/service.py @@ -143,6 +143,63 @@ class NavErosionService: payout_ratio = min(yield_ratio / 0.20, 1.0) return payout_ratio + def _calculate_dividend_trend(self, etf_data: Dict) -> Tuple[float, str]: + """ + Calculate dividend trend score and direction. + Returns a tuple of (trend_score, trend_direction) + trend_score: float between -1 and 1 + trend_direction: str ('Increasing', 'Decreasing', 'Stable', 'Unknown') + """ + try: + if not etf_data.get('dividends'): + logger.warning("No dividend data available for trend calculation") + return 0.0, "Unknown" + + # Convert dividends to DataFrame + dividends = pd.DataFrame.from_dict(etf_data['dividends'], orient='index', columns=['Dividends']) + dividends.index = pd.to_datetime(dividends.index) + + # Resample to monthly and calculate rolling averages + monthly_divs = dividends.resample('M')['Dividends'].sum() + if len(monthly_divs) < 6: # Need at least 6 months of data + return 0.0, "Unknown" + + # Calculate 3-month and 6-month moving averages + ma3 = monthly_divs.rolling(window=3).mean() + ma6 = monthly_divs.rolling(window=6).mean() + + # Calculate trend metrics + recent_ma3 = ma3.iloc[-3:].mean() + recent_ma6 = ma6.iloc[-6:].mean() + + # Calculate year-over-year growth + yearly_divs = dividends.resample('Y')['Dividends'].sum() + if len(yearly_divs) >= 2: + yoy_growth = (yearly_divs.iloc[-1] / yearly_divs.iloc[-2]) - 1 + else: + yoy_growth = 0 + + # Calculate trend score (-1 to 1) + ma_trend = (recent_ma3 / recent_ma6) - 1 if recent_ma6 > 0 else 0 + trend_score = (ma_trend * 0.7 + yoy_growth * 0.3) # Weighted combination + + # Normalize trend score to -1 to 1 range + trend_score = max(min(trend_score, 1), -1) + + # Determine trend direction + if abs(trend_score) < 0.05: + direction = "Stable" + elif trend_score > 0: + direction = "Increasing" + else: + direction = "Decreasing" + + return trend_score, direction + + except Exception as e: + logger.error(f"Error calculating dividend trend: {str(e)}") + return 0.0, "Unknown" + def analyze_etf_erosion_risk(self, tickers: List[str]) -> NavErosionAnalysis: """Analyze erosion risk for a list of ETFs.""" results = [] @@ -164,6 +221,9 @@ class NavErosionService: yield_risk, yield_components = self._calculate_yield_risk(etf_data, etf_type) structural_risk, structural_components = self._calculate_structural_risk(etf_data) + # Calculate dividend trend + trend_score, trend_direction = self._calculate_dividend_trend(etf_data) + # Calculate final risk scores final_nav_risk = round( nav_risk * self.NAV_RISK_WEIGHT + @@ -186,14 +246,15 @@ class NavErosionService: yield_risk_explanation=( f"Dividend stability: {yield_components['stability']:.1%}, " f"Growth: {yield_components['growth']:.1%}, " - f"Payout ratio: {yield_components['payout']:.1%}" + f"Payout ratio: {yield_components['payout']:.1%}, " + f"Trend: {trend_direction}" ), etf_age_years=etf_data.get('age_years', 3), max_drawdown=round(etf_data.get('max_drawdown', 0.0), 3), volatility=round(etf_data.get('volatility', 0.0), 3), sharpe_ratio=round(etf_data.get('sharpe_ratio', 0.0), 2), sortino_ratio=round(etf_data.get('sortino_ratio', 0.0), 2), - dividend_trend=round(etf_data.get('dividend_trend', 0.0), 3), + dividend_trend=trend_score, component_risks={ 'nav': nav_components, 'yield': yield_components,