Spaces:
Running
Running
File size: 4,659 Bytes
eb606e1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 |
import pandas as pd
from datetime import datetime
from typing import Optional, Dict, Any, List
from app.models.goal_models import SIPCalculationRequest, SIPCalculationResponse, RequiredSIPRequest, RequiredSIPResponse
class SIPCalculator:
"""Calculate SIP returns and goal-based investments"""
@staticmethod
def calculate_sip_maturity(monthly_amount: float, annual_return: float, years: int) -> float:
"""Calculate SIP maturity amount"""
if annual_return == 0:
return monthly_amount * years * 12
monthly_return = annual_return / 12 / 100
months = years * 12
maturity_amount = monthly_amount * (
((1 + monthly_return) ** months - 1) / monthly_return
) * (1 + monthly_return)
return maturity_amount
@staticmethod
def calculate_required_sip(target_amount: float, years: int, expected_return: float) -> float:
"""Calculate required SIP amount for a target"""
if expected_return == 0:
return target_amount / (years * 12)
monthly_return = expected_return / 12 / 100
months = years * 12
required_sip = target_amount / (
((1 + monthly_return) ** months - 1) / monthly_return
) / (1 + monthly_return)
return required_sip
@staticmethod
def calculate_fund_returns(nav_data: pd.DataFrame, investment_amount: float,
start_date: datetime, end_date: datetime) -> Optional[Dict[str, Any]]:
"""Calculate returns for a fund investment"""
try:
filtered_data = nav_data[
(nav_data['date'] >= pd.to_datetime(start_date)) &
(nav_data['date'] <= pd.to_datetime(end_date))
]
if len(filtered_data) < 2:
return None
start_nav = filtered_data.iloc[0]['nav']
end_nav = filtered_data.iloc[-1]['nav']
units = investment_amount / start_nav
final_value = units * end_nav
total_return = ((final_value - investment_amount) / investment_amount) * 100
return {
'start_nav': start_nav,
'end_nav': end_nav,
'units': units,
'final_value': final_value,
'total_return': total_return,
'investment_amount': investment_amount
}
except Exception as e:
print(f"Error calculating returns: {e}")
return None
@staticmethod
def get_sip_calculation(request: SIPCalculationRequest, include_yearly_breakdown: bool = False) -> SIPCalculationResponse:
"""Get SIP calculation with optional yearly breakdown"""
maturity_amount = SIPCalculator.calculate_sip_maturity(
request.monthly_amount, request.annual_return, request.years
)
total_invested = request.monthly_amount * request.years * 12
gains = maturity_amount - total_invested
return_multiple = maturity_amount / total_invested if total_invested > 0 else 0
yearly_breakdown = None
if include_yearly_breakdown:
yearly_breakdown = []
invested_cumulative = 0
for year in range(1, request.years + 1):
maturity_year = SIPCalculator.calculate_sip_maturity(
request.monthly_amount, request.annual_return, year
)
invested_year = request.monthly_amount * year * 12
yearly_breakdown.append({
'Year': year,
'Invested': invested_year,
'Maturity Value': maturity_year,
'Gains': maturity_year - invested_year
})
return SIPCalculationResponse(
maturity_amount=maturity_amount,
total_invested=total_invested,
gains=gains,
return_multiple=return_multiple,
yearly_breakdown=yearly_breakdown
)
@staticmethod
def get_required_sip(request: RequiredSIPRequest) -> RequiredSIPResponse:
"""Get required SIP amount for a target"""
required_sip = SIPCalculator.calculate_required_sip(
request.target_amount, request.years, request.expected_return
)
return RequiredSIPResponse(required_sip=required_sip) |