From 0712578445440394ebd72e93f51e0960bcdfb771 Mon Sep 17 00:00:00 2001 From: Pascal Date: Tue, 3 Jun 2025 23:55:53 +0200 Subject: [PATCH] introduce a bug for nav & yield erosion --- .../services/drip_service/no_drip_service.py | 43 ++++++------ ETF_Portal/services/drip_service/service.py | 67 ++++++++++--------- pages/ETF_Portfolio_Builder.py | 2 +- 3 files changed, 61 insertions(+), 51 deletions(-) diff --git a/ETF_Portal/services/drip_service/no_drip_service.py b/ETF_Portal/services/drip_service/no_drip_service.py index 2141408..5143562 100644 --- a/ETF_Portal/services/drip_service/no_drip_service.py +++ b/ETF_Portal/services/drip_service/no_drip_service.py @@ -1,4 +1,5 @@ -from typing import Dict, List, Optional, Tuple, Any +from __future__ import annotations +from typing import Dict, Optional, Tuple, Any import pandas as pd import numpy as np import logging @@ -113,10 +114,9 @@ class NoDRIPService: erosion_config = self._parse_erosion_config(config) # If erosion is requested but no proper erosion_level is provided, calculate it - if (config.erosion_type != "None" and - (not hasattr(config, 'erosion_level') or - not isinstance(config.erosion_level, dict) or - "per_ticker" not in config.erosion_level)): + if (erosion_config.erosion_type != "None" and + (not isinstance(erosion_config.erosion_level, dict) or + "per_ticker" not in erosion_config.erosion_level)): logger.info(f"Calculating erosion rates for No-DRIP with erosion type: {config.erosion_type}") tickers = list(ticker_data.keys()) @@ -238,9 +238,13 @@ class NoDRIPService: def _parse_erosion_config(self, config: DripConfig) -> ErosionConfig: """Parse and validate erosion configuration (reuse from DRIP service)""" - if not hasattr(config, 'erosion_level') or config.erosion_type == "None": + if config.erosion_type == "None": return ErosionConfig(erosion_type="None", erosion_level={}) + if not hasattr(config, 'erosion_level') or config.erosion_level is None: + # erosion_level not provided - will be calculated automatically later + return ErosionConfig(erosion_type=config.erosion_type, erosion_level={}) + # Check if erosion_level is already in the correct format if isinstance(config.erosion_level, dict) and "per_ticker" in config.erosion_level: return ErosionConfig( @@ -253,7 +257,7 @@ class NoDRIPService: erosion_level=config.erosion_level ) - def _calculate_erosion_from_analysis(self, tickers: List[str]) -> Dict: + def _calculate_erosion_from_analysis(self, tickers: list[str]) -> Dict: """Calculate erosion rates using NavErosionService (reuse from DRIP service)""" try: # Use NavErosionService to analyze the tickers @@ -269,10 +273,11 @@ class NoDRIPService: except Exception as e: logger.error(f"Error calculating erosion rates for No-DRIP: {str(e)}") - logger.warning("Falling back to no erosion") - return {"per_ticker": {ticker: {"nav": 0.0, "yield": 0.0} for ticker in tickers}} + logger.warning("Falling back to default erosion rates") + # Use reasonable default erosion rates instead of zero + return {"per_ticker": {ticker: {"nav": 6.0, "yield": 4.0} for ticker in tickers}} - def _create_distribution_schedule(self, ticker_data: Dict[str, TickerData], total_months: int) -> Dict[str, List[int]]: + 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 (reuse from DRIP service)""" schedule = {} @@ -288,7 +293,7 @@ class NoDRIPService: return schedule - def _initialize_simulation_state(self, ticker_data: Dict[str, TickerData]) -> Dict[str, Any]: + def _initialize_simulation_state(self, ticker_data: dict[str, TickerData]) -> dict[str, Any]: """Initialize simulation state variables""" return { 'original_shares': {ticker: data.shares for ticker, data in ticker_data.items()}, # Constant @@ -298,11 +303,11 @@ class NoDRIPService: } def _calculate_monthly_distributions( - self, - month: int, - state: Dict[str, Any], - ticker_data: Dict[str, TickerData], - distribution_schedule: Dict[str, List[int]] + 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 (reuse logic from DRIP service)""" monthly_income = 0.0 @@ -329,9 +334,9 @@ class NoDRIPService: def _apply_monthly_erosion( self, - state: Dict[str, Any], + state: dict[str, Any], erosion_config: ErosionConfig, - tickers: List[str] + tickers: list[str] ) -> None: """Apply monthly erosion to prices and yields""" if erosion_config.erosion_type == "None": @@ -411,7 +416,7 @@ class NoDRIPService: else: return True # Default to monthly for unknown - def _create_no_drip_result(self, monthly_data: List[NoDRIPMonthlyData], state: Dict[str, Any]) -> NoDRIPResult: + def _create_no_drip_result(self, monthly_data: list[NoDRIPMonthlyData], state: dict[str, Any]) -> NoDRIPResult: """Create final No-DRIP result object""" if not monthly_data: raise ValueError("No monthly data generated") diff --git a/ETF_Portal/services/drip_service/service.py b/ETF_Portal/services/drip_service/service.py index 03caed9..e12b1ec 100644 --- a/ETF_Portal/services/drip_service/service.py +++ b/ETF_Portal/services/drip_service/service.py @@ -1,4 +1,5 @@ -from typing import Dict, List, Optional, Tuple, Any +from __future__ import annotations +from typing import Dict, Optional, Tuple, Any import pandas as pd import numpy as np import logging @@ -92,10 +93,9 @@ class DRIPService: erosion_config = self._parse_erosion_config(config) # If erosion is requested but no proper erosion_level is provided, calculate it - if (config.erosion_type != "None" and - (not hasattr(config, 'erosion_level') or - not isinstance(config.erosion_level, dict) or - "per_ticker" not in config.erosion_level)): + if (erosion_config.erosion_type != "None" and + (not isinstance(erosion_config.erosion_level, dict) or + "per_ticker" not in erosion_config.erosion_level)): logger.info(f"Calculating erosion rates for erosion type: {config.erosion_type}") tickers = list(ticker_data.keys()) @@ -218,9 +218,13 @@ class DRIPService: def _parse_erosion_config(self, config: DripConfig) -> ErosionConfig: """Parse and validate erosion configuration""" - if not hasattr(config, 'erosion_level') or config.erosion_type == "None": + if config.erosion_type == "None": return ErosionConfig(erosion_type="None", erosion_level={}) + if not hasattr(config, 'erosion_level') or config.erosion_level is None: + # erosion_level not provided - will be calculated automatically later + return ErosionConfig(erosion_type=config.erosion_type, erosion_level={}) + # Check if erosion_level is already in the correct format if isinstance(config.erosion_level, dict) and "per_ticker" in config.erosion_level: return ErosionConfig( @@ -235,7 +239,7 @@ class DRIPService: erosion_level=config.erosion_level ) - def calculate_erosion_from_analysis(self, tickers: List[str]) -> Dict: + def calculate_erosion_from_analysis(self, tickers: list[str]) -> Dict: """ Calculate erosion rates using NavErosionService @@ -259,15 +263,16 @@ class DRIPService: except Exception as e: logger.error(f"Error calculating erosion rates: {str(e)}") - logger.warning("Falling back to no erosion") - return {"per_ticker": {ticker: {"nav": 0.0, "yield": 0.0} for ticker in tickers}} + logger.warning("Falling back to default erosion rates") + # Use reasonable default erosion rates instead of zero + return {"per_ticker": {ticker: {"nav": 6.0, "yield": 4.0} for ticker in tickers}} def _normalize_erosion_rate(self, erosion_level: float) -> float: """Convert erosion level (0-9) to monthly rate with validation""" rate = (erosion_level / self.MAX_EROSION_LEVEL) * self.MAX_MONTHLY_EROSION return min(max(0.0, rate), self.MAX_MONTHLY_EROSION) - def _create_distribution_schedule(self, ticker_data: Dict[str, TickerData], total_months: int) -> Dict[str, List[int]]: + 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 = {} @@ -283,7 +288,7 @@ class DRIPService: return schedule - def _initialize_simulation_state(self, ticker_data: Dict[str, TickerData]) -> Dict[str, Any]: + 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()}, @@ -293,11 +298,11 @@ class DRIPService: } def _calculate_monthly_distributions( - self, - month: int, - state: Dict[str, Any], - ticker_data: Dict[str, TickerData], - distribution_schedule: Dict[str, List[int]] + 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 @@ -324,9 +329,9 @@ class DRIPService: def _apply_monthly_erosion( self, - state: Dict[str, Any], + state: dict[str, Any], erosion_config: ErosionConfig, - tickers: List[str] + tickers: list[str] ) -> None: """Apply monthly erosion to prices and yields""" if erosion_config.erosion_type == "None": @@ -394,10 +399,10 @@ class DRIPService: logger.info(f" Yield: {original_yield:.2%} -> {state['current_yields'][ticker]:.2%}") def _reinvest_dividends( - self, - month: int, - state: Dict[str, Any], - distribution_schedule: Dict[str, List[int]] + self, + month: int, + state: dict[str, Any], + distribution_schedule: dict[str, list[int]] ) -> None: """Reinvest dividends for tickers that distributed in this month""" @@ -434,7 +439,7 @@ class DRIPService: else: return True # Default to monthly for unknown - def _create_drip_result(self, monthly_data: List[MonthlyData], state: Dict[str, Any]) -> DripResult: + 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") @@ -452,10 +457,10 @@ class DRIPService: # The main calculate_drip_vs_no_drip_comparison method is defined below def forecast_portfolio( - self, - portfolio_df: pd.DataFrame, + self, + portfolio_df: pd.DataFrame, config: DripConfig, - tickers: Optional[List[str]] = None + tickers: Optional[list[str]] = None ) -> DRIPPortfolioResult: """ Forecast DRIP growth for an entire portfolio. @@ -548,7 +553,7 @@ class DRIPService: self, portfolio_df: pd.DataFrame, config: DripConfig - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """ Calculate and compare DRIP vs No-DRIP strategies with detailed analysis. This method runs both simulations and displays comparison tables. @@ -636,10 +641,10 @@ class DRIPService: def _calculate_break_even_analysis( self, strategy_name: str, - monthly_data: List, + monthly_data: list, initial_investment: float, value_extractor: callable - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """ Calculate break-even analysis for a strategy. Break-even occurs when total value exceeds initial investment, @@ -705,8 +710,8 @@ class DRIPService: winner: str, advantage_amount: float, advantage_percentage: float, - drip_break_even: Dict[str, Any], - no_drip_break_even: Dict[str, Any] + drip_break_even: dict[str, Any], + no_drip_break_even: dict[str, Any] ) -> None: """Print detailed strategy comparison table""" diff --git a/pages/ETF_Portfolio_Builder.py b/pages/ETF_Portfolio_Builder.py index f90d83a..abfd6ee 100644 --- a/pages/ETF_Portfolio_Builder.py +++ b/pages/ETF_Portfolio_Builder.py @@ -2103,7 +2103,7 @@ if st.session_state.simulation_run and st.session_state.df_data is not None: config = DripConfig( months=12, erosion_type=st.session_state.get("erosion_type", "Conservative"), - erosion_level={} # Let the service calculate this automatically + erosion_level=None # Let the service calculate this automatically ) # Calculate DRIP vs No-DRIP comparison using the integrated method