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

cleaning unused files

Browse Source
This commit is contained in:
Pascal BIBEHE 2025-06-04 14:35:14 +02:00
parent 27ef418f84
commit c30e89f82c
7 changed files with 0 additions and 776 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
services/drip_service/__init__.py
View File

@ -1,4 +0,0 @@
from .service import DripService
from .models import DripConfig, DripResult, MonthlyData, PortfolioAllocation
__all__ = ['DripService', 'DripConfig', 'DripResult', 'MonthlyData', 'PortfolioAllocation']

23
services/drip_service/logger.py
View File

@ -1,23 +0,0 @@
import logging
import sys
def setup_logger():
# Create logger
logger = logging.getLogger('drip_service')
logger.setLevel(logging.DEBUG)
# Create console handler with formatting
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG)
# Create formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
# Add handler to logger
logger.addHandler(console_handler)
return logger
# Create logger instance
logger = setup_logger()

46
services/drip_service/models.py
View File

@ -1,46 +0,0 @@
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]

455
services/drip_service/service.py
View File

@ -1,455 +0,0 @@
from typing import Dict, List, Optional, Tuple, Any
import pandas as pd
import numpy as np
import traceback
from dataclasses import dataclass, field
from enum import Enum
from .models import PortfolioAllocation, MonthlyData, DripConfig, DripResult
from ..nav_erosion_service import NavErosionService
from .logger import logger
class DistributionFrequency(Enum):
"""Enum for distribution frequencies"""
MONTHLY = ("Monthly", 12)
QUARTERLY = ("Quarterly", 4)
SEMI_ANNUALLY = ("Semi-Annually", 2)
ANNUALLY = ("Annually", 1)
UNKNOWN = ("Unknown", 12)
def __init__(self, name: str, payments_per_year: int):
self.display_name = name
self.payments_per_year = payments_per_year
@dataclass
class TickerData:
"""Data structure for individual ticker information"""
ticker: str
price: float
annual_yield: float
shares: float
allocation_pct: float
distribution_freq: DistributionFrequency
@property
def market_value(self) -> float:
return self.price * self.shares
@property
def monthly_yield(self) -> float:
return self.annual_yield / 12
@property
def distribution_yield(self) -> float:
return self.annual_yield / self.distribution_freq.payments_per_year
class DripService:
"""Enhanced DRIP calculation service with improved performance and accuracy"""
def __init__(self) -> None:
self.DISTRIBUTION_FREQUENCIES = {freq.display_name: freq for freq in DistributionFrequency}
self.nav_erosion_service = NavErosionService()
def calculate_drip_growth(self, portfolio_df: pd.DataFrame, config: DripConfig) -> DripResult:
"""
Calculate DRIP growth for a portfolio over a specified period with enhanced accuracy.
Args:
portfolio_df: DataFrame containing portfolio allocation
config: DripConfig object with simulation parameters
Returns:
DripResult object containing the simulation results
"""
try:
# Validate inputs
self._validate_inputs(portfolio_df, config)
# Get erosion data from nav_erosion_service
erosion_data = self.nav_erosion_service.analyze_etf_erosion_risk(portfolio_df["Ticker"].tolist())
logger.info(f"Erosion data results: {erosion_data.results}")
# Initialize erosion rates dictionary
erosion_rates = {}
# Use erosion rates from nav_erosion_service
for ticker in portfolio_df["Ticker"]:
# Find the result for this ticker in erosion_data.results
result = next((r for r in erosion_data.results if r.ticker == ticker), None)
if result:
erosion_rates[ticker] = {
"nav": result.monthly_nav_erosion_rate,
"yield": result.monthly_yield_erosion_rate
}
logger.info(f"=== EROSION RATE DEBUG ===")
logger.info(f"Ticker: {ticker}")
logger.info(f"Erosion rates from nav_erosion_service:")
logger.info(f" NAV: {erosion_rates[ticker]['nav']:.4%}")
logger.info(f" Yield: {erosion_rates[ticker]['yield']:.4%}")
logger.info(f"=== END EROSION RATE DEBUG ===\n")
else:
# Use default erosion rates if not found
erosion_rates[ticker] = {
"nav": 0.05, # 5% per month (very high, for test)
"yield": 0.07 # 7% per month (very high, for test)
}
logger.info(f"=== EROSION RATE DEBUG ===")
logger.info(f"Ticker: {ticker}")
logger.info(f"Using default erosion rates:")
logger.info(f" NAV: {erosion_rates[ticker]['nav']:.4%}")
logger.info(f" Yield: {erosion_rates[ticker]['yield']:.4%}")
logger.info(f"=== END EROSION RATE DEBUG ===\n")
# Log the final erosion rates dictionary
logger.info(f"Final erosion rates dictionary: {erosion_rates}")
# Initialize portfolio data
ticker_data = self._initialize_ticker_data(portfolio_df)
# Pre-calculate distribution schedule for performance
distribution_schedule = self._create_distribution_schedule(ticker_data, config.months)
# Initialize simulation state
simulation_state = self._initialize_simulation_state(ticker_data)
monthly_data: List[MonthlyData] = []
# Run monthly simulation
for month in range(1, config.months + 1):
logger.info(f"\n=== Starting Month {month} ===")
logger.info(f"Initial state for month {month}:")
for ticker in ticker_data.keys():
logger.info(f" {ticker}:")
logger.info(f" Price: ${simulation_state['current_prices'][ticker]:.2f}")
logger.info(f" Yield: {simulation_state['current_yields'][ticker]:.2%}")
logger.info(f" Shares: {simulation_state['current_shares'][ticker]:.4f}")
month_result = self._simulate_month(
month,
simulation_state,
ticker_data,
erosion_rates,
distribution_schedule
)
monthly_data.append(month_result)
logger.info(f"Final state for month {month}:")
for ticker in ticker_data.keys():
logger.info(f" {ticker}:")
logger.info(f" Price: ${simulation_state['current_prices'][ticker]:.2f}")
logger.info(f" Yield: {simulation_state['current_yields'][ticker]:.2%}")
logger.info(f" Shares: {simulation_state['current_shares'][ticker]:.4f}")
logger.info(f"=== End Month {month} ===\n")
# Calculate final results
return self._create_drip_result(monthly_data, simulation_state)
except Exception as e:
logger.error(f"Error calculating DRIP growth: {str(e)}")
logger.error(traceback.format_exc())
raise
def _validate_inputs(self, portfolio_df: pd.DataFrame, config: DripConfig) -> None:
"""Validate input parameters"""
required_columns = ["Ticker", "Price", "Yield (%)", "Shares"]
missing_columns = [col for col in required_columns if col not in portfolio_df.columns]
if missing_columns:
raise ValueError(f"Missing required columns: {missing_columns}")
if config.months <= 0:
raise ValueError("Months must be positive")
if portfolio_df.empty:
raise ValueError("Portfolio DataFrame is empty")
def _initialize_ticker_data(self, portfolio_df: pd.DataFrame) -> Dict[str, TickerData]:
"""Initialize ticker data with validation"""
ticker_data = {}
for _, row in portfolio_df.iterrows():
ticker = row["Ticker"]
# Handle distribution frequency
dist_period = row.get("Distribution Period", "Monthly")
dist_freq = self.DISTRIBUTION_FREQUENCIES.get(dist_period, DistributionFrequency.MONTHLY)
ticker_data[ticker] = TickerData(
ticker=ticker,
price=max(0.01, float(row["Price"])), # Prevent zero/negative prices
annual_yield=max(0.0, float(row["Yield (%)"] / 100)), # Convert to decimal
shares=max(0.0, float(row["Shares"])),
allocation_pct=float(row.get("Allocation (%)", 0) / 100),
distribution_freq=dist_freq
)
return ticker_data
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 = {}
for ticker, data in ticker_data.items():
distribution_months = []
freq = data.distribution_freq
for month in range(1, total_months + 1):
if self._is_distribution_month(month, freq):
distribution_months.append(month)
schedule[ticker] = distribution_months
return schedule
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()},
'current_prices': {ticker: data.price for ticker, data in ticker_data.items()},
'current_yields': {ticker: data.annual_yield for ticker, data in ticker_data.items()},
'cumulative_income': 0.0
}
def _simulate_month(
self,
month: int,
state: Dict[str, Any],
ticker_data: Dict[str, TickerData],
erosion_rates: Dict[str, Dict[str, float]],
distribution_schedule: Dict[str, List[int]]
) -> MonthlyData:
"""Simulate a single month with improved accuracy"""
# Debug logging for erosion rates
logger.info(f"\n=== EROSION RATES DEBUG ===")
logger.info(f"Erosion rates dictionary: {erosion_rates}")
for ticker, rates in erosion_rates.items():
logger.info(f" {ticker}:")
logger.info(f" nav: {rates['nav']:.4%}")
logger.info(f" yield: {rates['yield']:.4%}")
logger.info(f"=== END EROSION RATES DEBUG ===\n")
# Apply erosion first
for ticker, rates in erosion_rates.items():
if ticker in state['current_prices']:
# Get monthly erosion rates (already in decimal form)
monthly_nav_erosion = rates['nav']
monthly_yield_erosion = rates['yield']
# Get current values
old_price = state['current_prices'][ticker]
old_yield = state['current_yields'][ticker]
# Debug logging
logger.info(f"\n=== EROSION CALCULATION DEBUG ===")
logger.info(f"Ticker: {ticker}")
logger.info(f"Raw erosion rates from nav_erosion_service:")
logger.info(f" monthly_nav_erosion: {monthly_nav_erosion:.4%}")
logger.info(f" monthly_yield_erosion: {monthly_yield_erosion:.4%}")
logger.info(f"Current values:")
logger.info(f" old_price: ${old_price:.4f}")
logger.info(f" old_yield: {old_yield:.4%}")
# Calculate new values
new_price = old_price * (1 - monthly_nav_erosion)
new_yield = old_yield * (1 - monthly_yield_erosion)
logger.info(f"Calculated new values:")
logger.info(f" new_price = ${old_price:.4f} * (1 - {monthly_nav_erosion:.4%})")
logger.info(f" new_price = ${old_price:.4f} * {1 - monthly_nav_erosion:.4f}")
logger.info(f" new_price = ${new_price:.4f}")
logger.info(f" new_yield = {old_yield:.4%} * (1 - {monthly_yield_erosion:.4%})")
logger.info(f" new_yield = {old_yield:.4%} * {1 - monthly_yield_erosion:.4f}")
logger.info(f" new_yield = {new_yield:.4%}")
# Apply the new values with bounds checking
state['current_prices'][ticker] = max(0.01, new_price) # Prevent zero/negative prices
state['current_yields'][ticker] = max(0.0, new_yield) # Prevent negative yields
logger.info(f"Final values after bounds checking:")
logger.info(f" final_price: ${state['current_prices'][ticker]:.4f}")
logger.info(f" final_yield: {state['current_yields'][ticker]:.4%}")
logger.info(f"=== END EROSION CALCULATION DEBUG ===\n")
# Log the actual erosion being applied
logger.info(f"Applied erosion to {ticker}:")
logger.info(f" NAV: {monthly_nav_erosion:.4%} -> New price: ${state['current_prices'][ticker]:.2f}")
logger.info(f" Yield: {monthly_yield_erosion:.4%} -> New yield: {state['current_yields'][ticker]:.2%}")
# Calculate monthly income from distributions using eroded values
monthly_income = self._calculate_monthly_distributions(
month, state, ticker_data, distribution_schedule
)
# Update cumulative income
state['cumulative_income'] += monthly_income
# Reinvest dividends (DRIP)
self._reinvest_dividends(month, state, distribution_schedule)
# Calculate total portfolio value with bounds checking
total_value = 0.0
for ticker in ticker_data.keys():
shares = state['current_shares'][ticker]
price = state['current_prices'][ticker]
if shares > 0 and price > 0:
total_value += shares * price
return MonthlyData(
month=month,
total_value=total_value,
monthly_income=monthly_income,
cumulative_income=state['cumulative_income'],
shares=state['current_shares'].copy(),
prices=state['current_prices'].copy(),
yields=state['current_yields'].copy()
)
def _calculate_monthly_distributions(
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
for ticker, data in ticker_data.items():
if month in distribution_schedule[ticker]:
shares = state['current_shares'][ticker]
price = state['current_prices'][ticker]
yield_rate = state['current_yields'][ticker]
# Calculate distribution amount using annual yield divided by payments per year
distribution_yield = yield_rate / data.distribution_freq.payments_per_year
distribution_amount = shares * price * distribution_yield
monthly_income += distribution_amount
return monthly_income
def _reinvest_dividends(
self,
month: int,
state: Dict[str, Any],
distribution_schedule: Dict[str, List[int]]
) -> None:
"""Reinvest dividends for tickers that distributed in this month"""
for ticker, distribution_months in distribution_schedule.items():
if month in distribution_months:
shares = state['current_shares'][ticker]
price = state['current_prices'][ticker]
yield_rate = state['current_yields'][ticker]
# Calculate dividend income using the correct distribution frequency
freq = self.DISTRIBUTION_FREQUENCIES.get(ticker, DistributionFrequency.MONTHLY)
dividend_income = shares * price * (yield_rate / freq.payments_per_year)
# Purchase additional shares
if price > 0:
new_shares = dividend_income / price
state['current_shares'][ticker] += new_shares
def _is_distribution_month(self, month: int, frequency: DistributionFrequency) -> bool:
"""Check if current month is a distribution month"""
if frequency == DistributionFrequency.MONTHLY:
return True
elif frequency == DistributionFrequency.QUARTERLY:
return month % 3 == 0
elif frequency == DistributionFrequency.SEMI_ANNUALLY:
return month % 6 == 0
elif frequency == DistributionFrequency.ANNUALLY:
return month % 12 == 0
else:
return True # Default to monthly for unknown
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")
final_data = monthly_data[-1]
return DripResult(
monthly_data=monthly_data,
final_portfolio_value=final_data.total_value,
total_income=final_data.cumulative_income,
total_shares=state['current_shares'].copy()
)
# Utility methods for analysis and comparison
def calculate_drip_vs_no_drip_comparison(
self,
portfolio_df: pd.DataFrame,
config: DripConfig
) -> Dict[str, Any]:
"""Calculate comparison between DRIP and no-DRIP scenarios"""
# Calculate DRIP scenario
drip_result = self.calculate_drip_growth(portfolio_df, config)
# Calculate no-DRIP scenario (dividends not reinvested)
no_drip_result = self._calculate_no_drip_scenario(portfolio_df, config)
# Calculate comparison metrics
drip_advantage = drip_result.final_portfolio_value - no_drip_result['final_value']
percentage_advantage = (drip_advantage / no_drip_result['final_value']) * 100
return {
'drip_final_value': drip_result.final_portfolio_value,
'no_drip_final_value': no_drip_result['final_value'],
'drip_advantage': drip_advantage,
'percentage_advantage': percentage_advantage,
'total_dividends_reinvested': drip_result.total_income,
'cash_dividends_no_drip': no_drip_result['total_dividends']
}
def _calculate_no_drip_scenario(self, portfolio_df: pd.DataFrame, config: DripConfig) -> Dict[str, float]:
"""Calculate scenario where dividends are not reinvested"""
ticker_data = self._initialize_ticker_data(portfolio_df)
erosion_data = self.nav_erosion_service.analyze_etf_erosion_risk(portfolio_df["Ticker"].tolist())
erosion_rates = {
result.ticker: {
"nav": result.monthly_nav_erosion_rate,
"yield": result.monthly_yield_erosion_rate
}
for result in erosion_data.results
}
state = self._initialize_simulation_state(ticker_data)
total_dividends = 0.0
for month in range(1, config.months + 1):
# Calculate dividends but don't reinvest
monthly_dividends = self._calculate_monthly_distributions(
month, state, ticker_data,
self._create_distribution_schedule(ticker_data, config.months)
)
total_dividends += monthly_dividends
# Apply erosion
for ticker, rates in erosion_rates.items():
if ticker in state['current_prices']:
# Get monthly erosion rates (already in decimal form)
monthly_nav_erosion = rates['nav']
monthly_yield_erosion = rates['yield']
# Apply NAV erosion (decrease price)
old_price = state['current_prices'][ticker]
new_price = old_price * (1 - monthly_nav_erosion)
state['current_prices'][ticker] = max(0.01, new_price) # Prevent zero/negative prices
# Apply yield erosion (decrease yield)
old_yield = state['current_yields'][ticker]
new_yield = old_yield * (1 - monthly_yield_erosion)
state['current_yields'][ticker] = max(0.0, new_yield) # Prevent negative yields
final_value = sum(
state['current_shares'][ticker] * state['current_prices'][ticker]
for ticker in ticker_data.keys()
)
return {
'final_value': final_value,
'total_dividends': total_dividends
}

8
services/nav_erosion_service/__init__.py
View File

@ -1,8 +0,0 @@
"""
Nav Erosion Service package
"""
from .service import NavErosionService
from .models import NavErosionResult
__all__ = ['NavErosionService', 'NavErosionResult']

31
services/nav_erosion_service/models.py
View File

@ -1,31 +0,0 @@
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

209
services/nav_erosion_service/service.py
View File

@ -1,209 +0,0 @@
from typing import Dict, List, Tuple
from .models import NavErosionConfig, NavErosionResult, NavErosionAnalysis
from enum import Enum
from dataclasses import dataclass
import streamlit as st
class ETFType(Enum):
INCOME = "Income"
GROWTH = "Growth"
BALANCED = "Balanced"
@dataclass
class NavErosionResult:
"""Result of NAV erosion analysis for a single ETF"""
ticker: str
nav_erosion_rate: float # Annual NAV erosion rate
yield_erosion_rate: float # Annual yield erosion rate
monthly_nav_erosion_rate: float # Monthly NAV erosion rate
monthly_yield_erosion_rate: float # Monthly yield erosion rate
risk_level: int
risk_explanation: str
max_drawdown: float
volatility: float
is_new_etf: bool
etf_age_years: float
@dataclass
class NavErosionAnalysis:
"""Complete NAV erosion analysis results"""
results: List[NavErosionResult]
portfolio_nav_risk: float = 0.0
portfolio_erosion_rate: float = 0.0
risk_summary: str = ""
class NavErosionService:
def __init__(self):
self.NAV_COMPONENT_WEIGHTS = {
'drawdown': 0.4,
'volatility': 0.3,
'sharpe': 0.15,
'sortino': 0.15
}
# Default erosion rates based on risk level (0-9)
self.RISK_TO_EROSION = {
0: 0.01, # 1% annual
1: 0.02, # 2% annual
2: 0.03, # 3% annual
3: 0.04, # 4% annual
4: 0.05, # 5% annual
5: 0.06, # 6% annual
6: 0.07, # 7% annual
7: 0.08, # 8% annual
8: 0.09, # 9% annual
9: 0.10 # 10% annual
}
def analyze_etf_erosion_risk(self, tickers: List[str]) -> NavErosionAnalysis:
"""Analyze NAV erosion risk for a list of ETFs"""
results = []
print("\n=== NAV EROSION SERVICE DEBUG ===")
print(f"Session state keys: {st.session_state.keys()}")
print(f"Erosion level from session state: {st.session_state.get('erosion_level')}")
for ticker in tickers:
# Get erosion rates from session state
erosion_level = st.session_state.get('erosion_level', {'nav': 5.0, 'yield': 5.0})
annual_nav_erosion = erosion_level['nav'] / 100 # Convert from percentage to decimal
annual_yield_erosion = erosion_level['yield'] / 100 # Convert from percentage to decimal
# Convert annual rates to monthly
monthly_nav_erosion = 1 - (1 - annual_nav_erosion) ** (1/12)
monthly_yield_erosion = 1 - (1 - annual_yield_erosion) ** (1/12)
print(f"\n=== NAV EROSION SERVICE DEBUG ===")
print(f"Ticker: {ticker}")
print(f"Session State Values:")
print(f" Annual NAV Erosion: {annual_nav_erosion:.4%}")
print(f" Annual Yield Erosion: {annual_yield_erosion:.4%}")
print(f" Monthly NAV Erosion: {monthly_nav_erosion:.4%}")
print(f" Monthly Yield Erosion: {monthly_yield_erosion:.4%}")
print(f"=== END NAV EROSION SERVICE DEBUG ===\n")
result = NavErosionResult(
ticker=ticker,
nav_erosion_rate=annual_nav_erosion,
yield_erosion_rate=annual_yield_erosion,
monthly_nav_erosion_rate=monthly_nav_erosion,
monthly_yield_erosion_rate=monthly_yield_erosion,
risk_level=5, # Arbitrary risk level
risk_explanation="Using erosion rates from session state",
max_drawdown=0.2,
volatility=0.25,
is_new_etf=False,
etf_age_years=1.0
)
results.append(result)
print(f"Created NavErosionResult for {ticker}:")
print(f" monthly_nav_erosion_rate: {result.monthly_nav_erosion_rate:.4%}")
print(f" monthly_yield_erosion_rate: {result.monthly_yield_erosion_rate:.4%}")
# Calculate portfolio-level metrics
portfolio_nav_risk = sum(r.risk_level for r in results) / len(results)
portfolio_erosion_rate = sum(r.nav_erosion_rate for r in results) / len(results)
analysis = NavErosionAnalysis(
results=results,
portfolio_nav_risk=portfolio_nav_risk,
portfolio_erosion_rate=portfolio_erosion_rate,
risk_summary="Portfolio has moderate NAV erosion risk"
)
print("\nFinal NavErosionAnalysis:")
for r in analysis.results:
print(f" {r.ticker}:")
print(f" monthly_nav_erosion_rate: {r.monthly_nav_erosion_rate:.4%}")
print(f" monthly_yield_erosion_rate: {r.monthly_yield_erosion_rate:.4%}")
print("=== END NAV EROSION SERVICE DEBUG ===\n")
return analysis
def _calculate_nav_risk(self, etf_data: Dict, etf_type: ETFType) -> Tuple[float, Dict]:
"""Calculate NAV risk components with ETF-type specific adjustments"""
components = {}
# Base risk calculation with ETF-type specific thresholds
if etf_data.get('max_drawdown') is not None:
if etf_type == ETFType.INCOME:
# Income ETFs typically have lower drawdowns
if etf_data['max_drawdown'] > 0.25:
components['drawdown'] = 7
elif etf_data['max_drawdown'] > 0.15:
components['drawdown'] = 5
elif etf_data['max_drawdown'] > 0.10:
components['drawdown'] = 3
else:
components['drawdown'] = 2
elif etf_type == ETFType.GROWTH:
# Growth ETFs typically have higher drawdowns
if etf_data['max_drawdown'] > 0.35:
components['drawdown'] = 7
elif etf_data['max_drawdown'] > 0.25:
components['drawdown'] = 5
elif etf_data['max_drawdown'] > 0.15:
components['drawdown'] = 3
else:
components['drawdown'] = 2
else: # BALANCED
# Balanced ETFs have moderate drawdowns
if etf_data['max_drawdown'] > 0.30:
components['drawdown'] = 7
elif etf_data['max_drawdown'] > 0.20:
components['drawdown'] = 5
elif etf_data['max_drawdown'] > 0.12:
components['drawdown'] = 3
else:
components['drawdown'] = 2
else:
components['drawdown'] = 4 # Default medium risk if no data
# Rest of the method remains unchanged
if etf_data.get('volatility') is not None:
if etf_data['volatility'] > 0.40:
components['volatility'] = 7
elif etf_data['volatility'] > 0.25:
components['volatility'] = 5
elif etf_data['volatility'] > 0.15:
components['volatility'] = 3
else:
components['volatility'] = 2
else:
components['volatility'] = 4
if etf_data.get('sharpe_ratio') is not None:
if etf_data['sharpe_ratio'] >= 2.0:
components['sharpe'] = 1
elif etf_data['sharpe_ratio'] >= 1.5:
components['sharpe'] = 2
elif etf_data['sharpe_ratio'] >= 1.0:
components['sharpe'] = 3
elif etf_data['sharpe_ratio'] >= 0.5:
components['sharpe'] = 4
else:
components['sharpe'] = 5
else:
components['sharpe'] = 4
if etf_data.get('sortino_ratio') is not None:
if etf_data['sortino_ratio'] >= 2.0:
components['sortino'] = 1
elif etf_data['sortino_ratio'] >= 1.5:
components['sortino'] = 2
elif etf_data['sortino_ratio'] >= 1.0:
components['sortino'] = 3
elif etf_data['sortino_ratio'] >= 0.5:
components['sortino'] = 4
else:
components['sortino'] = 5
else:
components['sortino'] = 4
# Calculate weighted NAV risk
nav_risk = sum(
components[component] * weight
for component, weight in self.NAV_COMPONENT_WEIGHTS.items()
)
return nav_risk, components