mlnsio/text2sql
0
1"""2This module defines a class, MFRating, which provides methods for calculating3the weighted rating and overall score for mutual funds based on various parameters.4 5"""6import logging7from typing import List, Dict, Any8import numpy as np9from django.db.models import Max, Min10from core.models import MutualFund, Stock11 12 13logger = logging.getLogger(__name__)14 15 16class MFRating:17 """18 This class provides methods for calculating the weighted stock rank rating and overall score for mutual funds based on various parameters.19 """20 21 def __init__(self, max_rank: int = 1000) -> None:22 self.max_rank = max_rank23 self.scores = {24 "stock_ranking_score": [10],25 "crisil_rank_score": [10],26 "churn_score": [10],27 "sharperatio_score": [10],28 "expenseratio_score": [10],29 "aum_score": [10],30 "alpha_score": [10],31 "beta_score": [10],32 }33 34 def get_weighted_score(self, values: List[float]) -> float:35 """36 Calculates the weighted rating based on the weights and values provided.37 """38 weights = []39 values = []40 for _, (weight, score) in self.scores.items():41 weights.append(weight)42 values.append(score)43 44 return np.average(values, weights=weights)45 46 def get_rank_rating(self, stock_ranks: List[int]) -> List[float]:47 """48 Calculates the rank rating based on the stock ranks and the maximum rank.49 """50 return [51 (self.max_rank - (rank if rank else self.max_rank)) / self.max_rank52 for rank in stock_ranks53 ]54 55 def get_overall_score(self, **kwargs) -> float:56 """57 It returns the overall weighted score for mutual funds based on various parameters.58 59 """60 61 stock_rankings = self.get_rank_rating(kwargs.get("stock_rankings"))62 # what np.average do?63 # Multiply each element in the stock_rankings array by its corresponding weights, then Sum up the results, then divide by the sum of the weights.64 # data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]65 # weights = [10, 9, 8, 7, 6, 5, 4, 3, 2, 1]66 #67 # Multiply each element in the data array by its corresponding weight:68 # [1*10, 2*9, 3*8, 4*7, 5*6, 6*5, 7*4, 8*3, 9*2, 10*1]69 # [10, 18, 24, 28, 30, 30, 28, 24, 18, 10]70 #71 # Sum up the results:72 # 10 + 18 + 24 + 28 + 30 + 30 + 28 + 24 + 18 + 10 = 22073 #74 # Sum up the weights:75 # 10 + 9 + 8 + 7 + 6 + 5 + 4 + 3 + 2 + 1 = 5576 #77 # Divide the sum of the weighted elements by the sum of the weights:78 # 220 / 55 = 4.079 self.scores["stock_ranking_score"].append(80 np.average(stock_rankings, weights=kwargs.get("stock_weights"))81 )82 self.scores["alpha_score"].append(kwargs.get("alpha", 0) / 100)83 self.scores["beta_score"].append((2 - kwargs.get("beta", 2)) / 2)84 self.scores["crisil_rank_score"].append(85 (kwargs.get("crisil_rank_score", 0)) / 586 )87 self.scores["churn_score"].append(kwargs.get("churn_rate", 0) / 100)88 self.scores["sharperatio_score"].append(kwargs.get("sharpe_ratio", 0) / 100)89 self.scores["expenseratio_score"].append(kwargs.get("expense_ratio", 0) / 100)90 max_aum, min_aum, aum = kwargs.get("aum_score", (1, 0, 0))91 self.scores["aum_score"].append((aum - min_aum) / (max_aum - min_aum))92 # Calculate the overall rating using weighted sum93 94 return self.get_weighted_score(self.scores)95 96 97class MutualFundScorer:98 def __init__(self) -> None:99 self.mf_scores = []100 101 def _get_stock_ranks(self, isin_ids: List[str]) -> List[int]:102 """Get stock ranks based on ISIN ids."""103 104 return list(105 Stock.objects.filter(isin_number__in=isin_ids)106 .order_by("rank")107 .values_list("rank", "isin_number")108 )109 110 def _get_mutual_funds(self) -> List[MutualFund]:111 """Get a list of top 30 mutual funds based on rank."""112 113 return MutualFund.objects.exclude(rank=None).order_by("rank")[:30]114 115 def _get_risk_measure(116 self, risk_measures: Dict[str, Any], key: str, year: str117 ) -> float:118 """119 Get value of the specified key from the risk_measures dictionary for the given year.120 """121 try:122 value = risk_measures.get(year, {}).get(key, 0)123 return float(value)124 except (TypeError, ValueError):125 return 0126 127 def _get_most_non_null_key(self, key, mutual_funds):128 """129 Get the year with the maximum number of non-None values for the specified key130 within the given mutual funds.131 """132 year_counts = {133 "for15Year": 0,134 "for10Year": 0,135 "for5Year": 0,136 "for3Year": 0,137 "for1Year": 0,138 }139 140 for mf in mutual_funds:141 risk_measures = mf.data["risk_measures"].get("fundRiskVolatility", {})142 143 for year in year_counts:144 if risk_measures.get(year, {}).get(key) is not None:145 year_counts[year] += 1146 147 most_non_null_year = max(year_counts, key=year_counts.get)148 return most_non_null_year149 150 def get_scores(self) -> List[Dict[str, Any]]:151 """Calculate scores for mutual funds and return the results."""152 153 logger.info("Calculating scores for mutual funds...")154 max_aum = MutualFund.objects.exclude(rank=None).aggregate(max_price=Max("aum"))[155 "max_price"156 ]157 min_aum = MutualFund.objects.exclude(rank=None).aggregate(min_price=Min("aum"))[158 "min_price"159 ]160 mutual_funds = self._get_mutual_funds()161 162 # Get the year with the maximum number of non-None values for sharpeRatio, alpha and beta163 sharpe_ratio_year = self._get_most_non_null_key("sharpeRatio", mutual_funds)164 alpha_year = self._get_most_non_null_key("alpha", mutual_funds)165 beta_year = self._get_most_non_null_key("beta", mutual_funds)166 for mf in mutual_funds:167 mf_rating = MFRating(168 max_rank=1000,169 )170 logger.info(f"Processing mutual fund: %s", mf.fund_name)171 holdings = (172 mf.data.get("holdings", {})173 .get("equityHoldingPage", {})174 .get("holdingList", [])175 )176 portfolio_holding_weights = {177 holding.get("isin"): (178 holding.get("weighting") if holding.get("weighting") else 0179 )180 for holding in holdings181 if holding.get("isin")182 }183 stock_ranks_and_weights = [184 (rank, portfolio_holding_weights[isin])185 for rank, isin in self._get_stock_ranks(186 portfolio_holding_weights.keys()187 )188 ]189 stock_ranks, stock_weights = zip(*stock_ranks_and_weights)190 sharpe_ratio = self._get_risk_measure(191 mf.data["risk_measures"].get("fundRiskVolatility", {}),192 "sharpeRatio",193 sharpe_ratio_year,194 )195 alpha = self._get_risk_measure(196 mf.data["risk_measures"].get("fundRiskVolatility", {}),197 "alpha",198 alpha_year,199 )200 beta = self._get_risk_measure(201 mf.data["risk_measures"].get("fundRiskVolatility", {}),202 "beta",203 beta_year,204 )205 overall_score = mf_rating.get_overall_score(206 stock_rankings=stock_ranks,207 stock_weights=stock_weights,208 churn_rate=mf.data["quotes"]["lastTurnoverRatio"]209 if mf.data["quotes"].get("lastTurnoverRatio")210 else 0,211 sharpe_ratio=sharpe_ratio,212 expense_ratio=mf.data["quotes"]["expenseRatio"],213 crisil_rank_score=mf.crisil_rank,214 aum_score=(max_aum, min_aum, mf.aum),215 alpha=alpha,216 beta=beta,217 )218 219 self.mf_scores.append(220 {221 "isin": mf.isin_number,222 "name": mf.fund_name,223 "rank": mf.rank,224 "sharpe_ratio": round(sharpe_ratio, 4),225 "churn_rate": mf.data["quotes"].get("lastTurnoverRatio", 0),226 "expense_ratio": mf.data["quotes"].get("expenseRatio", 0),227 "aum": mf.aum,228 "alpha": round(alpha, 4),229 "beta": round(beta, 4),230 "crisil_rank": mf.crisil_rank,231 "overall_score": round(overall_score, 4),232 }233 )234 logger.info("Finished calculating scores.")235 return sorted(self.mf_scores, key=lambda d: d["overall_score"], reverse=True)236