core/mfrating/score_calculator.py

"""
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)