From 40cf9aac63a860efc9712b5ec169c826651c7e06 Mon Sep 17 00:00:00 2001 From: Pascal Date: Wed, 28 May 2025 16:12:32 +0200 Subject: [PATCH] feat: Update DRIP Forecast implementation with improved erosion calculations and comparison features --- pages/ETF_Portfolio_Builder.py | 344 ++++++++++++++++++++++++- services/drip_service/__init__.py | 4 + services/drip_service/models.py | 46 ++++ services/drip_service/service.py | 280 ++++++++++++++++++++ services/nav_erosion_service/models.py | 31 +++ 5 files changed, 704 insertions(+), 1 deletion(-) create mode 100644 services/drip_service/__init__.py create mode 100644 services/drip_service/models.py create mode 100644 services/drip_service/service.py create mode 100644 services/nav_erosion_service/models.py diff --git a/pages/ETF_Portfolio_Builder.py b/pages/ETF_Portfolio_Builder.py index 110b3c7..a8df893 100644 --- a/pages/ETF_Portfolio_Builder.py +++ b/pages/ETF_Portfolio_Builder.py @@ -2261,4 +2261,346 @@ if st.session_state.simulation_run and st.session_state.df_data is not None: except Exception as e: st.error(f"Error displaying allocation details: {str(e)}") logger.error(f"Error in allocation display: {str(e)}") - logger.error(traceback.format_exc()) \ No newline at end of file + logger.error(traceback.format_exc()) + + with tab2: + st.subheader("📊 DRIP Forecast") + + # Get erosion settings from session state + erosion_type = st.session_state.get("erosion_type", "None") + erosion_level = st.session_state.get("erosion_level", 0) + + # Create DRIP configuration + from services.drip_service import DripConfig, DripService + + drip_config = DripConfig( + months=12, # Default to 12 months for initial view + erosion_type=erosion_type, + erosion_level=erosion_level + ) + + # Initialize DRIP service and calculate forecast + drip_service = DripService() + drip_result = drip_service.calculate_drip_growth(final_alloc, drip_config) + + # Display portfolio growth chart + st.subheader("Portfolio Growth") + fig1 = go.Figure() + fig1.add_trace(go.Scatter( + x=[data.month for data in drip_result.monthly_data], + y=[data.total_value for data in drip_result.monthly_data], + mode='lines', + name='Portfolio Value', + line=dict(color="#1f77b4", width=3) + )) + fig1.update_layout( + title="Portfolio Value Over Time", + xaxis_title="Month", + yaxis_title="Value ($)", + template="plotly_dark", + xaxis=dict(tickmode='linear', tick0=0, dtick=1) + ) + st.plotly_chart(fig1, use_container_width=True) + + # Display income growth chart + st.subheader("Income Growth") + fig2 = go.Figure() + fig2.add_trace(go.Scatter( + x=[data.month for data in drip_result.monthly_data], + y=[data.monthly_income for data in drip_result.monthly_data], + mode='lines', + name='Monthly Income', + line=dict(color="#ff7f0e", width=3) + )) + fig2.update_layout( + title="Monthly Income Over Time", + xaxis_title="Month", + yaxis_title="Monthly Income ($)", + template="plotly_dark", + xaxis=dict(tickmode='linear', tick0=0, dtick=1) + ) + st.plotly_chart(fig2, use_container_width=True) + + # Display detailed forecast table + st.subheader("DRIP Forecast Details") + + # Convert monthly data to DataFrame for display + forecast_data = [] + for data in drip_result.monthly_data: + row = { + "Month": data.month, + "Total Value ($)": data.total_value, + "Monthly Income ($)": data.monthly_income, + "Cumulative Income ($)": data.cumulative_income + } + # Add ticker-specific data + for ticker, shares in data.shares.items(): + row[f"{ticker} Shares"] = shares + row[f"{ticker} Price ($)"] = data.prices[ticker] + row[f"{ticker} Yield (%)"] = data.yields[ticker] * 100 + forecast_data.append(row) + + forecast_df = pd.DataFrame(forecast_data) + + # Format the data for display + display_forecast = forecast_df.copy() + display_forecast["Total Value ($)"] = display_forecast["Total Value ($)"].apply(lambda x: f"${x:,.2f}") + display_forecast["Monthly Income ($)"] = display_forecast["Monthly Income ($)"].apply(lambda x: f"${x:,.2f}") + display_forecast["Cumulative Income ($)"] = display_forecast["Cumulative Income ($)"].apply(lambda x: f"${x:,.2f}") + + # Format share counts and prices + share_columns = [col for col in display_forecast.columns if "Shares" in col] + price_columns = [col for col in display_forecast.columns if "Price" in col] + yield_columns = [col for col in display_forecast.columns if "Yield (%)" in col] + + for col in share_columns: + display_forecast[col] = display_forecast[col].apply(lambda x: f"{x:.4f}") + + for col in price_columns: + display_forecast[col] = display_forecast[col].apply(lambda x: f"${x:.2f}") + + for col in yield_columns: + display_forecast[col] = display_forecast[col].apply(lambda x: f"{x:.2f}%") + + # Create tabs for different views of the data + detail_tabs = st.tabs(["Summary View", "Full Details"]) + + with detail_tabs[0]: + st.dataframe(display_forecast[["Month", "Total Value ($)", "Monthly Income ($)", "Cumulative Income ($)"]], + use_container_width=True) + + with detail_tabs[1]: + # Group columns by ticker for better readability + tickers = list(drip_result.monthly_data[0].shares.keys()) + ticker_columns = {} + for ticker in tickers: + ticker_columns[ticker] = [ + f"{ticker} Shares", + f"{ticker} Price ($)", + f"{ticker} Yield (%)" + ] + + # Create ordered columns list + basic_columns = ["Month", "Total Value ($)", "Monthly Income ($)", "Cumulative Income ($)"] + ordered_columns = basic_columns.copy() + for ticker in tickers: + ordered_columns.extend(ticker_columns[ticker]) + + st.dataframe( + display_forecast[ordered_columns], + use_container_width=True, + height=500 + ) + + # Add comparison between DRIP and No-DRIP strategies + st.subheader("📊 1-Year DRIP vs. No-DRIP Comparison") + + # Add note about erosion effects if applicable + if erosion_type != "None" and isinstance(erosion_level, dict): + if erosion_level.get("use_per_ticker", False): + st.info(""" + This comparison factors in the custom per-ETF erosion rates. + Both strategies are affected by erosion, but DRIP helps mitigate losses by steadily acquiring more shares. + """) + else: + nav_annual = (1 - (1 - (erosion_level["global"]["nav"] / MAX_EROSION_LEVEL) * max_monthly_erosion)**12) * 100 + yield_annual = (1 - (1 - (erosion_level["global"]["yield"] / MAX_EROSION_LEVEL) * max_monthly_erosion)**12) * 100 + st.info(f""" + This comparison factors in: + - NAV Erosion: {nav_annual:.1f}% annually + - Yield Erosion: {yield_annual:.1f}% annually + + Both strategies are affected by erosion, but DRIP helps mitigate losses by steadily acquiring more shares. + """) + + # Calculate no-drip scenario (taking dividends as income) + initial_value = drip_result.monthly_data[0].total_value + initial_monthly_income = drip_result.monthly_data[0].monthly_income + annual_income = initial_monthly_income * 12 + + # Initialize ticker data dictionary + ticker_data_dict = {} + for _, row in final_alloc.iterrows(): + ticker = row["Ticker"] + ticker_data_dict[ticker] = { + "price": row["Price"], + "yield_annual": row["Yield (%)"] / 100, # Convert from % to decimal + "distribution": row.get("Distribution Period", "Monthly") + } + + # Get the final prices after erosion from the last month of the DRIP forecast + final_prices = {} + for ticker in tickers: + price_col = f"{ticker} Price ($)" + if price_col in forecast_df.columns: + final_prices[ticker] = forecast_df[price_col].iloc[-1] + else: + # Fallback to initial price if column doesn't exist + final_prices[ticker] = ticker_data_dict[ticker]["price"] + + # Extract initial shares for each ETF from month 1 + initial_shares = {ticker: forecast_df.iloc[0][f"{ticker} Shares"] for ticker in tickers} + + # Calculate the No-DRIP final value by multiplying initial shares by final prices + # This correctly accounts for NAV erosion while keeping shares constant + nodrip_final_value = sum(initial_shares[ticker] * final_prices[ticker] for ticker in tickers) + + # The final income should account for erosion but not compounding growth + # This requires simulation of the erosion that would have happened + if erosion_type != "None" and isinstance(erosion_level, dict): + # Initialize the current prices and yields from the final_alloc dataframe + current_prices = {} + current_yields = {} + + # Reconstruct ticker data from final_alloc + for _, row in final_alloc.iterrows(): + ticker = row["Ticker"] + current_prices[ticker] = row["Price"] + current_yields[ticker] = row["Yield (%)"] / 100 + + # Get the erosion rates for each ticker + if erosion_level.get("use_per_ticker", False): + ticker_nav_rates = {} + ticker_yield_rates = {} + for ticker in tickers: + ticker_settings = erosion_level["per_ticker"].get(ticker, {"nav": 0, "yield": 0}) + ticker_nav_rates[ticker] = ticker_settings["nav"] / MAX_EROSION_LEVEL * max_monthly_erosion + ticker_yield_rates[ticker] = ticker_settings["yield"] / MAX_EROSION_LEVEL * max_monthly_erosion + else: + # Use global rates for all tickers + global_nav = erosion_level["global"]["nav"] / MAX_EROSION_LEVEL * max_monthly_erosion + global_yield = erosion_level["global"]["yield"] / MAX_EROSION_LEVEL * max_monthly_erosion + ticker_nav_rates = {ticker: global_nav for ticker in tickers} + ticker_yield_rates = {ticker: global_yield for ticker in tickers} + + # Apply 12 months of erosion + for month in range(1, 13): + # Apply erosion to each ticker + for ticker in tickers: + # Apply NAV erosion + if ticker_nav_rates[ticker] > 0: + current_prices[ticker] *= (1 - ticker_nav_rates[ticker]) + + # Apply yield erosion + if ticker_yield_rates[ticker] > 0: + current_yields[ticker] *= (1 - ticker_yield_rates[ticker]) + + # Calculate final monthly income with eroded prices and yields but original shares + final_monthly_income_nodrip = sum( + (current_yields[ticker] / 12) * + (initial_shares[ticker] * current_prices[ticker]) + for ticker in tickers + ) + else: + # No erosion, so final income is the same as initial income + final_monthly_income_nodrip = initial_monthly_income + + nodrip_final_annual_income = final_monthly_income_nodrip * 12 + + # Get values for DRIP scenario from forecast + drip_final_value = drip_result.final_portfolio_value + drip_final_monthly_income = drip_result.monthly_data[-1].monthly_income + drip_annual_income_end = drip_final_monthly_income * 12 + + # Create comparison dataframe with withdrawn income for a more complete financial picture + + # For No-DRIP strategy, calculate cumulative withdrawn income (sum of monthly dividends) + # This is equivalent to the cumulative income in the DRIP forecast, but in No-DRIP it's withdrawn + withdrawn_income = 0 + monthly_dividends = [] + + # Reconstruct the monthly dividend calculation for No-DRIP + current_prices_monthly = {ticker: ticker_data_dict[ticker]["price"] for ticker in tickers} + current_yields_monthly = {ticker: ticker_data_dict[ticker]["yield_annual"] for ticker in tickers} + + for month in range(1, 13): + # Calculate dividends for this month based on current yields and prices + month_dividend = sum( + (current_yields_monthly[ticker] / 12) * + (initial_shares[ticker] * current_prices_monthly[ticker]) + for ticker in tickers + ) + withdrawn_income += month_dividend + monthly_dividends.append(month_dividend) + + # Apply erosion for next month + if erosion_type != "None": + for ticker in tickers: + # Apply NAV erosion + if ticker in ticker_nav_rates and ticker_nav_rates[ticker] > 0: + current_prices_monthly[ticker] *= (1 - ticker_nav_rates[ticker]) + + # Apply yield erosion + if ticker in ticker_yield_rates and ticker_yield_rates[ticker] > 0: + current_yields_monthly[ticker] *= (1 - ticker_yield_rates[ticker]) + + # Calculate total economic result + nodrip_total = nodrip_final_value + withdrawn_income + drip_total = drip_final_value + + # Display economic comparison + st.subheader("Economic Comparison") + + col1, col2 = st.columns(2) + + with col1: + st.markdown("#### No-DRIP KPIs") + st.metric("Final Portfolio Value", f"${nodrip_final_value:,.2f}") + st.metric("Total Income Withdrawn", f"${withdrawn_income:,.2f}") + st.metric("Final Monthly Income", f"${final_monthly_income_nodrip:,.2f}") + st.metric("Total Economic Result", f"${nodrip_total:,.2f}") + + with col2: + st.markdown("#### DRIP KPIs") + st.metric("Final Portfolio Value", f"${drip_final_value:,.2f}") + st.metric("Final Monthly Income", f"${drip_final_monthly_income:,.2f}") + st.metric("Total Economic Result", f"${drip_total:,.2f}") + + # Display monthly income comparison chart + st.subheader("Monthly Income Comparison") + + # Create monthly income comparison chart + fig3 = go.Figure() + + # Add No-DRIP monthly income line + fig3.add_trace(go.Scatter( + x=list(range(1, 13)), + y=monthly_dividends, + mode='lines', + name='No-DRIP Monthly Income', + line=dict(color='#ff7f0e', width=2) + )) + + # Add DRIP monthly income line + drip_monthly_income = [data.monthly_income for data in drip_result.monthly_data] + fig3.add_trace(go.Scatter( + x=list(range(1, 13)), + y=drip_monthly_income, + mode='lines', + name='DRIP Monthly Income', + line=dict(color='#1f77b4', width=2) + )) + + fig3.update_layout( + title="Monthly Income: DRIP vs No-DRIP Strategy", + xaxis_title="Month", + yaxis_title="Monthly Income ($)", + template="plotly_dark", + showlegend=True + ) + + st.plotly_chart(fig3, use_container_width=True) + + # Display advantages of DRIP strategy + st.subheader("DRIP Strategy Advantages") + + # Calculate percentage differences + value_diff_pct = ((drip_final_value - nodrip_final_value) / nodrip_final_value) * 100 + income_diff_pct = ((drip_final_monthly_income - final_monthly_income_nodrip) / final_monthly_income_nodrip) * 100 + + st.markdown(f""" + - **Portfolio Value**: DRIP strategy results in a {value_diff_pct:.1f}% higher final portfolio value + - **Monthly Income**: DRIP strategy generates {income_diff_pct:.1f}% higher monthly income + - **Risk Mitigation**: DRIP helps mitigate erosion effects by continuously acquiring more shares + - **Compounding Effect**: Reinvested dividends generate additional income through compounding + """) \ No newline at end of file diff --git a/services/drip_service/__init__.py b/services/drip_service/__init__.py new file mode 100644 index 0000000..6f00234 --- /dev/null +++ b/services/drip_service/__init__.py @@ -0,0 +1,4 @@ +from .service import DripService +from .models import DripConfig, DripResult, MonthlyData, PortfolioAllocation + +__all__ = ['DripService', 'DripConfig', 'DripResult', 'MonthlyData', 'PortfolioAllocation'] \ No newline at end of file diff --git a/services/drip_service/models.py b/services/drip_service/models.py new file mode 100644 index 0000000..6434058 --- /dev/null +++ b/services/drip_service/models.py @@ -0,0 +1,46 @@ +from dataclasses import dataclass +from typing import Dict, List, Optional +from datetime import datetime + +@dataclass +class PortfolioAllocation: + ticker: str + price: float + yield_annual: float + initial_shares: float + initial_allocation: float + distribution: str + +@dataclass +class MonthlyData: + month: int + total_value: float + monthly_income: float + cumulative_income: float + shares: Dict[str, float] + prices: Dict[str, float] + yields: Dict[str, float] + +@dataclass +class DripConfig: + months: int + erosion_type: str + erosion_level: Dict + dividend_frequency: Dict[str, int] = None + + def __post_init__(self): + if self.dividend_frequency is None: + self.dividend_frequency = { + "Monthly": 12, + "Quarterly": 4, + "Semi-Annually": 2, + "Annually": 1, + "Unknown": 12 # Default to monthly if unknown + } + +@dataclass +class DripResult: + monthly_data: List[MonthlyData] + final_portfolio_value: float + total_income: float + total_shares: Dict[str, float] \ No newline at end of file diff --git a/services/drip_service/service.py b/services/drip_service/service.py new file mode 100644 index 0000000..d9bc085 --- /dev/null +++ b/services/drip_service/service.py @@ -0,0 +1,280 @@ +from typing import Dict, List, Optional, Tuple, Any +import pandas as pd +import logging +import traceback +from .models import PortfolioAllocation, MonthlyData, DripConfig, DripResult + +# Configure logging +logger = logging.getLogger(__name__) + +class DripService: + def __init__(self) -> None: + self.MAX_EROSION_LEVEL = 9 + self.max_monthly_erosion = 1 - (0.1)**(1/12) # ~17.54% monthly for 90% annual erosion + self.dividend_frequency = { + "Monthly": 12, + "Quarterly": 4, + "Semi-Annually": 2, + "Annually": 1, + "Unknown": 12 # Default to monthly if unknown + } + + def calculate_drip_growth(self, portfolio_df: pd.DataFrame, config: DripConfig) -> DripResult: + """ + Calculate DRIP growth for a portfolio over a specified period. + + Args: + portfolio_df: DataFrame containing portfolio allocation + config: DripConfig object with simulation parameters + + Returns: + DripResult object containing the simulation results + """ + try: + # Initialize monthly data list + monthly_data: List[MonthlyData] = [] + + # Get initial values + initial_shares = self._calculate_initial_shares(portfolio_df) + initial_prices = dict(zip(portfolio_df["Ticker"], portfolio_df["Price"])) + initial_yields = dict(zip(portfolio_df["Ticker"], portfolio_df["Yield (%)"] / 100)) + + # Initialize tracking variables + current_shares = initial_shares.copy() + current_prices = initial_prices.copy() + current_yields = initial_yields.copy() + cumulative_income = 0.0 + + # Run simulation for each month + for month in range(1, config.months + 1): + # Calculate monthly income + monthly_income = sum( + (current_yields[ticker] / 12) * + (current_shares[ticker] * current_prices[ticker]) + for ticker in current_shares.keys() + ) + + # Update cumulative income + cumulative_income += monthly_income + + # Calculate total portfolio value + total_value = sum( + current_shares[ticker] * current_prices[ticker] + for ticker in current_shares.keys() + ) + + # Apply erosion if enabled + if config.erosion_type != "None": + current_prices, current_yields = self._apply_erosion( + current_prices, + current_yields, + config.erosion_type, + config.erosion_level + ) + + # Reinvest dividends + for ticker in current_shares.keys(): + dividend_income = (current_yields[ticker] / 12) * (current_shares[ticker] * current_prices[ticker]) + new_shares = dividend_income / current_prices[ticker] + current_shares[ticker] += new_shares + + # Store monthly data + monthly_data.append(MonthlyData( + month=month, + total_value=total_value, + monthly_income=monthly_income, + cumulative_income=cumulative_income, + shares=current_shares.copy(), + prices=current_prices.copy(), + yields=current_yields.copy() + )) + + # Calculate final values + final_portfolio_value = monthly_data[-1].total_value + total_income = monthly_data[-1].cumulative_income + total_shares = current_shares.copy() + + return DripResult( + monthly_data=monthly_data, + final_portfolio_value=final_portfolio_value, + total_income=total_income, + total_shares=total_shares + ) + + except Exception as e: + logger.error(f"Error calculating DRIP growth: {str(e)}") + logger.error(traceback.format_exc()) + raise + + def _calculate_initial_shares(self, portfolio_df: pd.DataFrame) -> Dict[str, float]: + """Calculate initial shares for each ETF.""" + return dict(zip(portfolio_df["Ticker"], portfolio_df["Shares"])) + + def _apply_erosion( + self, + prices: Dict[str, float], + yields: Dict[str, float], + erosion_type: str, + erosion_level: Dict[str, Any] + ) -> Tuple[Dict[str, float], Dict[str, float]]: + """ + Apply erosion to prices and yields based on configuration. + + Args: + prices: Dictionary of current prices + yields: Dictionary of current yields + erosion_type: Type of erosion to apply + erosion_level: Dictionary containing erosion levels + + Returns: + Tuple of updated prices and yields + """ + try: + updated_prices = prices.copy() + updated_yields = yields.copy() + + if erosion_type == "None": + return updated_prices, updated_yields + + if erosion_level.get("use_per_ticker", False): + # Apply per-ticker erosion rates + for ticker in prices.keys(): + if ticker in erosion_level: + ticker_erosion = erosion_level[ticker] + # Apply monthly erosion + nav_erosion = (ticker_erosion["nav"] / 9) * 0.1754 + yield_erosion = (ticker_erosion["yield"] / 9) * 0.1754 + + updated_prices[ticker] *= (1 - nav_erosion) + updated_yields[ticker] *= (1 - yield_erosion) + else: + # Apply global erosion rates + nav_erosion = (erosion_level["global"]["nav"] / 9) * 0.1754 + yield_erosion = (erosion_level["global"]["yield"] / 9) * 0.1754 + + for ticker in prices.keys(): + updated_prices[ticker] *= (1 - nav_erosion) + updated_yields[ticker] *= (1 - yield_erosion) + + return updated_prices, updated_yields + + except Exception as e: + logger.error(f"Error applying erosion: {str(e)}") + logger.error(traceback.format_exc()) + return prices, yields + + def _initialize_erosion_rates( + self, + tickers: List[str], + erosion_type: str, + erosion_level: Dict[str, Any] + ) -> Tuple[Dict[str, float], Dict[str, float]]: + """Initialize erosion rates for each ticker based on configuration.""" + ticker_nav_rates: Dict[str, float] = {} + ticker_yield_rates: Dict[str, float] = {} + + if erosion_type != "None" and isinstance(erosion_level, dict): + if erosion_level.get("use_per_ticker", False) and "per_ticker" in erosion_level: + global_nav = erosion_level["global"]["nav"] / self.MAX_EROSION_LEVEL * self.max_monthly_erosion + global_yield = erosion_level["global"]["yield"] / self.MAX_EROSION_LEVEL * self.max_monthly_erosion + + for ticker in tickers: + ticker_settings = erosion_level["per_ticker"].get(ticker, {"nav": 0, "yield": 0}) + ticker_nav_rates[ticker] = ticker_settings["nav"] / self.MAX_EROSION_LEVEL * self.max_monthly_erosion + ticker_yield_rates[ticker] = ticker_settings["yield"] / self.MAX_EROSION_LEVEL * self.max_monthly_erosion + else: + global_nav = erosion_level["global"]["nav"] / self.MAX_EROSION_LEVEL * self.max_monthly_erosion + global_yield = erosion_level["global"]["yield"] / self.MAX_EROSION_LEVEL * self.max_monthly_erosion + + for ticker in tickers: + ticker_nav_rates[ticker] = global_nav + ticker_yield_rates[ticker] = global_yield + else: + for ticker in tickers: + ticker_nav_rates[ticker] = 0 + ticker_yield_rates[ticker] = 0 + + return ticker_nav_rates, ticker_yield_rates + + def _create_ticker_data(self, portfolio_df: pd.DataFrame) -> Dict[str, Dict[str, Any]]: + """Create a dictionary of ticker-specific data.""" + ticker_data: Dict[str, Dict[str, Any]] = {} + for _, row in portfolio_df.iterrows(): + ticker = row["Ticker"] + ticker_data[ticker] = { + "price": row["Price"], + "yield_annual": row["Yield (%)"] / 100, + "initial_shares": row["Capital Allocated ($)"] / row["Price"], + "initial_allocation": row["Allocation (%)"] / 100, + "distribution": row.get("Distribution Period", "Monthly") + } + return ticker_data + + def _calculate_monthly_income( + self, + current_shares: Dict[str, float], + current_prices: Dict[str, float], + current_yields: Dict[str, float], + tickers: List[str] + ) -> float: + """Calculate expected monthly income based on current portfolio and yields.""" + return sum( + (current_yields[ticker] / 12) * + (current_shares[ticker] * current_prices[ticker]) + for ticker in tickers + ) + + def _create_month_data( + self, + month: int, + current_total_value: float, + monthly_income: float, + cumulative_income: float, + current_shares: Dict[str, float], + current_prices: Dict[str, float], + current_yields: Dict[str, float], + tickers: List[str] + ) -> MonthlyData: + """Create monthly data object.""" + return MonthlyData( + month=month, + total_value=current_total_value, + monthly_income=monthly_income, + cumulative_income=cumulative_income, + shares=current_shares.copy(), + prices=current_prices.copy(), + yields=current_yields.copy() + ) + + def _calculate_and_reinvest_dividends( + self, + month: int, + ticker_data: Dict[str, Dict[str, Any]], + current_shares: Dict[str, float], + current_prices: Dict[str, float], + current_yields: Dict[str, float], + ticker_yield_rates: Dict[str, float], + dividend_frequency: Dict[str, int] + ) -> float: + """Calculate dividends and reinvest them proportionally.""" + month_dividends: Dict[str, float] = {} + for ticker, data in ticker_data.items(): + freq = dividend_frequency[data["distribution"]] + if month % (12 / freq) == 0: + if ticker_yield_rates[ticker] > 0: + dividend = (current_yields[ticker] / freq) * current_shares[ticker] * current_prices[ticker] + else: + dividend = (data["yield_annual"] / freq) * current_shares[ticker] * current_prices[ticker] + else: + dividend = 0 + month_dividends[ticker] = dividend + + total_month_dividend = sum(month_dividends.values()) + + # Reinvest dividends proportionally + for ticker, data in ticker_data.items(): + if current_prices[ticker] > 0: + new_shares = (total_month_dividend * data["initial_allocation"]) / current_prices[ticker] + current_shares[ticker] += new_shares + + return total_month_dividend \ No newline at end of file diff --git a/services/nav_erosion_service/models.py b/services/nav_erosion_service/models.py new file mode 100644 index 0000000..62265bd --- /dev/null +++ b/services/nav_erosion_service/models.py @@ -0,0 +1,31 @@ +from dataclasses import dataclass +from typing import Dict, List, Optional +from datetime import datetime + +@dataclass +class NavErosionConfig: + max_erosion_level: int = 9 + max_monthly_erosion: float = 1 - (0.1)**(1/12) # ~17.54% monthly for 90% annual erosion + use_per_ticker: bool = False + global_nav_rate: float = 0 + per_ticker_rates: Dict[str, float] = None + +@dataclass +class NavErosionResult: + ticker: str + nav_erosion_rate: float + monthly_erosion_rate: float + annual_erosion_rate: float + risk_level: int # 0-9 scale + risk_explanation: str + max_drawdown: float + volatility: float + is_new_etf: bool + etf_age_years: Optional[float] + +@dataclass +class NavErosionAnalysis: + results: List[NavErosionResult] + portfolio_nav_risk: float # Average risk level + portfolio_erosion_rate: float # Weighted average erosion rate + risk_summary: str \ No newline at end of file