File size: 8,989 Bytes
38171fa |
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 |
"""
This module defines a class, MFRating, which provides methods for calculating
the weighted rating and overall score for mutual funds based on various parameters.
"""
import logging
from typing import List, Dict, Any
import numpy as np
from django.db.models import Max, Min
from core.models import MutualFund, Stock
logger = logging.getLogger(__name__)
class MFRating:
"""
This class provides methods for calculating the weighted stock rank rating and overall score for mutual funds based on various parameters.
"""
def __init__(self, max_rank: int = 1000) -> None:
self.max_rank = max_rank
self.scores = {
"stock_ranking_score": [10],
"crisil_rank_score": [10],
"churn_score": [10],
"sharperatio_score": [10],
"expenseratio_score": [10],
"aum_score": [10],
"alpha_score": [10],
"beta_score": [10],
}
def get_weighted_score(self, values: List[float]) -> float:
"""
Calculates the weighted rating based on the weights and values provided.
"""
weights = []
values = []
for _, (weight, score) in self.scores.items():
weights.append(weight)
values.append(score)
return np.average(values, weights=weights)
def get_rank_rating(self, stock_ranks: List[int]) -> List[float]:
"""
Calculates the rank rating based on the stock ranks and the maximum rank.
"""
return [
(self.max_rank - (rank if rank else self.max_rank)) / self.max_rank
for rank in stock_ranks
]
def get_overall_score(self, **kwargs) -> float:
"""
It returns the overall weighted score for mutual funds based on various parameters.
"""
stock_rankings = self.get_rank_rating(kwargs.get("stock_rankings"))
# what np.average do?
# 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.
# data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# weights = [10, 9, 8, 7, 6, 5, 4, 3, 2, 1]
#
# Multiply each element in the data array by its corresponding weight:
# [1*10, 2*9, 3*8, 4*7, 5*6, 6*5, 7*4, 8*3, 9*2, 10*1]
# [10, 18, 24, 28, 30, 30, 28, 24, 18, 10]
#
# Sum up the results:
# 10 + 18 + 24 + 28 + 30 + 30 + 28 + 24 + 18 + 10 = 220
#
# Sum up the weights:
# 10 + 9 + 8 + 7 + 6 + 5 + 4 + 3 + 2 + 1 = 55
#
# Divide the sum of the weighted elements by the sum of the weights:
# 220 / 55 = 4.0
self.scores["stock_ranking_score"].append(
np.average(stock_rankings, weights=kwargs.get("stock_weights"))
)
self.scores["alpha_score"].append(kwargs.get("alpha", 0) / 100)
self.scores["beta_score"].append((2 - kwargs.get("beta", 2)) / 2)
self.scores["crisil_rank_score"].append(
(kwargs.get("crisil_rank_score", 0)) / 5
)
self.scores["churn_score"].append(kwargs.get("churn_rate", 0) / 100)
self.scores["sharperatio_score"].append(kwargs.get("sharpe_ratio", 0) / 100)
self.scores["expenseratio_score"].append(kwargs.get("expense_ratio", 0) / 100)
max_aum, min_aum, aum = kwargs.get("aum_score", (1, 0, 0))
self.scores["aum_score"].append((aum - min_aum) / (max_aum - min_aum))
# Calculate the overall rating using weighted sum
return self.get_weighted_score(self.scores)
class MutualFundScorer:
def __init__(self) -> None:
self.mf_scores = []
def _get_stock_ranks(self, isin_ids: List[str]) -> List[int]:
"""Get stock ranks based on ISIN ids."""
return list(
Stock.objects.filter(isin_number__in=isin_ids)
.order_by("rank")
.values_list("rank", "isin_number")
)
def _get_mutual_funds(self) -> List[MutualFund]:
"""Get a list of top 30 mutual funds based on rank."""
return MutualFund.objects.exclude(rank=None).order_by("rank")[:30]
def _get_risk_measure(
self, risk_measures: Dict[str, Any], key: str, year: str
) -> float:
"""
Get value of the specified key from the risk_measures dictionary for the given year.
"""
try:
value = risk_measures.get(year, {}).get(key, 0)
return float(value)
except (TypeError, ValueError):
return 0
def _get_most_non_null_key(self, key, mutual_funds):
"""
Get the year with the maximum number of non-None values for the specified key
within the given mutual funds.
"""
year_counts = {
"for15Year": 0,
"for10Year": 0,
"for5Year": 0,
"for3Year": 0,
"for1Year": 0,
}
for mf in mutual_funds:
risk_measures = mf.data["risk_measures"].get("fundRiskVolatility", {})
for year in year_counts:
if risk_measures.get(year, {}).get(key) is not None:
year_counts[year] += 1
most_non_null_year = max(year_counts, key=year_counts.get)
return most_non_null_year
def get_scores(self) -> List[Dict[str, Any]]:
"""Calculate scores for mutual funds and return the results."""
logger.info("Calculating scores for mutual funds...")
max_aum = MutualFund.objects.exclude(rank=None).aggregate(max_price=Max("aum"))[
"max_price"
]
min_aum = MutualFund.objects.exclude(rank=None).aggregate(min_price=Min("aum"))[
"min_price"
]
mutual_funds = self._get_mutual_funds()
# Get the year with the maximum number of non-None values for sharpeRatio, alpha and beta
sharpe_ratio_year = self._get_most_non_null_key("sharpeRatio", mutual_funds)
alpha_year = self._get_most_non_null_key("alpha", mutual_funds)
beta_year = self._get_most_non_null_key("beta", mutual_funds)
for mf in mutual_funds:
mf_rating = MFRating(
max_rank=1000,
)
logger.info(f"Processing mutual fund: %s", mf.fund_name)
holdings = (
mf.data.get("holdings", {})
.get("equityHoldingPage", {})
.get("holdingList", [])
)
portfolio_holding_weights = {
holding.get("isin"): (
holding.get("weighting") if holding.get("weighting") else 0
)
for holding in holdings
if holding.get("isin")
}
stock_ranks_and_weights = [
(rank, portfolio_holding_weights[isin])
for rank, isin in self._get_stock_ranks(
portfolio_holding_weights.keys()
)
]
stock_ranks, stock_weights = zip(*stock_ranks_and_weights)
sharpe_ratio = self._get_risk_measure(
mf.data["risk_measures"].get("fundRiskVolatility", {}),
"sharpeRatio",
sharpe_ratio_year,
)
alpha = self._get_risk_measure(
mf.data["risk_measures"].get("fundRiskVolatility", {}),
"alpha",
alpha_year,
)
beta = self._get_risk_measure(
mf.data["risk_measures"].get("fundRiskVolatility", {}),
"beta",
beta_year,
)
overall_score = mf_rating.get_overall_score(
stock_rankings=stock_ranks,
stock_weights=stock_weights,
churn_rate=mf.data["quotes"]["lastTurnoverRatio"]
if mf.data["quotes"].get("lastTurnoverRatio")
else 0,
sharpe_ratio=sharpe_ratio,
expense_ratio=mf.data["quotes"]["expenseRatio"],
crisil_rank_score=mf.crisil_rank,
aum_score=(max_aum, min_aum, mf.aum),
alpha=alpha,
beta=beta,
)
self.mf_scores.append(
{
"isin": mf.isin_number,
"name": mf.fund_name,
"rank": mf.rank,
"sharpe_ratio": round(sharpe_ratio, 4),
"churn_rate": mf.data["quotes"].get("lastTurnoverRatio", 0),
"expense_ratio": mf.data["quotes"].get("expenseRatio", 0),
"aum": mf.aum,
"alpha": round(alpha, 4),
"beta": round(beta, 4),
"crisil_rank": mf.crisil_rank,
"overall_score": round(overall_score, 4),
}
)
logger.info("Finished calculating scores.")
return sorted(self.mf_scores, key=lambda d: d["overall_score"], reverse=True)
|