Pascal/ETF_Suite_Portal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

introduce a bug for nav & yield erosion

Browse Source
This commit is contained in:
Pascal BIBEHE 2025-06-03 23:55:53 +02:00
parent 30dc087ce3
commit 0712578445
3 changed files with 61 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
ETF_Portal/services/drip_service/no_drip_service.py
View File

@ -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")

67
ETF_Portal/services/drip_service/service.py
View File

@ -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"""

2
pages/ETF_Portfolio_Builder.py
View File

@ -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